swarms

# Swarms Swarms is a multi-agent orchestration framework. Everything is built from one primitive — `Agent` — which multi-agent structures compose. This document is verified against **swarms v14.0.0**. ## Golden rules 1. **Import from the top level**: `from swarms import Agent`, never `from swarms.structs.agent import Agent`. The one common exception is `PlannerWorkerSwarm` (see below). 2. **Every agent needs a unique `agent_name`** — memory files and swarm routing key on it. 3. **Default to `max_loops=1`.** Use a specific integer for production. Use `"auto"` only for genuinely open-ended work. 4. **Pass `tools=None`, not `tools=[]`.** An empty list breaks schema generation. 5. **Check `examples/`** — 586 runnable examples live there. One is probably close to what you need. 6. **Never set `streaming_on=True` and `streaming_callback` together.** Pick one. ## Setup ```bash pip install -U swarms ``` Set the key for whichever provider you use — any [LiteLLM](https://docs.litellm.ai/docs/providers) model string works: ```bash export OPENAI_API_KEY="sk-..." export ANTHROPIC_API_KEY="sk-ant-..." export GROQ_API_KEY="..." export WORKSPACE_DIR="agent_workspace" # where agent state and memory land ``` --- # Part 1 — The Agent ```python from swarms import Agent agent = Agent( agent_name="Analyst", agent_description="Analyzes market data and produces summaries.", system_prompt="You are a precise financial analyst.", model_name="gpt-5.4", max_loops=1, ) result = agent.run("Summarize the state of the semiconductor market.") ``` `Agent.__init__` accepts 90+ parameters. These are the ones that matter: | Parameter | Type | Default | Purpose | |---|---|---|---| | `agent_name` | `str` | `"swarm-worker-01"` | Unique identity; keys memory + routing | | `agent_description` | `str` | generic | How orchestrators decide to route to it | | `system_prompt` | `str` | built-in | Persona and instructions | | `model_name` | `str` | `"gpt-5.4"` | Any LiteLLM model string | | `max_loops` | `int \| "auto"` | `1` | Iterations, or autonomous mode | | `tools` | `list[Callable]` | `None` | Python functions the agent may call | | `temperature` | `float` | `0.5` | Sampling temperature | | `max_tokens` | `int` | model max | Output cap per call | | `top_p` | `float` | `None` | Nucleus sampling | | `context_length` | `int` | `None` | Token budget; triggers compression at 90% | | `output_type` | `str` | `"str-all-except-first"` | Return shape — see below | | `streaming_on` | `bool` | `False` | Stream tokens to stdout | | `streaming_callback` | `Callable` | `None` | Stream tokens to your function | | `interactive` | `bool` | `False` | REPL — prompts the user each loop | | `verbose` | `bool` | `False` | Debug logging | | `print_on` | `bool` | `True` | Print the final output | | `autosave` | `bool` | `False` | Persist agent state after each run | | `retry_attempts` | `int` | `3` | LLM call retries | | `reasoning_effort` | `str` | `None` | `minimal`/`low`/`medium`/`high`/`xhigh`/`ultra`/`max`/`none` | | `thinking_tokens` | `int` | `1024` | Extended thinking budget (Claude) | | `mcp_url` / `mcp_urls` | `str` / `list[str]` | `None` | MCP servers to load tools from | | `handoffs` | `list[Agent]` | `None` | Agents this one may delegate to | | `persistent_memory` | `bool` | `False` | Read/write `MEMORY.md` across restarts | | `context_compression` | `bool` | `True` | Auto-summarize near the context limit | | `plan_enabled` | `bool` | `False` | Plan before executing | | `mode` | `str` | `"standard"` | `"standard"`, `"fast"`, `"interactive"` | | `fallback_models` | `list[str]` | `None` | Models to try if the primary fails | **`output_type` options**: `"str"`, `"list"`, `"dict"`, `"json"`, `"yaml"`, `"final"`, `"last"`, `"all"`, `"basemodel"`, `"str-all-except-first"`, `"dict-all-except-first"`, `"dict-final"`, `"list-final"`. ### Running ```python agent.run(task="...") # standard agent.run(task="...", img="chart.png") # one image agent.run(task="...", imgs=["a.png", "b.png"]) # several images agent.run(task="...", n=3) # 3 independent samples await agent.arun("...") # async ``` `Agent.run` signature: `run(task=None, img=None, imgs=None, correct_answer=None, streaming_callback=None, n=1)`. ### Streaming ```python # To stdout agent = Agent(agent_name="Writer", model_name="gpt-5.4", streaming_on=True) agent.run("Write a haiku about distributed systems.") # To a callback (do NOT combine with streaming_on) def on_token(token: str) -> None: print(token, end="", flush=True) agent = Agent(agent_name="Writer", model_name="gpt-5.4", streaming_callback=on_token) agent.run("Write a haiku.") # Async streaming async for token in agent.arun_stream("Explain async/await."): print(token, end="", flush=True) ``` --- # Part 2 — Tools Any Python function with type hints and a docstring becomes a tool. The framework generates the OpenAI function schema automatically — **the docstring is the tool description the model reads, so write it for the model.** ```python from swarms import Agent def get_stock_price(ticker: str) -> str: """Fetch the current stock price for a ticker symbol. Args: ticker: Stock ticker symbol, e.g. 'AAPL'. Returns: The current price as a formatted string. """ import yfinance as yf return f"{ticker}: ${yf.Ticker(ticker).fast_info['last_price']:.2f}" agent = Agent( agent_name="StockAnalyst", model_name="gpt-5.4", tools=[get_stock_price], max_loops=3, # needs > 1 so it can act on the tool result ) agent.run("What are Apple and Microsoft trading at?") ``` **`max_loops` must exceed 1 for tool use** — loop 1 calls the tool, loop 2 uses the result. Related knobs: `tool_call_summary=True` (summarize tool output), `show_tool_execution_output=True` (print raw returns), `tool_retry_attempts` (retries on tool failure). ### MCP servers ```python agent = Agent( agent_name="MCPAgent", model_name="gpt-5.4", mcp_url="http://localhost:8000/sse", # or: mcp_urls=["http://localhost:8000/sse", "http://localhost:8001/sse"] max_loops=3, ) ``` Inspect what a server exposes before wiring it up: ```python from swarms.tools.mcp_manager import MCPManager mgr = MCPManager(mcp_url="http://localhost:8000/sse") print(mgr.list_tool_names()) schemas = mgr.get_tools() # aget_tools() for the async form ``` ### Handoffs Give an agent a roster it can delegate to. It receives a `handoff_task` tool automatically. ```python triage = Agent( agent_name="Triage", model_name="gpt-5.4", handoffs=[billing_agent, technical_agent, refunds_agent], max_loops=3, ) triage.run("My invoice is wrong and the app won't load.") ``` --- # Part 3 — Autonomous mode (`max_loops="auto"`) The agent runs plan → execute → reflect until it decides it is finished, with **16 built-in tools** available: | Group | Tools | |---|---| | Planning | `create_plan`, `think`, `subtask_done`, `complete_task`, `respond_to_user` | | Files | `create_file`, `update_file`, `read_file`, `list_directory`, `delete_file` | | System | `run_bash`, `grep` | | Delegation | `create_sub_agent`, `assign_task`, `check_sub_agent_status`, `cancel_sub_agent_tasks` | ```python agent = Agent( agent_name="Researcher", model_name="gpt-5.4", max_loops="auto", tools=[search_web], # your tools stack on top of the built-ins persistent_memory=True, context_compression=True, context_length=32000, ) agent.run("Research the top 5 vector databases and write compare.md") ``` Restrict the built-in set with `selected_tools` (default `"all"`): ```python agent = Agent( agent_name="ReadOnly", max_loops="auto", selected_tools=["create_plan", "think", "read_file", "grep", "complete_task"], ) ``` Inspect the full list at runtime with `agent.get_all_selected_tools()`. ⚠️ **`run_bash` and `delete_file` are real.** In autonomous mode the agent can modify and delete files and execute shell commands. Scope `selected_tools` and set `WORKSPACE_DIR` deliberately. --- # Part 4 — Memory and conversation ### Persistent memory `persistent_memory=True` reads `{WORKSPACE_DIR}/agents/{agent_name}/MEMORY.md` on startup and appends to it each response. It is **off by default** — set it in every process that should share the memory. ```python agent = Agent(agent_name="ProjectAssistant", model_name="gpt-5.4", persistent_memory=True) agent.run("My project is called Helios. Remember that.") # Later process, same agent_name and the flag set again → it remembers. ``` ### Context compression `context_compression=True` (default) fires at 90% of `context_length`, summarizing history in place so long sessions never hit the wall. Leave it on for anything long-running. ### Conversation ```python from swarms import Conversation conv = Conversation( name="my-conversation", # note: `name`, not `agent_name` system_prompt="You are helpful.", time_enabled=True, token_count=True, ) conv.add("user", "What is 2+2?") conv.add("assistant", "4.") conv.return_history_as_string() conv.search("2+2") conv.compact(summary="User asked arithmetic. Answer: 4.") # archives, then collapses conv.save_as_json("conv.json") ``` --- # Part 5 — Multi-agent architectures ## Choosing one | Situation | Use | |---|---| | Single task | `Agent` | | Linear A→B→C | `SequentialWorkflow` | | Same task, many agents at once | `ConcurrentWorkflow` | | Custom mix of sequential + parallel | `AgentRearrange` | | Dependency graph / fan-out-fan-in | `GraphWorkflow` | | Many models, one synthesized answer | `MixtureOfAgents` | | Manager delegates to specialists | `HierarchicalSwarm` | | Open discussion | `GroupChat` | | Discrete decision by consensus | `MajorityVoting` | | Quality-critical evaluation | `CouncilAsAJudge` | | Structured adversarial debate | `DebateWithJudge` | | Deep multi-stage research | `HeavySwarm` | | Route each task to the best agent | `MultiAgentRouter` | | Plan then execute with workers | `PlannerWorkerSwarm` | | Don't know yet | `SwarmRouter(swarm_type="auto")` or `AutoSwarmBuilder` | ## SequentialWorkflow Each agent's output becomes the next agent's context. ```python from swarms import Agent, SequentialWorkflow pipeline = SequentialWorkflow(