Supervisor pattern

In a supervisor (router) graph, a router reads the request and selects the worker agents to run. The selected workers run in parallel, and a gather node combines their results.

The example is examples/langgraph-integration/design_patterns/supervisor/toy/main.py. Its router is an LLM; its two workers are AsyncFlow tasks that sleep for 3 s each, so a run that selects both shows them running at the same time.

flowchart LR
  START((start)) --> R[llm_router]
  R -->|Send| A[agent_A]
  R -->|Send| B[agent_B]
  A --> G[gather]
  B --> G
  G --> END((end))

Run it

The router and the gather node use OpenRouter: set OPEN_ROUTER_API_KEY, then run make examples-supervisor-toy. CI, which has no credentials, runs the example with a scripted chat model in place of OpenRouter (tests/integration/test_design_patterns_offline.py); tests/integration/test_integration.py runs it live when FLOWGENTIC_RUN_LLM_INTEGRATION=1 is set.

The state

The workers write into results, whose reducer merges the dictionaries that parallel workers return:

class GraphState(BaseModel):
    query: str = Field(..., description="User query to route")
    routing_decision: Optional[List[str]] = Field(
        default=None, description="List of agents to route to"
    )
    routing_rationale: Optional[str] = Field(
        default=None, description="Decision to move to a set of nodes"
    )
    results: Annotated[Dict[str, str], operator.or_] = Field(
        default_factory=dict, description="Merge dicts from parallel agents"
    )
    final_summary: Optional[str] = Field(
        default=None, description="Combined summary from gather node"
    )
    messages: Annotated[List[BaseMessage], add_messages] = []

The router

create_llm_router(responsibilities, model) (flowgentic.langGraph.utils.supervisor) returns a node that asks the model which agents should handle state.query, with structured output, and stores its choice in routing_decision and routing_rationale. The example runs it as an EXECUTION_BLOCK:

        # Define the routing prompt template
        routing_prompt_agents_responsibility = """

- agent_A: Handles data processing and analysis tasks
- agent_B: Handles question answering and information retrieval
- both: When the query requires both processing AND answering

"""

        # Define the model for routing
        router_model = ChatLLMProvider(
            provider="OpenRouter", model="google/gemini-2.5-pro"
        )

        # Create and decorate the router function using the factory
        llm_router = agents_manager.execution_wrappers.asyncflow(
            create_llm_router(routing_prompt_agents_responsibility, router_model),
            flow_type=AsyncFlowType.EXECUTION_BLOCK,
        )

The workers

@agents_manager.execution_wrappers.asyncflow(
    flow_type=AsyncFlowType.FUNCTION_TASK
)
async def agent_a(state: GraphState) -> dict:
    start = time.perf_counter()
    logging.info(f"📊 agent_a START - Processing data for: '{state.query}'")
    await asyncio.sleep(3.0)  # simulate work
    elapsed = (time.perf_counter() - start) * 1000
    logging.info(f"📊 agent_a END   took_ms={elapsed:.1f}")
    return {"results": {"agent_A": f"Data analysis complete: {state.query}"}}

@agents_manager.execution_wrappers.asyncflow(
    flow_type=AsyncFlowType.FUNCTION_TASK
)
async def agent_b(state: GraphState) -> dict:
    start = time.perf_counter()
    logging.info(f"💬 agent_b START - Answering: '{state.query}'")
    await asyncio.sleep(3.0)  # simulate work
    elapsed = (time.perf_counter() - start) * 1000
    logging.info(f"💬 agent_b END   took_ms={elapsed:.1f}")
    return {
        "results": {
            "agent_B": f"Answer: {state.query} - Parallelism is executing multiple tasks simultaneously!"
        }
    }

The graph

supervisor_fan_out returns one Send per agent in routing_decision, so LangGraph runs the selected workers in parallel. The path_map lists the agents the router may choose:

graph.add_node("llm_router", llm_router)
graph.add_node("agent_A", agent_a)
graph.add_node("agent_B", agent_b)
graph.add_node("gather", gather)

# Edges
graph.add_edge(START, "llm_router")
graph.add_conditional_edges(
    "llm_router", supervisor_fan_out, path_map=["agent_A", "agent_B"]
)
graph.add_edge("agent_A", "gather")
graph.add_edge("agent_B", "gather")
graph.add_edge("gather", END)

app = graph.compile()

When both workers ran, the gather node asks a second model, built with langchain.agents.create_agent, to combine their outputs; with one worker, it passes the result through.

For an agent that hands work to another agent through a tool call, see Hand off to another agent.