{
  "schema_version": 1,
  "skill_name": "langgraph",
  "evals": [
    {
      "id": "pattern-selection",
      "prompt": "We are building an agent that handles customer support tickets: it needs to classify the ticket, call a specialist tool for the issue type, and sometimes escalate to a human. I keep reading about supervisor patterns, swarm patterns, and hierarchical patterns. How do I choose the right orchestration pattern?",
      "expected_output": "A pattern-selection analysis grounded in the workflow's structure rather than pattern-name enthusiasm: the response examines the workflow's control flow — the ticket must be routed by type, specialist sub-agents do bounded work, and human escalation is an interrupt — and maps it to the pattern that fits: a supervisor pattern where a router node decides the specialist and checks the result, with explicit state passing, rather than a free-running swarm where agents autonomously hand off work, because the workflow has a defined decision point and bounded sub-tasks. It explains the distinguishing questions: is the next step decided centrally (supervisor) or by agents themselves (swarm), does the graph have a fixed skeleton with choices (state graph with conditional edges) or recursive spawning (hierarchical), and where human-in-the-loop interrupts live. It prescribes sketching the control flow before choosing the pattern and names the gotcha: reaching for a swarm when the workflow is a deterministic pipeline adds nondeterminism and observability cost.",
      "assertions": [
        "The response maps the workflow's control flow to a concrete pattern choice",
        "Supervisor and swarm patterns are distinguished by who decides the next step",
        "Human-in-the-loop interrupts are placed in the design",
        "The response warns against free-running swarm patterns for deterministic routed workflows",
        "It prescribes sketching control flow before pattern selection"
      ]
    },
    {
      "id": "state-schema-design",
      "prompt": "I am designing a LangGraph agent that researches a topic, drafts a report, and revises it after review. The agents need to share the research findings and the draft, but I keep hearing that shared mutable state causes bugs in LangGraph. How should I design the state schema?",
      "expected_output": "A state-schema design that separates the concerns of shared data from per-step data: the response explains LangGraph's state model — a shared state object that nodes annotate, with reducers controlling how updates merge — and prescribes modeling the fields the whole graph needs (the topic, the research findings, the draft, review feedback) with typed annotations and explicit reducers where messages or lists accumulate, while transient per-node data that should not persist stays local to the node. It explains the common bugs: using a plain list field without a reducer so each node overwrites prior messages, mutating shared state in place instead of returning updates, and stuffing node-local scratch data into the shared state where it pollutes downstream nodes. The response prescribes the reducer choice (add for accumulating lists, replace for single-value updates, and a custom reducer for merging dicts) and shows how to inspect the state at each step for debugging.",
      "assertions": [
        "The response separates graph-shared state from per-node transient data",
        "Reducers are explained and prescribed for accumulating or merging fields",
        "The overwriting-list bug and in-place mutation pitfall are called out",
        "Typed annotations with reducer behavior are part of the design",
        "State inspection per step is prescribed for debugging"
      ]
    },
    {
      "id": "subgraph-composition",
      "prompt": "My agent has three independent phases — research, drafting, and review — and each phase is itself a multi-node graph. I want to compose them so each phase stays reusable and testable. How do I structure this with subgraphs, and where do I get it wrong?",
      "expected_output": "A subgraph-composition design where each phase is a self-contained graph with its own internal nodes and a narrow contract with the parent: the response prescribes defining each phase as a compiled subgraph whose input and output are explicit typed states, then composing them in the parent graph as single nodes that pass and receive only the agreed fields. It explains the common failure modes: coupling phases through the shared state by reading fields the phase does not own, making the subgraph's internal nodes reachable from outside (breaking encapsulation and making tests brittle), and mismatched state schemas between the parent and subgraph that surface as silent drops or type errors. The response covers testing each phase independently with its own fixtures and the parent test that verifies the handoff between phases, and it shows how the composition stays legible when each subgraph is treated as a node.",
      "assertions": [
        "Each phase is a self-contained subgraph with a narrow input-output contract",
        "The parent graph composes subgraphs as single nodes passing only agreed fields",
        "Encapsulation failures (external access to internal nodes, shared-state coupling) are flagged",
        "State-schema mismatches between parent and subgraph are called out",
        "Independent phase tests plus a handoff test are prescribed"
      ]
    },
    {
      "id": "human-in-the-loop-interrupt",
      "prompt": "My agent drafts an expense report and should pause for a human to approve it before submitting. If the approval fails or the reviewer edits the draft, the agent must revise and re-pause. How do I implement this interrupt pattern without losing the agent's state?",
      "expected_output": "A human-in-the-loop implementation built on interrupts and checkpoints: the response prescribes using the graph's interrupt mechanism at the approval node so the graph pauses with its state intact, then resumes when the human decision arrives, with the checkpointing layer persisting the full state so a process restart resumes the same run. It covers the design decisions: the interrupt payload (what the human sees and the structured decision input they return), validating the resumed input before continuing, the branch after the interrupt (approved proceeds, rejected or edited returns to the revision node with the feedback added to state), and the guardrails against the loop spinning: an explicit revision budget with a cap on re-pauses. It also explains the debugging angle: after an interrupt, inspecting the checkpointed state is how you verify nothing was lost.",
      "assertions": [
        "The interrupt mechanism is used with checkpointing so state survives pauses and restarts",
        "The interrupt payload and structured human decision input are designed",
        "Resumed input is validated before the graph continues",
        "The approve-revise-repause branch is implemented with a revision budget cap",
        "Checkpointed state inspection verifies nothing is lost"
      ]
    },
    {
      "id": "production-debugging",
      "prompt": "Our LangGraph agent works in tests but in production it sometimes ends in the wrong node: a tool result is missing from state, and a conditional edge routes to the error path even though the tool succeeded. How do I debug state and routing issues in a running graph?",
      "expected_output": "A debugging procedure that makes the graph's execution observable: the response prescribes inspecting the state and event stream at each step — using the graph's streaming or state-inspection facilities to see the exact state before and after each node, verifying what the tool call actually returned versus what the node wrote to state, and checking the conditional edge's routing function against that state to see why it chose the error path. It separates the failure classes: a node that did not write its output to state (missing field or wrong key), a reducer that overwrote a previous value, a conditional router reading the wrong field or applying the wrong predicate, and a tool result that never landed because the tool node errored or was skipped. The response prescribes reproducing with the production-shaped input, adding targeted logging at the state transitions, and writing a regression test that pins the routing decision.",
      "assertions": [
        "The debug procedure inspects state and events at each node transition",
        "It verifies what the tool returned versus what the node wrote to state",
        "Conditional routing functions are checked against the actual state",
        "Failure classes are separated: missing writes, reducer overwrites, wrong routing field",
        "A regression test pins the routing decision"
      ]
    }
  ]
}

# Architecture — LangGraph Core Concepts

This reference covers LangGraph's foundational architecture. Read this first if you're new to LangGraph or building a system from scratch.

## Overview

LangGraph is a low-level orchestration framework that models agent workflows as **directed graphs**. It is inspired by Google's [Pregel](https://research.google/pubs/pub37252/) and [Apache Beam](https://beam.apache.org/), with a public interface drawing from [NetworkX](https://networkx.org/). Unlike linear chain frameworks (LangChain Expression Language, traditional pipelines), LangGraph supports **cycles** — essential for agent tool-calling loops, iterative refinement, and multi-agent coordination.

```
User → [Agent Node] → {tool call?} → [Tool Node] → [Agent Node] → ... → Response
                    ↘ (no tool) → Response
```

## Core Abstractions

### State

The graph's shared, persistent data. All nodes read from and write to the same state object. Defined as a `TypedDict` or Pydantic schema.

**Key pattern — reducer annotations:** Fields that multiple nodes write to in parallel need a reducer. The `operator.add` reducer is the most common — it concatenates lists:

```python
from typing import Annotated, TypedDict
import operator

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]  # parallel-safe append
    next_agent: str                          # single writer only
    resolution_notes: Annotated[list[str], operator.add]
```

**Rule of thumb:** If multiple nodes can write to the same key, it needs a reducer. If only one node ever writes (e.g., a shared `topic` field), no reducer needed.

### Nodes

Python functions (or runnable objects) that receive state and return state updates:

```python
def my_node(state: AgentState) -> dict:
    # Read from state, do work, return updates
    return {"messages": [AIMessage(content="hello")]}
```

Nodes can be:
- **LLM calls** — invoke a model, return output
- **Tool nodes** — execute tool calls from an LLM
- **Python functions** — deterministic logic, transformations, validation
- **Subgraphs** — nested LangGraph graphs (see multi-agent-hierarchical reference)
- **Agents** — LangChain agents wrapped as a single node

### Edges

Edges define how execution flows between nodes:

- **`add_edge(start, end)`** — unconditionally traverse from start to end
- **`add_conditional_edges(source, router, path_map)`** — dynamic routing based on state

```python
# Standard edge
builder.add_edge("node_a", "node_b")

# Conditional edge
builder.add_conditional_edges(
    "analyzer",
    route_based_on_state,       # function that returns a node name
    {"billing": "billing_node", "tech": "tech_node", END: END}
)
```

## Two APIs

### Graph API (`StateGraph`)

Full control. You explicitly add nodes and wire edges with method calls.

```python
from langgraph.graph import StateGraph, START, END, MessagesState

builder = StateGraph(MessagesState)
builder.add_node("agent", agent_node)
builder.add_node("tools", ToolNode([search, calculator]))
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", should_continue, {"tools": "tools", END: END})
builder.add_edge("tools", "agent")
graph = builder.compile()
```

### Functional API (`@task` + `@entrypoint`)

Simpler decorator-based approach for linear or tree-shaped workflows:

```python
from langgraph.func import entrypoint, task

@task
def research(topic: str) -> str:
    return llm.invoke(f"Research {topic}").content

@task
def write_report(research: str) -> str:
    return llm.invoke(f"Write report: {research}").content

@entrypoint()
def workflow(topic: str):
    r = research(topic).result()
    return write_report(r).result()
```

**When to use each:**
- **Graph API** — complex routing, cycles, multi-agent, subgraphs, conditional flows
- **Functional API** — sequential chains, parallel fan-out patterns, simpler orchestration

## The Agent Loop (Tool-Calling)

The most common LangGraph pattern is the agent tool-calling loop:

```
1. LLM node receives messages
2. LLM returns response (may contain tool_calls)
3. Conditional edge: if tool_calls → ToolNode; if no tool_calls → END
4. ToolNode executes tools, returns ToolMessages
5. Edge back to LLM node
6. Repeat from 1
```

```python
from langgraph.prebuilt import ToolNode, tools_condition

builder = StateGraph(MessagesState)
builder.add_node("llm", llm_call)
builder.add_node("tools", ToolNode([search, calculator]))
builder.add_conditional_edges("llm", tools_condition, {"tools": "tools", END: END})
builder.add_edge("tools", "llm")
```

`tools_condition` is a prebuilt router: returns "tools" if the last message has `tool_calls`, otherwise returns `END`.

## Streaming

LangGraph supports multiple streaming modes:

```python
# Stream all events (most detailed)
for event in graph.stream_events(inputs, version="v3"):
    if event["method"] == "updates":
        print(event["params"]["data"])

# Stream values only (final state per superstep)
for chunk in graph.stream(inputs):
    print(chunk)

# Stream individual tokens from LLM nodes
for chunk in graph.stream(inputs, stream_mode="messages"):
    print(chunk)
```

## Key Architecture Patterns from the Official Guide

| Pattern | Structure | Best For |
|---------|-----------|----------|
| **Prompt Chaining** | Sequential LLM calls | Translation, verification, step-by-step generation |
| **Parallelization** | Fan-out → aggregate | Multi-criteria scoring, parallel research tasks |
| **Routing** | Classify → dispatch | Customer support triage, content-type routing |
| **Orchestrator-Worker** | Plan → fan-out → synthesize | Report writing, code generation across files |
| **Evaluator-Optimizer** | Generate → evaluate → loop | Iterative refinement, quality gates |
| **Agent** | LLM ↔ tool calls | General autonomous agents |

## Design Principles

1. **State is the source of truth** — all inter-node communication happens through state, not through side channels or global variables.
2. **Nodes are pure-ish** — a node receives state, does work, returns updates. It should not depend on state that isn't passed to it.
3. **Reducers prevent conflicts** — any state key written by multiple nodes in parallel MUST have a reducer.
4. **Start simple, add complexity only when needed** — a single agent with good prompts beats a multi-agent system with bad routing. Add agents only when a single prompt or toolset becomes unwieldy.
5. **Use `Send()` for dynamic fan-out** — when you don't know how many workers you'll need at compile time, use `Send()` to spawn workers dynamically from the orchestrator node.
6. **Subgraph state isolation** — subgraphs with different state schemas need a wrapper function to transform state at the boundary. Shared-schema subgraphs can be added directly as nodes.

# Evals — Evaluating Multi-Agent Systems

This reference covers methodology for evaluating multi-agent LangGraph systems. Evals are critical — without them, you're debugging routing behavior in production.

## Why Multi-Agent Evals Are Different

Single-agent evals test output quality against a rubric. Multi-agent evals must also test:
- **Routing accuracy** — Did the right agent(s) handle each domain?
- **Resolution coverage** — Were ALL parts of a multi-domain request addressed?
- **Handoff correctness** — Were handoffs clean, with full context propagation?
- **Recovery behavior** — Did the system degrade gracefully on errors?

## Building an Eval Dataset

### Using LangSmith

```python
from langsmith import Client

ls_client = Client()

dataset = ls_client.create_dataset(
    dataset_name="multi-agent-routing-evals",
    description="Routing and resolution evaluation dataset",
)

ls_client.create_examples(
    dataset_id=dataset.id,
    inputs=[
        {"question": "I need to change my payment method to a credit card."},
        {"question": "My SSO integration is returning error code SAML-401."},
        {"question": "I want to upgrade to Enterprise and also fix my broken SSO."},
        {"question": "Can you tell me who my account manager is?"},
    ],
    outputs=[
        {
            "expected_agents": ["billing"],
            "must_mention": ["payment", "credit card"],
        },
        {
            "expected_agents": ["tech_support"],
            "must_mention": ["SSO", "SAML"],
        },
        {
            "expected_agents": ["tech_support", "account"],
            "must_mention": ["SSO", "upgrade"],
        },
        {
            "expected_agents": ["account"],
            "must_mention": ["account manager"],
        },
    ],
)
```

### Dataset Design Principles

- **Cover all routing paths** — single-domain requests, multi-domain requests, edge cases
- **Include ambiguous requests** — ones that could route to multiple agents
- **Include negative cases** — requests that should not route to certain agents
- **5-20 examples minimum** — enough to catch regressions, not so many that eval is slow

## Evals

### 1. LLM-as-Judge (Routing Quality)

```python
from langsmith import evaluate
from openevals.llm import create_llm_as_judge

ROUTING_QUALITY_PROMPT = """\
Customer query: {inputs[question]}
Expected domains: {reference_outputs[expected_agents]}
Agent response: {outputs[final_response]}
Resolution notes: {outputs[resolution_notes]}

Rate 0.0-1.0 on whether the correct specialist agents handled the request
and the response fully addressed the customer's needs.
Return ONLY: {{"score": <float>, "reasoning": "<explanation>"}}"""

routing_judge = create_llm_as_judge(
    prompt=ROUTING_QUALITY_PROMPT,
    model="anthropic:claude-sonnet-4-5-20250929",
    feedback_key="routing_quality",
)
```

### 2. Resolution Coverage (Custom Evaluator)

Measures whether the final response mentions all required topics:

```python
def resolution_coverage(inputs: dict, outputs: dict, reference_outputs: dict) -> dict:
    text = outputs.get("final_response", "").lower()
    notes = " ".join(outputs.get("resolution_notes", [])).lower()
    combined = text + " " + notes
    must_mention = reference_outputs.get("must_mention", [])
    hits = sum(1 for t in must_mention if t.lower() in combined)
    return {
        "key": "resolution_coverage",
        "score": hits / len(must_mention) if must_mention else 1.0,
    }
```

### 3. Agent Routing Accuracy (Custom Evaluator)

Measures whether the correct agents were invoked:

```python
def agent_routing_accuracy(inputs: dict, outputs: dict, reference_outputs: dict) -> dict:
    notes = " ".join(outputs.get("resolution_notes", [])).lower()
    expected = reference_outputs.get("expected_agents", [])
    hits = sum(1 for agent in expected if agent.lower() in notes)
    return {
        "key": "routing_accuracy",
        "score": hits / len(expected) if expected else 1.0,
    }
```

## Running Evaluations

### Pattern Comparison

Run both patterns against the same dataset to compare:

```python
# Supervisor target function
def supervisor_target(inputs: dict) -> dict:
    result = supervisor_graph.invoke({
        "messages": [HumanMessage(content=inputs["question"])],
        "current_agent": "",
        "resolution_notes": [],
    })
    return {
        "final_response": result["messages"][-1].content,
        "resolution_notes": result.get("resolution_notes", []),
    }

# Swarm target function
def swarm_target(inputs: dict) -> dict:
    result = swarm_graph.invoke({
        "messages": [HumanMessage(content=inputs["question"])],
        "current_agent": "",
        "resolution_notes": [],
    })
    return {
        "final_response": result["messages"][-1].content,
        "resolution_notes": result.get("resolution_notes", []),
    }

# Run both
supervisor_results = evaluate(
    supervisor_target,
    data="multi-agent-routing-evals",
    evaluators=[routing_judge, resolution_coverage, agent_routing_accuracy],
    experiment_prefix="supervisor-v1",
    max_concurrency=2,
)

swarm_results = evaluate(
    swarm_target,
    data="multi-agent-routing-evals",
    evaluators=[routing_judge, resolution_coverage, agent_routing_accuracy],
    experiment_prefix="swarm-v1",
    max_concurrency=2,
)
```

## What to Watch

| Metric | What it catches | Action if it drops |
|--------|----------------|--------------------|
| **Routing accuracy** | Wrong agent handling a domain | Fix routing prompt or handoff logic |
| **Resolution coverage** | Multi-domain requests only partially addressed | Add explicit multi-domain routing logic |
| **Token cost per request** | Supervisor re-routing waste or context bloat | Consider fast-path or swarm migration |
| **Handoff chain length** | Swarm ping-pong or routing confusion | Add recursion guard, fix agent prompts |

## Eval-Driven Development Workflow

1. **Write the eval before the second agent** — building the routing accuracy eval is the first step after the triage agent
2. **Run on every PR** — the `routing_accuracy` evaluator is the canary. If it drops, your routing prompt or handoff logic regressed
3. **Compare patterns side-by-side in LangSmith** — make the supervisor-vs-swarm decision with data, not intuition
4. **Add examples from production misroutes** — every routing error in production should become a new eval example

# Hierarchical Multi-Agent — Subgraphs and Nested Teams

Hierarchical multi-agent systems use **subgraphs** — standalone LangGraph graphs composed as nodes within a parent graph. Each subgraph has its own state, its own nodes, and its own control flow, but communicates with the parent through defined state channels or function calls.

This pattern maps to organizational team structures: a parent supervisor delegates to team leads, who each manage their own specialist agents.

## Architecture

```
[Parent Supervisor]
       │
       ▼
┌──────────────┐     ┌──────────────┐
│  Research    │     │  Writing     │
│  Team        │     │  Team        │
│  (subgraph)  │     │  (subgraph)  │
│              │     │              │
│ Search →     │     │ Draft →      │
│ Analyze →    │     │ Review →     │
│ Synthesize   │     │ Polish       │
└──────────────┘     └──────────────┘
       │                      │
       └──────────┬───────────┘
                  ▼
         [Parent Supervisor]
                  │
                  ▼
            [Final Output]
```

## Two Subgraph Wiring Patterns

### Pattern A: Different State Schemas (Call Inside a Node)

Use when the parent and subgraph have **no shared state keys**. Write a wrapper function that transforms parent state to subgraph input and subgraph output back to parent state:

```python
from langgraph.graph.state import StateGraph, START

# Subgraph with its own state schema
class ResearchState(TypedDict):
    topic: str
    findings: list[str]

def search_node(state: ResearchState):
    results = search(state["topic"])
    return {"findings": [results]}

def analyze_node(state: ResearchState):
    analysis = analyze_findings(state["findings"])
    return {"findings": state["findings"] + [analysis]}

research_builder = StateGraph(ResearchState)
research_builder.add_node(search_node)
research_builder.add_node(analyze_node)
research_builder.add_edge(START, "search_node")
research_builder.add_edge("search_node", "analyze_node")
research_subgraph = research_builder.compile()

# Parent graph with different state
class ParentState(TypedDict):
    query: str
    answer: str

def call_research_team(state: ParentState):
    # Transform parent → subgraph state
    subgraph_input = {"topic": state["query"], "findings": []}
    subgraph_output = research_subgraph.invoke(subgraph_input)
    # Transform subgraph → parent state
    return {"answer": subgraph_output["findings"][-1]}

parent_builder = StateGraph(ParentState)
parent_builder.add_node("research", call_research_team)
parent_builder.add_edge(START, "research")
parent_graph = parent_builder.compile()
```

### Pattern B: Shared State Keys (Add Subgraph as Node)

Use when the parent and subgraph **share state keys** (e.g., both use `messages`). Pass the compiled subgraph directly to `add_node` — no wrapper needed:

```python
# Subgraph that reads/writes shared state
class SharedState(MessagesState):
    extra: str

def sub_agent_node(state: SharedState):
    result = some_agent.invoke({"messages": state["messages"]})
    return {"messages": result["messages"][-1:]}

sub_builder = StateGraph(SharedState)
sub_builder.add_node("sub_agent", sub_agent_node)
sub_builder.add_edge(START, "sub_agent")
subgraph = sub_builder.compile()

# Parent graph — add subgraph directly as a node
parent_builder = StateGraph(SharedState)
parent_builder.add_node("my_subgraph", subgraph)  # compiled graph as node
parent_builder.add_edge(START, "my_subgraph")
parent_graph = parent_builder.compile()
```

## Subgraph Persistence Modes

Subgraph persistence is controlled by the `checkpointer` parameter on `.compile()`.

| Mode | `checkpointer=` | Behavior | Use Case |
|------|----------------|----------|----------|
| **Per-invocation** (default) | `None` | Each call starts fresh, inherits parent's checkpointer for interrupts within a single call | Multi-agent systems where subagents handle independent one-off requests |
| **Per-thread** | `True` | State accumulates across calls on the same thread | Research assistant that builds context over several exchanges |
| **Stateless** | `False` | No checkpointing — runs like a plain function | Simple transformations, no durability needed |

### Per-Invocation (Default — Recommended)

```python
subgraph = builder.compile()  # checkpointer=None, inherits from parent
```

- Supports interrupts and durable execution within a single call
- Each call starts fresh — no memory across calls
- Supports parallel calls to the same subgraph without conflicts
- Right choice for most multi-agent systems

### Per-Thread

```python
subgraph = builder.compile(checkpointer=True)
```

- Subgraph remembers previous interactions on the same thread
- **Does NOT support parallel calls** — same-namespace checkpoint conflicts
- Use `ToolCallLimitMiddleware` when wrapping as tools to prevent parallel invocation
- Requires namespace isolation when multiple per-thread subgraphs exist

**Namespace isolation pattern:**

```python
def create_sub_agent(model, *, name, **kwargs):
    """Wrap an agent with a unique node name for namespace isolation."""
    agent = create_agent(model=model, name=name, **kwargs)
    return (
        StateGraph(MessagesState)
        .add_node(name, agent)   # unique name → stable namespace
        .add_edge("__start__", name)
        .compile()
    )

fruit_agent = create_sub_agent("gpt-4o-mini", name="fruit_agent",
    tools=[fruit_info], prompt="...", checkpointer=True)
veggie_agent = create_sub_agent("gpt-4o-mini", name="veggie_agent",
    tools=[veggie_info], prompt="...", checkpointer=True)
```

### Stateless

```python
subgraph = builder.compile(checkpointer=False)
```

- Runs like a plain function call
- No interrupts, no durable execution
- If the process crashes mid-run, the subgraph cannot recover

## Streaming from Subgraphs

```python
# Using the stream.subgraphs projection (recommended)
stream = graph.stream_events(inputs, version="v3")
for subgraph in stream.subgraphs:
    print(subgraph.graph_name, subgraph.path)
    for snapshot in subgraph.values:
        print(subgraph.path, snapshot)

# Using raw event protocol
for event in stream:
    if event["method"] == "updates":
        print(event["params"]["namespace"], event["params"]["data"])
```

## Inspecting Subgraph State

```python
# Requires parent graph compiled with checkpointer
config = {"configurable": {"thread_id": "1"}}
state = graph.get_state(config, subgraphs=True)

# Access subgraph state
subgraph_state = state.tasks[0].state  # first subgraph's current state
```

## Hierarchical Teams Pattern (Full)

This pattern combines a supervisor node with nested subgraph teams:

```python
from langgraph.graph import StateGraph, START, END

# 1. Create team subgraphs
research_team = create_research_subgraph(llm, tools)
writing_team = create_writing_subgraph(llm, tools)

# 2. Create parent state
class ManagerState(MessagesState):
    current_team: str
    final_output: str

# 3. Manager node decides which team to activate
def manager_node(state: ManagerState):
    decision = routing_llm.invoke([
        SystemMessage(content="Route to: research_team, writing_team, or DONE"),
        *state["messages"],
    ])
    return {"current_team": decision.next_agent}

# 4. Wire parent graph with subgraph nodes
builder = StateGraph(ManagerState)
builder.add_node("manager", manager_node)
builder.add_node("research_team", research_team)   # subgraph as node
builder.add_node("writing_team", writing_team)     # subgraph as node
builder.add_edge(START, "manager")
builder.add_conditional_edges(
    "manager",
    lambda s: s.get("current_team", "DONE"),
    {"research_team": "research_team", "writing_team": "writing_team", "DONE": END},
)
builder.add_edge("research_team", "manager")
builder.add_edge("writing_team", "manager")
graph = builder.compile()
```

## Design Principles

1. **Each subgraph is independently testable** — you can `.invoke()` any subgraph in isolation before composing it into the parent.
2. **State boundaries are interface contracts** — the state keys a subgraph reads and writes define its public API. Document them.
3. **Prefer shared state schemas (Pattern B)** when subgraphs operate on the same data types (messages, documents). The wrapper-less composition is simpler and faster.
4. **Use per-invocation persistence by default** — only opt into per-thread when a subagent genuinely needs cross-call memory.
5. **Name isolation for per-thread subgraphs** — two per-thread subgraphs need different namespace prefixes. The `create_sub_agent` wrapper pattern ensures this.

# Supervisor Pattern — Centralized Multi-Agent Routing

The supervisor pattern uses a dedicated routing node to coordinate specialist agents. Every message passes through the supervisor, which classifies intent and routes to the appropriate specialist. After the specialist responds, control returns to the supervisor for the next routing decision.

This pattern is the right choice when **accuracy matters more than latency**, domain boundaries are ambiguous, and you need a centralized audit trail of every routing decision.

## Architecture

```
[User] → [Supervisor] → [Billing Agent] → [Supervisor] → [Response]
                    ↘ [Tech Support] → ↗
                    ↘ [Account Mgmt] → ↗
```

The supervisor is a dedicated LLM node with **structured output** (`with_structured_output(RoutingDecision)`) — no regex, no string parsing, just typed routing decisions.

## Comparative Metrics

| Metric | Supervisor | Swarm |
|--------|------------|-------|
| Avg latency (single-domain) | ~4.2s | ~2.8s |
| Avg latency (handoff required) | ~9.1s | ~5.4s |
| LLM calls (single-domain) | 2 (route + specialist) | 1 (specialist only) |
| LLM calls (handoff required) | 4 (route×2 + specialist×2) | 2 (specialist×2) |
| Avg tokens per request | ~2,800 | ~1,900 |
| Routing accuracy | ~94% | ~91% |

## Implementation

### 1. Define State

```python
from typing import Annotated, TypedDict
from langgraph.graph import MessagesState
import operator

class MultiAgentState(MessagesState):
    current_agent: str                           # which specialist is active
    resolution_notes: Annotated[list[str], operator.add]  # audit trail
    handoff_count: int                           # recursion guard
```

### 2. Define Routing Schema (Pydantic)

```python
from pydantic import BaseModel, Field

class RoutingDecision(BaseModel):
    next_agent: str = Field(
        description="Next agent: 'billing', 'tech_support', 'account', or 'DONE'"
    )
    reasoning: str = Field(description="Why this agent was chosen")
```

### 3. Create the Supervisor Node

```python
routing_llm = llm.with_structured_output(RoutingDecision)

def supervisor(state: MultiAgentState) -> dict:
    notes = "\n".join(state.get("resolution_notes", []))
    history_context = f"\n\nAlready resolved:\n{notes}" if notes else ""

    response = routing_llm.invoke([
        SystemMessage(
            content="You are a multi-agent supervisor. Analyze the conversation "
                    "and decide which specialist should handle the next step.\n\n"
                    "Available agents:\n"
                    "- billing: invoices, payments, discounts, pricing\n"
                    "- tech_support: technical issues, SSO, integrations, bugs\n"
                    "- account: plan changes, upgrades, account administration\n"
                    "- DONE: the customer's request has been fully addressed\n\n"
                    "Do NOT re-route to an agent that has already handled "
                    "its portion of the request." + history_context
        ),
        *state["messages"],
    ])
    return {"current_agent": response.next_agent}
```

### 4. Create Specialist Agents

Each specialist has a focused system prompt and domain-specific tools. Using `create_agent` from `langchain.agents`:

```python
from langchain.agents import create_agent

billing_agent = create_agent(
    llm,
    tools=[lookup_billing_info, apply_discount],
    system_prompt="You are a billing specialist. Help customers with invoices, "
           "payments, discounts, and plan pricing. "
           "Customer ID is 'C-1042' unless otherwise specified.",
)

tech_support_agent = create_agent(
    llm,
    tools=[diagnose_sso, check_system_status],
    system_prompt="You are a technical support specialist. Help customers "
           "diagnose and resolve technical issues.",
)

account_agent = create_agent(
    llm,
    tools=[lookup_account_details, update_plan],
    system_prompt="You are an account management specialist. Help customers "
           "with plan changes, upgrades, and account administration.",
)
```

### 5. Specialist Node Wrappers

```python
def billing_node(state: MultiAgentState) -> dict:
    result = billing_agent.invoke({"messages": state["messages"]})
    return {
        "messages": result["messages"][-1:],
        "resolution_notes": [
            f"Billing: {result['messages'][-1].content[:200]}"
        ],
    }

def tech_support_node(state: MultiAgentState) -> dict:
    result = tech_support_agent.invoke({"messages": state["messages"]})
    return {
        "messages": result["messages"][-1:],
        "resolution_notes": [
            f"Tech Support: {result['messages'][-1].content[:200]}"
        ],
    }

def account_node(state: MultiAgentState) -> dict:
    result = account_agent.invoke({"messages": state["messages"]})
    return {
        "messages": result["messages"][-1:],
        "resolution_notes": [
            f"Account: {result['messages'][-1].content[:200]}"
        ],
    }
```

### 6. Wire the Graph

```python
from langgraph.graph import StateGraph, START, END

def route_to_agent(state: MultiAgentState) -> str:
    agent = state.get("current_agent", "DONE")
    if agent == "DONE":
        return "end"
    return agent

builder = StateGraph(MultiAgentState)

builder.add_node("supervisor", supervisor)
builder.add_node("billing", billing_node)
builder.add_node("tech_support", tech_support_node)
builder.add_node("account", account_node)

builder.add_edge(START, "supervisor")

builder.add_conditional_edges(
    "supervisor",
    route_to_agent,
    {
        "billing": "billing",
        "tech_support": "tech_support",
        "account": "account",
        "end": END,
    },
)

# Each specialist returns to supervisor
builder.add_edge("billing", "supervisor")
builder.add_edge("tech_support", "supervisor")
builder.add_edge("account", "supervisor")

supervisor_graph = builder.compile()
```

## Failure Modes

### Routing Loops
The supervisor routes to billing → billing responds → supervisor routes to billing again → repeats.

**Fix:** Include resolution notes in the supervisor's context so it can see what's already been addressed. The `resolution_notes` accumulator (with `operator.add` reducer) tracks what each specialist has done.

### Supervisor Bottleneck
Every interaction requires a routing LLM call — even when the intent is obvious ("I need to change my password" doesn't need a routing decision).

**Fix:** Add a fast-path classifier before the supervisor:

```python
FAST_PATH = {
    "password": "tech_support",
    "invoice": "billing",
    "upgrade": "account",
    "downgrade": "account",
}

def fast_path_or_supervisor(state: MultiAgentState) -> dict:
    last_msg = state["messages"][-1].content.lower()
    for keyword, agent in FAST_PATH.items():
        if keyword in last_msg:
            return {"current_agent": agent}
    return supervisor(state)
```

### Token Waste on Re-routing
The supervisor pattern doubles token spend on routing calls compared to swarm.

**Fix:** Track total tokens per pattern across sessions in LangSmith. If token costs exceed the value of the routing accuracy gain, consider switching to swarm.

## When to Use

- Routing accuracy is the most important metric
- Domain boundaries are ambiguous (billing vs account overlap)
- You need a centralized audit trail of every decision
- You're iterating on routing logic and want to change it in one place
- You have fewer than 5 specialists and latency is acceptable

# Swarm Pattern — Direct Agent-to-Agent Handoffs

The swarm pattern eliminates the central supervisor. Agents hand off directly to each other using `Command` objects returned from handoff tools. When an agent calls a handoff tool, `Command(goto=target, graph=Command.PARENT)` tells LangGraph to navigate to a different node in the parent graph.

This pattern is the right choice when **latency is the primary constraint**, domain boundaries are clear, and agents rarely misroute on their own.

## Architecture

```
[User] → [Triage Agent] → [Billing Agent] → [Response]
                │                    │
                ├──→ [Tech Support] ─┘
                └──→ [Account Mgmt]
```

No central orchestrator. Each agent decides whether to handle the request itself or hand off to another specialist using a `Command`.

## Comparative Metrics

| Metric | Swarm | Supervisor |
|--------|-------|------------|
| Avg latency (single-domain) | ~2.8s | ~4.2s |
| Avg latency (handoff required) | ~5.4s | ~9.1s |
| LLM calls (single-domain) | 1 (specialist only) | 2 (route + specialist) |
| LLM calls (handoff required) | 2 (specialist×2) | 4 (route×2 + specialist×2) |
| Avg tokens per request | ~1,900 | ~2,800 |
| Routing accuracy | ~91% | ~94% |

## Implementation

### 1. Define State

```python
from typing import Annotated, TypedDict
from langgraph.graph import MessagesState
import operator

class SwarmState(MessagesState):
    current_agent: str
    resolution_notes: Annotated[list[str], operator.add]
    handoff_count: int       # recursion guard — increment on each handoff
```

### 2. Create Handoff Tool Factory

The handoff tool returns a `Command` that tells LangGraph to navigate to a different node:

```python
from langgraph.types import Command
from langchain_core.tools import tool

def make_handoff_tool(target_agent: str, description: str):
    """Factory that creates a handoff tool for transferring to another agent."""

    @tool(f"transfer_to_{target_agent}")
    def handoff(reason: str) -> Command:
        """Transfer the conversation to another specialist agent."""
        return Command(
            goto=target_agent,
            update={"current_agent": target_agent},
            graph=Command.PARENT,
        )

    handoff.__doc__ = description
    return handoff
```

### 3. Create Handoff Tools

```python
transfer_to_billing = make_handoff_tool(
    "billing",
    "Transfer to the billing specialist for invoices, payments, or discounts.",
)
transfer_to_tech = make_handoff_tool(
    "tech_support",
    "Transfer to technical support for SSO, integrations, or system issues.",
)
transfer_to_account = make_handoff_tool(
    "account",
    "Transfer to account management for plan changes or upgrades.",
)
```

### 4. Create Agents with Handoff Tools

The **triage agent** has no domain tools — it only routes:

```python
from langchain.agents import create_agent

triage_agent = create_agent(
    llm,
    tools=[transfer_to_billing, transfer_to_tech, transfer_to_account],
    system_prompt="You are a triage agent. Analyze the customer's request and "
           "transfer to the appropriate specialist. Do not answer questions "
           "yourself — always transfer. If multiple issues exist, transfer "
           "to the most urgent one first.",
)
```

**Specialist agents** have domain tools PLUS handoff tools for other domains:

```python
billing_swarm_agent = create_agent(
    llm,
    tools=[lookup_billing_info, apply_discount,
           transfer_to_tech, transfer_to_account],
    system_prompt="You are a billing specialist. Help with invoices, payments, "
           "and discounts. If the customer has unresolved issues outside your "
           "domain, transfer to the appropriate specialist.",
)

tech_swarm_agent = create_agent(
    llm,
    tools=[diagnose_sso, check_system_status,
           transfer_to_billing, transfer_to_account],
    system_prompt="You are a technical support specialist. Help with technical "
           "issues, SSO, and integrations. If the customer has unresolved "
           "issues outside your domain, transfer to the appropriate specialist.",
)

account_swarm_agent = create_agent(
    llm,
    tools=[lookup_account_details, update_plan,
           transfer_to_billing, transfer_to_tech],
    system_prompt="You are an account management specialist. Help with plan "
           "changes and upgrades. If the customer has unresolved issues "
           "outside your domain, transfer to the appropriate specialist.",
)
```

### 5. Create Node Wrappers

```python
def triage_node(state: SwarmState) -> Command:
    result = triage_agent.invoke({"messages": state["messages"]})
    return result  # Command from the handoff tool

def billing_swarm_node(state: SwarmState) -> dict:
    result = billing_swarm_agent.invoke({"messages": state["messages"]})
    return {
        "messages": result["messages"][-1:],
        "resolution_notes": [
            f"Billing: {result['messages'][-1].content[:200]}"
        ],
    }

def tech_swarm_node(state: SwarmState) -> dict:
    result = tech_swarm_agent.invoke({"messages": state["messages"]})
    return {
        "messages": result["messages"][-1:],
        "resolution_notes": [
            f"Tech Support: {result['messages'][-1].content[:200]}"
        ],
    }

def account_swarm_node(state: SwarmState) -> dict:
    result = account_swarm_agent.invoke({"messages": state["messages"]})
    return {
        "messages": result["messages"][-1:],
        "resolution_notes": [
            f"Account: {result['messages'][-1].content[:200]}"
        ],
    }
```

### 6. Wire the Graph

```python
from langgraph.graph import StateGraph, START, END
from typing import Literal

def route_after_agent(
    state: SwarmState,
) -> Literal["billing", "tech_support", "account", "__end__"]:
    # If the last message is an AIMessage without tool_calls, we're done
    messages = state.get("messages", [])
    if messages:
        last_msg = messages[-1]
        if isinstance(last_msg, AIMessage) and not last_msg.tool_calls:
            return "__end__"
    # Otherwise, continue with the current agent (handoff via Command)
    current = state.get("current_agent", "")
    if current in ("billing", "tech_support", "account"):
        return current
    return "__end__"

swarm_builder = StateGraph(SwarmState)

swarm_builder.add_node("triage", triage_node)
swarm_builder.add_node("billing", billing_swarm_node)
swarm_builder.add_node("tech_support", tech_swarm_node)
swarm_builder.add_node("account", account_swarm_node)

swarm_builder.add_edge(START, "triage")

# Each specialist can route to any other specialist or end
for node in ["billing", "tech_support", "account"]:
    swarm_builder.add_conditional_edges(
        node, route_after_agent,
        ["billing", "tech_support", "account", END],
    )

swarm_graph = swarm_builder.compile()
```

## Key Details

### Command-based Routing

The `Command` object is the key mechanism. When an agent calls a handoff tool, the tool returns `Command(goto=target_agent, graph=Command.PARENT)` which:
1. Updates `current_agent` in state
2. Tells LangGraph to navigate to the target node in the parent graph
3. Carries any additional state updates via `update=`

### Recursion Guard

The swarm pattern has no natural recursion limit. Two agents can ping-pong indefinitely. Add a handoff counter:

```python
from langgraph.types import Command

def make_handoff_tool(target_agent, description, max_handoffs=3):
    @tool(f"transfer_to_{target_agent}")
    def handoff(reason: str, _state: SwarmState = None) -> Command:
        """Transfer the conversation to another specialist agent."""
        # You'd need to access state — use ToolRuntime or inject via closure
        return Command(
            goto=target_agent,
            update={"current_agent": target_agent},
            graph=Command.PARENT,
        )
    handoff.__doc__ = description
    return handoff
```

Track `handoff_count` in state and check it in `route_after_agent` — after 3, force END.

### Context Propagation

When Agent A hands off to Agent B, the response from Agent A must be in the message history. The `Command.update` should carry the relevant messages. Without this, Agent B sees the original user message but has no context about what Agent A already did.

## Failure Modes

### Context Loss on Handoff
Agent A resolves part of the issue and hands off to Agent B. Agent B sees the original message but doesn't know what Agent A already did.

**Fix:** Ensure `Command.update` includes the resolution context. The `resolution_notes` accumulator is one approach; forwarding the last AIMessage is another.

### Swarm Ping-Pong
Agent A doesn't know the answer → hands off to Agent B. Agent B also doesn't know → hands off back to Agent A. Repeats until recursion limit.

**Fix:** Track `handoff_count` in state. Hard limit at 3, then escalate to human or fallback agent.

### Lost Messages During Handoff
The handoff tool returns a `Command(graph=Command.PARENT)`, but the specialist agent's internal tool-calling loop messages don't propagate to the parent graph.

**Fix:** Ensure your `Command.update` includes the relevant messages. LLMs expect tool calls to be paired with `ToolMessage` responses. If you break that pairing, the next agent sees malformed history.

## When to Use

- Latency is your primary constraint
- Domain boundaries are clear and agents rarely misroute
- Requests often span multiple domains (latency savings compound)
- You want agents to maintain conversational context through handoffs
- You have good per-agent evals to catch misroutes

# Persistence — Checkpointers and Stores

LangGraph provides two complementary persistence systems: **checkpointers** for short-term, thread-scoped memory and **stores** for long-term, cross-thread memory.

## Checkpointer vs Store

| Dimension | Checkpointer | Store |
|-----------|-------------|-------|
| Persists | Graph state snapshots | Application-defined key-value data |
| Scope | A single thread | Across threads |
| Memory type | Short-term, thread-scoped | Long-term, cross-thread |
| Use for | Conversation continuity, HITL, time travel, fault tolerance | User preferences, facts, shared knowledge |
| Access | Pass `thread_id` in graph config | Read/write from nodes or application code |

## Checkpointers (Short-Term Memory)

A checkpointer saves snapshots of graph state after each superstep (node execution). This enables:
- **Conversation continuity** — resume a thread where it left off
- **Human-in-the-loop** — pause for input, then resume
- **Time travel** — replay from any checkpoint
- **Fault tolerance** — recover from crashes

### Backends

```python
from langgraph.checkpoint.memory import MemorySaver   # in-memory (dev only)
from langgraph.checkpoint.sqlite import SqliteSaver   # file-based (dev)
from langgraph.checkpoint.postgres import PostgresSaver  # production
```

### Usage

```python
checkpointer = MemorySaver()  # or PostgresSaver.from_conn_string(...)
checkpointer.setup()          # creates tables for persistent backends

graph = builder.compile(checkpointer=checkpointer)

# Each thread gets a unique thread_id
config = {"configurable": {"thread_id": "thread-123"}}

result = graph.invoke(
    {"messages": [{"role": "user", "content": "Hi, my name is Bob."}]},
    config=config,
)

# Resume on the same thread — graph remembers previous messages
result2 = graph.invoke(
    {"messages": [{"role": "user", "content": "What's my name?"}]},
    config=config,
)
```

### Agent Server Handles Persistence Automatically

When deploying via LangSmith Agent Server, you do not need to implement or configure checkpointers manually. The server handles persistence infrastructure.

## Stores (Long-Term Memory)

A store persists key-value data outside graph state, accessible across threads and sessions.

### Usage

```python
from langgraph.store.memory import InMemoryStore

store = InMemoryStore()

# Write to store from a node
def remember_user(state: MessagesState, store: InMemoryStore):
    user_id = extract_user_id(state["messages"])
    store.put(
        ("users", user_id),
        "preferences",
        {"theme": "dark", "language": "en"},
    )
    return {"messages": [AIMessage(content="Saved your preferences!")]}

# Read from store
config = {"configurable": {"thread_id": "1"}}
result = graph.invoke(inputs, config=config, store=store)
```

### When to Use Store vs Checkpointer

| Need | Use |
|------|-----|
| Resume a conversation mid-stream | Checkpointer |
| Undo/redo across steps (time travel) | Checkpointer |
| Pause for human approval | Checkpointer |
| Remember user preferences across sessions | Store |
| Share data between unrelated threads | Store |
| Learn facts that persist beyond conversation | Store |

## Checkpointer Troubleshooting

### `thread_id` too long (PostgresSaver)

Keep `thread_id` under 255 characters. Use UUID or hash:

```python
import uuid
config = {"configurable": {"thread_id": str(uuid.uuid4())[:255]}}
```

### `MemorySaver` doesn't persist between restarts

In-memory checkpointers are lost on process restart. Use `PostgresSaver` or `SqliteSaver` for persistence.

### Checkpoints growing unboundedly

Long conversations accumulate checkpoints. Prune periodically or set retention:

```python
# PostgresSaver — add a cron job to delete old checkpoints:
# DELETE FROM langgraph_checkpoints WHERE created_at < NOW() - INTERVAL '7 days'
```

### State access from parent to subgraph

Subgraphs manage their own checkpoint namespace. Use **Store** for data that needs to cross graph boundaries, or configure the subgraph to write to the parent checkpoint.

## Advanced: Subgraph Persistence Modes

See `references/multi-agent-hierarchical.md` for the full subgraph persistence reference. Summary:

| Mode | `checkpointer=` | Behavior |
|------|----------------|----------|
| Per-invocation (default) | `None` | Fresh each call, inherits parent checkpointer for HITL |
| Per-thread | `True` | State accumulates across calls |
| Stateless | `False` | No checkpointing, runs like plain function |

# Production — Deployment, Observability, and Common Failures

Deploying LangGraph workflows to production requires careful infrastructure planning. This reference covers deployment via LangSmith Agent Server, observability with LangSmith, and the most common production failure modes with fixes.

## Deployment

LangGraph apps are deployed via **LangSmith Agent Server** — a purpose-built platform for long-running, stateful workflows.

### Key Capabilities

| Feature | Details |
|---------|---------|
| Persistence | Handled automatically by Agent Server — no manual checkpointer config needed |
| Scaling | Horizontal scaling with stateful persistence |
| Monitoring | Built-in LangSmith tracing and evaluation |
| CLI | `langgraph deploy` for deployment management |
| Docker | Custom Dockerfiles supported via `langgraph.json` configuration |

### Deployment Architecture

```
Local Dev → langgraph deploy → Agent Server → Production
                                    │
                                    ▼
                            LangSmith Tracing
                                    │
                                    ▼
                        Observability + Evaluation
```

## Observability with LangSmith

### Per-Node Tracing

Decorate every node with `@traceable` for isolated spans:

```python
from langsmith import traceable

@traceable(name="billing_node", run_type="chain")
def billing_node(state: MultiAgentState) -> dict:
    # ... node implementation
    return result
```

### Tag Traces for Comparison

```python
with tracing_context(
    metadata={"pattern": "supervisor", "agents_available": 3},
    tags=["production", "multi-agent-v1"],
):
    result = graph.invoke(inputs)
```

### Key Metrics to Watch

1. **Routing accuracy** — Open the supervisor span, check if the chosen agent matches the actual domain. Log misroutes as negative feedback.
2. **Handoff chains** (swarm) — Trace the full `triage → tech → billing` path. Longer than 3 hops = routing problem.
3. **Token waste on re-routing** — The supervisor pattern doubles token spend on routing calls. Track total tokens per pattern and compare.

## Common Production Failure Modes

### 1. Routing Loops (Supervisor)

Supervisor routes to billing → billing responds → supervisor routes to billing again → repeats.

**Diagnosis:** LangSmith traces show the same `supervisor → billing → supervisor → billing` pattern repeating.

**Fix:** Include resolution notes in the supervisor's context so it can see what's already been addressed. Add a `handoff_count` to state and check it in the routing function.

```python
class MultiAgentState(MessagesState):
    current_agent: str
    resolution_notes: Annotated[list[str], operator.add]
    handoff_count: int

def route_to_agent(state: MultiAgentState) -> str:
    if state["handoff_count"] >= 5:
        return "end"  # force escalation
    agent = state.get("current_agent", "DONE")
    if agent == "DONE":
        return "end"
    return agent
```

### 2. Context Loss on Handoff (Swarm)

Agent A resolves part of the issue and hands off to Agent B. Agent B sees the original message but has no context about what Agent A already did.

**Fix:** Propagate resolution context through `Command.update`. The `resolution_notes` accumulator ensures each specialist's work is visible to the next.

### 3. Supervisor Bottleneck

Every interaction requires a routing LLM call — even for obvious intents.

**Fix:** Add a fast-path classifier (keyword matching or small model) for unambiguous intents before the supervisor LLM call.

### 4. Swarm Ping-Pong

Agent A doesn't know the answer → hands off to Agent B → Agent B doesn't know → hands off back. Repeats until recursion limit.

**Fix:** Track `handoff_count` in state. After 3, force escalation to a human or fallback agent.

### 5. Lost Messages During Handoff

Handoff tool returns `Command(graph=Command.PARENT)` but the specialist's tool-calling loop messages don't propagate to the parent.

**Fix:** Ensure `Command.update` includes the relevant messages. LLMs expect `ToolCall` ↔ `ToolMessage` pairing. Breaking this pairing causes malformed history errors.

### 6. Per-Thread Subgraph Parallel Calls

An LLM calls a per-thread subgraph tool multiple times in parallel. Both writes hit the same checkpoint namespace → conflict.

**Fix:** Use `ToolCallLimitMiddleware` or configure the model to prevent parallel tool calls for per-thread subgraph tools.

### 7. Checkpoint Bloat

Long conversations accumulate thousands of checkpoints, increasing latency and storage costs.

**Fix:** Prune old checkpoints periodically. With PostgresSaver, set up a cron job to delete checkpoints older than N days.

### 8. MemorySaver Not Persisting Restarts

Local development with `MemorySaver` loses all state when the process restarts.

**Fix:** Use a persistent backend (`PostgresSaver` / `SqliteSaver`) for anything that needs to survive restarts.

## Production Readiness Checklist

- [ ] Checkpointer uses a persistent backend (not `MemorySaver`)
- [ ] All nodes have `@traceable` decorators for observability
- [ ] Routing has a recursion guard (`handoff_count` limit)
- [ ] Supervisor includes resolution notes in context (prevents loops)
- [ ] Fast-path classifier exists for unambiguous intents
- [ ] Evals pipeline runs on every PR (routing accuracy + resolution coverage)
- [ ] Checkpoint pruning cron job configured
- [ ] Per-thread subgraph tools have parallel call limits
- [ ] Error handling for tool failures (graceful degradation, not crash)
- [ ] Streaming implemented for real-time UX

# Troubleshooting — Common Failures and Fixes

This reference catalogs the most common LangGraph failure modes, their symptoms, root causes, and fixes. Organized by pattern and symptom for quick lookup.

## Supervisor Pattern Failures

| Symptom | Likely Cause | Fix |
|---------|-------------|-----|
| Same agent called repeatedly in a loop | Routing loop — supervisor doesn't know the agent already handled it | Include `resolution_notes` in supervisor context |
| Slow response times under load | Supervisor bottleneck — even obvious intents get a routing LLM call | Add fast-path classifier for unambiguous intents |
| High token costs per request | Token waste on re-routing | Compare supervisor vs swarm token costs in LangSmith |
| Agent handles wrong domain | Routing accuracy regression | Check routing prompt, add negative feedback to eval dataset |

### Routing Loop — Full Diagnosis

1. Open LangSmith trace
2. Look for repeating `supervisor → billing → supervisor → billing` pattern
3. Check if `resolution_notes` is being passed to the supervisor node
4. If missing, the supervisor has no memory of what's been done
5. Fix: ensure supervisor prompt includes resolution notes:

```python
def supervisor(state: MultiAgentState) -> dict:
    notes = "\n".join(state.get("resolution_notes", []))
    history_context = f"\n\nAlready resolved:\n{notes}" if notes else ""
    # ... rest of supervisor
```

## Swarm Pattern Failures

| Symptom | Likely Cause | Fix |
|---------|-------------|-----|
| Infinite handoff loop | Swarm ping-pong — no recursion guard | Track `handoff_count`, hard limit at 3 |
| Customer asked the same question twice | Context loss on handoff — Agent B doesn't know what Agent A did | Propagate resolution context via `Command.update` |
| Next agent sees malformed history | Lost messages during handoff — ToolMessage pairing broken | Ensure `Command.update` includes paired messages |
| Agent routes to wrong specialist | Routing accuracy issue | Add more specific guidance in system prompt |

### Swarm Ping-Pong — Fix

```python
def route_after_agent(state: SwarmState) -> str:
    if state.get("handoff_count", 0) >= 3:
        return "human_escalation"  # or END with a fallback
    # ... rest of routing logic
```

### Context Loss — Fix

When a handoff tool returns `Command`, include what was done in the update:

```python
@tool
def transfer_to_billing(reason: str, context: dict = None) -> Command:
    update = {"current_agent": "billing", "handoff_count": context["count"] + 1}
    return Command(goto="billing", update=update, graph=Command.PARENT)
```

## Subgraph Failures

| Symptom | Likely Cause | Fix |
|---------|-------------|-----|
| Subgraph doesn't remember previous calls | Per-invocation default | Compile with `checkpointer=True` for per-thread memory |
| Tool calls fail with checkpoint conflict | Parallel calls to per-thread subgraph | Add `ToolCallLimitMiddleware` |
| Parallel subgraph calls overwrite each other | Same namespace conflict | Use `create_sub_agent` wrapper with unique names |
| Subgraph crashes without recovery | Stateless mode (`checkpointer=False`) | Use per-invocation (default) for durability |
| Parent can't see subgraph state | Different checkpoint namespaces | Use Store for cross-boundary data |

## Persistence Failures

| Symptom | Likely Cause | Fix |
|---------|-------------|-----|
| `thread_id` too long error | Postgres column length limit | Keep under 255 chars, use UUID |
| State lost on restart | Using in-memory `MemorySaver` | Switch to `PostgresSaver` or `SqliteSaver` |
| Checkpoints growing unbounded | No retention policy | Add cron job to prune old checkpoints |
| `interrupt()` has no effect | No checkpointer configured | Compile parent graph with checkpointer |

## Stream and Event Failures

| Symptom | Likely Cause | Fix |
|---------|-------------|-----|
| Subgraph events not visible in stream | Using wrong event protocol | Use `stream.subgraphs` projection or filter by namespace |
| Stream hangs | Graph in infinite loop | Add recursion guard or max iteration limit |
| Event namespace is empty string | Parent-level event, not subgraph | Filter by namespace: `[]` = parent, `[...]` = subgraph |

## Debugging Techniques

### 1. Check the Trace

LangSmith traces show every node execution, including:
- Input/output state per node
- Tool calls and their results
- Routing decisions (from structured output)
- Handoff chains (from Command objects)

### 2. Isolate the Subgraph

Test any subgraph independently before composing it:

```python
# Test subgraph in isolation
result = subgraph.invoke({"topic": "test"})
print(result)
```

### 3. Reduce Parallelism

When debugging, set `max_concurrency=1` to serialize execution:

```python
result = graph.invoke(inputs, {"max_concurrency": 1})
```

### 4. Add Logging to Nodes

Wrap nodes with print statements or pass through LangSmith:

```python
def debug_node(state: State) -> dict:
    print(f"enter {node_name}: keys={list(state.keys())}")
    result = actual_node(state)
    print(f"exit {node_name}: keys={list(result.keys())}")
    return result
```

### 5. Use `get_state` for Subgraph Inspection

```python
state = graph.get_state(config, subgraphs=True)
for task in state.tasks:
    if task.state:
        print(f"Subgraph state: {task.state}")
```

## Verification Checklist

When deploying a new multi-agent system, verify these in order:

- [ ] Single agent works independently (no routing dependencies)
- [ ] Supervisor/swarm routes correctly for single-domain requests
- [ ] Multi-domain requests are decomposed and handled completely
- [ ] Handoffs preserve context (no repeated questions)
- [ ] Recursion guard prevents infinite loops (supervisor and swarm)
- [ ] All errors are caught and produce graceful degradation
- [ ] Persistence works across thread resumptions
- [ ] Subgraphs are independently testable
- [ ] Eval dataset covers all routing paths
- [ ] LangSmith traces are readable and complete

#!/usr/bin/env python3
"""
LangGraph Eval Generator.

Generate evaluation datasets and run evaluators for multi-agent systems.
Supports two modes:
  1. Dataset creation — build a LangSmith eval dataset from a JSON/YAML spec
  2. Evaluator generation — produce Python code for routing accuracy and
     resolution coverage evaluators

Usage:
    # Generate a dataset from a spec file
    python lg-eval-generator.py dataset --name my-evals --spec examples.json

    # Generate evaluator code
    python lg-eval-generator.py evaluator --name my-evals --output ./evals

    # Generate an example spec file to fill in
    python lg-eval-generator.py example-spec --output ./examples.json
"""

import argparse
import json
import os
from typing import Any, Dict, List


EXAMPLE_SPEC = [
    {
        "question": "I need to change my payment method to a credit card.",
        "expected_agents": ["billing"],
        "must_mention": ["payment", "credit card"],
    },
    {
        "question": "My SSO integration is returning error code SAML-401.",
        "expected_agents": ["tech_support"],
        "must_mention": ["SSO", "SAML"],
    },
    {
        "question": "I want to upgrade to Enterprise and also fix my broken SSO.",
        "expected_agents": ["tech_support", "account"],
        "must_mention": ["SSO", "upgrade"],
    },
    {
        "question": "Can you tell me who my account manager is?",
        "expected_agents": ["account"],
        "must_mention": ["account manager"],
    },
    {
        "question": "What's the status of my refund?",
        "expected_agents": ["billing"],
        "must_mention": ["refund"],
    },
    {
        "question": "Reset my password and downgrade my plan.",
        "expected_agents": ["tech_support", "account"],
        "must_mention": ["password", "downgrade"],
    },
]


def generate_dataset(name: str, spec_path: str, output_path: str):
    """Generate LangSmith eval dataset creation script from a spec file."""
    with open(spec_path) as f:
        if spec_path.endswith(".json"):
            examples = json.load(f)
        else:
            # Assume JSON for simplicity — YAML support can be added
            examples = json.loads(f.read())

    # Validate
    for i, ex in enumerate(examples):
        if "question" not in ex:
            raise ValueError(f"Example {i} missing 'question'")
        if "expected_agents" not in ex:
            raise ValueError(f"Example {i} missing 'expected_agents'")

    dataset_code = f'''\
\"\"\"
Eval dataset: {name}

Generated by lg-eval-generator.py
Run this script to create the LangSmith dataset and run evaluators.
\"\"\"
from langsmith import Client, evaluate
from openevals.llm import create_llm_as_judge


# ── Dataset ────────────────────────────────────────────────────────────────

EXAMPLES = {json.dumps(examples, indent=2)}


def create_dataset(client: Client, dataset_name: str = "{name}"):
    \"\"\"Create or update the LangSmith eval dataset.\"\"\"
    try:
        dataset = client.create_dataset(
            dataset_name=dataset_name,
            description="Multi-agent routing and resolution evaluation dataset",
        )
    except Exception:
        # Dataset may already exist — use it
        dataset = [d for d in client.list_datasets() if d.name == dataset_name][0]

    inputs = [{{"question": ex["question"]}} for ex in EXAMPLES]
    outputs = [
        {{
            "expected_agents": ex.get("expected_agents", []),
            "must_mention": ex.get("must_mention", []),
        }}
        for ex in EXAMPLES
    ]

    client.create_examples(
        dataset_id=dataset.id,
        inputs=inputs,
        outputs=outputs,
    )
    print(f"Dataset '{{dataset_name}}' ready with {{len(EXAMPLES)}} examples")
    return dataset_name


# ── Evaluators ─────────────────────────────────────────────────────────────

ROUTING_QUALITY_PROMPT = """\\\\
Customer query: {inputs[question]}
Expected domains: {reference_outputs[expected_agents]}
Agent response: {outputs[final_response]}
Resolution notes: {outputs[resolution_notes]}

Rate 0.0-1.0 on whether the correct specialist agents handled the request
and the response fully addressed the customer's needs.
Return ONLY: {{"score": <float>, "reasoning": "<explanation>"}}"""

routing_judge = create_llm_as_judge(
    prompt=ROUTING_QUALITY_PROMPT,
    model="anthropic:claude-sonnet-4-5-20250929",
    feedback_key="routing_quality",
)


def resolution_coverage(inputs: dict, outputs: dict, reference_outputs: dict) -> dict:
    \"\"\"Did the agents address all parts of the customer's request?\"\"\"
    text = outputs.get("final_response", "").lower()
    notes = " ".join(outputs.get("resolution_notes", [])).lower()
    combined = text + " " + notes
    must_mention = reference_outputs.get("must_mention", [])
    hits = sum(1 for t in must_mention if t.lower() in combined)
    return {{
        "key": "resolution_coverage",
        "score": hits / len(must_mention) if must_mention else 1.0,
    }}


def agent_routing_accuracy(inputs: dict, outputs: dict, reference_outputs: dict) -> dict:
    \"\"\"Were the correct agents invoked?\"\"\"
    notes = " ".join(outputs.get("resolution_notes", [])).lower()
    expected = reference_outputs.get("expected_agents", [])
    hits = sum(1 for agent in expected if agent.lower() in notes)
    return {{
        "key": "routing_accuracy",
        "score": hits / len(expected) if expected else 1.0,
    }}


# ── Target Functions ───────────────────────────────────────────────────────

def supervisor_target(inputs: dict) -> dict:
    \"\"\"Invoke supervisor-pattern graph and return results for eval.

    Replace this with your actual graph invocation.
    \"\"\"
    from langchain_core.messages import HumanMessage

    # Import your graph — adjust the import path
    # from my_project.supervisor_graph import supervisor_graph

    # result = supervisor_graph.invoke({{
    #     "messages": [HumanMessage(content=inputs["question"])],
    #     "current_agent": "",
    #     "resolution_notes": [],
    # }})
    # return {{
    #     "final_response": result["messages"][-1].content,
    #     "resolution_notes": result.get("resolution_notes", []),
    # }}

    raise NotImplementedError("Replace with your supervisor graph invocation")


def swarm_target(inputs: dict) -> dict:
    \"\"\"Invoke swarm-pattern graph and return results for eval.

    Replace this with your actual graph invocation.
    \"\"\"
    from langchain_core.messages import HumanMessage

    # Import your graph — adjust the import path
    # from my_project.swarm_graph import swarm_graph

    # result = swarm_graph.invoke({{
    #     "messages": [HumanMessage(content=inputs["question"])],
    #     "current_agent": "",
    #     "resolution_notes": [],
    # }})
    # return {{
    #     "final_response": result["messages"][-1].content,
    #     "resolution_notes": result.get("resolution_notes", []),
    # }}

    raise NotImplementedError("Replace with your swarm graph invocation")


def run_evaluations():
    \"\"\"Run both supervisor and swarm evaluations for comparison.\"\"\"
    client = Client()

    dataset_name = create_dataset(client)

    print("\\\\nRunning supervisor evaluation...")
    supervisor_results = evaluate(
        supervisor_target,
        data=dataset_name,
        evaluators=[routing_judge, resolution_coverage, agent_routing_accuracy],
        experiment_prefix="supervisor-v1",
        max_concurrency=2,
        client=client,
    )

    print("\\\\nRunning swarm evaluation...")
    swarm_results = evaluate(
        swarm_target,
        data=dataset_name,
        evaluators=[routing_judge, resolution_coverage, agent_routing_accuracy],
        experiment_prefix="swarm-v1",
        max_concurrency=2,
        client=client,
    )

    print("\\\\n=== Results ===")
    print(f"Supervisor: {{supervisor_results}}")
    print(f"Swarm: {{swarm_results}}")


if __name__ == "__main__":
    import sys
    if "--compare" in sys.argv:
        run_evaluations()
    else:
        create_dataset(Client())
        print("Dataset created. Run with --compare to evaluate patterns.")
'''

    output_path = output_path or f"{name}_eval.py"
    with open(output_path, "w") as f:
        f.write(dataset_code)

    print(f"Eval script generated: {output_path}")
    print(f"Dataset name: {name}")
    print(f"Examples: {len(examples)}")


def generate_evaluator(name: str, output_dir: str):
    """Generate standalone evaluator module."""
    os.makedirs(output_dir, exist_ok=True)
    path = os.path.join(output_dir, f"{name}_evaluators.py")

    code = '''\
"""Multi-agent evaluators for routing accuracy and resolution coverage."""

from typing import Any, Dict


def resolution_coverage(
    inputs: Dict[str, Any],
    outputs: Dict[str, Any],
    reference_outputs: Dict[str, Any],
) -> Dict[str, Any]:
    """Evaluate whether all required topics were addressed in the response.

    Args:
        inputs: The original input (e.g., {"question": "..."})
        outputs: The graph output (e.g., {"final_response": "...", "resolution_notes": [...]})
        reference_outputs: Expected results (e.g., {"must_mention": ["...", "..."]})

    Returns:
        {"key": "resolution_coverage", "score": 0.0-1.0}
    """
    text = outputs.get("final_response", "").lower()
    notes = " ".join(outputs.get("resolution_notes", [])).lower()
    combined = text + " " + notes

    must_mention = reference_outputs.get("must_mention", [])
    if not must_mention:
        return {"key": "resolution_coverage", "score": 1.0}

    hits = sum(1 for term in must_mention if term.lower() in combined)
    return {"key": "resolution_coverage", "score": hits / len(must_mention)}


def agent_routing_accuracy(
    inputs: Dict[str, Any],
    outputs: Dict[str, Any],
    reference_outputs: Dict[str, Any],
) -> Dict[str, Any]:
    """Evaluate whether the correct specialist agents were invoked.

    Args:
        inputs: The original input
        outputs: Graph output with resolution_notes containing "AgentName: ..." entries
        reference_outputs: Expected agents (e.g., {"expected_agents": ["billing", "tech"]})

    Returns:
        {"key": "routing_accuracy", "score": 0.0-1.0}
    """
    notes = " ".join(outputs.get("resolution_notes", [])).lower()
    expected = reference_outputs.get("expected_agents", [])
    if not expected:
        return {"key": "routing_accuracy", "score": 1.0}

    hits = sum(1 for agent in expected if agent.lower() in notes)
    return {"key": "routing_accuracy", "score": hits / len(expected)}


def handoff_chain_length(
    inputs: Dict[str, Any],
    outputs: Dict[str, Any],
    reference_outputs: Dict[str, Any] = None,
) -> Dict[str, Any]:
    """Measure the number of handoffs in a swarm pattern execution.

    Long chains (>3) indicate routing problems or ping-pong behavior.

    Returns:
        {"key": "handoff_chain_length", "score": N} where N is the handoff count
    """
    count = outputs.get("handoff_count", 0)
    # Score = 1.0 if <= 3 handoffs, decreasing linearly after that
    score = min(1.0, 3.0 / max(count, 1)) if count > 0 else 1.0
    return {"key": "handoff_chain_length", "score": score, "info": {"count": count}}


def routing_efficiency(
    inputs: Dict[str, Any],
    outputs: Dict[str, Any],
    reference_outputs: Dict[str, Any] = None,
) -> Dict[str, Any]:
    """Evaluate routing efficiency — token cost vs accuracy tradeoff.

    Returns:
        {"key": "routing_efficiency", "score": 0.0-1.0}
    """
    # This is a placeholder — real implementation would compare token counts
    # from LangSmith traces against routing accuracy
    messages = outputs.get("messages", [])
    total_tokens = sum(
        getattr(m, "usage_metadata", {}).get("total_tokens", 0)
        for m in messages
        if hasattr(m, "usage_metadata")
    )
    return {"key": "routing_efficiency", "score": 1.0, "info": {"estimated_tokens": total_tokens}}
'''
    with open(path, "w") as f:
        f.write(code)
    print(f"Evaluator module generated: {path}")


def generate_example_spec(output_path: str):
    """Generate an example JSON spec file to fill in."""
    with open(output_path, "w") as f:
        json.dump(EXAMPLE_SPEC, f, indent=2)
    print(f"Example spec generated: {output_path}")
    print("Fill in the question/expected_agents/must_mention fields for your domain.")


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="LangGraph Eval Generator")
    subparsers = parser.add_subparsers(dest="mode", help="Mode: dataset, evaluator, or example-spec")

    # Dataset mode
    ds_parser = subparsers.add_parser("dataset", help="Generate eval dataset creation script")
    ds_parser.add_argument("--name", required=True, help="Dataset name")
    ds_parser.add_argument("--spec", required=True, help="Path to JSON spec file")
    ds_parser.add_argument("--output", default=None, help="Output Python file path")

    # Evaluator mode
    ev_parser = subparsers.add_parser("evaluator", help="Generate evaluator module")
    ev_parser.add_argument("--name", required=True, help="Evaluator name prefix")
    ev_parser.add_argument("--output", default=".", help="Output directory")

    # Example spec mode
    ex_parser = subparsers.add_parser("example-spec", help="Generate example JSON spec")
    ex_parser.add_argument("--output", default="eval_examples.json", help="Output path")

    args = parser.parse_args()

    if args.mode == "dataset":
        generate_dataset(args.name, args.spec, args.output)
    elif args.mode == "evaluator":
        generate_evaluator(args.name, args.output)
    elif args.mode == "example-spec":
        generate_example_spec(args.output)
    else:
        parser.print_help()

#!/usr/bin/env python3
"""
LangGraph Supervisor Pattern Scaffold Generator.

Generates a complete supervisor-based multi-agent project with:
- State definitions (MultiAgentState with routing fields)
- Routing decision schema (Pydantic)
- Supervisor node with structured output
- Specialist agent wrappers
- Graph assembly with conditional edges
- Main entry point and example invocation

Usage:
    python lg-supervisor-scaffold.py --name customer-service --agents billing,tech,account
    python lg-supervisor-scaffold.py --name research --agents search,summarize,fact-check --output ./my-project
"""

import argparse
import os
from typing import List


def snake_case(name: str) -> str:
    return name.replace("-", "_").replace(" ", "_").lower()


def pascal_case(name: str) -> str:
    return "".join(word.capitalize() for word in name.replace("-", " ").replace("_", " ").split())


AGENT_TEMPLATE = """\
from langchain.agents import create_agent

{tool_defs}

{agent_name}_agent = create_agent(
    llm,
    tools=[{tool_list}],
    system_prompt="{system_prompt}",
)
"""

NODE_TEMPLATE = """\
def {agent_name}_node(state: {state_class}) -> dict:
    \"\"\"{agent_name} specialist node.\"\"\"
    result = {agent_name}_agent.invoke({{"messages": state["messages"]}})
    return {{
        "messages": result["messages"][-1:],
        "resolution_notes": [
            f"{agent_name_display}: {{result['messages'][-1].content[:200]}}"
        ],
    }}
"""


def generate_project(project_name: str, agents: List[str], output_dir: str):
    pname = snake_case(project_name)
    state_class = pascal_case(project_name) + "State"
    dir_path = os.path.join(output_dir, pname)
    os.makedirs(dir_path, exist_ok=True)

    # Generate agent names with display labels
    agent_names = [snake_case(a) for a in agents]
    agent_labels = [a.replace("-", " ").title() for a in agents]

    # state.py
    state_code = f'''\
\"\"\"State definitions for {project_name} multi-agent system.\"\"\"
from typing import Annotated, TypedDict
from langgraph.graph import MessagesState
from pydantic import BaseModel, Field
import operator


class {state_class}(MessagesState):
    """Shared state across all agents in the {project_name} system."""

    current_agent: str
    """Which specialist agent is currently active."""

    resolution_notes: Annotated[list[str], operator.add]
    """Audit trail of what each agent resolved. Use operator.add for parallel-safety."""

    handoff_count: int = 0
    """Recursion guard — incremented on each routing decision."""


class RoutingDecision(BaseModel):
    """Structured output schema for the supervisor routing node."""

    next_agent: str = Field(
        description="The next agent to handle the request: "
        "{agent_choices} or 'DONE'"
    )
    reasoning: str = Field(description="Why this agent was chosen")
'''

    agent_choices = ", ".join(f"'{n}'" for n in agent_names)
    state_code = state_code.replace("{agent_choices}", agent_choices)

    utils_code = f'''\
\"\"\"Utility functions for {project_name}.\"\"\"
from langgraph.types import Command
from langchain_core.tools import tool


def make_handoff_tool(target_agent: str, description: str):
    \"\"\"Factory that creates a handoff tool for transferring to another agent.\"\"\"

    @tool(f"transfer_to_{{target_agent}}")
    def handoff(reason: str) -> Command:
        \"\"\"Transfer the conversation to another specialist agent.\"\"\"
        return Command(
            goto=target_agent,
            update={{"current_agent": target_agent}},
            graph=Command.PARENT,
        )

    handoff.__doc__ = description
    return handoff
'''

    # graph.py
    agent_uppers = [snake_case(a).upper() for a in agents]
    fast_path_entries = "\n    ".join(
        f'FAST_PATH[{a!r}] = "{n}"'
        for a, n in zip(agents, agent_names)
    )

    node_funcs = "\n\n\n".join(
        NODE_TEMPLATE.format(
            agent_name=name,
            agent_name_display=label,
            state_class=state_class,
        )
        for name, label in zip(agent_names, agent_labels)
    )

    graph_code = f'''\
\"\"\"Graph assembly for {project_name} multi-agent system (supervisor pattern).\"\"\"

from langchain_core.messages import SystemMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

from .state import {state_class}, RoutingDecision

# Initialize LLM — swap provider as needed
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
routing_llm = llm.with_structured_output(RoutingDecision)

# Fast-path routing for unambiguous intents
FAST_PATH = {{}}
{fast_path_entries}


def supervisor(state: {state_class}) -> dict:
    \"\"\"Central routing node. Classifies intent and delegates to the appropriate specialist.\"\"\"

    # Try fast-path first
    if state["messages"]:
        last_msg = state["messages"][-1].content.lower()
        for keyword, agent in FAST_PATH.items():
            if keyword in last_msg:
                return {{"current_agent": agent}}

    # Full routing with context
    notes = "\\n".join(state.get("resolution_notes", []))
    history_context = f"\\n\\nAlready resolved:\\n{{notes}}" if notes else ""

    agent_descriptions = "\\n".join(
        f"- {{name}}: {{desc}}"
        for name, desc in [
            {agent_descriptions}
        ]
    )

    response = routing_llm.invoke([
        SystemMessage(
            content="You are a multi-agent supervisor. Analyze the conversation "
                    "and decide which specialist should handle the next step.\\n\\n"
                    f"Available agents:\\n{{agent_descriptions}}\\n"
                    "- DONE: the request has been fully addressed\\n\\n"
                    "Do NOT re-route to an agent that has already handled "
                    "its portion of the request." + history_context
        ),
        *state["messages"],
    ])
    return {{"current_agent": response.next_agent}}


# Specialist agents — import or define your agents here
# See templates in assets/templates/ for full agent definitions
{node_funcs}


def route_to_agent(state: {state_class}) -> str:
    \"\"\"Reads current_agent from state and returns the target node name.\"\"\"
    agent = state.get("current_agent", "DONE")
    if agent == "DONE" or state.get("handoff_count", 0) >= 5:
        return "end"
    return agent


def build_graph() -> StateGraph:
    \"\"\"Assemble and compile the supervisor multi-agent graph.\"\"\"
    builder = StateGraph({state_class})

    # Add nodes
    builder.add_node("supervisor", supervisor)
    {node_additions}

    # Wire edges
    builder.add_edge(START, "supervisor")

    builder.add_conditional_edges(
        "supervisor",
        route_to_agent,
        {{
            {route_map}
            "end": END,
        }},
    )

    # Each specialist returns to supervisor
    {return_edges}

    # Compile with checkpointer for persistence
    checkpointer = MemorySaver()
    return builder.compile(checkpointer=checkpointer)
'''

    node_additions = "\n    ".join(
        f'builder.add_node("{name}", {name}_node)'
        for name in agent_names
    )
    route_map = ",\n            ".join(
        f'            "{name}": "{name}"'
        for name in agent_names
    )
    return_edges = "\n    ".join(
        f'builder.add_edge("{name}", "supervisor")'
        for name in agent_names
    )
    agent_descriptions = ",\n        ".join(
        f'("{n}", "{l} specialist")'
        for n, l in zip(agent_names, agent_labels)
    )

    graph_code = graph_code.replace("{node_additions}", node_additions)
    graph_code = graph_code.replace("{route_map}", route_map)
    graph_code = graph_code.replace("{return_edges}", return_edges)
    graph_code = graph_code.replace("{agent_descriptions}", agent_descriptions)

    # main.py
    main_code = f'''\
\"\"\"{project_name} — Supervisor Multi-Agent System\"\"\"
from graph import build_graph


def main():
    graph = build_graph()

    config = {{"configurable": {{"thread_id": "example-1"}}}}

    user_query = input("What would you like help with? ")
    result = graph.invoke(
        {{
            "messages": [{{"role": "user", "content": user_query}}],
            "current_agent": "",
            "resolution_notes": [],
            "handoff_count": 0,
        }},
        config=config,
    )

    print("\\n=== Result ===")
    for msg in result["messages"]:
        if hasattr(msg, "content") and msg.content:
            print(f"{{msg.type}}: {{msg.content[:200]}}")

    if result.get("resolution_notes"):
        print("\\n=== Resolution Notes ===")
        for note in result["resolution_notes"]:
            print(f"  - {{note}}")


if __name__ == "__main__":
    main()
'''

    # Write files
    with open(os.path.join(dir_path, "state.py"), "w") as f:
        f.write(state_code)
    with open(os.path.join(dir_path, "utils.py"), "w") as f:
        f.write(utils_code)
    with open(os.path.join(dir_path, "graph.py"), "w") as f:
        f.write(graph_code)
    with open(os.path.join(dir_path, "main.py"), "w") as f:
        f.write(main_code)

    print(f"Project generated at: {dir_path}")
    print(f"Files: state.py, utils.py, graph.py, main.py")
    print(f"Agents: {', '.join(agent_names)}")


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Generate LangGraph supervisor pattern scaffold")
    parser.add_argument("--name", required=True, help="Project name (e.g., customer-service)")
    parser.add_argument("--agents", required=True, help="Comma-separated agent names (e.g., billing,tech,account)")
    parser.add_argument("--output", default=".", help="Output directory (default: current)")
    args = parser.parse_args()

    agents = [a.strip() for a in args.agents.split(",")]
    generate_project(args.name, agents, args.output)

#!/usr/bin/env python3
"""
LangGraph Swarm Pattern Scaffold Generator.

Generates a complete swarm-based multi-agent project with:
- State definitions with handoff tracking
- Handoff tool factory (Command-based)
- Triage agent for initial routing
- Specialist agents with domain tools + handoff tools
- Conditional routing with recursion guard
- Node wrappers for each agent

Usage:
    python lg-swarm-scaffold.py --name support --agents billing,tech,account
    python lg-swarm-scaffold.py --name triage --agents search,summarize --triage-only
"""

import argparse
import os
from typing import List


def snake_case(name: str) -> str:
    return name.replace("-", "_").replace(" ", "_").lower()


def pascal_case(name: str) -> str:
    return "".join(word.capitalize() for word in name.replace("-", " ").replace("_", " ").split())


SWARM_HANDOFF_TOOLS = """\
from langgraph.types import Command
from langchain_core.tools import tool


def make_handoff_tool(target_agent: str, description: str):
    \"\"\"Factory that creates a handoff tool for transferring to another agent.

    The tool returns a Command that tells LangGraph to navigate to a different
    node in the parent graph, updating current_agent and the handoff counter.
    \"\"\"

    @tool(f"transfer_to_{target_agent}")
    def handoff(reason: str) -> Command:
        \"\"\"Transfer the conversation to another specialist agent.\"\"\"
        return Command(
            goto=target_agent,
            update={"current_agent": target_agent},
            graph=Command.PARENT,
        )

    handoff.__doc__ = description
    return handoff
"""


def generate_swarm_project(project_name: str, agents: List[str], output_dir: str, triage_only: bool = False):
    pname = snake_case(project_name)
    state_class = pascal_case(project_name) + "State"
    dir_path = os.path.join(output_dir, pname)
    os.makedirs(dir_path, exist_ok=True)

    agent_names = [snake_case(a) for a in agents]
    agent_labels = [a.replace("-", " ").title() for a in agents]

    # state.py
    state_code = f'''\
\"\"\"State definitions for {project_name} swarm multi-agent system.\"\"\"
from typing import Annotated, TypedDict
from langgraph.graph import MessagesState
import operator


class {state_class}(MessagesState):
    """Shared state across all agents in the {project_name} swarm."""

    current_agent: str = ""
    """Which specialist agent is currently active. Empty = triage phase."""

    resolution_notes: Annotated[list[str], operator.add]
    """Audit trail of what each agent resolved."""

    handoff_count: int = 0
    """Recursion guard — incremented on each handoff. Hard limit at 3."""
'''

    # handoff_tools.py
    handoff_tool_defs = ""
    handoff_tool_imports = "\n".join(
        f'transfer_to_{name} = make_handoff_tool(\n'
        f'    "{name}",\n'
        f'    "Transfer to the {label.lower()} specialist for {label.lower()} issues.",\n'
        f')'
        for name, label in zip(agent_names, agent_labels)
    )
    handoff_code = SWARM_HANDOFF_TOOLS + "\n\n" + handoff_tool_imports

    # agents.py
    if triage_only:
        agent_code = f'''\
\"\"\"Agent definitions for {project_name} swarm.

Triage-only mode: the triage agent routes to specialists who handle the request.
Specialists may or may not have handoff tools depending on the use case.
\"\"\"
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from .handoff_tools import (
    {", ".join(f"transfer_to_{n}" for n in agent_names)}
)

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

# Triage agent — only routes, never answers
triage_agent = create_agent(
    llm,
    tools=[{", ".join(f"transfer_to_{n}" for n in agent_names)}],
    system_prompt=(
        "You are a triage agent. Analyze the request and transfer "
        "to the appropriate specialist using the transfer tools. "
        "Do NOT try to answer questions yourself — always transfer."
    ),
)
'''
    else:
        # Full swarm: each specialist gets handoff tools for all OTHER agents
        agent_handoff_imports = "\n".join(
            f'from .handoff_tools import transfer_to_{n}'
            for n in agent_names
        )
        agent_code = f'''\
\"\"\"Agent definitions for {project_name} swarm.\"\"\"
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
{agent_handoff_imports}

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

# Triage agent — only routes, never answers
triage_agent = create_agent(
    llm,
    tools=[{", ".join(f"transfer_to_{n}" for n in agent_names)}],
    system_prompt=(
        "You are a triage agent. Analyze the request and transfer "
        "to the appropriate specialist. Do NOT answer questions "
        "yourself — always transfer. If multiple issues exist, "
        "transfer to the most urgent one first."
    ),
)
'''

        for name, label in zip(agent_names, agent_labels):
            other_handoffs = [f"transfer_to_{n}" for n in agent_names if n != name]
            handoff_str = ",\n    ".join(other_handoffs)

            agent_code += f'''

# {label} specialist
{name}_agent = create_agent(
    llm,
    tools=[
        # Add domain tools here
        # e.g., lookup_{name}_info, do_{name}_action,
        {handoff_str},
    ],
    system_prompt=(
        "You are a {label.lower()} specialist. "
        "Help with {label.lower()} issues. "
        "If the customer has issues outside your domain, "
        "transfer to the appropriate specialist."
    ),
)
'''

    # graph.py
    route_code = f'''\
\"\"\"Graph assembly for {project_name} swarm multi-agent system.\"\"\"
from typing import Literal

from langchain_core.messages import AIMessage
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

from .state import {state_class}
from .agents import triage_agent, {", ".join(f"{name}_agent" for name in agent_names)}
'''

    route_code += f"""

# Node wrappers
def triage_node(state: {state_class}) -> Command:
    result = triage_agent.invoke({{"messages": state["messages"]}})
    return result  # Command from handoff tool

"""

    for name, label in zip(agent_names, agent_labels):
        route_code += f"""
def {name}_node(state: {state_class}) -> dict:
    result = {name}_agent.invoke({{"messages": state["messages"]}})
    return {{
        "messages": result["messages"][-1:],
        "resolution_notes": [f"{label}: {{result['messages'][-1].content[:200]}}"],
    }}

"""

    route_code += f"""
def route_after_agent(
    state: {state_class},
) -> Literal[{', '.join(f'"{n}"' for n in agent_names)}, "__end__"]:
    \"\"\"Route to the next agent or end based on state and handoff count.\"\"\"

    # Recursion guard
    if state.get("handoff_count", 0) >= 3:
        return "__end__"

    messages = state.get("messages", [])
    if messages:
        last_msg = messages[-1]
        if isinstance(last_msg, AIMessage) and not last_msg.tool_calls:
            return "__end__"  # No tool calls = done

    current = state.get("current_agent", "")
    if current in ({', '.join(f'"{n}"' for n in agent_names)}):
        return current
    return "__end__"


def build_graph() -> StateGraph:
    \"\"\"Assemble and compile the swarm multi-agent graph.\"\"\"
    builder = StateGraph({state_class})

    # Add nodes
    builder.add_node("triage", triage_node)
    {chr(10) + '    '.join(f'builder.add_node("{n}", {n}_node)' for n in agent_names)}

    # Wire edges
    builder.add_edge(START, "triage")

    # Each specialist can route to any other specialist or end
    for node in [{', '.join(f'"{n}"' for n in agent_names)}]:
        builder.add_conditional_edges(
            node,
            route_after_agent,
            [{', '.join(f'"{n}"' for n in agent_names)}, END],
        )

    checkpointer = MemorySaver()
    return builder.compile(checkpointer=checkpointer)
"""

    # Write files
    with open(os.path.join(dir_path, "state.py"), "w") as f:
        f.write(state_code)
    with open(os.path.join(dir_path, "handoff_tools.py"), "w") as f:
        f.write(handoff_code)
    with open(os.path.join(dir_path, "agents.py"), "w") as f:
        f.write(agent_code)
    with open(os.path.join(dir_path, "graph.py"), "w") as f:
        f.write(route_code)

    # Add Command import to graph.py
    graph_path = os.path.join(dir_path, "graph.py")
    with open(graph_path) as f:
        content = f.read()
    content = content.replace(
        "from .agents import",
        "from langgraph.types import Command\n\nfrom .agents import"
    )
    with open(graph_path, "w") as f:
        f.write(content)

    print(f"Swarm project generated at: {dir_path}")
    print(f"Files: state.py, handoff_tools.py, agents.py, graph.py")
    print(f"Agents: {', '.join(agent_names)}")
    if triage_only:
        print("Mode: triage-only (specialists handle requests without further handoffs)")


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Generate LangGraph swarm pattern scaffold")
    parser.add_argument("--name", required=True, help="Project name (e.g., support)")
    parser.add_argument("--agents", required=True, help="Comma-separated agent names (e.g., billing,tech,account)")
    parser.add_argument("--output", default=".", help="Output directory (default: current)")
    parser.add_argument("--triage-only", action="store_true",
                        help="Triage routes to specialists who handle without further handoffs")
    args = parser.parse_args()

    agents = [a.strip() for a in args.agents.split(",")]
    generate_swarm_project(args.name, agents, args.output, args.triage_only)