# Multi-Agent Patterns in Python with 10xGraph

> Build supervisor, swarm, manual handoff and plan-act-reflect multi-agent systems in 10xGraph where each agent runs on a different company's LLM.

Source: https://10xgraph.com/docs/examples/multi-agent
Last updated: 2026-10-28

This example shows four multi-agent shapes: a supervisor team, a swarm, a manual handoff graph and a plan, act and reflect loop. In every file the agents run on models from different companies (Anthropic, OpenAI, Google and DeepSeek), because each agent is its own `Agent` with its own model string.

## What you will build

Four multi-agent systems, each with a different shape of control flow. A supervisor routes work to specialists. A swarm lets peers hand work to each other. A manual graph shows how to wire handoffs yourself. A plan, act and reflect loop separates planning from review.

In every file the agents run on models from different companies. That is possible because each agent is its own `Agent` with its own model string, and 10xGraph converts messages for each provider.

## Which pattern when

| Pattern | File | Use it when | Control flow |
|---|---|---|---|
| Supervisor team | `supervisor.py` | One coordinator should decide who works next and when to stop | Supervisor, worker, supervisor, ... until `FINISH` |
| Swarm | `swarm.py` | Specialists know best who should take over, and you want no central router | Agent hands off directly to a peer |
| Manual handoff | `handoff_manual.py` | You need custom edges, tool nodes or state beyond what `SwarmAgent` wires | `create_handoff_tool` on your own `StateGraph` |
| Plan, act, reflect | `plan_act_reflect.py` | A task needs tool use followed by a critique pass and possible retry | PLAN, ACT, REFLECT, repeat until `[DONE]` |

## Which agent uses which model

> **Every agent uses a different company's model**
> Within each file, the agents are spread across providers. The point is that one graph can mix OpenAI, Anthropic, Google and an OpenAI-compatible endpoint such as DeepSeek. It is a choice made in these examples, not a requirement.

| File | Agent | Company | Model |
|---|---|---|---|
| `supervisor.py` | Supervisor | Anthropic | `claude-sonnet-5-5` |
| | RESEARCHER | OpenAI | `gpt-5-mini` |
| | ANALYST | Google | `gemini-2.5-flash` |
| | WRITER | DeepSeek | `deepseek-chat` |
| `swarm.py` | TRIAGE | OpenAI | `gpt-5-mini` |
| | RESEARCHER | Anthropic | `claude-sonnet-5-5` |
| | WRITER | Google | `gemini-2.5-flash` |
| `handoff_manual.py` | coordinator | Google | `gemini-2.5-flash` |
| | researcher | DeepSeek | `deepseek-chat` |
| | writer | Anthropic | `claude-sonnet-5-5` |
| `plan_act_reflect.py` | PLAN | Anthropic | `claude-sonnet-5-5` |
| | ACT | none (runs tools) | not applicable |
| | REFLECT | Google | `gemini-2.5-flash` |

DeepSeek is reached through its OpenAI-compatible endpoint. 10xGraph treats an unknown model name as an OpenAI-style model, so `base_url` and `api_key` on the `Agent` are all it needs.

## Key concepts

- `SupervisorTeamAgent(supervisor_model=..., workers=\{...\})`: builds the supervisor from a model string and accepts pre-built worker agents.
- `WorkerConfig(agent=..., description=...)`: a worker and the description the supervisor reads when choosing.
- `SwarmAgent(members=\{...\}, entry=...)`: wires peer handoffs and injects a `transfer_to_<member>` tool into each member.
- `SwarmMemberConfig(agent=..., can_handoff_to=[...], description=...)`: one member and its allowed targets.
- `create_handoff_tool(name, description)`: a tool that moves control to the node `name` without running.
- `PlanActReflectAgent(model, plan_model=..., reflect_model=..., tools=[...], max_iterations=...)`: planner, tool step and reviewer loop.
- `Agent(model=..., base_url=..., api_key=...)`: per-agent provider settings.

## Run it

```bash
git clone https://github.com/10xGraph/10xGraph.git
cd 10xGraph/agentflow
```

```bash
pip install "10xgraph[openai,anthropic,google-genai]" python-dotenv
export OPENAI_API_KEY="..."
export ANTHROPIC_API_KEY="..."
export GEMINI_API_KEY="..."
export DEEPSEEK_API_KEY="..."
```

The `openai` extra also covers DeepSeek. Each script needs only the keys for the models it uses, as listed in its docstring.

```bash
python examples/multi_agent/supervisor.py
python examples/multi_agent/swarm.py
python examples/multi_agent/handoff_manual.py
python examples/multi_agent/plan_act_reflect.py
```

Each script prints every assistant reply with the model that produced it.

## How the code works

### supervisor.py

Workers are normal agents. Only the supervisor is built from a string.

```python
researcher = Agent(
    model="gpt-5-mini",  # OpenAI
    system_prompt=[
        {"role": "system", "content": "You research facts. Use web_search, then report findings."}
    ],
    tool_node=ToolNode([web_search]),
)
```

The researcher runs on OpenAI and the analyst on Google. Each has its own tool node and system prompt.

```python
writer = Agent(
    model="deepseek-chat",  # DeepSeek, through its OpenAI-compatible endpoint
    base_url="https://api.deepseek.com",
    api_key=os.environ["DEEPSEEK_API_KEY"],
    system_prompt=[
        {"role": "system", "content": "You write a concise summary from the work done so far."}
    ],
)
```

The writer uses DeepSeek through its OpenAI-compatible endpoint, so `base_url` and `api_key` are set on this agent only.

```python
team = SupervisorTeamAgent(
    supervisor_model="claude-sonnet-5-5",  # Anthropic
    workers={
        "RESEARCHER": WorkerConfig(agent=researcher, description="Finds facts with web search."),
        "ANALYST": WorkerConfig(agent=analyst, description="Does calculations and comparisons."),
        "WRITER": WorkerConfig(agent=writer, description="Writes the final answer for the user."),
    },
    max_rounds=6,  # Upper bound on supervisor -> worker hops per turn.
)
```

The supervisor runs on Anthropic. It replies with one worker name or `FINISH`, and its choice is not added to the conversation. The `description` of each worker is what it reads when choosing. A worker that has tools gets its own tool loop. `max_rounds` caps the hops per turn.

```python
    for msg in result["messages"]:
        if msg.role == "assistant" and msg.text():
            model = (msg.metadata or {}).get("model") or "unknown model"
            print(f"[{model}] {msg.text()[:300]}\n")
```

Each reply is printed with the model that produced it, read from `msg.metadata["model"]`. This is how you can see the mix of providers in one run.

### swarm.py

Members are pre-built agents too: triage on OpenAI, researcher on Anthropic, writer on Google.

```python
swarm = SwarmAgent(
    members={
        "TRIAGE": SwarmMemberConfig(
            agent=triage,
            can_handoff_to=["RESEARCHER", "WRITER"],
            description="Routes requests to the right specialist.",
        ),
        "RESEARCHER": SwarmMemberConfig(
            agent=researcher,
            can_handoff_to=["WRITER"],
            description="Researches a topic and gathers facts.",
        ),
        # No can_handoff_to: the writer ends the run when it replies without a handoff.
        "WRITER": SwarmMemberConfig(agent=writer, description="Writes and saves articles."),
    },
    entry="TRIAGE",
)
```

`can_handoff_to` limits who each member can pass work to, and `None` (the default) means everyone else. Do not add handoff tools yourself, because `SwarmAgent` injects a `transfer_to_<member>` tool into every member. A member that replies without a handoff or tool call ends the run, so the writer finishes the conversation.

### handoff_manual.py

Use this when `SwarmAgent`'s wiring is not what you need. The target name in `create_handoff_tool` must equal a node name in the graph and is used as written, so the nodes here are lowercase on purpose.

```python
def route_to(tool_node: str):
    """Go to the agent's tool node when the last reply has tool calls, else finish."""

    def route(state: AgentState) -> str:
        last = state.context[-1] if state.context else None
        return tool_node if last and last.role == "assistant" and last.tools_calls else END
```

This router sends control to the agent's tool node when the last reply has tool calls, and ends otherwise.

```python
    for name, (agent, tools) in agents.items():
        graph.add_node(name, agent)
        graph.add_node(f"{name}_tools", ToolNode(tools))
        graph.add_conditional_edges(
            name, route_to(f"{name}_tools"), {f"{name}_tools": f"{name}_tools", END: END}
        )
        # After a regular tool, return to the same agent. A transfer tool skips this edge
        # because its Command already names the next node.
        graph.add_edge(f"{name}_tools", name)
```

Each agent has a `ToolNode` holding its own tools and transfer tools. The tool node sees a `transfer_to_<name>` call and returns a jump to that node, so the transfer tool itself never runs. The `\{name\}_tools -> \{name\}` edge returns control to the same agent after an ordinary tool. A transfer tool skips that edge because its `Command` already names the next node.

```python
    graph.set_entry_point("coordinator")
    return graph
```

### plan_act_reflect.py

```python
agent = PlanActReflectAgent(
    model="claude-sonnet-5-5",  # Fallback for any phase without its own model.
    plan_model="claude-sonnet-5-5",  # Anthropic plans and chooses tools.
    reflect_model="gemini-2.5-flash",  # Google reviews the result.
    tools=[search_flights, search_hotels],
    max_iterations=3,
)
```

`model` is the fallback for a phase with no override. PLAN is an LLM that decides which tools to call, ACT is the `ToolNode` that executes them with no model call, and REFLECT is an LLM that reviews the work. REFLECT ends the loop by including `[DONE]`, or sends it back to PLAN until `max_iterations`. Here Anthropic plans and Google reflects.

Extra keyword arguments such as `base_url` apply to both model phases, so pick providers that can share them. Use `supervisor.py` when each agent needs its own endpoint.

## Full code

**supervisor.py**

```python
"""Supervisor team: one model routes work to specialist workers, each from a different company.

SupervisorTeamAgent builds the supervisor itself from `supervisor_model` (Anthropic here).
Each worker is an ordinary Agent you construct, so it carries its own model, tools and
endpoint: OpenAI, Gemini and DeepSeek below. The supervisor answers with a worker name or
FINISH, and the graph follows.

Run: python examples/multi_agent/supervisor.py
     (needs ANTHROPIC_API_KEY, OPENAI_API_KEY, GEMINI_API_KEY, DEEPSEEK_API_KEY)
"""

import asyncio
import os

from dotenv import load_dotenv

from tenxgraph.core.graph import Agent, ToolNode
from tenxgraph.core.state import Message
from tenxgraph.prebuilt.agent import SupervisorTeamAgent, WorkerConfig

load_dotenv()

def web_search(query: str) -> str:
    """Search the web and return a short list of findings."""
    return (
        f"Findings for '{query}': global EV sales grew about 25% year over year; "
        "China accounts for roughly 60% of new registrations; battery pack prices fell 14%."
    )

def calculate_growth(start: float, end: float) -> str:
    """Return the percentage change from start to end."""
    return f"{(end - start) / start * 100:.1f}% change"

researcher = Agent(
    model="gpt-5-mini",  # OpenAI
    system_prompt=[
        {"role": "system", "content": "You research facts. Use web_search, then report findings."}
    ],
    tool_node=ToolNode([web_search]),
)

analyst = Agent(
    model="gemini-2.5-flash",  # Google
    system_prompt=[
        {"role": "system", "content": "You analyse numbers. Use calculate_growth when useful."}
    ],
    tool_node=ToolNode([calculate_growth]),
)

writer = Agent(
    model="deepseek-chat",  # DeepSeek, through its OpenAI-compatible endpoint
    base_url="https://api.deepseek.com",
    api_key=os.environ["DEEPSEEK_API_KEY"],
    system_prompt=[
        {"role": "system", "content": "You write a concise summary from the work done so far."}
    ],
)

team = SupervisorTeamAgent(
    supervisor_model="claude-sonnet-5-5",  # Anthropic
    workers={
        "RESEARCHER": WorkerConfig(agent=researcher, description="Finds facts with web search."),
        "ANALYST": WorkerConfig(agent=analyst, description="Does calculations and comparisons."),
        "WRITER": WorkerConfig(agent=writer, description="Writes the final answer for the user."),
    },
    max_rounds=6,  # Upper bound on supervisor -> worker hops per turn.
)

async def main() -> None:
    app = team.compile()
    result = await app.ainvoke(
        {"messages": [Message.text_message("Brief me on the electric vehicle market.")]},
        config={"thread_id": "supervisor-1", "recursion_limit": 40},
    )
    for msg in result["messages"]:
        if msg.role == "assistant" and msg.text():
            model = (msg.metadata or {}).get("model") or "unknown model"
            print(f"[{model}] {msg.text()[:300]}\n")
    await app.aclose()

if __name__ == "__main__":
    asyncio.run(main())
```

**swarm.py**

```python
"""Swarm: peer agents hand work to each other, and each peer runs on a different company's model.

SwarmAgent injects a `transfer_to_<member>` tool into every member. When a model calls
one, control moves to that member. No supervisor is involved. Triage runs on OpenAI, the
researcher on Anthropic and the writer on Gemini.

Run: python examples/multi_agent/swarm.py
     (needs OPENAI_API_KEY, ANTHROPIC_API_KEY, GEMINI_API_KEY)
"""

import asyncio

from dotenv import load_dotenv

from tenxgraph.core.graph import Agent, ToolNode
from tenxgraph.core.state import Message
from tenxgraph.prebuilt.agent import SwarmAgent, SwarmMemberConfig

load_dotenv()

def web_search(query: str) -> str:
    """Search the web and return findings."""
    return f"Findings for '{query}': solid-state batteries promise 2x energy density by 2030."

def save_document(title: str, body: str) -> str:
    """Save a finished document."""
    return f"Saved '{title}' ({len(body)} characters)."

triage = Agent(
    model="gpt-5-mini",  # OpenAI
    system_prompt=[
        {
            "role": "system",
            "content": "Route the request. Research questions go to the researcher; "
            "writing-only requests go to the writer. Do not answer yourself.",
        }
    ],
)

researcher = Agent(
    model="claude-sonnet-5-5",  # Anthropic
    system_prompt=[
        {
            "role": "system",
            "content": "Research with web_search. When you have findings, hand off to the writer.",
        }
    ],
    tool_node=ToolNode([web_search]),
)

writer = Agent(
    model="gemini-2.5-flash",  # Google
    system_prompt=[
        {"role": "system", "content": "Write a short article from the conversation, then save it."}
    ],
    tool_node=ToolNode([save_document]),
)

swarm = SwarmAgent(
    members={
        "TRIAGE": SwarmMemberConfig(
            agent=triage,
            can_handoff_to=["RESEARCHER", "WRITER"],
            description="Routes requests to the right specialist.",
        ),
        "RESEARCHER": SwarmMemberConfig(
            agent=researcher,
            can_handoff_to=["WRITER"],
            description="Researches a topic and gathers facts.",
        ),
        # No can_handoff_to: the writer ends the run when it replies without a handoff.
        "WRITER": SwarmMemberConfig(agent=writer, description="Writes and saves articles."),
    },
    entry="TRIAGE",
)

async def main() -> None:
    app = swarm.compile()
    result = await app.ainvoke(
        {"messages": [Message.text_message("Write a short article on solid-state batteries.")]},
        config={"thread_id": "swarm-1", "recursion_limit": 30},
    )
    for msg in result["messages"]:
        if msg.role == "assistant" and msg.text():
            model = (msg.metadata or {}).get("model") or "unknown model"
            print(f"[{model}] {msg.text()[:300]}\n")
    await app.aclose()

if __name__ == "__main__":
    asyncio.run(main())
```

**handoff_manual.py**

```python
"""Manual handoff: wire agent-to-agent transfers on your own StateGraph with create_handoff_tool.

Use this when SwarmAgent's wiring is not what you need. Each agent gets a ToolNode that
holds its regular tools and transfer tools. When the model calls `transfer_to_<node>`,
the tool node returns a Command that jumps to that node without running the tool.
Agents: coordinator on Gemini, researcher on DeepSeek, writer on Anthropic.

Run: python examples/multi_agent/handoff_manual.py
     (needs GEMINI_API_KEY, DEEPSEEK_API_KEY, ANTHROPIC_API_KEY)
"""

import asyncio
import os

from dotenv import load_dotenv

from tenxgraph.core.graph import Agent, StateGraph, ToolNode
from tenxgraph.core.state import AgentState, Message
from tenxgraph.prebuilt.tools import create_handoff_tool
from tenxgraph.utils.constants import END

load_dotenv()

def search_web(query: str) -> str:
    """Search the web for information."""
    return (
        f"Results for '{query}': quantum computers use qubits; error correction is the main hurdle."
    )

def write_document(title: str, content: str) -> str:
    """Save a document."""
    return f"Document '{title}' saved ({len(content)} characters)."

# The target name in create_handoff_tool must equal a node name in the graph. It is used as
# written, so these nodes are lowercase on purpose.
coordinator = Agent(
    model="gemini-2.5-flash",  # Google
    system_prompt=[
        {
            "role": "system",
            "content": "You coordinate. Send research to the researcher and writing to the writer. "
            "When the writer returns, give the user a one-line wrap-up.",
        }
    ],
    tool_node="coordinator_tools",
)
researcher = Agent(
    model="deepseek-chat",  # DeepSeek
    base_url="https://api.deepseek.com",
    api_key=os.environ["DEEPSEEK_API_KEY"],
    system_prompt=[
        {"role": "system", "content": "Research with search_web, then transfer to the writer."}
    ],
    tool_node="researcher_tools",
)
writer = Agent(
    model="claude-sonnet-5-5",  # Anthropic
    system_prompt=[
        {
            "role": "system",
            "content": "Write a short article, save it with write_document, "
            "then transfer back to the coordinator.",
        }
    ],
    tool_node="writer_tools",
)

def route_to(tool_node: str):
    """Go to the agent's tool node when the last reply has tool calls, else finish."""

    def route(state: AgentState) -> str:
        last = state.context[-1] if state.context else None
        return tool_node if last and last.role == "assistant" and last.tools_calls else END

    return route

def build_graph() -> StateGraph:
    graph = StateGraph()
    agents = {
        "coordinator": (
            coordinator,
            [
                create_handoff_tool("researcher", "Hand off a research task."),
                create_handoff_tool("writer", "Hand off a writing task."),
            ],
        ),
        "researcher": (
            researcher,
            [search_web, create_handoff_tool("writer", "Hand off findings to the writer.")],
        ),
        "writer": (
            writer,
            [write_document, create_handoff_tool("coordinator", "Return to the coordinator.")],
        ),
    }
    for name, (agent, tools) in agents.items():
        graph.add_node(name, agent)
        graph.add_node(f"{name}_tools", ToolNode(tools))
        graph.add_conditional_edges(
            name, route_to(f"{name}_tools"), {f"{name}_tools": f"{name}_tools", END: END}
        )
        # After a regular tool, return to the same agent. A transfer tool skips this edge
        # because its Command already names the next node.
        graph.add_edge(f"{name}_tools", name)

    graph.set_entry_point("coordinator")
    return graph

async def main() -> None:
    app = build_graph().compile()
    result = await app.ainvoke(
        {
            "messages": [
                Message.text_message("Research quantum computing and write a short article.")
            ]
        },
        config={"thread_id": "handoff-1", "recursion_limit": 30},
    )
    for msg in result["messages"]:
        if msg.role == "assistant" and msg.text():
            model = (msg.metadata or {}).get("model") or "unknown model"
            print(f"[{model}] {msg.text()[:300]}\n")
    await app.aclose()

if __name__ == "__main__":
    asyncio.run(main())
```

**plan_act_reflect.py**

```python
"""Plan, act, reflect: a planner and a reviewer on different companies' models.

PlanActReflectAgent loops PLAN -> ACT -> REFLECT until the reflector says [DONE] or
`max_iterations` is reached. PLAN is an LLM that decides which tools to call. ACT is the
ToolNode that runs them (no model). REFLECT is an LLM that reviews the work. `plan_model`
and `reflect_model` set a model per phase: Anthropic plans and Gemini reflects.

Run: python examples/multi_agent/plan_act_reflect.py   (needs ANTHROPIC_API_KEY, GEMINI_API_KEY)
"""

import asyncio

from dotenv import load_dotenv

from tenxgraph.core.state import Message
from tenxgraph.prebuilt.agent import PlanActReflectAgent

load_dotenv()

def search_flights(origin: str, destination: str) -> str:
    """Search flights between two cities."""
    return f"{origin} -> {destination}: 08:10 at 240 EUR, 13:45 at 190 EUR, 19:30 at 165 EUR."

def search_hotels(city: str, nights: int) -> str:
    """Search hotels in a city."""
    return f"{city}, {nights} nights: Hotel Aurora 110 EUR/night, Casa Verde 85 EUR/night."

agent = PlanActReflectAgent(
    model="claude-sonnet-5-5",  # Fallback for any phase without its own model.
    plan_model="claude-sonnet-5-5",  # Anthropic plans and chooses tools.
    reflect_model="gemini-2.5-flash",  # Google reviews the result.
    tools=[search_flights, search_hotels],
    max_iterations=3,
)

async def main() -> None:
    app = agent.compile()
    result = await app.ainvoke(
        {
            "messages": [
                Message.text_message("Plan a 3-night trip from Berlin to Lisbon under 600 EUR.")
            ]
        },
        config={"thread_id": "par-1", "recursion_limit": 40},
    )
    for msg in result["messages"]:
        if msg.role == "assistant" and msg.text():
            model = (msg.metadata or {}).get("model") or "unknown model"
            print(f"[{model}] {msg.text()[:300]}\n")
    await app.aclose()

if __name__ == "__main__":
    asyncio.run(main())
```

Source on GitHub: [examples/multi_agent](https://github.com/10xGraph/10xGraph/tree/main/agentflow/examples/multi_agent)

## Expected output

Illustrative output of `supervisor.py`:

```text
[gpt-5-mini-2025-08-07] Findings: global EV sales grew about 25% year over year, with China at roughly 60%...

[gemini-2.5-flash] Growth of 25% on a base of 14M units implies about 3.5M additional vehicles...

[deepseek-chat] Electric vehicle sales are growing quickly, led by China. Battery costs fell 14%...
```

## Using a different model

Change the model string on any agent and install the matching extra. Swapping the supervisor to OpenAI is `supervisor_model="gpt-5-mini"` with `pip install "10xgraph[openai]"`. Swapping a worker to Gemini is `model="gemini-2.5-flash"` with `pip install "10xgraph[google-genai]"`.

## Next steps

- [Structured output](/docs/examples/structured-output) makes a worker return typed data.
- [Observability](/docs/examples/observability) shows which agent and model handled each step.
- [Callbacks](/docs/examples/callbacks) adds validation and cost tracking around every model call.
- [Research assistant](/docs/examples/research-assistant) and [Customer support](/docs/examples/customer-support) build on these patterns.

## Frequently asked questions

### Can agents in the same graph use different LLM providers?

Yes. Each worker or member is a separate Agent with its own model string, so one graph can mix OpenAI, Anthropic, Google and OpenAI-compatible endpoints such as DeepSeek.

### What is the difference between SupervisorTeamAgent and SwarmAgent?

In a supervisor team one model decides which worker runs next and control returns to it after each worker. In a swarm the agents hand work directly to each other with transfer_to tools and no central router.

### How do I give one agent a custom base_url without affecting the others?

Set base_url and api_key on that Agent when you build it. SupervisorTeamAgent and SwarmAgent take pre-built Agent instances, so settings stay per agent. PlanActReflectAgent shares extra keyword arguments across its phases.

### Does PlanActReflectAgent call a model in the act step?

No. ACT is a ToolNode that runs the tools the planner chose. Only PLAN and REFLECT call a model, so plan_model and reflect_model are the two per-phase settings.