Multi-Agent Patterns in Python with 10xGraph

"""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: 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())

"""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: 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())