Architecting Multi-Agent Teams with LangGraph: Supervisor Patterns, Worker Isolation, and Sub-Graph Composition

Two patterns for multi-agent coordination: a Python supervisor that routes between specialist worker nodes via explicit state evaluation, and a typed graph engine that composes child sub-graphs with isolated state boundaries. When each earns its complexity, and the failure modes that make both patterns break.

HimanshuJuly 22, 2025Updated September 08, 2026

Every hop in a multi-agent graph is a network round trip, a fresh set of input tokens, and one more place for state to go stale.

A graph earns that cost only when the problem needs iterative self-correction, role-isolated tool permissions, or state you can inspect and resume at each step. Otherwise it is one LLM call with extra hops.

Two patterns cover most real cases. The first is a Python supervisor that routes between specialist workers using state evaluation at temperature=0.0. The second is a typed graph engine that composes child sub-graphs with isolated execution boundaries. Each has specific failure modes the other avoids.

Pattern 1: Star topology with a central supervisor

Step 1: Define the shared state schema

The state schema is the single source of truth for routing decisions. Every field must be explicitly typed. The supervisor reads structured dictionary keys, not natural language.

python

from typing import TypedDict, Optional
from langgraph.graph import StateGraph, END
 
class AgentState(TypedDict):
    query: str
    research_results: list[str]
    draft: Optional[str]
    review_comments: list[str]
    iterations: int
    max_iterations: int
    next_action: Optional[str]
    messages: list[dict]

Step 2: Implement the supervisor and worker nodes

The supervisor runs at temperature=0.0 and returns exactly one token from a closed vocabulary. If the model returns anything else, routing fails with a missing key exception.

python

def call_llm(prompt: str, model: str = "gpt-4o", temperature: float = 0.7) -> str:
    """Direct LLM call. Replace with your actual client."""
    raise NotImplementedError
 
class Agents:
    def __init__(self, llm_model: str = "gpt-4o"):
        self.llm_model = llm_model
 
    def supervisor_agent(self, state: AgentState) -> dict:
        """Routes to the correct worker based on current state content."""
        supervisor_prompt = f"""
        Analyze the current pipeline state and decide the next action.
        
        State summary:
        - Research results: {len(state.get('research_results', []))} items collected
        - Draft exists: {bool(state.get('draft'))}
        - Review comments: {len(state.get('review_comments', []))} items
        - Iterations completed: {state.get('iterations', 0)} / {state.get('max_iterations', 2)}
        
        Rules:
        1. If research is empty or insufficient, return 'research'.
        2. If research exists and no draft is written, return 'draft'.
        3. If a draft exists without review feedback, return 'review'.
        4. If review comments exist and iterations < max_iterations, return 'revise'.
        5. If draft is reviewed and validated, or max_iterations is reached, return 'finish'.
        
        Output EXACTLY one word: research, draft, review, revise, finish.
        """
        action = call_llm(supervisor_prompt, model=self.llm_model, temperature=0.0).strip().lower()
        return {"next_action": action}
 
    def researcher_agent(self, state: AgentState) -> dict:
        """Gathers domain evidence for the topic."""
        research_prompt = f"""
        Find 3-5 concise technical facts about: "{state['query']}".
        Focus on concrete architecture details and code-level constraints.
        Existing findings: {state.get('research_results', [])}
        """
        new_research = call_llm(research_prompt, model=self.llm_model).strip()
        updated_results = state.get("research_results", []) + [new_research]
        return {"research_results": updated_results}
 
    def writer_agent(self, state: AgentState) -> dict:
        """Drafts from gathered research."""
        research_summary = "\n".join(state.get("research_results", []))
        writer_prompt = f"""
        Draft a 200-300 word technical summary on: "{state['query']}".
        Research context: {research_summary}
        Target: Senior software engineers.
        """
        draft = call_llm(writer_prompt, model=self.llm_model).strip()
        return {"draft": draft}
 
    def reviewer_agent(self, state: AgentState) -> dict:
        """Evaluates draft for accuracy and clarity gaps."""
        comments = call_llm(f"""
        Critically review this draft for: "{state['query']}".
        If sufficient, state "No specific changes needed."
        Otherwise, provide bulleted critique.
        Draft: {state.get('draft', '')}
        """, model=self.llm_model).strip()
        return {"review_comments": state.get("review_comments", []) + [comments]}
 
    def reviser_agent(self, state: AgentState) -> dict:
        """Applies review feedback and increments iteration count."""
        revised = call_llm(f"""
        Revise this draft to address all feedback while preserving factual precision.
        Query: "{state['query']}"
        Original: {state.get('draft', '')}
        Feedback: {"\n".join(state.get('review_comments', []))}
        """, model=self.llm_model).strip()
        return {
            "draft": revised,
            "review_comments": [],
            "iterations": state.get("iterations", 0) + 1,
        }

Step 3: Connect nodes in a star topology

All coordination flows through the supervisor. No worker routes directly to another worker.

python

team_agents = Agents()
workflow = StateGraph(AgentState)
 
workflow.add_node("supervisor", team_agents.supervisor_agent)
workflow.add_node("researcher", team_agents.researcher_agent)
workflow.add_node("writer", team_agents.writer_agent)
workflow.add_node("reviewer", team_agents.reviewer_agent)
workflow.add_node("reviser", team_agents.reviser_agent)
 
workflow.set_entry_point("supervisor")
 
workflow.add_conditional_edges(
    "supervisor",
    lambda state: state["next_action"],
    {
        "research": "researcher",
        "draft": "writer",
        "review": "reviewer",
        "revise": "reviser",
        "finish": END,
    }
)
 
for node in ["researcher", "writer", "reviewer", "reviser"]:
    workflow.add_edge(node, "supervisor")
 
app = workflow.compile()

Step 4: Stream state transitions

python

initial_state: AgentState = {
    "query": "Explain transformer attention in distributed inference.",
    "research_results": [],
    "draft": None,
    "review_comments": [],
    "iterations": 0,
    "max_iterations": 2,
    "next_action": None,
    "messages": [],
}
 
for event in app.stream(initial_state):
    for node_name, state_update in event.items():
        print(f"Node: {node_name} | Delta: {list(state_update.keys())}")

Pattern 2: Typed sub-graph engine with isolated worker state

When workers need distinct execution contexts, the star topology breaks down: workers contaminate each other's state through shared fields. The alternative is a graph registry where each worker runs as an isolated sub-graph with its own state boundary.

Step 1: Define the graph engine

typescript

// core/graphEngine.ts
import { WorkflowState } from './workflowState';
 
export type NodeFunction = (state: WorkflowState) => Promise<WorkflowState>;
export type EdgeTarget = string | null | ((state: WorkflowState) => string | null);
 
export interface WorkflowGraph {
    id: string;
    startNode: string;
    nodes: Map<string, NodeFunction>;
    edges: Map<string, EdgeTarget>;
}
 
export const graphRegistry = new Map<string, WorkflowGraph>();
 
export async function runWorkflowGraph(graphId: string, state: WorkflowState): Promise<WorkflowState> {
    const graph = graphRegistry.get(graphId);
    if (!graph) throw new Error(`Graph '${graphId}' not registered.`);
 
    let currentNode: string | null = graph.startNode;
 
    while (currentNode !== null) {
        const nodeFn = graph.nodes.get(currentNode);
        if (!nodeFn) throw new Error(`Node '${currentNode}' not found in graph '${graphId}'.`);
 
        state = await nodeFn(state);
 
        const edgeTarget = graph.edges.get(currentNode);
        currentNode = typeof edgeTarget === 'function' ? edgeTarget(state) : edgeTarget ?? null;
    }
 
    return state;
}

Step 2: Orchestrator with bounded sub-task queue

typescript

// graphs/mainOrchestratorGraph.ts
const decomposeTaskNode: NodeFunction = async (state) => {
    const prompt = `Break down this task into 2-4 concrete sub-tasks: "${state.overallTask}"
    Return JSON: {"sub_tasks": [{"id": "...", "description": "...", "requiredWorkerGraph": "..."}]}`;
    
    const response = await llmCall(prompt, state.contextData);
    const parsed = JSON.parse(response);
    state.subTaskQueue = parsed.sub_tasks;
    return state;
};
 
const expertRouterNode: NodeFunction = async (state) => {
    if (state.subTaskQueue.length > 0) {
        state.currentSubTask = state.subTaskQueue.shift()!;
        state.metadata.nextManagerAction = 'invokeWorker';
    } else {
        state.metadata.nextManagerAction = 'synthesize';
    }
    return state;
};
 
const invokeWorkerGraphNode: NodeFunction = async (state) => {
    if (!state.currentSubTask) throw new Error("No sub-task to execute.");
    
    const workerGraphId = state.currentSubTask.requiredWorkerGraph;
    if (!graphRegistry.has(workerGraphId)) {
        throw new Error(`Worker graph '${workerGraphId}' not registered.`);
    }
 
    // Construct isolated sub-state: worker cannot modify parent queue or metadata
    let workerState: WorkflowState = {
        overallTask: state.currentSubTask.description,
        currentSubTask: { ...state.currentSubTask, output: null, status: 'PENDING' },
        subTaskQueue: [],
        subTaskHistory: [],
        metadata: {},
        contextData: state.contextData,
    };
 
    workerState = await runWorkflowGraph(workerGraphId, workerState);
 
    state.currentSubTask.output = workerState.currentSubTask?.output ?? null;
    state.currentSubTask.status = workerState.currentSubTask?.status ?? 'FAILED';
    state.subTaskHistory.push({ ...state.currentSubTask });
    state.currentSubTask = null;
    return state;
};

Step 3: Worker isolation boundary

Only contextData and the input payload cross the parent-worker boundary. Workers cannot modify the parent's queue or pollute state used by sibling nodes:

Field	Orchestrator state	Worker sub-state
`overallTask`	User prompt	Sub-task instruction
`subTaskHistory`	Global execution log	Fresh empty array
`metadata`	Orchestrator queue control	Fresh empty dictionary
`contextData`	Inherited from caller	Passed down directly

What breaks both patterns

Unstructured supervisor output. If the supervisor prompt fails to enforce a closed vocabulary and the LLM returns "I think we should research more", the conditional edge lookup fails with a missing key exception. Enforce with JSON schema validation or response_format.

Shared state contamination in the star topology. When workers write output into unstructured message lists rather than explicit dictionary keys, downstream nodes must parse natural language to find parameters, and dropped context is silent.

Unbounded revision cycles. Without a programmatic ceiling on max_iterations, a reviewer that always returns "needs improvement" runs until API quota is exhausted. Check iteration bounds inside the edge function, not just inside the LLM prompt.

Registry dispatch errors. The sub-graph engine must validate that requiredWorkerGraph identifiers exist in graphRegistry before invocation. A missing graph at runtime produces an opaque error that doesn't point to the decomposition step that hallucinated the graph name.

When this is the wrong choice

The task is sequential and deterministic. Plain functions are faster, cheaper, and easier to test. A supervisor that always routes research, draft, review, finish in that order is a for loop paying LLM prices per iteration.
One call with structured output already solves it. If JSON schema enforcement on a single call gets the answer, the graph adds a state schema, a router, and four failure modes for nothing.
You are on an interactive latency budget. Every hop is a model round trip. Two hops at 300ms each already exceed most interactive SLAs, and the supervisor pattern spends a hop on routing before any work happens.
The workers share one state object anyway. If nothing needs isolated execution context, the sub-graph engine buys you a registry, a boundary table, and dispatch errors that point at the wrong step.