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.
| Fact | Value |
|---|---|
| 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 |
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.
Prerequisites
- 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
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.
1pip install langgraph langchain langchain-anthropic2export ANTHROPIC_API_KEY="sk-ant-..."34from typing import TypedDict, Optional56class DealState(TypedDict):7 request: str8 company_research: Optional[str]9 quote: Optional[str]10 draft_email: Optional[str]11 next_step: Optional[str]Expected result: DealState importable with no errors; it's just a typed dict shape at this point, nothing runs yet.
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.
1from langchain_anthropic import ChatAnthropic23model = ChatAnthropic(model="claude-sonnet-5-5")45CRM = {"Acme Corp": {"employees": 1200, "industry": "logistics", "tier": "enterprise"}}67def research_node(state: DealState) -> dict:8 name = state["request"].split("for ")[-1].split(",")[0].strip()9 record = CRM.get(name, {"employees": 0, "industry": "unknown", "tier": "smb"})10 return {"company_research": f"{name}: {record}"}1112def pricing_node(state: DealState) -> dict:13 resp = model.invoke(14 f"Given this company research: {state['company_research']}\n"15 f"and this request: {state['request']}\n"16 "Compute a per-seat annual price and a discount band. One short paragraph."17 )18 return {"quote": resp.content}Expected result: Two callable functions, each testable alone by passing in a hand-built DealState dict.
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.
1def drafting_node(state: DealState) -> dict:2 resp = model.invoke(3 f"Write a short proposal email using this quote: {state['quote']}\n"4 f"Original request: {state['request']}"5 )6 return {"draft_email": resp.content}78def supervisor_node(state: DealState) -> dict:9 if not state.get("company_research"):10 return {"next_step": "research"}11 if not state.get("quote"):12 return {"next_step": "pricing"}13 if not state.get("draft_email"):14 return {"next_step": "drafting"}15 return {"next_step": "done"}Expected result: supervisor_node returns {'next_step': 'research'} when called on a fresh state with only 'request' set.
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.
1from langgraph.graph import StateGraph, START, END2from langgraph.checkpoint.memory import InMemorySaver34graph = StateGraph(DealState)5graph.add_node("supervisor", supervisor_node)6graph.add_node("research", research_node)7graph.add_node("pricing", pricing_node)8graph.add_node("drafting", drafting_node)910graph.add_edge(START, "supervisor")11graph.add_conditional_edges(12 "supervisor",13 lambda state: state["next_step"],14 {"research": "research", "pricing": "pricing", "drafting": "drafting", "done": END},15)16graph.add_edge("research", "supervisor")17graph.add_edge("pricing", "supervisor")18graph.add_edge("drafting", "supervisor")1920app = graph.compile(checkpointer=InMemorySaver())Expected result: app.compile() succeeds with no errors; app is a runnable graph.
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.
1config = {"configurable": {"thread_id": "deal-acme-001"}}2result = app.invoke(3 {"request": "Put together a quote for Acme Corp, 200 seats, annual plan"},4 config=config,5)6print(result["draft_email"])Expected result: A printed proposal email, and result also contains company_research and quote from the earlier hops.
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.
1def pricing_node(state: DealState) -> dict:2 try:3 resp = model.invoke(f"...{state['company_research']}...")4 return {"quote": resp.content}5 except Exception as e:6 return {"quote": None, "next_step": "needs_human", "error": str(e)}78result = 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 working example
1"""Deal-desk orchestrator: a LangGraph supervisor routes one request through2research, pricing, and drafting sub-agents.3Run: python deal_desk_orchestrator.py (pip install langgraph langchain langchain-anthropic)4"""5from typing import TypedDict, Optional67from langchain_anthropic import ChatAnthropic8from langgraph.graph import StateGraph, START, END9from langgraph.checkpoint.memory import InMemorySaver1011model = ChatAnthropic(model="claude-sonnet-5-5")1213CRM = {"Acme Corp": {"employees": 1200, "industry": "logistics", "tier": "enterprise"}}141516class DealState(TypedDict):17 request: str18 company_research: Optional[str]19 quote: Optional[str]20 draft_email: Optional[str]21 next_step: Optional[str]22 error: Optional[str]232425def research_node(state: DealState) -> dict:26 name = state["request"].split("for ")[-1].split(",")[0].strip()27 record = CRM.get(name, {"employees": 0, "industry": "unknown", "tier": "smb"})28 return {"company_research": f"{name}: {record}"}293031def pricing_node(state: DealState) -> dict:32 try:33 resp = model.invoke(34 f"Given this company research: {state['company_research']}\n"35 f"and this request: {state['request']}\n"36 "Compute a per-seat annual price and a discount band. One short paragraph."37 )38 return {"quote": resp.content}39 except Exception as e:40 return {"quote": None, "next_step": "needs_human", "error": str(e)}414243def drafting_node(state: DealState) -> dict:44 resp = model.invoke(45 f"Write a short proposal email using this quote: {state['quote']}\n"46 f"Original request: {state['request']}"47 )48 return {"draft_email": resp.content}495051def supervisor_node(state: DealState) -> dict:52 if state.get("next_step") == "needs_human":53 return {"next_step": "needs_human"}54 if not state.get("company_research"):55 return {"next_step": "research"}56 if not state.get("quote"):57 return {"next_step": "pricing"}58 if not state.get("draft_email"):59 return {"next_step": "drafting"}60 return {"next_step": "done"}616263graph = StateGraph(DealState)64graph.add_node("supervisor", supervisor_node)65graph.add_node("research", research_node)66graph.add_node("pricing", pricing_node)67graph.add_node("drafting", drafting_node)6869graph.add_edge(START, "supervisor")70graph.add_conditional_edges(71 "supervisor",72 lambda state: state["next_step"],73 {74 "research": "research",75 "pricing": "pricing",76 "drafting": "drafting",77 "needs_human": END,78 "done": END,79 },80)81graph.add_edge("research", "supervisor")82graph.add_edge("pricing", "supervisor")83graph.add_edge("drafting", "supervisor")8485app = graph.compile(checkpointer=InMemorySaver())8687if __name__ == "__main__":88 config = {"configurable": {"thread_id": "deal-acme-001"}, "recursion_limit": 15}89 result = app.invoke(90 {"request": "Put together a quote for Acme Corp, 200 seats, annual plan"},91 config=config,92 )93 if result.get("error"):94 print(f"Pricing failed, routed to human review: {result['error']}")95 else:96 print(result["draft_email"])97Common mistakes
Why it's a problem: Letting every sub-agent see the full original request instead of just the state fields it needs
How to avoid: 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.
Why it's a problem: Hardcoding a fixed sequence (research then pricing then drafting) instead of routing through a supervisor
How to avoid: A fixed sequence can't skip steps or add a human-review branch later. Route every hop through one supervisor function that reads state.
Why it's a problem: Forgetting add_edge back to the supervisor after a worker node, so the graph dead-ends
How to avoid: 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.
Why it's a problem: Running without a recursion_limit while a supervisor bug is still possible
How to avoid: A supervisor that never returns next_step='done' loops forever. Set recursion_limit in the invoke config during development.
Why it's a problem: Treating a sub-agent's exception as the whole run's failure
How to avoid: 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.
Still stuck?
Copy one of these prompts to get a personalized, step-by-step explanation.
I have a LangGraph supervisor orchestrating research, pricing, and drafting nodes over a shared DealState (code below). Add a fourth node, legal_review, that the supervisor routes to whenever company_research mentions a regulated industry, before drafting runs. [paste complete_code]
In Claude Code, open deal_desk_orchestrator.py and ask: 'Add a retry node that the supervisor routes to when pricing sets next_step to needs_human, and retry pricing once before giving up.'
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.
Talk to an Expert
Our team has built 1,000+ apps. Get personalized help with your project.
Book a free consultation