# How to Orchestrate AI Agents with a Supervisor Pattern - Tool: Any LLM or framework - Difficulty: Advanced - Time required: 45-60 min - Compatibility: Python 3.10+, langgraph 1.2+, langchain 1.4+, an Anthropic API key, any OS - Last updated: November 2026 ## TL;DR AI agent orchestration means one controller routes a task to the right specialist agent instead of one model trying to do everything. This guide builds a deal-desk system: a supervisor graph in LangGraph that routes a sales request to a research sub-agent, a pricing sub-agent, and a drafting sub-agent, in that order, sharing state between them. You'll use StateGraph, conditional edges, and a checkpointer. Budget 45-60 minutes and a Python environment. ## What you'll build You'll build a deal-desk orchestrator for a B2B sales team. A rep types a request like 'put together a quote for Acme Corp, 200 seats, annual plan,' and a supervisor node decides what happens next: first a research sub-agent that pulls company size and industry from a CRM-shaped lookup, then a pricing sub-agent that computes a quote and a discount band, then a drafting sub-agent that writes the proposal email. The supervisor isn't just a switch statement: it re-reads the shared state after every sub-agent finishes and decides the next hop, so a request that needs legal review routes there instead of straight to drafting. Orchestration is the layer above individual agents: it owns the state that flows between them, the order they run in, and what happens when one of them fails or needs another pass. LangGraph models this explicitly as a graph, not a hidden loop inside a framework, which is why it's the right tool when you need the routing logic itself to be inspectable and debuggable. Choose a supervisor graph over a single agent with a huge tool list when the sub-tasks need genuinely different context, prompts, or models (a cheap model for research, a stronger one for drafting), or when you want to unit-test each stage in isolation. If every sub-task actually just needs a different tool and the same context, you want the simpler pattern in how-to-give-an-ai-agent-tools instead. ## Before you start - Python 3.10+ and a terminal - An Anthropic API key (ANTHROPIC_API_KEY) - pip install langgraph langchain langchain-anthropic - Basic comfort with Python classes and TypedDict ## Step-by-step guide ### 1. Install LangGraph and define the shared state Install langgraph alongside langchain and the Anthropic provider package. Define a TypedDict that every node reads from and writes to: the original request, the research findings, the computed quote, and the final draft. This shared state is what makes orchestration different from three agents that never talk to each other; each sub-agent only adds its own field and leaves the rest alone. ``` pip install langgraph langchain langchain-anthropic export ANTHROPIC_API_KEY="sk-ant-..." from typing import TypedDict, Optional class DealState(TypedDict): request: str company_research: Optional[str] quote: Optional[str] draft_email: Optional[str] next_step: Optional[str] ``` **Expected result:** DealState importable with no errors; it's just a typed dict shape at this point, nothing runs yet. ### 2. Write the research and pricing sub-agent nodes Each sub-agent is a plain function that takes the current state and returns a partial update, the shape every LangGraph node must return. research_node calls a mock CRM lookup (swap for a real Salesforce or HubSpot call later) and fills company_research. pricing_node reads company_research, not the raw request, which is the point: it only sees what research already resolved, so its prompt stays small and focused. ``` from langchain_anthropic import ChatAnthropic model = ChatAnthropic(model="claude-sonnet-5-5") CRM = {"Acme Corp": {"employees": 1200, "industry": "logistics", "tier": "enterprise"}} def research_node(state: DealState) -> dict: name = state["request"].split("for ")[-1].split(",")[0].strip() record = CRM.get(name, {"employees": 0, "industry": "unknown", "tier": "smb"}) return {"company_research": f"{name}: {record}"} def pricing_node(state: DealState) -> dict: resp = model.invoke( f"Given this company research: {state['company_research']}\n" f"and this request: {state['request']}\n" "Compute a per-seat annual price and a discount band. One short paragraph." ) return {"quote": resp.content} ``` **Expected result:** Two callable functions, each testable alone by passing in a hand-built DealState dict. ### 3. Write the drafting node and the supervisor drafting_node turns the quote into a proposal email. supervisor_node is the orchestration logic itself: it looks at which fields are already filled in state and returns the name of the next node to run, as a plain string. Keeping that decision in one function, reading state instead of a counter, is what lets you add a fourth stage (legal review, say) later without touching research or pricing. ``` def drafting_node(state: DealState) -> dict: resp = model.invoke( f"Write a short proposal email using this quote: {state['quote']}\n" f"Original request: {state['request']}" ) return {"draft_email": resp.content} def supervisor_node(state: DealState) -> dict: if not state.get("company_research"): return {"next_step": "research"} if not state.get("quote"): return {"next_step": "pricing"} if not state.get("draft_email"): return {"next_step": "drafting"} return {"next_step": "done"} ``` **Expected result:** supervisor_node returns {'next_step': 'research'} when called on a fresh state with only 'request' set. ### 4. Assemble the graph with conditional routing Build the StateGraph, add all four nodes, and wire the supervisor's output to add_conditional_edges: a routing function reads next_step from state and returns the matching node name, or END when next_step is 'done'. Every worker node routes back to the supervisor instead of to each other, so the supervisor is the only place that decides what runs next. ``` from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import InMemorySaver graph = StateGraph(DealState) graph.add_node("supervisor", supervisor_node) graph.add_node("research", research_node) graph.add_node("pricing", pricing_node) graph.add_node("drafting", drafting_node) graph.add_edge(START, "supervisor") graph.add_conditional_edges( "supervisor", lambda state: state["next_step"], {"research": "research", "pricing": "pricing", "drafting": "drafting", "done": END}, ) graph.add_edge("research", "supervisor") graph.add_edge("pricing", "supervisor") graph.add_edge("drafting", "supervisor") app = graph.compile(checkpointer=InMemorySaver()) ``` **Expected result:** app.compile() succeeds with no errors; app is a runnable graph. ### 5. Run the orchestration end to end Invoke the compiled graph with a thread_id in its config so the checkpointer can resume it later if it crashes mid-run. The graph loops through supervisor, research, supervisor, pricing, supervisor, drafting, supervisor until next_step is 'done', then stops. Print the final state to see every field each sub-agent contributed. ``` config = {"configurable": {"thread_id": "deal-acme-001"}} result = app.invoke( {"request": "Put together a quote for Acme Corp, 200 seats, annual plan"}, config=config, ) print(result["draft_email"]) ``` **Expected result:** A printed proposal email, and result also contains company_research and quote from the earlier hops. ### 6. Harden it: cap loops and isolate sub-agent failures Two things turn this from a demo into something safe to run unattended. First, pass recursion_limit in the invoke config so a supervisor bug that never sets next_step to 'done' fails loudly instead of looping forever. Second, wrap each worker node's body in try/except and write an error field into state instead of raising, so the supervisor can route to a human-review node on failure instead of crashing the whole graph. ``` def pricing_node(state: DealState) -> dict: try: resp = model.invoke(f"...{state['company_research']}...") return {"quote": resp.content} except Exception as e: return {"quote": None, "next_step": "needs_human", "error": str(e)} result = app.invoke(initial_state, config={**config, "recursion_limit": 15}) ``` **Expected result:** A pricing failure sets next_step to needs_human instead of raising, and a runaway loop stops at 15 steps with a clear GraphRecursionError. ## Complete code example File: `deal_desk_orchestrator.py` ```python """Deal-desk orchestrator: a LangGraph supervisor routes one request through research, pricing, and drafting sub-agents. Run: python deal_desk_orchestrator.py (pip install langgraph langchain langchain-anthropic) """ from typing import TypedDict, Optional from langchain_anthropic import ChatAnthropic from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import InMemorySaver model = ChatAnthropic(model="claude-sonnet-5-5") CRM = {"Acme Corp": {"employees": 1200, "industry": "logistics", "tier": "enterprise"}} class DealState(TypedDict): request: str company_research: Optional[str] quote: Optional[str] draft_email: Optional[str] next_step: Optional[str] error: Optional[str] def research_node(state: DealState) -> dict: name = state["request"].split("for ")[-1].split(",")[0].strip() record = CRM.get(name, {"employees": 0, "industry": "unknown", "tier": "smb"}) return {"company_research": f"{name}: {record}"} def pricing_node(state: DealState) -> dict: try: resp = model.invoke( f"Given this company research: {state['company_research']}\n" f"and this request: {state['request']}\n" "Compute a per-seat annual price and a discount band. One short paragraph." ) return {"quote": resp.content} except Exception as e: return {"quote": None, "next_step": "needs_human", "error": str(e)} def drafting_node(state: DealState) -> dict: resp = model.invoke( f"Write a short proposal email using this quote: {state['quote']}\n" f"Original request: {state['request']}" ) return {"draft_email": resp.content} def supervisor_node(state: DealState) -> dict: if state.get("next_step") == "needs_human": return {"next_step": "needs_human"} if not state.get("company_research"): return {"next_step": "research"} if not state.get("quote"): return {"next_step": "pricing"} if not state.get("draft_email"): return {"next_step": "drafting"} return {"next_step": "done"} graph = StateGraph(DealState) graph.add_node("supervisor", supervisor_node) graph.add_node("research", research_node) graph.add_node("pricing", pricing_node) graph.add_node("drafting", drafting_node) graph.add_edge(START, "supervisor") graph.add_conditional_edges( "supervisor", lambda state: state["next_step"], { "research": "research", "pricing": "pricing", "drafting": "drafting", "needs_human": END, "done": END, }, ) graph.add_edge("research", "supervisor") graph.add_edge("pricing", "supervisor") graph.add_edge("drafting", "supervisor") app = graph.compile(checkpointer=InMemorySaver()) if __name__ == "__main__": config = {"configurable": {"thread_id": "deal-acme-001"}, "recursion_limit": 15} result = app.invoke( {"request": "Put together a quote for Acme Corp, 200 seats, annual plan"}, config=config, ) if result.get("error"): print(f"Pricing failed, routed to human review: {result['error']}") else: print(result["draft_email"]) ``` ## Common mistakes - **Letting every sub-agent see the full original request instead of just the state fields it needs** — undefined Fix: Pass only what each node needs (pricing reads company_research, not the raw request). Smaller context per node means fewer irrelevant tool calls and cheaper runs. - **Hardcoding a fixed sequence (research then pricing then drafting) instead of routing through a supervisor** — undefined Fix: A fixed sequence can't skip steps or add a human-review branch later. Route every hop through one supervisor function that reads state. - **Forgetting add_edge back to the supervisor after a worker node, so the graph dead-ends** — undefined Fix: Every worker node needs an edge back to the node that decides what's next, or the graph has no way to continue or terminate correctly. - **Running without a recursion_limit while a supervisor bug is still possible** — undefined Fix: A supervisor that never returns next_step='done' loops forever. Set recursion_limit in the invoke config during development. - **Treating a sub-agent's exception as the whole run's failure** — undefined Fix: Catch it inside the node, write an error field, and route to a human-review or retry node. One bad pricing call shouldn't lose the research step that already succeeded. ## Best practices - Keep the supervisor's routing logic in one function that only reads state, never in scattered if/else checks across worker nodes. - Give each worker node the narrowest state slice it needs; don't let every node read the full object out of convenience. - Always compile with a checkpointer, even InMemorySaver for development, so you can resume or inspect a run that stalled. - Set recursion_limit explicitly once you add any conditional loop; the default exists but a stated limit documents your intent. - Route failures to a named next_step ('needs_human', 'retry') instead of letting exceptions propagate out of invoke(). - Use a cheaper model for research-style lookups and a stronger one only for the step that actually needs judgment, like pricing or drafting. ## Use cases ### undefined ### undefined ### undefined ## Frequently asked questions ### What's the difference between AI agent orchestration and a multi-agent system? A multi-agent system is the set of specialized agents; orchestration is the layer that decides which one runs, in what order, and with what shared state. You can have multiple agents with no real orchestration (a simple round-robin) or sophisticated orchestration over just two agents. ### Why use a supervisor function instead of just chaining nodes with add_edge? A fixed chain can't skip steps, retry one stage, or branch to a human-review path based on what a sub-agent found. A supervisor reads state and decides the next hop every time, so the graph's shape can change at runtime. ### Does every sub-agent need to be a separate LLM call? No. research_node in this guide is a plain Python function with no model call at all. Only give a node its own LLM call when the task actually needs language understanding or generation; a lookup or calculation is cheaper and more reliable as plain code. ### How do I debug which path the supervisor actually took? Call app.get_graph().draw_mermaid() to see the full edge map, and log next_step at the top of supervisor_node. LangGraph's checkpointer also lets you inspect the state history for a given thread_id after the run finishes. ### Can I orchestrate agents built on different frameworks, like one CrewAI crew and one raw OpenAI Agents SDK agent? Yes, as long as each one is wrapped in a plain function that takes and returns the shared state shape. LangGraph doesn't care what runs inside a node, only that the node's return value merges into state correctly. ### What does running this cost per request? With Claude Sonnet 5.5 ($2/M input, $10/M output) and two model calls (pricing and drafting) at a few thousand tokens each, a full research-to-draft run typically costs a few cents, well under the orchestration overhead of most alternatives. ### Can someone build this orchestration layer for me? Yes. RapidDev builds multi-agent orchestration on LangGraph and the other major SDKs, including the checkpointing and failure-routing most tutorials skip. A free 30-minute consultation is the fastest way to scope your version. --- Source: https://www.rapidevelopers.com/build-ai-agent/how-to-orchestrate-ai-agents © RapidDev — https://www.rapidevelopers.com/build-ai-agent/how-to-orchestrate-ai-agents