Multi-Agent Architectures

When to Go Multi-Agent

Single agents work well up to a complexity threshold. Beyond that, multi-agent systems win:

Key signals to go multi-agent:

• Task naturally decomposes into distinct specializations
• Different parts of the task benefit from different models
• Task is too long for a single context window
• You want parallel execution of independent sub-tasks

The Supervisor Pattern

flowchart TD
    U([User Task]) --> SUP["SUPERVISOR LLM\nTracks overall goal\nDecides next worker\nAggregates results"]
    SUP --> RES["RESEARCHER\nweb_search · fetch_page"]
    SUP --> ANA["ANALYST\ncalculate · chart"]
    SUP --> WRI["WRITER\nformat · save_doc"]
    RES --> SUP
    ANA --> SUP
    WRI --> SUP
    SUP --> FIN([Complete])
    style SUP fill:#EEF0F7,stroke:#6366F1,color:#0F172A
    style RES fill:#f0fdf9,stroke:#0D9488,color:#0F172A
    style ANA fill:#f0fdf9,stroke:#0D9488,color:#0F172A
    style WRI fill:#f0fdf9,stroke:#0D9488,color:#0F172A

Implementing the Supervisor with LangGraph

The supervisor node acts as the orchestrator, it reads the full progress summary and decides which worker to invoke next, or whether the task is complete.

# src/agents/multi_agent/supervisor.py
from langgraph.graph import StateGraph, END
from langchain_core.messages import HumanMessage, AIMessage
from typing import Literal

# Worker agents (each is a compiled LangGraph or Pydantic AI agent)
from .researcher import researcher_agent
from .analyst import analyst_agent
from .writer import writer_agent

WORKERS = {
    "researcher": researcher_agent,
    "analyst": analyst_agent,
    "writer": writer_agent,
}

class SupervisorState(TypedDict):
    task: str
    messages: Annotated[list[BaseMessage], add_messages]
    next_worker: str | None
    worker_results: dict[str, str]  # worker_name → result; builds up as each worker completes
    final_report: str | None

# Use a capable model for the supervisor: it needs to reason about task decomposition
supervisor_llm = ChatOpenAI(model="gpt-5.6-sol", temperature=0)

SUPERVISOR_SYSTEM = """You are a supervisor managing a team of AI agents.

Available workers:
- researcher: Searches the web for factual information
- analyst: Analyzes data and produces quantitative insights
- writer: Produces well-structured written content

Given the task and what has been done so far, decide the next worker to call.
When the task is complete, respond with FINISH.

Respond with JSON: {"next": "researcher" | "analyst" | "writer" | "FINISH", "instruction": "specific instruction for this worker"}"""

def supervisor_node(state: SupervisorState) -> dict:
    """Supervisor decides the next action."""
    # Summarize completed work so the supervisor can reason about what's left
    progress = "\n".join([
        f"{worker}: {result[:200]}..."
        for worker, result in state["worker_results"].items()
    ])
    
    response = supervisor_llm.invoke([
        {"role": "system", "content": SUPERVISOR_SYSTEM},
        {"role": "user", "content": f"""Task: {state['task']}

Progress so far:
{progress or 'Nothing done yet.'}

What should we do next?"""}
    ])
    
    import json
    decision = json.loads(response.content)
    
    if decision["next"] == "FINISH":
        return {"next_worker": "FINISH"}
    
    # Pass a targeted instruction to the chosen worker via the messages channel
    return {
        "next_worker": decision["next"],
        "messages": [HumanMessage(content=decision["instruction"])],
    }

def route_to_worker(state: SupervisorState) -> str:
    """Route to the chosen worker or END."""
    return "finish" if state["next_worker"] == "FINISH" else state["next_worker"]

def make_worker_node(worker_name: str, agent):
    """Factory: create a node that runs a worker agent."""
    def worker_node(state: SupervisorState) -> dict:
        # Each worker only sees the instruction it was given, not the full supervisor state
        last_instruction = state["messages"][-1].content
        
        # Run the worker agent with a fresh state
        result = agent.invoke({
            "messages": [HumanMessage(content=last_instruction)],
            "task_status": "running",
            "tool_call_count": 0,
            "final_answer": None,
        })
        
        # Extract the last AI message as the worker's deliverable
        ai_msgs = [m for m in result["messages"] if isinstance(m, AIMessage)]
        worker_output = ai_msgs[-1].content if ai_msgs else "No output"
        
        # Merge this worker's result into the shared results dict without overwriting others
        return {
            "worker_results": {**state["worker_results"], worker_name: worker_output},
        }
    
    worker_node.__name__ = f"{worker_name}_node"
    return worker_node

def finish_node(state: SupervisorState) -> dict:
    """Compile worker results into final output."""
    sections = "\n\n".join([
        f"## {name.title()} Findings\n{result}"
        for name, result in state["worker_results"].items()
    ])
    
    # Use the supervisor LLM to synthesize: it already understands the overall task goal
    final = supervisor_llm.invoke([
        {"role": "user", "content": f"Compile these into a cohesive final report:\n\n{sections}"}
    ])
    
    return {"final_report": final.content}

# Build the supervisor graph
supervisor_builder = StateGraph(SupervisorState)
supervisor_builder.add_node("supervisor", supervisor_node)
supervisor_builder.add_node("finish", finish_node)

for name, agent in WORKERS.items():
    supervisor_builder.add_node(name, make_worker_node(name, agent))
    supervisor_builder.add_edge(name, "supervisor")  # workers always report back to the supervisor

supervisor_builder.set_entry_point("supervisor")
supervisor_builder.add_conditional_edges(
    "supervisor",
    route_to_worker,
    # explicit routing table so LangGraph knows all possible destinations
    {"researcher": "researcher", "analyst": "analyst", "writer": "writer", "finish": "finish"}
)
supervisor_builder.add_edge("finish", END)

supervisor = supervisor_builder.compile()

Defining Specialized Workers

Each worker should have a narrow, focused system prompt:

# src/agents/multi_agent/researcher.py
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent
from langchain_community.tools.tavily_search import TavilySearchResults

researcher_agent = create_react_agent(
    # gpt-5.6-luna is cost-effective for research: most effort is in tool calls, not reasoning
    model=ChatOpenAI(model="gpt-5.6-luna", temperature=0),
    tools=[TavilySearchResults(max_results=8)],  # only the tools this role needs
    state_modifier="""You are a research specialist. Your only job is to find accurate, 
current information from the web. 
    
Rules:
- Always search for information rather than relying on memory
- Search multiple times with different queries to ensure thoroughness
- Return factual information with source URLs
- Do not draw conclusions, that's the analyst's job
- Be thorough and precise""",
)

# src/agents/multi_agent/analyst.py
analyst_agent = create_react_agent(
    model=ChatOpenAI(model="gpt-5.6-sol", temperature=0),  # use a more capable model for analysis reasoning
    tools=[],  # analyst works with provided information, no external tools prevents scope creep
    state_modifier="""You are a data analyst. You receive research findings and produce 
structured analysis.

Produce:
- Key trends and patterns
- Quantitative insights where possible
- Comparative assessments
- Risk factors
- Clear, numbered conclusions""",
)

Running the Multi-Agent System

async def run_market_analysis(topic: str) -> str:
    initial_state: SupervisorState = {
        "task": f"Produce a comprehensive market analysis for: {topic}",
        "messages": [],
        "next_worker": None,
        "worker_results": {},   # starts empty; populated as each worker completes
        "final_report": None,
    }
    
    result = await supervisor.ainvoke(initial_state)
    return result["final_report"]

# Usage
report = await run_market_analysis("AI agent development tools 2025-2026")
print(report)

The supervisor will automatically:

1. Call the researcher to gather information
2. Call the analyst to process the research
3. Call the writer to produce the final report
4. Compile everything into a cohesive document

And if any worker fails, the supervisor receives the error and can decide to retry, skip, or adapt its plan.