Master multi-agent orchestration with LangGraph, AutoGen teams, and OpenAI handoffs. Learn DAG-style routing, typed shared state, protocol boundaries, and human-in-the-loop controls for reliable AI systems.
Agent failure handling gives one worker a recovery plan. Multi-agent orchestration starts when one agent has too many responsibilities and the job needs typed nodes, explicit edges, persistent state, and approval points.
Multi-agent orchestration splits a hard workflow into specialized steps with explicit dependencies. A graph makes control flow clearer when one model call shouldn't plan, retrieve, decide, and act alone. It becomes safer only when it also limits capabilities, validates merged evidence, and binds approvals to exact side effects.
One release engineer shouldn't inspect eval results, check canary metrics, verify rollout policy, and write the incident update while other releases wait. Strong platform teams assign bounded roles: one worker reads evals, another checks service health, a third validates policy, and a coordinator owns the handoff.
Long-running agent workflows face a similar coordination problem. As one transcript accumulates instructions, tool results, and intermediate decisions inside a limited context window, relevant state becomes harder to isolate. Multi-agent orchestration splits the work into specialized agents connected by an explicit workflow graph. Instead of one prompt that tries to do everything, you build a small pipeline where a triage agent classifies the request, a deployment-state agent queries the release store, and a response formatter composes the reply.
You have already practiced the pieces this pattern needs: structured outputs, ReAct loops, human-in-the-loop (HITL) approval gates, memory, and failure recovery. If those feel fuzzy, review the prerequisite lessons first. Here, the next step is wiring multiple specialized agents into a graph so they collaborate on a single job without hiding state in a transcript.
AutoGen[1] and MetaGPT[2] are useful early examples of role-based coordination. AutoGen evaluated multi-agent conversation patterns across coding, math, and question answering. MetaGPT organized software agents around explicit standard operating procedures (SOPs). Together, they made the systems-design questions concrete: who owns state, which worker acts next, and how does the runtime stop?
Before you reach for a graph of agents, take the cost seriously. First-party engineering reports now make the boundary clearer: parallel exploration can pay off, while coupled decisions and unbounded writes make multi-agent systems harder to control.
Cognition, the team behind Devin[3], published a widely read argument titled "Don't Build Multi-Agents."[4] Their position: when you split work across parallel subagents, each subagent acts on its own slice of context. Because every action carries implicit decisions, those decisions conflict when the subagents can't see each other's full traces. The system can look parallel while producing inconsistent, hard-to-reconcile output. Their recommended default is a single agent with strong context engineering, and where you must split work, share the complete trace rather than isolated messages.
In a 2026 follow-up, Cognition updated that position with a narrower pattern they said was working in practice: multiple agents may contribute intelligence, but writes stay single-threaded.[5] That refinement is central to production design. Parallel research, retrieval, and proposal branches don't justify parallel traffic shifts, data deletion, or incident-publication writes.
One day after Cognition's original June 2025 post, Anthropic published "How we built our multi-agent research system"[6] describing a different choice for their Research feature: a lead agent that spawns 3 to 5 subagents, each exploring a different part of the question in its own context window. Their internal research eval showed this setup outperformed a single Opus 4 agent by 90.2%. The catch they state plainly: multi-agent systems use about 15 times more tokens than chats, so the task value has to justify the larger token budget. They also name where multi-agent is a poor fit: domains where every agent must share the same context or agent dependencies are heavy. They cite coding as exactly that kind of poor fit, because subtasks are tightly coupled and agents struggle to coordinate in real time.
Read together, Cognition's warning and refinement plus Anthropic's research system describe the same trade-off from different task shapes:
| Signal | Favors a single strong agent | Favors multiple agents |
|---|---|---|
| Subtask coupling | Tight, decisions depend on each other | Loose, subtasks explore independent slices |
| Context sharing | Every step needs the full trace | Each branch needs only its own slice |
| Parallelism | Little, the work is sequential | High, branches run at the same time |
| Context size | Fits one window with good engineering | Exceeds one window, needs compression |
| Task value | Routine, cost-sensitive | High value, can pay a much larger token budget |
| External writes | One bounded writer | Parallel evidence only, then one controlled writer |
The practical rule for 2026: reach for multi-agent only when subtasks are genuinely separable and parallel, and when the task value justifies the token and failure-surface cost. Otherwise a single agent with disciplined context engineering is cheaper and easier to debug. The rest of the design assumes you have cleared that bar and now need to wire the graph correctly.
A small admission check keeps that decision concrete: parallel branches win only when work is independent, worth the extra execution, and able to merge through a clear contract.
1from dataclasses import dataclass
2
3@dataclass(frozen=True)
4class TaskShape:
5 independent_branches: int
6 needs_shared_trace: bool
7 value_justifies_extra_cost: bool
8 merge_contract_defined: bool
9
10def choose_orchestration(task: TaskShape) -> str:
11 if (
12 task.independent_branches >= 2
13 and not task.needs_shared_trace
14 and task.value_justifies_extra_cost
15 and task.merge_contract_defined
16 ):
17 return "parallel graph"
18 return "single scoped agent"
19
20research = TaskShape(4, False, True, True)
21code_edit = TaskShape(3, True, True, False)
22
23print("research:", choose_orchestration(research))
24print("coupled code edit:", choose_orchestration(code_edit))1research: parallel graph
2coupled code edit: single scoped agent
In the simplest case, an agent pipeline is a straight line: the user asks a question, agent A looks it up, agent B analyzes it, and agent C replies. For a narrow task like "What does this runbook say?", a chain works fine. But production-release questions are messy. An on-call engineer might ask, "Can we promote reranker-v17, and did the canary burn error budget?" That touches evals, deployment state, and policy at the same time. A rigid assembly line forces you to process these in sequence even when they could run in parallel.
A naive pipeline processes steps sequentially, handing output directly from one task to the next. The visual below compares that rigid path with a DAG route that can branch by intent and merge results before the final answer.
Linear chains break the moment you need:
A Directed Acyclic Graph (DAG) models your agent workflow as nodes (agents or functions) and edges (dependencies). It defines routing rules: "If the Triage Agent says 'deployment health', go to the Health Analyst. If it says 'promotion request', go to the Policy Agent." A DAG gives you branching and parallelism without back-edges.
Once you add retry loops, review loops, or approval loops, you're no longer dealing with a strict DAG. You're dealing with a state machine or a more general graph. LangGraph supports both. In the DAG route above, the triage node can fan out to specialists and every successful branch still converges on the formatter without cycling.
The control flow is explicit. Given the same state and routing decisions, you can inspect why execution moved from one node to the next. That's much easier to debug than an open-ended conversation loop where an LLM silently decides the next speaker.
Explicit routing isn't authorization. Keep read-only evidence branches separate from any node that can shift traffic, roll back a service, or send an external notification. A write-capable node must still require policy checks, a bound approval when needed, and an idempotency key.
LangGraph is a graph-based framework for multi-agent orchestration. Its core primitives are a state schema, node functions, and edges.[7]
A shared TypedDict (a Python type hint that adds static typing to dictionaries) flows through the graph as the memory every agent reads and updates. The schema keeps the release-health graph compact: message history, the current routing step, retrieved deployment data, and an error count. In production, sensitive or large release records should remain in an access-controlled store; graph state should carry a scoped reference plus the verified fields downstream nodes need.
1from typing import Annotated
2from typing_extensions import TypedDict
3from langchain_core.messages import AIMessage, AnyMessage
4from langgraph.graph.message import add_messages
5
6DeploymentData = dict[str, str]
7
8class AgentState(TypedDict):
9 # Messages accumulate using the add_messages reducer
10 messages: Annotated[list[AnyMessage], add_messages]
11 current_step: str
12 deployment_data: DeploymentData | None
13 service_health: str | None
14 final_answer: str | None
15 error_count: int
16 next_worker: str | NoneMessage reducer: The
Annotated[list[AnyMessage], add_messages]pattern tells LangGraph to merge updates by appending rather than overwriting. If two parallel branches both append messages, you get both sets in the final list.
Nodes are standard Python functions that transform the graph's state. The functions below implement a deployment lookup and a health analysis step. Each takes the current AgentState, performs external read calls (like a release-store query or metrics API call), and returns a dictionary of state updates. LangGraph merges those updates into the global state automatically.
In the snippet, deployment_db and metrics_api are application adapters. The complete local smoke test after the routing section uses in-memory adapters so you can run the graph without API keys.
1def lookup_node(state: AgentState) -> dict[str, object]:
2 """Deployment lookup agent: queries the release store."""
3 query = state["messages"][-1].content
4 # Example: using a search tool or database wrapper
5 results = deployment_db.search(query)
6 return {
7 "deployment_data": results,
8 "current_step": "health_check",
9 # Appends to messages list due to Annotated[list, add_messages]
10 "messages": [AIMessage(content=f"Found release {results['release_id']}")],
11 }
12
13def health_node(state: AgentState) -> dict[str, object]:
14 """Health analyst: checks service metrics for the release."""
15 deployment = state["deployment_data"]
16 if deployment is None:
17 raise RuntimeError("Missing deployment data")
18 health = metrics_api.health(deployment["release_id"])
19 return {
20 "service_health": health.status,
21 "final_answer": f"Release {deployment['release_id']} health is {health.status}.",
22 "current_step": "format",
23 "messages": [AIMessage(content=f"Service health: {health.status}")],
24 }Notice that each node returns only the fields it changes. LangGraph patches those into the existing state rather than replacing the whole object. That means the deployment_data produced by lookup_node is still visible when health_node runs, even though health_node doesn't return deployment_data again.
Conditional edges define the graph's control flow based on the current state. A triage node can write a routing decision into shared state, and a small router function can read that decision and return the next node name. That keeps the classification logic inside a node while keeping the edge function deterministic and easy to inspect.
1from typing import Literal
2from langchain_core.messages import AIMessage
3from langgraph.graph import StateGraph, START, END
4
5Route = Literal["lookup_agent", "faq_agent", "policy_agent"]
6
7def triage_node(state: AgentState) -> dict[str, object]:
8 user_query = state["messages"][-1].content.lower()
9
10 if any(token in user_query for token in ("promote", "rollback", "policy")):
11 route: Route = "policy_agent"
12 elif any(token in user_query for token in ("deploy", "release", "canary")):
13 route = "lookup_agent"
14 else:
15 route = "faq_agent"
16
17 return {"next_worker": route}
18
19def faq_node(state: AgentState) -> dict[str, object]:
20 answer = llm.invoke(
21 f"Answer briefly from the FAQ: {state['messages'][-1].content}"
22 )
23 return {"final_answer": answer.content, "current_step": "format"}
24
25def policy_node(state: AgentState) -> dict[str, object]:
26 answer = llm.invoke(
27 f"Explain the rollout policy for: {state['messages'][-1].content}"
28 )
29 return {"final_answer": answer.content, "current_step": "format"}
30
31def format_node(state: AgentState) -> dict[str, object]:
32 return {
33 "current_step": "complete",
34 "messages": [AIMessage(content=state["final_answer"] or "No answer generated yet.")],
35 }
36
37def router(state: AgentState) -> Route:
38 return state["next_worker"] or "faq_agent"
39
40# Build the graph
41builder = StateGraph(AgentState)
42builder.add_node("triage", triage_node)
43builder.add_node("lookup_agent", lookup_node)
44builder.add_node("faq_agent", faq_node)
45builder.add_node("policy_agent", policy_node)
46builder.add_node("health_analyst", health_node)
47builder.add_node("response_formatter", format_node)
48
49builder.add_edge(START, "triage")
50
51# Conditional edges handle dynamic routing
52builder.add_conditional_edges("triage", router)
53
54# Standard edges define fixed transitions and the final fan-in
55builder.add_edge("lookup_agent", "health_analyst")
56builder.add_edge("health_analyst", "response_formatter")
57builder.add_edge("faq_agent", "response_formatter")
58builder.add_edge("policy_agent", "response_formatter")
59builder.add_edge("response_formatter", END)
60
61app = builder.compile()If you invoke this graph with the query "Can we release search-api?", the trace looks like this:
triage_node reads the query, writes "lookup_agent" into next_worker."lookup_agent", so LangGraph calls lookup_node.lookup_node fetches release data, writes deployment_data, and sets current_step to "health_check".lookup_agent -> health_analyst triggers health_node.health_node queries metrics, writes service_health, stores final_answer, and sets current_step to "format".health_analyst -> response_formatter triggers format_node.format_node reads final_answer and appends the final message.response_formatter -> END terminates the run.Production tip: Keep router functions deterministic. If the routing decision depends on an LLM call, perform that call inside a node (like
triage_node), write the result into state, and let the edge function read it. That way you can log the exact routing decision without re-running the LLM during debugging.
Routing also needs a capability boundary. A requester mentioning promotion may route to a policy lookup and review proposal, but it must not shift production traffic merely because a specialist node exists.
1ROUTES = {
2 "release": ("eval_read", "deployment_read", "draft_summary"),
3 "promotion": ("eval_read", "policy_read", "propose_promotion", "human_review"),
4}
5
6def route_for(message: str) -> tuple[str, ...]:
7 intent = "promotion" if "promote" in message.lower() else "release"
8 return ROUTES[intent]
9
10path = route_for("Please promote reranker-v17")
11assert "shift_traffic" not in path
12print("route:", " -> ".join(path))
13print("external write included:", "shift_traffic" in path)1route: eval_read -> policy_read -> propose_promotion -> human_review
2external write included: FalseThe same graph can run as a complete local smoke test. It uses fake release and health data, but the LangGraph state, nodes, conditional edge, fixed edges, and invocation are real:
1from typing import Annotated, Literal
2from typing_extensions import TypedDict
3
4from langchain_core.messages import AIMessage, AnyMessage, HumanMessage
5from langgraph.graph import END, START, StateGraph
6from langgraph.graph.message import add_messages
7
8DeploymentData = dict[str, str]
9
10class AgentState(TypedDict):
11 messages: Annotated[list[AnyMessage], add_messages]
12 current_step: str
13 deployment_data: DeploymentData | None
14 service_health: str | None
15 final_answer: str | None
16 error_count: int
17 next_worker: str | None
18
19DEPLOYMENTS = {"search-api": {"service_id": "search-api", "release_id": "rel-7421"}}
20HEALTH = {"rel-7421": "error_budget_ok"}
21Route = Literal["lookup_agent", "faq_agent", "policy_agent"]
22
23def triage_node(state: AgentState) -> dict[str, object]:
24 user_query = state["messages"][-1].content.lower()
25 if any(token in user_query for token in ("promote", "rollback", "policy")):
26 route: Route = "policy_agent"
27 elif any(token in user_query for token in ("deploy", "release", "canary")):
28 route = "lookup_agent"
29 else:
30 route = "faq_agent"
31 return {"next_worker": route}
32
33def router(state: AgentState) -> Route:
34 return state["next_worker"] or "faq_agent"
35
36def lookup_node(state: AgentState) -> dict[str, object]:
37 query = state["messages"][-1].content
38 service_id = "search-api" if "search" in query else "unknown"
39 deployment = DEPLOYMENTS[service_id]
40 return {
41 "deployment_data": deployment,
42 "current_step": "health_check",
43 "messages": [AIMessage(content=f"Found release {deployment['release_id']}")],
44 }
45
46def health_node(state: AgentState) -> dict[str, object]:
47 deployment = state["deployment_data"]
48 if deployment is None:
49 raise ValueError("deployment_data must be populated before health_node runs")
50 health = HEALTH[deployment["release_id"]]
51 return {
52 "service_health": health,
53 "final_answer": f"Release {deployment['release_id']} health is {health}.",
54 "current_step": "format",
55 "messages": [AIMessage(content=f"Service health: {health}")],
56 }
57
58def faq_node(state: AgentState) -> dict[str, object]:
59 return {"final_answer": "FAQ answer", "current_step": "format"}
60
61def policy_node(state: AgentState) -> dict[str, object]:
62 return {"final_answer": "Policy answer", "current_step": "format"}
63
64def format_node(state: AgentState) -> dict[str, object]:
65 return {
66 "current_step": "complete",
67 "messages": [
68 AIMessage(content=state["final_answer"] or "No answer generated yet.")
69 ],
70 }
71
72builder = StateGraph(AgentState)
73builder.add_node("triage", triage_node)
74builder.add_node("lookup_agent", lookup_node)
75builder.add_node("faq_agent", faq_node)
76builder.add_node("policy_agent", policy_node)
77builder.add_node("health_analyst", health_node)
78builder.add_node("response_formatter", format_node)
79
80builder.add_edge(START, "triage")
81builder.add_conditional_edges("triage", router)
82builder.add_edge("lookup_agent", "health_analyst")
83builder.add_edge("health_analyst", "response_formatter")
84builder.add_edge("faq_agent", "response_formatter")
85builder.add_edge("policy_agent", "response_formatter")
86builder.add_edge("response_formatter", END)
87
88app = builder.compile()
89result = app.invoke({
90 "messages": [HumanMessage(content="Can we release search-api?")],
91 "current_step": "start",
92 "deployment_data": None,
93 "service_health": None,
94 "final_answer": None,
95 "error_count": 0,
96 "next_worker": None,
97})
98
99print("service_health:", result["service_health"])
100print("final message:", result["messages"][-1].content)
101print("route completed:", result["service_health"] == "error_budget_ok")1service_health: error_budget_ok
2final message: Release rel-7421 health is error_budget_ok.
3route completed: TrueOnce you can build a single graph, the next question is how to arrange agents when the problem grows. Start with dependency shape, then choose how much routing authority stays centralized.
One LLM acts as a manager that delegates tasks to specialized workers. That pattern works when the sub-steps aren't known in advance. In our release system, a supervisor might look at the conversation history and decide whether to call the deployment lookup agent, the policy agent, or a human escalation path.
1from langchain_core.messages import SystemMessage
2
3def supervisor_node(state: AgentState) -> dict[str, object]:
4 """Supervisor decides which worker to invoke next."""
5 response = llm.invoke([
6 SystemMessage(content="""You are a supervisor managing these workers:
7 - deployment_lookup: Queries the release store
8 - health_analyst: Checks service health metrics
9 - policy_agent: Explains rollout and rollback rules
10 - response_formatter: Writes the final operator reply
11
12 Based on the conversation, decide which worker to invoke next,
13 or respond with FINISH if the task is complete."""),
14 *state["messages"]
15 ])
16 return {"messages": [response], "next_worker": response.content}Supervisors offer simple, centralized control. When a promotion request turns out to need both deployment lookup and policy explanation, the supervisor can sequence them explicitly rather than hard-coding every combination in the graph edges.
Supervisors become a bottleneck and a single point of failure. Every step costs an extra LLM call, which adds latency and token cost. Supervisors also rely on the LLM outputting structured routing instructions (like "deployment_lookup"), which can be brittle with smaller models.
For complex enterprise workflows, one supervisor can become a wide routing bottleneck. A hierarchy introduces supervisors of supervisors. In a large AI platform, a "Release Manager" agent might manage an "Evaluation Team" (a sub-supervisor coordinating benchmark and regression checks) and an "Operations Team" (a sub-supervisor coordinating canary health and rollback readiness).
Hierarchical structures scale for massive tasks and offer clear separation of concerns. No single manager has to remember every detail of every sub-team.
Deep hierarchies add latency. Handoff dilution can occur when the original operator intent loses detail as it passes through multiple manager layers. A request that needs only one release-store lookup might now take three LLM calls just to decide who does the lookup.
Agents hand off to each other directly without a long-lived supervisor. OpenAI's experimental Swarm repo[8] made this pattern easy to study, and the current OpenAI Agents SDK exposes the same idea through first-class handoffs.[9][10] Handoffs are represented to the model as transfer tools, and input filters can control what history the receiving agent sees.[10] In the example below, the triage agent doesn't solve the task itself. It selects the specialist that should take over.
1import asyncio
2from agents import Agent, Runner
3
4release_agent = Agent(
5 name="Release Agent",
6 handoff_description="Handles model-promotion questions and rollout plans.",
7 instructions="You explain release status clearly and concisely.",
8)
9
10health_agent = Agent(
11 name="Health Agent",
12 handoff_description="Handles service metrics and canary-health updates.",
13 instructions="You look up deployment health and explain current signals.",
14)
15
16triage_agent = Agent(
17 name="Triage Agent",
18 instructions="Route the operator to one specialist.",
19 handoffs=[release_agent, health_agent],
20)
21
22async def main():
23 result = await Runner.run(
24 triage_agent,
25 "Can we promote reranker-v17 from shadow to canary?",
26 )
27 print(result.final_output)
28 print(f"Answered by: {result.last_agent.name}")
29
30asyncio.run(main())If the orchestrator should keep ownership of the conversation and only call specialists for sub-tasks, use specialists as tools instead of handoffs.
There's no supervisor bottleneck, agents are loosely coupled, and it's easy to add new specialists.
It's harder to debug because the flow is less explicit. Circular handoffs (A -> B -> A) are possible, so peer-to-peer flows need trace logging and hard turn limits.
This pattern uses a parallel fan-out for "embarrassingly parallel" tasks, followed by an aggregation step. It's ideal for tasks like checking a release from multiple perspectives (offline evals, canary metrics, rollback readiness). In the topology visual above, this is the fan-out shape: one router emits independent analyst tasks, and a reducer merges their findings into one release recommendation.
To implement a true map-reduce pattern in current LangGraph, the cleanest option is the Send API. It lets one node emit a variable number of parallel tasks, each with its own payload, and then merge the branch outputs through reducers.[7]
1import operator
2from typing import Annotated
3from typing_extensions import TypedDict
4from langgraph.graph import StateGraph, START, END
5from langgraph.types import Send
6
7class MapReduceState(TypedDict):
8 items: list[str]
9 findings: Annotated[list[str], operator.add]
10 combined_report: str
11
12class AnalystTask(TypedDict):
13 item: str
14
15def fan_out_node(state: MapReduceState) -> dict[str, object]:
16 return {"items": ["eval_gate", "canary_health", "rollback_plan"]}
17
18def analyst_node(state: AnalystTask) -> dict[str, object]:
19 item = state["item"]
20 return {"findings": [f"{item}: checked and ready"]}
21
22def route_to_analysts(state: MapReduceState) -> list[Send]:
23 return [Send("analyst", {"item": item}) for item in state["items"]]
24
25def reducer_node(state: MapReduceState) -> dict[str, object]:
26 return {"combined_report": "\n".join(state["findings"])}
27
28builder = StateGraph(MapReduceState)
29builder.add_node("fan_out", fan_out_node)
30builder.add_node("analyst", analyst_node)
31builder.add_node("reducer", reducer_node)
32
33builder.add_edge(START, "fan_out")
34builder.add_conditional_edges("fan_out", route_to_analysts, ["analyst"])
35builder.add_edge("analyst", "reducer")
36builder.add_edge("reducer", END)
37
38graph = builder.compile()
39
40result = graph.invoke({"items": [], "findings": [], "combined_report": ""})
41print(result["combined_report"])
42print("finding count:", len(result["findings"]))
43print("contains canary health:", "canary_health" in result["combined_report"])1eval_gate: checked and ready
2canary_health: checked and ready
3rollback_plan: checked and ready
4finding count: 3
5contains canary health: TrueOne practical detail: parallel branches should either write disjoint state keys or use reducers for shared keys such as Annotated[list[str], operator.add]. Otherwise the runtime has no safe way to merge their updates.
The reducer is also a trust boundary. It should admit only findings for the current tenant and request, with recorded evidence, before any formatter or decision node consumes them.
1from dataclasses import dataclass
2
3@dataclass(frozen=True)
4class Finding:
5 tenant_id: str
6 claim: str
7 source_ref: str
8 verified: bool
9
10def reduce_findings(findings: list[Finding], tenant_id: str) -> list[str]:
11 accepted: list[str] = []
12 for finding in findings:
13 if finding.tenant_id != tenant_id:
14 raise ValueError("tenant mismatch")
15 if not finding.verified or not finding.source_ref:
16 raise ValueError("missing verified evidence")
17 accepted.append(finding.claim)
18 return accepted
19
20safe = [Finding("tenant-7", "canary error budget ok", "metrics:evt-81", True)]
21foreign = [Finding("tenant-8", "promotion eligible", "evals:92", True)]
22
23print("accepted:", reduce_findings(safe, "tenant-7"))
24try:
25 reduce_findings(safe + foreign, "tenant-7")
26except ValueError as exc:
27 print("blocked merge:", exc)1accepted: ['canary error budget ok']
2blocked merge: tenant mismatchReducers must also reject incompatible scalar writes. Two branches may append independent findings; they shouldn't silently choose different traffic targets or rollback decisions.
1def merge_updates(updates: list[dict[str, object]]) -> dict[str, object]:
2 merged_findings: list[str] = []
3 proposed_traffic: str | None = None
4 for update in updates:
5 merged_findings.extend(update.get("findings", []))
6 if "proposed_traffic" in update:
7 candidate = str(update["proposed_traffic"])
8 if proposed_traffic is not None and candidate != proposed_traffic:
9 raise ValueError("conflicting traffic proposals")
10 proposed_traffic = candidate
11 return {"findings": merged_findings, "proposed_traffic": proposed_traffic}
12
13parallel_reads = [{"findings": ["eval verified"]}, {"findings": ["policy found"]}]
14print("findings:", merge_updates(parallel_reads)["findings"])
15try:
16 merge_updates([{"proposed_traffic": "10%"}, {"proposed_traffic": "25%"}])
17except ValueError as exc:
18 print("blocked write merge:", exc)1findings: ['eval verified', 'policy found']
2blocked write merge: conflicting traffic proposalsIn production, agents crash, APIs time out, and servers restart. You can't rely on in-memory state. LangGraph's InMemorySaver is convenient for local development, but production deployments usually use a durable checkpointer such as Postgres. The current docs use PostgresSaver.from_conn_string(...); one common pattern is wrapping it in a context manager so the runtime can persist each step and resume by thread_id after a crash.[11]
1from langgraph.checkpoint.postgres import PostgresSaver
2
3DB_URI = "postgresql://postgres:postgres@localhost:5442/postgres?sslmode=disable"
4
5with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
6 # Run once when initializing the checkpoint tables
7 # checkpointer.setup()
8
9 app = graph.compile(checkpointer=checkpointer)
10
11 config = {"configurable": {"thread_id": "user-123-conversation-456"}}
12 result = app.invoke(initial_state, config=config)Production tip: Use a durable checkpointer for workflows that must survive restarts or pause for approval. Reusing the same
thread_idlets the runtime resume from the latest checkpoint and inspect prior state snapshots during debugging.
One more durability rule matters in production: resumed runs replay from a checkpoint boundary instead of jumping back to the exact Python line where execution paused. That means side effects and non-deterministic work should either live in LangGraph tasks or be made idempotent.[11]
The storage boundary, not the model, must enforce idempotency. If the execution node is replayed after a timeout, the same key retrieves the first result rather than applying a second traffic shift.
1processed_promotions: dict[str, str] = {}
2
3def promote_release(idempotency_key: str, traffic_percent: int) -> str:
4 if idempotency_key in processed_promotions:
5 return f"reused {processed_promotions[idempotency_key]}"
6 promotion_id = f"promotion-{len(processed_promotions) + 1}"
7 processed_promotions[idempotency_key] = promotion_id
8 return f"created {promotion_id} for {traffic_percent}% traffic"
9
10key = "promotion:reranker-v17:25pct:v3"
11print("first run:", promote_release(key, 25))
12print("resumed run:", promote_release(key, 25))
13print("promotions applied:", len(processed_promotions))1first run: created promotion-1 for 25% traffic
2resumed run: reused promotion-1
3promotions applied: 1Autonomous agents are rarely fully trusted in production. You need breakpoints where a human can approve, modify, or reject an agent's plan. A large traffic shift, safety-filter change, or rollback that affects a critical endpoint are all cases where the graph should pause and wait. Approval must apply to one exact proposed action, not to the conversation in general.
LangGraph supports this natively through interrupt(), backed by a checkpointer.[12] Static interrupt_before / interrupt_after breakpoints still exist, but the current docs position them as debugging aids rather than production approval flows. In a production HITL flow, the node emits a structured approval request, the caller sees that payload under __interrupt__, and the graph resumes with Command(resume=...).
1import hashlib
2import json
3from typing import Literal
4from langgraph.types import Command, interrupt
5
6def action_digest(action: dict[str, object]) -> str:
7 encoded = json.dumps(action, sort_keys=True, separators=(",", ":")).encode()
8 return hashlib.sha256(encoded).hexdigest()
9
10def approval_node(state: dict[str, object]) -> Command[Literal["execute", "cancel"]]:
11 action = state["proposed_action"]
12 digest = action_digest(action)
13 decision = interrupt({
14 "question": "Approve this promotion?",
15 "action": action,
16 "action_digest": digest,
17 "release_version": state["release_version"],
18 "idempotency_key": state["idempotency_key"],
19 "evidence_refs": state["evidence_refs"],
20 })
21 approved = (
22 decision.get("approved") is True
23 and decision.get("action_digest") == digest
24 and decision.get("release_version") == state["release_version"]
25 )
26 return Command(goto="execute" if approved else "cancel")
27
28app = graph.compile(checkpointer=checkpointer)
29config = {"configurable": {"thread_id": "promotion-review-reranker-v17"}}
30
31# First call pauses and exposes the interrupt payload to the caller
32paused = app.invoke(inputs, config=config)
33print(paused["__interrupt__"])
34
35# The UI resumes with the exact displayed digest and release version
36decision = {
37 "approved": True,
38 "action_digest": displayed_digest,
39 "release_version": displayed_release_version,
40}
41app.invoke(Command(resume=decision), config=config)The downstream execute node must re-read the authoritative release version immediately before writing, require another review if it changed, and send the same idempotency key to the deployment service. One subtle runtime detail matters a lot in production: after resuming, LangGraph restarts the node from the top rather than continuing from the exact line of the interrupt(). Any side effects before the pause need to be idempotent or moved after the interrupt.
1import hashlib
2import json
3
4def proposal(traffic_percent: int, release_version: int) -> dict[str, object]:
5 action = {
6 "kind": "promotion",
7 "model_id": "reranker-v17",
8 "traffic_percent": traffic_percent,
9 }
10 digest = hashlib.sha256(json.dumps(action, sort_keys=True).encode()).hexdigest()
11 return {"action": action, "digest": digest, "release_version": release_version}
12
13def can_execute(packet: dict[str, object], decision: dict[str, object], live_version: int) -> bool:
14 return (
15 decision.get("approved") is True
16 and decision.get("digest") == packet["digest"]
17 and decision.get("release_version") == packet["release_version"] == live_version
18 )
19
20reviewed = proposal(25, 3)
21decision = {"approved": True, "digest": reviewed["digest"], "release_version": 3}
22changed_traffic = proposal(50, 3)
23
24print("reviewed action admitted:", can_execute(reviewed, decision, live_version=3))
25print("changed traffic admitted:", can_execute(changed_traffic, decision, live_version=3))
26print("stale release admitted:", can_execute(reviewed, decision, live_version=4))1reviewed action admitted: True
2changed traffic admitted: False
3stale release admitted: False
When orchestrating multiple agents, the choice of coordination contract shapes the whole architecture. In practice, three common patterns appear repeatedly: typed shared state, message-oriented coordination, and supervisor-based routing.
In a shared state model (used by LangGraph), a single typed object acts as a whiteboard. Every node reads from and writes to this schema. That makes intermediate variables explicit: if an early node fetches a release record, a node deep in the graph can still access it without repackaging it into a chat message. The downside is that the schema needs discipline. As the graph grows, the state can become bloated, and careless updates can overwrite important fields.
In a message-oriented model (common in AutoGen AgentChat teams), the conversation transcript becomes the main contract between agents. Agents coordinate by reading prior messages and producing new ones, which feels natural for delegation and review loops. Typing is weaker: the transcript can grow quickly, and downstream agents only know what the conversation history includes.
In a supervisor model, a central router or manager decides which specialist agent should act next and tracks overall progress. Use this when you want explicit control over delegation and completion criteria, but expect a central bottleneck and another policy layer to debug.
| Approach | Framework | Pros | Cons |
|---|---|---|---|
| Shared State | LangGraph | Typed contracts, easy inspection of intermediates, clean checkpointing. | Schema discipline required, state can bloat over time. |
| Message-Oriented Teams | AutoGen | Natural delegation loops, flexible roles, transcript is easy to inspect. | Weaker typing, prompt history grows, context only exists if the transcript carries it. |
| Supervisor Routing | CrewAI / custom orchestrators | Clear delegation control, centralized progress tracking, easy policy insertion. | Coordinator can become a latency bottleneck and single point of failure. |
Shared state and message passing solve coordination inside one runtime. Production systems often need something else: interoperability between independent systems. Two standards matter here, but they solve different problems.
MCP (Model Context Protocol)[13], introduced by Anthropic in late 2024, standardizes how an assistant connects to tools, resources, and prompts exposed by an MCP server. It's best thought of as a tool and context protocol, not a full remote-agent delegation protocol.
A2A (Agent-to-Agent Protocol)[14][15], announced by Google in April 2025 and now developed as a broader protocol community standard, enables direct communication between autonomous agents from different providers. A2A focuses on task delegation, status updates, and result sharing between independent agent systems. This lets one vendor's planner agent delegate work to another vendor's specialist agent without flattening that specialist into a single tool call.
The two protocols solve different layers of the agent interoperability problem. MCP is a context and capability protocol: it lets one host discover and invoke tools, read resources, and reuse prompts from compliant local or remote servers. A2A is a remote delegation protocol: it lets one autonomous agent hand off a complete sub-task to another independent agent (possibly from a different company), poll for progress, and receive structured results or intermediate state.
Without this distinction, teams often end up wrapping every remote capability as an ad-hoc tool call and then wonder why long-running workflows, approvals, and status updates become awkward to implement.
| Concern | MCP (tool/context protocol) | A2A (agent delegation protocol) |
|---|---|---|
| Tool or data access | Strong fit (native) | Not the primary goal |
| Remote specialist agent | Usually wrapped as a tool call | Strong fit (first-class task handoff) |
| Long-running task status | Tool-specific custom polling logic | Built into the protocol (task objects + updates) |
| Cross-vendor interoperability | Tool and data format interoperability | Full agent-to-agent task delegation |
| Human-in-the-loop integration | Handled by the calling host | Can be modeled as explicit review tasks |
In practice, many production systems use both layers together: MCP to give each agent detailed local tools and context, and A2A when one agent's planner needs to delegate work to a separate specialist agent running in another environment or organization. The references at the end link to the current protocol specs and community discussions.
Microsoft's AutoGen[1] popularized a conversation-based multi-agent pattern where agents communicate through messages. That design space overlaps with work like ChatDev[16], where agents role-play as software developers in a collaborative environment. In current AgentChat releases, chat-style teams such as SelectorGroupChat, RoundRobinGroupChat, and Swarm coexist with directed workflows through GraphFlow.[17]
For these team abstractions, the main contract is still the shared conversation history plus team policies such as speaker selection and termination conditions. The snippet below uses SelectorGroupChat, which is the clearest example of message-oriented coordination.
This snippet requires autogen-agentchat, autogen-ext[openai], and OpenAI credentials. It uses GPT-5.6 Luna as a cost-sensitive selector model. AutoGen accepts explicit model_info when its installed release doesn't yet recognize a new OpenAI model ID, which keeps capability checks separate from framework release cadence.[18][19] It constructs a team; running it calls the configured model.
1from autogen_agentchat.agents import AssistantAgent
2from autogen_agentchat.conditions import MaxMessageTermination, TextMentionTermination
3from autogen_agentchat.teams import SelectorGroupChat
4from autogen_ext.models.openai import OpenAIChatCompletionClient
5
6model_client = OpenAIChatCompletionClient(
7 model="gpt-5.6-luna",
8 model_info={
9 "vision": True,
10 "function_calling": True,
11 "json_output": True,
12 "family": "gpt-5",
13 "structured_output": True,
14 },
15)
16
17# Define specialized agents
18researcher = AssistantAgent(
19 "Researcher",
20 description="Finds relevant facts and sources",
21 model_client=model_client,
22 system_message="You research topics using web search.",
23)
24
25analyst = AssistantAgent(
26 "Analyst",
27 description="Synthesizes research into a concise answer",
28 model_client=model_client,
29 system_message="You analyze research findings and end with TERMINATE when done.",
30)
31
32termination = TextMentionTermination("TERMINATE") | MaxMessageTermination(max_messages=12)
33
34# SelectorGroupChat uses the model to pick the next speaker
35team = SelectorGroupChat(
36 [researcher, analyst],
37 model_client=model_client,
38 termination_condition=termination,
39)If you need explicit graph routing inside AutoGen instead of model-selected turn-taking, use GraphFlow rather than a selector-based team.
Choosing the right orchestration framework depends heavily on the use case. While all three frameworks support multi-agent patterns, they prioritize different approaches: LangGraph enforces strict graph-based execution, AutoGen spans conversation-centric teams and GraphFlow workflows, and MetaGPT structures agents around rigid software engineering roles.
| Feature | LangGraph | AutoGen | MetaGPT |
|---|---|---|---|
| Control Flow | Explicit Graph / State Machine | Conversation Teams or directed GraphFlow | SOPs (Standard Operating Procedures) |
| State | Typed Shared State | Shared Conversation History, optionally constrained by flow edges | Shared Environment / Artifacts |
| Best For | Structured, engineering-heavy workflows | Conversation teams, handoffs, or graph workflows inside AgentChat | Software dev, strictly defined roles |
| Complexity | High (Write code for every node) | Medium to High (team presets plus GraphFlow) | High (Class-based definitions) |
Choose LangGraph when you need strict control over routing, retries, checkpoints, and approvals. AutoGen fits conversation-centric team abstractions, especially when GraphFlow gives you enough structure without dropping to a lower-level state machine. MetaGPT fits rigid software roles and SOP-driven artifact generation.
Multi-agent orchestration is useful when it separates independent reads and keeps external writes behind one controlled execution boundary. These are three plausible production designs, not claims about deployed performance.
A coordinator routes independent eval, canary-health, and rollback-readiness checks in parallel, then merges their source-backed findings. A permitted deployment service, not a free-form specialist, shifts traffic only after evidence and policy checks satisfy release rules. The graph can draft a release note or send an exception to review without giving every worker write authority.
When an alert fires, independent branches can retrieve recent deploys, error-budget burn, and runbook guidance scoped to the service and tenant. A reducer validates sources and produces a proposed action. If policy requires approval, the reviewer sees the exact action, evidence references, and action digest; an incident service publishes the update only after revalidation with an idempotency key.
When a nightly eval reports a regression, a triage node classifies the failure. Metric analysis can produce a rollback proposal, while a communication branch drafts the stakeholder update. Rolling back or disabling a route remains an authorized action bound to verified deployment state. After a bounded number of failed proposals, the graph escalates with evidence instead of continuing indefinitely.
Check that you can defend these design choices in a system review or interview:
| Skill | What good looks like |
|---|---|
| When to go multi-agent | Argue, with the Cognition and Anthropic positions, why multi-agent fits separable parallel work but hurts tightly coupled tasks given the much larger token budget. |
| Chain limits | Explain why linear agent chains fail for complex workflows with conditional routing, parallel work, or cycles. |
| Graph shape | Distinguish DAG-style workflows from cyclic state-machine orchestration. |
| LangGraph implementation | Build a StateGraph with typed state, nodes, reducers, fixed edges, and conditional edges. |
| Team topology | Compare supervisor, hierarchical, swarm, and map-reduce patterns by control, latency, and debuggability. |
| Coordination contract | Explain typed shared state versus message-oriented coordination. |
| Protocol boundary | Distinguish MCP tool/context access from A2A remote-agent delegation. |
| Production control | Add checkpointing, termination guards, evidence validation, and approvals bound to exact risky actions. |
How do you handle infinite loops in a cyclic agent graph? Once you allow retries or review loops, you're no longer dealing with a strict DAG. LangGraph can model cyclic graphs and state machines, but production systems need hard termination through recursion_limit, step_count, max_turns, max retries, deadlines, or a forced transition to failure handling or human review.
When would you choose a supervisor pattern over a flat swarm pattern? Choose a supervisor when you need centralized control, strict process adherence, policy insertion, or arbitration between workers with overlapping capabilities. A swarm or peer-to-peer handoff pattern fits loosely coupled specialists, but it's harder to debug because routing decisions are decentralized.
Why is typed shared state often easier than message passing for structured workflows? It isn't universally better, but it gives downstream nodes reliable fields such as deployment_data, approval_status, policy_result, and error_count without asking them to infer state from a growing transcript. Message-oriented teams can work well, but they need stronger serialization, summaries, and termination policies as history grows.
How does human-in-the-loop approval fit into graph architecture? Model it as an explicit approval node. In LangGraph, that node can call interrupt() to emit a structured review payload and pause execution through a checkpointer. The payload should contain the proposed action digest, evidence, source version, and idempotency key. After a human decision resumes the workflow with Command(resume=...), the execution node still revalidates authoritative state before writing.
| Symptom | Cause | Fix |
|---|---|---|
| Multi-agent prototype is slower and worse than one good prompt. | Graph was split into too many tiny agents with no real parallelism or separation of concerns. | Collapse adjacent steps, keep one agent for tightly coupled reasoning, and add workers only where roles or tools truly differ. |
| Routing burns tokens before useful work starts. | Every handoff or branch decision uses a large model, even for rules you can write down. | Use deterministic routing for explicit rules, and reserve model-based routing for real language ambiguity. |
| Retry loop never exits cleanly. | Cyclic graph has no hard guard such as recursion_limit, retry budget, or deadline. | Add termination counters in state and force failure handling or human review when the limit is hit. |
| State grows until every node becomes expensive. | Large records, raw tool payloads, or full transcripts are copied into shared state. | Store big objects by ID, summarize long histories, and keep only fields downstream nodes read. |
| Downstream worker breaks after an upstream wording change. | Agents are handing off text instead of typed state fields or schema-constrained outputs. | Put structured contracts between workers: typed state keys, validated JSON, or reducer-backed message formats. |
| Approved promotion changes before execution. | Approval was recorded as a Boolean instead of being bound to an action digest and source version. | Revalidate the exact proposal at execution time and issue the write through an idempotency key. |
| Remote integrations are awkward and brittle. | MCP tool access and A2A delegation were treated as interchangeable. | Use MCP for local tool and context access, and A2A when one independent agent system must delegate to another. |
If you allow cyclic graphs (for example, "if the release lookup fails, try again with a different query"), you must enforce termination. Common guards include:
recursion_limit in the graph invoke or stream config.step_count or error_count field in state that increments each retry.max_turns guard in conversation loops that forces a transition to failure handling or human review.Without one of these, two agents can pass a task back and forth forever.
1def lookup_with_budget(statuses: list[str], max_attempts: int) -> str:
2 for attempt, status in enumerate(statuses[:max_attempts], start=1):
3 print(f"attempt {attempt}: {status}")
4 if status == "verified":
5 return "continue to formatter"
6 return "escalate for review"
7
8outcome = lookup_with_budget(["not_found", "timeout", "not_found"], max_attempts=2)
9print("outcome:", outcome)1attempt 1: not_found
2attempt 2: timeout
3outcome: escalate for reviewA frequent failure mode is passing the entire message history to every agent. If the conversation has twenty turns, every node sees all twenty even when it only needs the latest release ID. Fixes include:
messages list for the transcript but adding dedicated state keys (like deployment_data) for data that downstream nodes need directly.The state can expose a minimal verified projection while keeping sensitive record content behind the application authorization boundary:
1release_store = {
2 "releases/tenant-7/reranker-v17": {
3 "owner_email": "[email protected]",
4 "status": "canary_clean",
5 "release_id": "rel-7421",
6 }
7}
8
9def project_for_graph(release_ref: str, tenant_id: str) -> dict[str, str]:
10 if not release_ref.startswith(f"releases/{tenant_id}/"):
11 raise PermissionError("cross-tenant reference")
12 record = release_store[release_ref]
13 return {"release_ref": release_ref, "verified_status": record["status"]}
14
15state_update = project_for_graph("releases/tenant-7/reranker-v17", "tenant-7")
16print("state update:", state_update)
17print("contains owner email:", "owner_email" in state_update)1state update: {'release_ref': 'releases/tenant-7/reranker-v17', 'verified_status': 'canary_clean'}
2contains owner email: FalseUsing an LLM to decide the next step when a simple if/else or regex would suffice increases cost and reduces reliability. Use deterministic routing for rules you can write down, and use LLM-based routing only when the classification genuinely requires language understanding.
To check your understanding, try this small design exercise:
Task: Sketch a LangGraph StateGraph with at least three nodes and one conditional edge. Define the TypedDict state, name the nodes, and write the routing logic. Then answer these questions:
interrupt() if the traffic shift exceeds 10%?release_ref, minimal verified deployment fields, policy_result, operator_request, proposed_action, approval_status, and final_answer.State schema defines the contract between agents. Keep it strict, typed, tenant-scoped, and limited to necessary evidence or references.The Advanced Agents and Retrieval section ends with a concrete orchestration stack: specialized agents wired into explicit graphs, workflow-shaped topology, typed shared state, durable progress, bounded approvals for risky actions, and a clear admission check for when not to build a multi-agent system at all.
Answer every question, then check your score. Score above 75% to mark this lesson complete.
10 questions remaining.
AutoGen: Enabling Next-Gen LLM Applications via Multi-Agent Conversation.
Wu, Q., et al. · 2023
MetaGPT: Meta Programming for A Multi-Agent Collaborative Framework.
Hong, S., et al. · 2023
Introducing Devin, the first AI software engineer
Cognition Labs · 2024
Don't Build Multi-Agents
Yan, W. (Cognition) · 2025
Multi-Agents: What's Actually Working
Yan, W. (Cognition) · 2026
How we built our multi-agent research system
Hadfield, J., Zhang, B., Lien, K., et al. (Anthropic) · 2025
LangGraph Graph API
LangChain · 2026
Swarm: Educational Framework for Multi-Agent Orchestration.
OpenAI · 2024
OpenAI Agents SDK
OpenAI · 2025
OpenAI Agents SDK Handoffs
OpenAI · 2026
LangGraph Persistence
LangChain · 2026
LangGraph Interrupts
LangChain · 2024
Introducing the Model Context Protocol
Anthropic · 2024
A2A Protocol Specification
A2A Project · 2025
A2A Protocol Ships v1.0: Production-Ready Standard for Agent-to-Agent Communication
A2A Protocol Community · 2026
Communicative Agents for Software Development.
Qian, C., et al. · 2023
GraphFlow (Workflows)
Microsoft AutoGen · 2026
GPT-5.6 Luna Model
OpenAI · 2026
Model Clients
Microsoft AutoGen · 2026
Questions and insights from fellow learners.