Routing, Fan-Out, ReAct, Critique and Approval: Five LangGraph Patterns

Longer prompts rarely fix an unreliable AI feature. When a system must browse, write code, run compliance checks or edit customer-facing text, one non-deterministic model call is too fragile. Structure helps: the model reasons where judgment is needed while code controls routing, loops and stopping. Below are five such patterns, each with a runnable LangGraph example in Python and the caveats to fix before production.

Why a graph fits agent workflows

Conventional programs run in a straight line. Agents need loops, conditional branches and persistent state: if generated code fails a test, the system must capture the error, go back and retry.

LangGraph models this as a directed graph:

• Nodes are Python functions doing one thing, such as a SQL query or a model call.
• Edges pick the next node, directly or via a routing function.
• State is a shared typed structure passed between nodes; each node returns only the keys it changes.

For a deeper tour of these primitives, see LangGraph in practice: state, nodes and edges.

Pattern 1: the router

A router is a classifier at the entry point. Instead of sending everything to a large, expensive model, a lightweight step forwards each request to a specialized model, sub-graph or local tool.

                  ┌───> [Specialized Coding Agent] ───> [END]
[START] ──> [Router]
                  └───> [General Knowledge Agent] ───> [END]

Use it to cut latency and cost, or to match intents to specialized tools.

A small model at temperature 0 labels the query coding or general, and add_conditional_edges maps the label to a handler node. route_decision falls back to general if the model returns anything unexpected.

from typing import TypedDict, Literal
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END

# 1. Define the shared state
class RouterState(TypedDict):
    query: str
    route: str
    response: str

# Use a fast, cost-effective model for classification
model = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# 2. Define the Nodes
def classify_query(state: RouterState):
    prompt = f"""Classify the following user query into one of two categories: 'coding' or 'general'.
    Respond with exactly one word, either 'coding' or 'general'.
    Query: {state['query']}"""

    response = model.invoke([HumanMessage(content=prompt)])
    classification = response.content.strip().lower()
    return {"route": classification}

def handle_coding(state: RouterState):
    return {"response": "Executing advanced syntax processing and code compilation logic..."}

def handle_general(state: RouterState):
    return {"response": "Processing casual conversation or general knowledge search..."}

# 3. Define Conditional Routing Logic
def route_decision(state: RouterState) -> Literal["coding", "general"]:
    return state["route"] if state["route"] in ["coding", "general"] else "general"

# 4. Construct the Graph
workflow = StateGraph(RouterState)

workflow.add_node("classifier", classify_query)
workflow.add_node("coding_agent", handle_coding)
workflow.add_node("general_agent", handle_general)

workflow.add_edge(START, "classifier")
workflow.add_conditional_edges("classifier", route_decision, {
    "coding": "coding_agent",
    "general": "general_agent"
})
workflow.add_edge("coding_agent", END)
workflow.add_edge("general_agent", END)

# Compile and Run
app = workflow.compile()
result = app.invoke({"query": "How do I implement a binary search tree in Python?"})
print(f"Route Taken: {result['route']}\nResponse: {result['response']}")

The model names were current when the example was written; substitute your provider's current ones. Structured output is more robust than parsing one word.

Pattern 2: orchestrator and workers

For tasks too broad for one prompt, an orchestrator splits the goal into independent sub-tasks, workers complete them, and a synthesizer merges the results.

                           ┌───> [Worker A: Section 1] ───┐
[START] ──> [Orchestrator] ├───> [Worker B: Section 2] ───┼───> [Synthesizer] ───> [END]
                           └───> [Worker C: Section 3] ───┘

It suits long-form content such as reports, and multi-source research.

The orchestrator requests a JSON list of two sub-topics, the workers node writes a paragraph for each, and the synthesizer combines them.

import json
from typing import TypedDict, List
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END

class OrchestratorState(TypedDict):
    topic: str
    tasks: List[str]
    worker_outputs: List[str]
    final_report: str

model = ChatOpenAI(model="gpt-4o", temperature=0.2)

def orchestrator_plan(state: OrchestratorState):
    prompt = f"Create a JSON list of exactly two sub-topics needed to write a comprehensive guide about: {state['topic']}. Return ONLY a valid JSON list of strings."
    response = model.invoke([HumanMessage(content=prompt)])
    tasks = json.loads(response.content.strip())
    return {"tasks": tasks, "worker_outputs": []}

def worker_execute(state: OrchestratorState):
    outputs = []
    for task in state["tasks"]:
        prompt = f"Write a brief, highly technical paragraph explaining: {task}"
        response = model.invoke([HumanMessage(content=prompt)])
        outputs.append(response.content)
    return {"worker_outputs": outputs}

def synthesize_report(state: OrchestratorState):
    combined_context = "\n\n".join(state["worker_outputs"])
    prompt = f"Combine the following sections into a cohesive newsletter update regarding {state['topic']}:\n\n{combined_context}"
    response = model.invoke([HumanMessage(content=prompt)])
    return {"final_report": response.content}

# Graph Construction
orchestrator_flow = StateGraph(OrchestratorState)

orchestrator_flow.add_node("orchestrator", orchestrator_plan)
orchestrator_flow.add_node("workers", worker_execute)
orchestrator_flow.add_node("synthesizer", synthesize_report)

orchestrator_flow.add_edge(START, "orchestrator")
orchestrator_flow.add_edge("orchestrator", "workers")
orchestrator_flow.add_edge("workers", "synthesizer")
orchestrator_flow.add_edge("synthesizer", END)

app = orchestrator_flow.compile()
output = app.invoke({"topic": "Quantum Computing Security Implications"})
print(output["final_report"])

Two caveats. This worker node loops through tasks sequentially, so nothing runs in parallel; LangGraph's Send API can dispatch one worker per task, collecting results via a state reducer. And models sometimes wrap JSON in Markdown fences, so validate the plan with structured output instead of trusting json.loads on raw text.

Pattern 3: ReAct, reasoning and acting in a loop

ReAct alternates reasoning with actions: the model assesses the situation, calls a tool such as a search or database query, observes the result, and stops once it can answer.

               ┌────────────────────────┐
               ▼                        │
[START] ──> [Reasoner (Thought)] ───> (Should Call Tool?) ───> [Tool Executor (Act)]
               │
               └─ (Has Final Answer) ──> [END]

It fits research, support and debugging agents, where needed data cannot be predicted.

The reasoner asks for ACTION: call_stock_api or FINAL: ..., including the last observation. The tool returns a mocked quote, and the edge back to reasoner closes the loop. should_continue caps it at three iterations.

from typing import TypedDict, Literal
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END

class ReActState(TypedDict):
    user_input: str
    agent_thought: str
    tool_output: str
    final_answer: str
    loop_count: int

model = ChatOpenAI(model="gpt-4o", temperature=0)

def reason(state: ReActState):
    loop_count = state.get("loop_count", 0) + 1
    tool_context = f"\nTool Observation: {state.get('tool_output', '')}" if loop_count > 1 else ""

    prompt = f"""You are a ReAct agent. Your goal is to find the current stock price of AAPL.
    Current Loop: {loop_count} {tool_context}

    Decide your next step. You must respond in one of two ways:
    1. If you need data, say: 'ACTION: call_stock_api'
    2. If you have the data, provide the answer starting with: 'FINAL: [your answer]'

    User Request: {state['user_input']}"""

    response = model.invoke([HumanMessage(content=prompt)]).content.strip()

    if "FINAL:" in response:
        return {"final_answer": response.replace("FINAL:", "").strip(), "loop_count": loop_count, "agent_thought": "done"}
    else:
        return {"agent_thought": "call_tool", "loop_count": loop_count}

def call_tool(state: ReActState):
    print("-> System: Executing external stock database API call...")
    mock_api_result = "$185.40 USD (Up 1.2% today)"
    return {"tool_output": mock_api_result}

def should_continue(state: ReActState) -> Literal["call_tool", "end"]:
    # Hard loop-break guardrail to prevent infinite execution loops
    if state["agent_thought"] == "call_tool" and state["loop_count"] < 3:
        return "call_tool"
    return "end"

react_flow = StateGraph(ReActState)
react_flow.add_node("reasoner", reason)
react_flow.add_node("tool_executor", call_tool)

react_flow.add_edge(START, "reasoner")
react_flow.add_conditional_edges("reasoner", should_continue, {
    "call_tool": "tool_executor",
    "end": END
})
react_flow.add_edge("tool_executor", "reasoner")

app = react_flow.compile()
result = app.invoke({"user_input": "What is the market status of Apple right now?", "loop_count": 0})
print(f"\nFinal Agent Resolution:\n{result['final_answer']}")

If the cap hits before a FINAL: answer, final_answer is never set and the last print raises a KeyError, so handle that case. Real systems usually use native tool calling instead of string markers. For the TypeScript equivalent, see bounded agentic loops for LLM tool use.

Pattern 4: evaluator and optimizer

A model judging its own work tends to rubber-stamp it, so an optimizer generates and revises while a separate, stronger evaluator critiques.

┌───> [Optimizer (Generate/Refine)] ───> [Evaluator (Critique)]
│                                               │
└──────────────── (If Rejected) ────────────────┼───> [Approved] ───> [END

It suits drafting, code generation, translation and strict quality or regulatory rules.

A cheaper model at temperature 0.7 writes a slogan, folding in feedback on later passes. The evaluator at temperature 0 checks that it contains future or smart and replies in a fixed ACCEPTED / FEEDBACK format; routing_gate sends rejections back.

from typing import TypedDict, Literal
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END

class EvaluationState(TypedDict):
    task: str
    draft: str
    feedback: str
    accepted: bool
    iterations: int

generator_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
evaluator_llm = ChatOpenAI(model="gpt-4o", temperature=0)

def generate_draft(state: EvaluationState):
    iterations = state.get("iterations", 0) + 1
    feedback_context = f"\nPrevious Feedback to incorporate: {state.get('feedback', '')}" if iterations > 1 else ""

    prompt = f"""Write a catchy, 3-sentence marketing slogan for: '{state['task']}'.
    {feedback_context}
    Provide ONLY the slogan."""

    response = generator_llm.invoke([HumanMessage(content=prompt)]).content.strip()
    return {"draft": response, "iterations": iterations}

def evaluate_draft(state: EvaluationState):
    prompt = f"""Review the following marketing slogan for the product '{state['task']}':
    Slogan: "{state['draft']}"

    CRITERIA: The slogan must include the exact word 'future' or 'smart'.

    Respond in EXACTLY the following format:
    ACCEPTED: True or False
    FEEDBACK: [If rejected, explain what needs fixing. If accepted, leave blank.]"""

    response = evaluator_llm.invoke([HumanMessage(content=prompt)]).content.strip()
    accepted = "ACCEPTED: True" in response
    feedback = response.split("FEEDBACK:")[-1].strip() if not accepted else ""
    return {"accepted": accepted, "feedback": feedback}

def routing_gate(state: EvaluationState) -> Literal["refine", "approve"]:
    if state["accepted"] or state["iterations"] >= 3:
        return "approve"
    return "refine"

eval_flow = StateGraph(EvaluationState)
eval_flow.add_node("generator", generate_draft)
eval_flow.add_node("evaluator", evaluate_draft)

eval_flow.add_edge(START, "generator")
eval_flow.add_edge("generator", "evaluator")
eval_flow.add_conditional_edges("evaluator", routing_gate, {
    "refine": "generator",
    "approve": END
})

app = eval_flow.compile()
result = app.invoke({"task": "Eco-friendly Electric Skateboards", "iterations": 0})
print(f"Final Slogan: {result['draft']}\nTotal Iterations: {result['iterations']}")

The gate also approves after three iterations regardless, so the output may be unaccepted; keep the accepted flag with the result. A keyword rule like this is cheaper and more reliable checked in code.

Pattern 5: human in the loop

For risky operations such as dropping tables, spending money or emailing clients, checkpointing lets the graph stop before a sensitive node, persist state and wait for approval.

[START] ──> [Stager] ──> ⛔ (State Saved to DB / Graph Pauses)
                          │
  [DevOps Manager Clicks "Approve"]
                          │
                          ▼
                [Executor (Run Production Deploy)] ──> [END]

Use it for migrations, deployments, payments or bulk email.

stager prepares a command and executor runs it. Compiling with a checkpointer and interrupt_before=["executor"] halts after staging. Runs are keyed by thread_id, so get_state shows saved values and the pending ('executor',) step; update_state records approval and invoke(None, config) resumes.

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

class DeploymentState(TypedDict):
    command: str
    approved: bool
    execution_log: str

# 1. Initialize thread checkpoint memory saver
memory = MemorySaver()

def stage_deployment(state: DeploymentState):
    print("-> System: Staging server deployment commands...")
    return {"command": "sudo systemctl restart production_api"}

def execute_deployment(state: DeploymentState):
    print("-> System: Execution approved. Running command on production servers...")
    return {"execution_log": f"Successfully executed: {state['command']}"}

hitl_flow = StateGraph(DeploymentState)
hitl_flow.add_node("stager", stage_deployment)
hitl_flow.add_node("executor", execute_deployment)

hitl_flow.add_edge(START, "stager")
hitl_flow.add_edge("stager", "executor")
hitl_flow.add_edge("executor", END)

# CRITICAL: Define the interrupt checkpoint before the executor node runs
app = hitl_flow.compile(checkpointer=memory, interrupt_before=["executor"])

# --- SIMULATING THE ACTIVE DEPLOYMENT WORKFLOW ---

config = {"configurable": {"thread_id": "prod_deploy_001"}}

# 1. Kick off the graph execution
initial_state = app.invoke({"command": "", "approved": False}, config)

# Verify the graph successfully halted its progress
print(f"\n[Current Graph State]: {app.get_state(config).values}")
print(f"[Next Pending Steps]: {app.get_state(config).next}") # Next step will say: ('executor',)

print("\n--- Halting Execution. Waiting for DevOps Manager Review... ---\n")

# 2. Simulate Human Reviewing the State and Updating with Approval
app.update_state(config, {"approved": True}, as_node="stager")

# 3. Resume execution thread seamlessly from the exact checkpoint
final_output = app.invoke(None, config)
print(f"[Final System Output]: {final_output['execution_log']}")

Three cautions. MemorySaver is in-memory only; pauses that survive restarts need a database-backed checkpointer. executor never checks approved, so add a check or a conditional edge that ends the run on rejection. Newer LangGraph releases also offer an interrupt() function, so check current docs for the recommended approach.

Choosing a pattern

Reliability comes from matching structure to the problem, not from bigger models or longer prompts:

• Router: many request types with different cost or skill needs.
• Orchestrator and workers: a large task that splits into independent parts.
• ReAct: the required information is only discoverable at run time.
• Evaluator and optimizer: output must meet explicit quality criteria.
• Human in the loop: an action is expensive or irreversible.

Patterns compose: a router can dispatch to a ReAct agent whose final action awaits approval. Keep three guardrails everywhere: cap every loop, validate model output that code depends on, and record whether a result was approved or ran out of attempts. Then the model need not be right first time, because the workflow lets it route, test, retry and defer to people.