How to Implement a Supervisor-Worker Multi-Agent Pattern

Quick answer

The short version: define shared graph state, have a structured-output planner create focused sections, and use LangGraph’s Send API to pass one section to each worker. Let each worker return its result under completed_sections, and configure a reducer so parallel results accumulate rather than overwrite one another.

Then have a synthesizer join the collected sections into final_report. Connect the graph from START to the orchestrator, route its conditional edge to the worker node, send worker outputs to the synthesizer, and finish at END.

Compile the graph, invoke it with a topic, and inspect the final report.

The documented Python setup installs langchain_core, langchain-anthropic, and langgraph, sets ANTHROPIC_API_KEY, and initializes ChatAnthropic. This is a provider-specific example, so check that your selected model supports the planner’s structured output and tool-calling requirements.

The demonstrated synthesis step concatenates worker content; if your workflow needs accuracy checks or conflict resolution, those are not provided by this example.

Step-by-step guide

Set up the documented Python environment

Install the documented packages in your Python environment:

pip install langchain_core langchain-anthropic langgraph

Set ANTHROPIC_API_KEY in your environment to your Anthropic API key, then initialize the chat model:

from langchain_anthropic import ChatAnthropic

model = ChatAnthropic(model="claude-sonnet-4-6")

This example uses Anthropic; the setup is provider-specific. Choose a chat model that supports both structured outputs and tool calling, and check compatibility for your provider and model before moving on to the workflow.

Sources: Workflows and agents — Creating workers in LangGraph.

Define shared workflow state and specialist roles

Keep the workflow’s inputs and outputs in one shared state. The reducer on completed_sections lets results from multiple workers accumulate instead of overwriting one another.

from typing import Annotated
import operator
from typing_extensions import TypedDict

class State(TypedDict):
    topic: str
    sections: list
    completed_sections: Annotated[list, operator.add]
    final_report: str

Have the supervisor’s structured planner return sections that can be assigned individually. In its prompt, specify the report topic, what each section should cover, and that the plan should divide the work into focused tasks.

Give workers a prompt that defines their specialist role, asks them to complete only their assigned section, and tells them to return the result as section content. Keep the report topic available in their task context so their contributions stay relevant.

Sources: Workflows and agents — Creating workers in LangGraph.

Plan work with the supervisor

Define a structured response model for the plan, then bind it to the chat model you initialized. Ask for focused sections that workers can handle separately.

from pydantic import BaseModel, Field

class Sections(BaseModel):
    sections: list[str] = Field(description="Focused report sections")

planner = model.with_structured_output(Sections)

def orchestrator(state: State):
    plan = planner.invoke(
        f"Plan focused sections for a report about: {state['topic']}"
    )
    return {"sections": plan.sections}

The planner returns a Sections object, and the orchestrator puts its section list into shared state for dispatch. Before moving on, check that the list contains the tasks you expect; if the plan is too broad, refine the planning prompt to request narrower sections.

Sources: Workflows and agents — Creating workers in LangGraph.

Dispatch section tasks to workers

Send each planned section to the same worker node with its own task input. The worker uses that section as its assignment and adds its response to completed_sections.

from langgraph.types import Send

def assign_workers(state: State):
    return [
        Send("llm_call", {"section": section})
        for section in state["sections"]
    ]

def llm_call(state):
    response = model.invoke(
        f"Write the report section for: {state['section']}"
    )
    return {"completed_sections": [response.content]}

Use assign_workers as the orchestrator’s conditional routing function. It creates one task per planned section; each task passes its section to llm_call.

Check that the worker returns a list under completed_sections, matching the reducer-backed state field from the previous step, so results can accumulate rather than replace one another.

Sources: Workflows and agents — Creating workers in LangGraph.

Collect results and synthesize the report

Use the reducer-backed completed_sections field from the shared state to gather worker outputs. Then join the collected section text into final_report:

def synthesizer(state: State):
    report = "\n\n".join(state["completed_sections"])
    return {"final_report": report}

The blank lines separate sections in the assembled report. This synthesizer concatenates the workers’ results; it does not review their accuracy or resolve conflicting content, so treat those as separate needs if your workflow requires them.

Sources: Workflows and agents — Creating workers in LangGraph.

Wire, compile, and invoke the graph

Connect the nodes in execution order, then compile the graph. This assumes the state and node functions from the earlier steps are defined.

from langgraph.graph import StateGraph, START, END

builder = StateGraph(State)
builder.add_node("orchestrator", orchestrator)
builder.add_node("llm_call", llm_call)
builder.add_node("synthesizer", synthesizer)
builder.add_edge(START, "orchestrator")
builder.add_conditional_edges("orchestrator", assign_workers, ["llm_call"])
builder.add_edge("llm_call", "synthesizer")
builder.add_edge("synthesizer", END)
graph = builder.compile()

Invoke it with a topic, then inspect final_report:

result = graph.invoke({"topic": "Create a report on LLM scaling laws"})
print(result["final_report"])

You should get a report assembled from the worker sections. If the output is missing or incomplete, check the routed node names and confirm workers return their content in completed_sections.

Sources: Workflows and agents — Creating workers in LangGraph.

Check the completed workflow

Invoke the compiled graph with a topic, then inspect both the collected worker outputs and the synthesized report:

result = graph.invoke({"topic": "Create a report on LLM scaling laws"})
print(result["completed_sections"])
print(result["final_report"])

Check that completed_sections contains worker content and final_report includes those sections. The configured flow should run from the orchestrator to the dispatched workers, then to the synthesizer; if results are missing, check the node names, worker return field, and reducer configuration.

Sources: Workflows and agents — Creating workers in LangGraph.

Diagnose planning and aggregation issues

If the supervisor fails to produce a usable plan, confirm that the selected chat model supports structured outputs and tool calling. Compatibility can vary by provider and model, so check the integration’s support for your specific choice; then inspect the planner’s result to make sure it contains the expected section list.

If worker results are missing, check that completed_sections uses the Annotated[list, operator.add] reducer and that each worker returns its content under that exact state key as a list. After invocation, inspect result["completed_sections"] and result["final_report"]: the first should contain worker outputs, and the second should contain the synthesized sections.

If outputs are present but the report is empty, check that the synthesizer reads completed_sections and writes final_report.

Sources: Workflows and agents — Creating workers in LangGraph.

Next steps

A supervisor-worker graph is a practical fit when a task can be divided into focused sections and you want a central node to coordinate the work. In this LangGraph example, the planner creates those sections, workers handle them, and a reducer-backed state field collects their outputs for synthesis.

After invoking the graph, check that completed_sections contains worker content and that final_report contains the assembled result. If planning fails, verify the model’s structured-output and tool-calling support; if outputs go missing, check the reducer and the workers’ return field.

The documented synthesizer only joins results, so add a separate review or conflict-handling step if your application needs one.