⚠️ Before You Start

Risk: Medium

Duration: 1-2 days

Last Validated: Not yet validated (draft)

Kind: How-to

Risk Note: Unbounded routing loops cause runaway LLM API costs and request timeouts

Build a supervisor with workers and hop limits in LangGraph

Draft outline: not validated

Drafted with AI from official sources on 16 Oct 2026. Verified and expanded under AGT-06-T05 on 23 Oct 2026. Do not use on a live system until it is written, dry-run by someone other than the author and signed off (planned 2026-12-18). Author: Sanjay Sarvesh · dry-run: Yasar · sign-off: James. Work items: AGT-06-T05, AGT-06-T06, AGT-06-T07, AGT-06-T08.

1. Summary

This how-to guides the implementation of a hierarchical multi-agent supervisor topology using LangGraph StateGraph and the Command control primitive. It configures a central supervisor node that dynamically delegates sub-tasks to specialized worker agents, trims conversational context between hops, and enforces a strict hop limit to prevent unbounded routing loops. The outcome is a resilient, cost-controlled multi-agent system with traceable delegation spans and graceful fallback on limit exhaustion.

2. When to use and when not to

Use this how-to when:

• Complex workflows require distinct specialized capabilities (e.g. web research, data transformation, code generation) where a single agent with dozens of tools degrades selection accuracy.
• Execution requires dynamic sequential or conditional delegation where the next worker depends on prior worker outputs.
• The multi-agent topology decision guide (PB-AGT-06-01) selects the Supervisor pattern over peer-to-peer handoffs or linear pipelines.

Do not use it when:

• Tasks follow a static, deterministic sequence of steps: use a linear pipeline graph (START -> step_a -> step_b -> END) without supervisor routing overhead.
• Workers execute independent sub-tasks concurrently over partitioned inputs: use the LangGraph Send API (Map-Reduce pattern) instead.
• Agents collaborate through open group discussion or peer-to-peer delegation without a central coordinator: use PB-AGT-06-03 (Build group chat, handoff and Magentic orchestrations) or peer handoffs.
• The task is a single-turn reasoning job with a handful of tools: use a standard ReAct task agent (PB-AGT-01-02).

3. What you must know first

1. Dynamic routing with Command. A node function returns Command(goto=target_node, update={...}) to perform state updates and transition control in a single atomic step, eliminating separate conditional edge functions. Annotating the return type as Command[Literal["worker_a", "worker_b", "__end__"]] allows LangGraph to infer graph edges and render Mermaid diagrams accurately.
2. Subgraphs and Command.PARENT. When workers are structured as independent subgraphs, routing back to the supervisor or parent graph requires specifying Command(graph=Command.PARENT, goto="supervisor", update={...}). Any shared state keys updated across graph boundaries require an explicit reducer (e.g. operator.add).
3. Hop counter (soft stop) vs engine recursion limit (hard stop). A state-managed hop counter (state["hops"] += 1) provides a controlled soft stop: when state["hops"] >= MAX_HOPS, the supervisor terminates delegation, synthesizes a partial answer from available notes, and routes to END. The LangGraph engine's recursion_limit (configured via {"recursion_limit": N}) serves as a hard safety net that halts execution and raises GraphRecursionError if an unexpected cycle occurs.
4. Tool request-response integrity. In message-driven multi-agent graphs, every AIMessage containing tool_calls requires a corresponding ToolMessage with the matching tool_call_id. Updating messages without providing the matching ToolMessage corrupts conversation state and causes model API errors.
5. Context engineering between hops. Passing full conversation histories to every worker bloats token usage, increases inference latency, and causes prompt distraction. The supervisor must isolate worker context by passing only the necessary sub-task prompt and aggregating structured outputs.

4. Prerequisites and access

Item	Detail
Model Endpoint	Azure OpenAI or OpenAI model deployment supporting structured outputs (with_structured_output(RoutingDecision)) or tool calling (e.g. gpt-4o, gpt-4o-mini).
Development Tools	Python 3.10+ with langgraph at the template lock-file version (no lower than 0.2.46, the first release with Command), langchain-core, langchain-openai and pydantic>=2.0.
Template Repository	Supervisor agent template cloned.
Observability	LangSmith API key with LANGSMITH_TRACING=true to capture hierarchical parent-child agent spans.
Worker Definitions	Pydantic input/output schemas and tools for each specialized worker agent (e.g. researcher, coder).
Configuration	Maximum hop limit (MAX_HOPS, value agreed with the agent owner) and an explicit recursion_limit in the run config, set above 2 × MAX_HOPS + 1 (the default varies by LangGraph version; never rely on it).

Set up the environment variables:

export OPENAI_API_KEY="${OPENAI_API_KEY}"
export LANGSMITH_TRACING="true"
export LANGSMITH_API_KEY="${LANGSMITH_API_KEY}"
export LANGSMITH_PROJECT="multi-agent-supervisor"

5. Steps

Step 1: Define team state with task, shared notes, and hop counter

Define a typed dictionary representing the graph state. Include the root task prompt, an additive list of worker outputs using operator.add, a tracking field for completed steps, and an integer hop counter.

import operator
from typing import Annotated, TypedDict

class TeamState(TypedDict):
    task: str
    notes: Annotated[list[str], operator.add]
    completed_workers: Annotated[list[str], operator.add]
    final_response: str
    hops: int

Check: Instantiating TeamState with initial values succeeds without type errors.

Step 2: Define structured output routing schema with Pydantic

Create a Pydantic model for the supervisor's decision. Restrict next_worker to allowed worker names or FINISH using Literal. Include a reasoning field and instructions for the selected worker.

from typing import Literal
from pydantic import BaseModel, Field

class RouterDecision(BaseModel):
    reasoning: str = Field(description="Explanation of why this worker is selected or why the task is finished")
    next_worker: Literal["researcher", "coder", "FINISH"] = Field(
        description="The next worker to invoke, or FINISH if the task is complete"
    )
    worker_instruction: str = Field(
        description="Specific instructions and scoped context for the selected worker"
    )

Check: Validating a sample dictionary against RouterDecision parses expected worker literals and rejects unknown agent names.

Step 3: Implement the supervisor node with hop limit enforcement and Command routing

Implement the supervisor function. Check if state["hops"] >= MAX_HOPS. If reached, halt delegation, synthesize a response indicating limit exhaustion, and route to END. Otherwise, invoke the LLM with structured output, increment hops, and return a Command targeting the chosen worker.

from langchain_openai import ChatOpenAI
from langgraph.graph import END
from langgraph.types import Command

MAX_HOPS = 6

def supervisor_node(state: TeamState) -> Command[Literal["researcher", "coder", "__end__"]]:
    if state.get("hops", 0) >= MAX_HOPS:
        return Command(
            goto=END,
            update={
                "final_response": "Task concluded: maximum delegation hop limit reached.",
                "notes": ["Supervisor halted delegation at hop limit."]
            }
        )

    llm = ChatOpenAI(model="gpt-4o", temperature=0).with_structured_output(RouterDecision)
    prompt = (
        f"You are a supervisor managing a team (researcher, coder).\n"
        f"Task: {state['task']}\n"
        f"Completed steps: {state.get('completed_workers', [])}\n"
        f"Notes so far: {state.get('notes', [])}\n"
        f"Select the next worker or FINISH if the goal is met."
    )
    decision = llm.invoke(prompt)

    if decision.next_worker == "FINISH":
        return Command(
            goto=END,
            update={"final_response": decision.reasoning}
        )

    return Command(
        goto=decision.next_worker,
        update={
            "hops": state.get("hops", 0) + 1,
            "completed_workers": [decision.next_worker]
        }
    )

Check: Calling supervisor_node({"task": "test", "hops": 6, ...}) returns Command(goto=END) without making an LLM API call.

Step 4: Implement worker nodes with context trimming and return reporting

Implement specialized worker nodes. Ensure workers receive only the relevant task subset, perform their specific domain logic, and return state updates to the shared notes channel. Workers must never route to other workers directly.

def researcher_node(state: TeamState) -> dict:
    # Context trimming: extract only the task objective
    task = state["task"]
    # Execute worker logic (e.g. search tools or domain LLM)
    result = f"Research findings for: {task}"
    return {"notes": [result]}

def coder_node(state: TeamState) -> dict:
    notes = state.get("notes", [])
    # Worker operates on summarized research findings rather than raw turn history
    result = f"Code generated based on {len(notes)} research notes"
    return {"notes": [result]}

Check: Invoking researcher_node returns a dictionary containing a list update for notes.

Step 5: Assemble the graph with Command transitions and workers-to-supervisor edges

Build the graph using StateGraph. Add the supervisor and worker nodes. Add an edge from START to supervisor, and edges from every worker back to supervisor.

from langgraph.graph import START, StateGraph

builder = StateGraph(TeamState)
builder.add_node("supervisor", supervisor_node)
builder.add_node("researcher", researcher_node)
builder.add_node("coder", coder_node)

builder.add_edge(START, "supervisor")
builder.add_edge("researcher", "supervisor")
builder.add_edge("coder", "supervisor")

graph = builder.compile()

Check: Calling graph.get_graph().nodes contains "supervisor", "researcher", and "coder".

Step 6: Configure compilation with checkpointer and engine recursion limit

Compile the graph with a checkpointer for conversational durability and pass a top-level recursion_limit in the invocation config as a hard safety net. Ensure recursion_limit is set higher than 2 * MAX_HOPS + 1 to accommodate round-trip super-steps.

from langgraph.checkpoint.memory import InMemorySaver

checkpointer = InMemorySaver()
app = builder.compile(checkpointer=checkpointer)

config = {
    "configurable": {"thread_id": "team-run-001"},
    "recursion_limit": 25
}

Check: The config dictionary contains recursion_limit at the top level and thread_id inside configurable.

Step 7: Test normal routing, hop limit exhaustion, and worker failure handling

Execute test invocations for:

1. Normal completion: Supervisor delegates to researcher, then coder, then selects FINISH.
2. Hop limit containment: Simulate recursive loop to verify soft stop triggers at MAX_HOPS before hitting GraphRecursionError.
3. Worker failure recovery: Add retry policies (RetryPolicy) on worker nodes for transient errors.

Check: All test scenarios complete with predictable state outputs and expected terminal states.

Snippets to add

• snippets/templates/supervisor_agent/graph.py: Complete supervisor StateGraph implementation with Command routing (python)
• snippets/templates/supervisor_agent/routing.py: Pydantic routing schemas and validation helpers (python)
• snippets/templates/supervisor_agent/test_supervisor.py: Unit and containment integration tests (python)

6. Verify it works

Execute the test suite to verify supervisor behavior, routing decisions, and limit enforcement:

python -m pytest snippets/templates/supervisor_agent/test_supervisor.py -v

Expected verification checks:

1. Happy Path Routing:
2. Input: "Write a python script to calculate Fibonacci sequence based on research."
3. Execution path: START -> supervisor -> researcher -> supervisor -> coder -> supervisor -> END.
4. Expected Output: final_response is populated, completed_workers equals ["researcher", "coder"], and hops < MAX_HOPS.
5. Hop Limit Soft Containment:
6. Input with forced cyclic mock supervisor:
7. Verify that execution stops cleanly when hops == MAX_HOPS.
8. Output contains "maximum delegation hop limit reached" in final_response.
9. Verify no GraphRecursionError is raised.
10. Hard Recursion Safety:
11. Invoke with config={"recursion_limit": 5} and verify LangGraph enforces GraphRecursionError if soft stop fails.
12. LangSmith Trace Verification:
13. Inspect the trace in LangSmith. Confirm that the parent run contains separate child spans for "supervisor", "researcher", and "coder", with token counts recorded per node.

7. Known issues and fixes

Symptom	Likely cause	Fix
Supervisor repeatedly calls the same worker in an infinite loop	Worker outputs lack clear progress signals, or supervisor prompt does not track completed steps	Record completed_workers in state. Add explicit prompt rules forbidding re-delegation to the same worker without new requirements.
Malformed message history error (OpenAI BadRequestError, HTTP 400)	An AIMessage with tool_calls was emitted without a corresponding ToolMessage with matching tool_call_id	Ensure workers return a ToolMessage(tool_call_id=..., content=...) for every tool invocation before returning control.
Worker hallucinates or confuses past agent context	Entire accumulated message history is passed to every worker	Implement context trimming: pass only the current subtask prompt and concise extracted notes to worker inputs.
Unhandled GraphRecursionError raised by LangGraph	Engine recursion_limit was reached before the soft hop limit	Ensure recursion_limit is set substantially higher than round-trip steps: recursion_limit > (2 * MAX_HOPS + 1).
Pydantic ValidationError on supervisor output	LLM structured output returned invalid JSON or unlisted agent literal	Use with_structured_output(RouterDecision) with strict JSON mode, and catch parse errors with a fallback routing edge.

8. Monitoring and alerts

Signal	Alert when	Action
Runs ending by hop limit	More than 5 percent of runs (catalog; window TO VERIFY)	Inspect supervisor prompt instructions and worker outputs for routing ambiguity or failure to converge.
Engine GraphRecursionError count	Greater than 0 occurrences	Critical incident: investigate infinite routing loops and verify MAX_HOPS soft stop is executing.
Average hops per completed task	Above the 7-day baseline (threshold TO VERIFY)	Review worker prompts and tool accuracy to ensure sub-tasks are resolved efficiently.
Token consumption per multi-agent request	Exceeds the agent's token budget per run (value TO VERIFY with the agent owner)	Audit worker context sizes; enforce stricter context trimming and intermediate summarization.

9. AWS and open-source equivalents

AWS Equivalents

• Amazon Bedrock AgentCore Runtime: The same LangGraph graph runs on AgentCore Runtime.
• Amazon Bedrock Agents Classic (maintenance mode): Managed supervisor/collaborator multi-agent collaboration. Note: Bedrock Agents is now branded Bedrock Agents Classic and is in maintenance mode.

Open-Source Equivalents

• Microsoft Semantic Kernel: GroupChatOrchestration with a custom group chat manager (experimental; replaces the unmaintained AgentGroupChat).
• OpenAI Agents SDK: Similar routing with handoffs and input filters (handoffs=[agent_a, agent_b], input_filter=...).

10. Validation record

Pending. Planned dry-run scenario: A second engineer builds the supervisor graph from the code snippets in a sandbox environment and executes the three verification test cases (happy path, hop limit soft stop, and worker error recovery). The validation passes when all three tests complete with expected outputs, zero unhandled GraphRecursionError exceptions occur, and LangSmith traces confirm clean hierarchical spans for supervisor and worker executions.

11. Sources

• LangGraph: Use the graph API (RetryPolicy, Send, Command)
• LangChain: Multi-agent handoffs
• LangChain: Multi-agent
• LangGraph: GraphRecursionError and recursion_limit
• Amazon Bedrock Agents: Multi-agent collaboration
• Microsoft Semantic Kernel: GroupChatOrchestration migration guide

📄 Snippet CODE

📋 Copy ↗ Raw ×