Mastering Langgraph

LangGraph Development Guide

Build stateful AI agents and workflows by defining graphs of nodes (steps) connected by edges (transitions).

Contents

• Quick Start
• Common Build Scenarios
• Core Principles
• Development Workflow
• Common Pitfalls
• Environment Setup
• Quick Verification
• API Essentials
• Next Steps

Quick Start

Minimal chatbot with memory:

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AnyMessage
from typing_extensions import TypedDict, Annotated
import operator

# 1. Define state
class State(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]  # Append mode

# 2. Define node
llm = ChatOpenAI(model="gpt-4")

def chat(state: State) -> dict:
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

# 3. Build graph
graph = StateGraph(State)
graph.add_node("chat", chat)
graph.add_edge(START, "chat")
graph.add_edge("chat", END)

# 4. Compile with memory
chain = graph.compile(checkpointer=InMemorySaver())

# 5. Invoke with thread_id for persistence
result = chain.invoke(
    {"messages": [HumanMessage(content="Hello!")]},
    config={"configurable": {"thread_id": "user-123"}}
)
print(result["messages"][-1].content)

Key patterns:

• Annotated[list, operator.add] — append to list instead of replace
• InMemorySaver() — enables memory across invocations
• thread_id — identifies conversation for persistence

Common Build Scenarios

Simple Chatbot / Q&A

The Quick Start above covers this. Add more nodes for preprocessing or postprocessing as needed.

Tool-Using Agent

Agent that calls external tools (APIs, calculators, search) in a loop until task complete. → See references/tool-agent-pattern.md

Structured Workflow

Multi-step pipeline with conditional branches, parallel execution, or prompt chaining. → See references/workflow-patterns.md

Agent with Long-Term Memory

Persist conversation across sessions, enable time-travel debugging, survive crashes. → See references/persistence-memory.md

Human-in-the-Loop

Pause for human approval, correction, or additional input mid-workflow. → See references/hitl-patterns.md

Debugging / Production Monitoring

Unit test nodes, visualize graphs, trace with LangSmith. → See references/debugging-monitoring.md

Multi-Agent Systems

Build supervisor or swarm-based multi-agent workflows with handoff tools. → See references/multi-agent-patterns.md

Production Deployment

Deploy to LangGraph Platform (cloud/self-hosted) or custom infrastructure. → See references/production-deployment.md

New to LangGraph?

Learn core concepts: State, Nodes, Edges, Graph APIs. → See references/core-api.md

Core Principles

1. Keep State Raw

Store facts, not formatted prompts. Each node can format data as needed.

# ✓ Good: raw data
class State(TypedDict):
    user_question: str
    retrieved_docs: list[str]
    intent: str

# ✗ Bad: pre-formatted
class State(TypedDict):
    full_prompt: str  # Mixes data with formatting

2. Single-Purpose Nodes

Each node does one thing. Name it descriptively.

# ✓ Good: clear responsibilities
graph.add_node("classify_intent", classify_intent)
graph.add_node("search_knowledge", search_knowledge)
graph.add_node("generate_response", generate_response)

3. Explicit Routing

Use conditional edges for decisions. Don't hide routing logic inside nodes.

def route_by_intent(state) -> str:
    if state["intent"] == "billing":
        return "billing_handler"
    return "general_handler"

graph.add_conditional_edges("classify", route_by_intent, 
    ["billing_handler", "general_handler"])

4. Use Aggregators for Lists

Any list field that accumulates values needs operator.add:

class State(TypedDict):
    messages: Annotated[list, operator.add]      # ✓ Appends
    current_step: str                             # Replaces (no annotation)

5. Handle Errors Deliberately

Development Workflow

1. Define Steps — Break task into discrete operations (each becomes a node) 2. Categorize Steps — LLM call? Data retrieval? Action? User input? 3. Design State — TypedDict with all needed fields; keep it raw 4. Implement Nodes — def node(state) -> dict for each step 5. Connect Graph — add_node(), add_edge(), add_conditional_edges() 6. Compile & Test — graph.compile(), test with sample inputs

Common Pitfalls

1. Forgetting operator.add on Lists

Symptom: Messages disappear, only last message retained.

# ✗ Wrong: messages: list[AnyMessage]
# ✓ Fix: messages: Annotated[list[AnyMessage], operator.add]

2. Missing thread_id for Memory

Symptom: Agent forgets previous turns.

# ✓ Fix: Always pass config with thread_id
chain.invoke(input, config={"configurable": {"thread_id": "unique-id"}})

3. Not Compiling Before Invoke

Symptom: AttributeError on graph object.

# ✗ Wrong: graph.invoke(input)
# ✓ Fix: chain = graph.compile(); chain.invoke(input)

4. Non-Deterministic Nodes Without @task

Symptom: Different results on resume from checkpoint.

from langgraph.func import task

@task  # Wrap for durable execution
def fetch_data(state):
    return {"data": requests.get(url).json()}

5. Circular Imports with Type Hints

Symptom: ImportError when defining state classes.

# ✓ Fix: Use string annotations
from __future__ import annotations

Environment Setup

# Core
pip install -U langgraph

# LLM providers (pick one or more)
pip install langchain-openai
pip install langchain-anthropic

# Production persistence
pip install langgraph-checkpoint-postgres

# Observability
pip install langsmith

Environment variables:

export OPENAI_API_KEY="sk-..."
export ANTHROPIC_API_KEY="sk-ant-..."
export LANGSMITH_API_KEY="ls-..."
export LANGSMITH_TRACING=true

Quick Verification

Before Building

• [ ] python -c "import langgraph; print(langgraph.__version__)" works
• [ ] LLM API key set (OPENAI_API_KEY or ANTHROPIC_API_KEY)
• [ ] Optional: LANGSMITH_API_KEY for tracing

After Building

• [ ] Graph compiles without error: chain = graph.compile()
• [ ] Visualization renders: print(chain.get_graph().draw_mermaid())
• [ ] Invoke succeeds with sample input: chain.invoke({...})
• [ ] Lists accumulate correctly (verify operator.add annotations)
• [ ] Memory persists across invocations (test same thread_id twice)
• [ ] Conditional routing works as expected (test each branch)

API Essentials

# Imports
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from typing_extensions import TypedDict, Annotated
import operator

# State with append-mode list
class State(TypedDict):
    messages: Annotated[list, operator.add]

# Node signature
def node(state: State) -> dict:
    return {"messages": [new_message]}

# Graph construction
graph = StateGraph(State)
graph.add_node("name", node_fn)
graph.add_edge(START, "name")
graph.add_edge("name", END)

# Conditional routing
graph.add_conditional_edges("from", router_fn, ["option1", "option2", END])

# Compile and run
chain = graph.compile(checkpointer=InMemorySaver())
result = chain.invoke(input, config={"configurable": {"thread_id": "id"}})

# Visualization
print(chain.get_graph().draw_mermaid())

For detailed API reference → See references/core-api.md

Next Steps

• Tool agents: references/tool-agent-pattern.md
• Workflows: references/workflow-patterns.md
• Persistence: references/persistence-memory.md
• Human-in-the-loop: references/hitl-patterns.md
• Testing/Monitoring: references/debugging-monitoring.md
• Multi-agent: references/multi-agent-patterns.md
• Production: references/production-deployment.md
• Core concepts: references/core-api.md
• Official docs: references/official-resources.md

Mastering LangGraph Agent Skill

    

Build stateful AI agents and agentic workflows with LangGraph in Python. This skill provides comprehensive guidance for tool-using agents, branching workflows, conversation memory, human-in-the-loop oversight, multi-agent systems, and production deployment.

Table of Contents

• Overview
• Key Concepts
• Quick Start
• Reference Documentation
• Requirements
• Installation
• License

Overview

This skill covers essential LangGraph patterns for building production-ready AI agents:

Key Concepts

Quick Start

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AnyMessage
from typing_extensions import TypedDict, Annotated
import operator

# Define state with append-mode messages
class State(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]

# Create chat node
llm = ChatOpenAI(model="gpt-4")

def chat(state: State) -> dict:
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

# Build and compile graph
graph = StateGraph(State)
graph.add_node("chat", chat)
graph.add_edge(START, "chat")
graph.add_edge("chat", END)

chain = graph.compile(checkpointer=InMemorySaver())

# Invoke with thread_id for memory persistence
result = chain.invoke(
    {"messages": [HumanMessage(content="Hello!")]},
    config={"configurable": {"thread_id": "user-123"}}
)

Reference Documentation

This skill includes detailed reference guides for each major topic:

Requirements

• Python: >= 3.9
• LangGraph: pip install langgraph
• LLM Provider: OpenAI, Anthropic, or other supported providers

Installing with Skilz (Universal Installer)

The recommended way to install this skill across different AI coding agents is using the skilz universal installer. This skill supports the Agent Skill Standard, which means it works with 14+ coding agents including Claude Code, OpenAI Codex, Cursor, and Gemini.

Install Skilz

pip install skilz

Install from Git

You can use either -g or --git with HTTPS or SSH URLs:

# HTTPS URL
skilz install -g https://github.com/SpillwaveSolutions/mastering-langgraph-agent-skill

# SSH URL
skilz install --git git@github.com:SpillwaveSolutions/mastering-langgraph-agent-skill.git

Install from SkillzWave Marketplace

skilz install SpillwaveSolutions_mastering-langgraph-agent-skill/mastering-langgraph

Agent-Specific Installation

Claude Code

# Install to user home (available in all projects)
skilz install -g https://github.com/SpillwaveSolutions/mastering-langgraph-agent-skill

# Install to current project only
skilz install -g https://github.com/SpillwaveSolutions/mastering-langgraph-agent-skill --project

OpenCode

# User-level install
skilz install -g https://github.com/SpillwaveSolutions/mastering-langgraph-agent-skill --agent opencode

# Project-level install
skilz install -g https://github.com/SpillwaveSolutions/mastering-langgraph-agent-skill --project --agent opencode

Gemini CLI

# Project-level install (Gemini only supports project level)
skilz install -g https://github.com/SpillwaveSolutions/mastering-langgraph-agent-skill --agent gemini

OpenAI Codex

# User-level install
skilz install -g https://github.com/SpillwaveSolutions/mastering-langgraph-agent-skill --agent codex

# Project-level install
skilz install -g https://github.com/SpillwaveSolutions/mastering-langgraph-agent-skill --project --agent codex

Other Supported Agents

Skilz supports 14+ coding agents including Windsurf, Qwen Code, Aidr, and more. For the full list of supported platforms, visit:

• SkillzWave Platforms
• skilz-cli GitHub Repository
• Skill Listing

License

MIT

---

SkillzWave - Largest Agentic Marketplace for AI Agent Skills | SpillWave - Leaders in AI Agent Development

LangGraph Core API

Foundational concepts for building LangGraph applications.

Contents

• State
• Nodes
• Edges
• Graph Construction
• Compile and Invoke
• Graph API vs Functional API

---

State

State is a shared mutable dictionary that carries data through the workflow. All nodes read from and write to this state.

Defining State

Use TypedDict to define your state schema:

from typing_extensions import TypedDict, Annotated
from langchain_core.messages import AnyMessage
import operator

class MyState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    query: str
    results: list[str]
    step_count: int

Aggregation Modes

# Replace mode (default)
class State(TypedDict):
    current_step: str  # Each update overwrites

# Append mode  
class State(TypedDict):
    messages: Annotated[list, operator.add]  # Each update appends

State Design Principles

Keep state raw: Store data, not formatted prompts.

# ✓ Good
class State(TypedDict):
    user_question: str
    retrieved_docs: list[str]
    intent: str

# ✗ Bad  
class State(TypedDict):
    full_prompt: str  # Mixes data with formatting

---

Nodes

Nodes are Python functions that perform one step of the workflow.

Node Signature

def my_node(state: MyState) -> dict:
    # Read from state
    query = state["query"]
    
    # Do work
    result = process(query)
    
    # Return updates (only changed fields)
    return {"results": [result]}

Return Values

1. Dictionary (most common): Updates merged into state

def node(state) -> dict:
    return {"field": "new_value"}

2. Command: Update state AND control routing

from langgraph.types import Command

def node(state) -> Command:
    return Command(
        update={"field": "value"},
        goto="next_node"  # Override normal edge
    )

Node Best Practices

1. Single purpose: One node = one responsibility 2. Descriptive names: classify_intent not step1 3. No side effects: Don't modify global state 4. Handle errors: Return error info in state or raise for debugging

---

Edges

Edges define transitions between nodes.

Special Markers

from langgraph.graph import START, END

# START: Entry point (where graph begins)
# END: Exit point (where graph terminates)

Unconditional Edges

graph.add_edge("node_a", "node_b")
graph.add_edge(START, "first_node")
graph.add_edge("last_node", END)

Conditional Edges

def route_by_type(state: MyState) -> str:
    if state["type"] == "urgent":
        return "urgent_handler"
    elif state["type"] == "normal":
        return "normal_handler"
    return END

graph.add_conditional_edges(
    "classifier",
    route_by_type,
    ["urgent_handler", "normal_handler", END]
)

Conditional Edge Patterns

Binary branch:

def should_continue(state) -> str:
    return END if state["done"] else "process_more"

graph.add_conditional_edges("check", should_continue, ["process_more", END])

Loop back:

def check_complete(state) -> str:
    return END if state["iterations"] >= 3 else "iterate_again"

graph.add_conditional_edges("process", check_complete, ["iterate_again", END])
graph.add_edge("iterate_again", "process")  # Creates cycle

---

Graph Construction

from langgraph.graph import StateGraph, START, END

# 1. Initialize with state type
graph = StateGraph(MyState)

# 2. Add nodes
graph.add_node("node_a", node_a_function)
graph.add_node("node_b", node_b_function)

# 3. Add edges
graph.add_edge(START, "node_a")
graph.add_edge("node_a", "node_b")
graph.add_conditional_edges("node_b", router_fn, ["node_c", END])
graph.add_edge("node_c", END)

# 4. Compile
chain = graph.compile()

Complete Example

from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict

class CounterState(TypedDict):
    count: int
    message: str

def increment(state: CounterState) -> dict:
    return {"count": state["count"] + 1}

def check_limit(state: CounterState) -> str:
    return END if state["count"] >= 5 else "increment"

graph = StateGraph(CounterState)
graph.add_node("increment", increment)
graph.add_edge(START, "increment")
graph.add_conditional_edges("increment", check_limit, ["increment", END])

chain = graph.compile()
result = chain.invoke({"count": 0, "message": ""})
# result["count"] == 5

---

Compile and Invoke

Basic Compilation

chain = graph.compile()

With Persistence

from langgraph.checkpoint.memory import InMemorySaver

chain = graph.compile(checkpointer=InMemorySaver())

Invoking

# Without persistence
result = chain.invoke({"query": "hello", "messages": []})

# With persistence
result = chain.invoke(
    {"query": "hello", "messages": []},
    config={"configurable": {"thread_id": "user-123"}}
)

Testing Specific Nodes

node = chain.nodes["node_name"]
output = node.invoke({"query": "test"})

Visualization

print(chain.get_graph().draw_mermaid())
chain.get_graph().draw_mermaid_png(output_file_path="graph.png")

---

Graph API vs Functional API

Graph API (Declarative)

graph = StateGraph(State)
graph.add_node("step1", step1_fn)
graph.add_node("step2", step2_fn)
graph.add_edge(START, "step1")
graph.add_edge("step1", "step2")
graph.add_edge("step2", END)
chain = graph.compile()

Functional API (Imperative)

from langgraph.func import entrypoint, task

@task
def step1(inputs):
    return {"result": process1(inputs)}

@task
def step2(inputs):
    return {"result": process2(inputs)}

@entrypoint()
def workflow(inputs):
    r1 = step1(inputs)
    r2 = step2(r1)
    return r2

Both compile to the same execution engine.

Debugging and Monitoring

Test nodes, visualize graphs, trace executions, and diagnose issues.

Contents

• Unit Testing Nodes
• Graph Visualization
• LangSmith Tracing
• Step-by-Step Debugging
• Common Issues and Fixes

---

Unit Testing Nodes

Direct Testing

def test_classify_node():
    state = {"query": "billing help", "messages": []}
    result = classify_node(state)
    
    assert "classification" in result
    assert result["classification"] in ["billing", "technical", "general"]

Via Compiled Graph

def test_node_via_graph():
    chain = graph.compile()
    node = chain.nodes["classify"]
    result = node.invoke({"query": "billing", "messages": []})
    assert result["classification"] == "billing"

Mocking LLM

from unittest.mock import Mock, patch

def test_with_mock_llm():
    mock_llm = Mock()
    mock_llm.invoke.return_value.content = "Mocked response"
    
    with patch("my_module.llm", mock_llm):
        result = generate_response({"query": "test"})
    
    assert result["response"] == "Mocked response"

---

Graph Visualization

Mermaid (Text)

chain = graph.compile()
print(chain.get_graph().draw_mermaid())

PNG Image

chain.get_graph().draw_mermaid_png(output_file_path="graph.png")

Jupyter

from IPython.display import Image, display
display(Image(chain.get_graph().draw_mermaid_png()))

---

LangSmith Tracing

Setup

export LANGSMITH_API_KEY="ls-..."
export LANGSMITH_TRACING=true
export LANGSMITH_PROJECT="my-agent"

Selective Tracing

import langsmith as ls

with ls.tracing_context(project_name="experiment-1", enabled=True):
    result = chain.invoke(input_state, config=config)

Adding Metadata

config = {
    "configurable": {"thread_id": "user-123"},
    "metadata": {"user_id": "123", "version": "1.0.0"}
}
result = chain.invoke(input_state, config=config)

---

Step-by-Step Debugging

Stream Each Step

for event in chain.stream(input_state, config=config):
    print(f"Node: {list(event.keys())}")
    print(f"Output: {event}")

Interrupt After Node

chain = graph.compile(
    checkpointer=InMemorySaver(),
    interrupt_after=["classify"]
)

partial = chain.invoke(input_state, config=config)
state = chain.get_state(config)
print(f"Classification: {state.values.get('classification')}")

# Continue
final = chain.invoke(None, config=config)

---

Common Issues and Fixes

1. Messages Disappearing

Symptom: Only last message retained.

# ✗ Wrong
messages: list[AnyMessage]

# ✓ Fix
messages: Annotated[list[AnyMessage], operator.add]

2. No Memory Between Turns

Symptom: Agent forgets previous turns.

# ✓ Fix: Add checkpointer + thread_id
chain = graph.compile(checkpointer=InMemorySaver())
chain.invoke(input, config={"configurable": {"thread_id": "id"}})

3. Wrong Node Executed

Debug: Log router decision.

def route_debug(state) -> str:
    result = "node_a" if state["condition"] else "node_b"
    print(f"Routing: {state['condition']} → {result}")
    return result

4. Tool Not Called

Causes: Tool not bound, poor docstring.

# Check binding
print(model_with_tools.kwargs.get("tools"))

# Improve docstring
@tool
def search(query: str) -> str:
    """Search knowledge base. Use for ANY factual question."""
    return results

5. Infinite Loop

Fix: Add iteration limit.

def should_continue(state) -> str:
    if state.get("iterations", 0) >= 10:
        return END
    # ... normal logic

6. State Update Not Applied

Cause: Key mismatch.

class State(TypedDict):
    result: str

def node(state) -> dict:
    return {"results": "value"}  # ✗ Typo: "results" vs "result"

Debug Checklist

• [ ] Visualize: chain.get_graph().draw_mermaid()
• [ ] Check state keys match node returns
• [ ] Verify operator.add on list fields
• [ ] Confirm checkpointer + thread_id for memory
• [ ] Enable LangSmith tracing
• [ ] Use stream() for step-by-step view

Human-in-the-Loop Patterns

Pause agents for human approval, correction, or additional input.

Contents

• Overview
• The interrupt() Function
• Approval Workflows
• Correction Workflows
• Requesting Additional Input
• Best Practices

---

Overview

Requirement: HITL requires a checkpointer for state persistence.

---

The interrupt() Function

from langgraph.types import interrupt

def review_node(state):
    draft = state["draft"]
    
    # Pause and ask human
    human_response = interrupt({
        "draft": draft,
        "question": "Approve or provide feedback."
    })
    
    # Resumes here after human responds
    if human_response.get("approved"):
        return {"status": "approved"}
    return {"feedback": human_response.get("feedback")}

Resuming After Interrupt

config = {"configurable": {"thread_id": "review-123"}}

# Initial invocation (pauses at interrupt)
result = chain.invoke({"draft": "Hello world..."}, config=config)
# Returns interrupt payload

# Human provides response
human_input = {"approved": True}
final = chain.invoke(human_input, config=config)

---

Approval Workflows

Draft → Review → Send

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import interrupt
from typing_extensions import TypedDict

class EmailState(TypedDict):
    request: str
    draft: str
    approved: bool
    sent: bool

def generate_draft(state: EmailState) -> dict:
    draft = llm.invoke(f"Write email for: {state['request']}")
    return {"draft": draft.content}

def human_review(state: EmailState) -> dict:
    response = interrupt({
        "draft": state["draft"],
        "action": "approve_or_edit"
    })
    
    if response.get("approved"):
        return {"approved": True}
    elif response.get("edited_draft"):
        return {"draft": response["edited_draft"], "approved": True}
    return {"approved": False}

def send_email(state: EmailState) -> dict:
    if state["approved"]:
        send(state["draft"])
        return {"sent": True}
    return {"sent": False}

def route_after_review(state: EmailState) -> str:
    return "send" if state["approved"] else "__end__"

graph = StateGraph(EmailState)
graph.add_node("generate", generate_draft)
graph.add_node("review", human_review)
graph.add_node("send", send_email)

graph.add_edge(START, "generate")
graph.add_edge("generate", "review")
graph.add_conditional_edges("review", route_after_review, ["send", "__end__"])
graph.add_edge("send", END)

chain = graph.compile(checkpointer=InMemorySaver())

---

Correction Workflows

Rollback and Retry

# Find checkpoint before mistake
for state in chain.get_state_history(config):
    print(f"{state.config['configurable']['checkpoint_id']}: {state.next}")

# Resume from earlier checkpoint
corrected_config = {
    "configurable": {
        "thread_id": "task-123",
        "checkpoint_id": "checkpoint-before-mistake"
    }
}

chain.update_state(corrected_config, {"classification": "correct_value"})
result = chain.invoke(None, config=corrected_config)

Mid-Execution Verification

def classification_node(state) -> dict:
    classification = llm.invoke(f"Classify: {state['input']}").content
    
    if state.get("confidence", 1.0) < 0.8:
        human_check = interrupt({
            "proposed": classification,
            "question": "Is this correct?"
        })
        if not human_check.get("correct"):
            classification = human_check.get("corrected_value")
    
    return {"classification": classification}

---

Requesting Additional Input

def process_order(state) -> dict:
    if not state.get("shipping_address"):
        address = interrupt({
            "question": "Please provide shipping address",
            "required_fields": ["street", "city", "zip"]
        })
        return {"shipping_address": address}
    
    return {"order_status": "processing"}

---

Best Practices

1. Always Use Checkpointer

# ✓ Required for interrupt()
chain = graph.compile(checkpointer=InMemorySaver())

2. Clear Interrupt Payloads

# ✓ Clear and actionable
interrupt({
    "draft": state["draft"],
    "action": "approve_or_reject",
    "instructions": "Review for tone and accuracy"
})

3. Handle All Response Cases

def review_node(state):
    response = interrupt({"draft": state["draft"]})
    
    if response.get("approved"):
        return {"status": "approved"}
    elif response.get("rejected"):
        return {"status": "rejected", "reason": response.get("reason")}
    elif response.get("edited"):
        return {"draft": response["edited"], "status": "revised"}
    return {"status": "pending"}

4. Audit Trail

def audited_review(state):
    response = interrupt({"draft": state["draft"]})
    
    return {
        "approved": response.get("approved"),
        "reviewer": response.get("reviewer_id"),
        "review_timestamp": response.get("timestamp")
    }

Multi-Agent Patterns

Build multi-agent systems where specialized agents collaborate through supervisors or swarm-based handoffs.

Contents

• Architecture Overview
• Supervisor Pattern
• Swarm Pattern
• Nested Hierarchies
• Custom Handoff Tools
• Choosing a Pattern

---

Architecture Overview

Multi-agent systems use a "divide-and-conquer" approach:

• Specialized agents handle specific domains with curated tools
• Coordinator routes tasks to the appropriate expert
• Shared state carries context between agents

---

Supervisor Pattern

A central supervisor routes tasks to specialized agents. Uses langgraph-supervisor package.

Installation

pip install langgraph-supervisor

Basic Supervisor

from langchain_openai import ChatOpenAI
from langgraph_supervisor import create_supervisor
from langgraph.prebuilt import create_react_agent

model = ChatOpenAI(model="gpt-4o")

# Define specialized tools
def add(a: float, b: float) -> float:
    """Add two numbers."""
    return a + b

def multiply(a: float, b: float) -> float:
    """Multiply two numbers."""
    return a * b

def web_search(query: str) -> str:
    """Search the web for information."""
    return f"Results for: {query}"

# Create specialized agents
math_agent = create_react_agent(
    model=model,
    tools=[add, multiply],
    name="math_expert",
    prompt="You are a math expert. Always use one tool at a time."
)

research_agent = create_react_agent(
    model=model,
    tools=[web_search],
    name="research_expert",
    prompt="You are a researcher with web search access. Do not do math."
)

# Create supervisor workflow
workflow = create_supervisor(
    agents=[research_agent, math_agent],
    model=model,
    prompt=(
        "You are a team supervisor managing research and math experts. "
        "For current events, use research_expert. For math, use math_expert."
    )
)

# Compile and run
app = workflow.compile()
result = app.invoke({
    "messages": [{"role": "user", "content": "What's 25 * 4?"}]
})

for message in result["messages"]:
    print(f"{message.type}: {message.content}")

Output Modes

Control how much agent history to include:

# Include full message history from workers
workflow = create_supervisor(
    agents=[agent1, agent2],
    output_mode="full_history"
)

# Include only final response from workers
workflow = create_supervisor(
    agents=[agent1, agent2],
    output_mode="last_message"
)

Custom Handoff Tool Names

from langgraph_supervisor import create_handoff_tool

workflow = create_supervisor(
    [research_agent, math_agent],
    tools=[
        create_handoff_tool(
            agent_name="math_expert",
            name="assign_to_math",
            description="Assign math problems to the math expert"
        ),
        create_handoff_tool(
            agent_name="research_expert",
            name="assign_to_research",
            description="Assign research tasks to the research expert"
        )
    ],
    model=model,
)

---

Swarm Pattern

Agents hand off control directly to each other. Uses langgraph-swarm package.

Installation

pip install langgraph-swarm

Basic Swarm

from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.prebuilt import create_react_agent
from langgraph_swarm import create_handoff_tool, create_swarm

model = ChatOpenAI(model="gpt-4o")

def add(a: int, b: int) -> int:
    """Add two numbers"""
    return a + b

# Create agents with handoff tools
alice = create_react_agent(
    model,
    [add, create_handoff_tool(agent_name="Bob")],
    prompt="You are Alice, an addition expert.",
    name="Alice",
)

bob = create_react_agent(
    model,
    [create_handoff_tool(
        agent_name="Alice",
        description="Transfer to Alice, she can help with math"
    )],
    prompt="You are Bob, you speak like a pirate.",
    name="Bob",
)

# Create swarm with persistence
checkpointer = InMemorySaver()
workflow = create_swarm(
    [alice, bob],
    default_active_agent="Alice"
)
app = workflow.compile(checkpointer=checkpointer)

# Multi-turn conversation
config = {"configurable": {"thread_id": "1"}}

turn_1 = app.invoke(
    {"messages": [{"role": "user", "content": "I'd like to speak to Bob"}]},
    config,
)
# Bob is now active

turn_2 = app.invoke(
    {"messages": [{"role": "user", "content": "What's 5 + 7?"}]},
    config,
)
# Bob transfers to Alice, Alice calculates 12

Customer Support Example

from langgraph_swarm import create_handoff_tool, create_swarm
from langgraph.prebuilt import create_react_agent

# Handoff tools for specialized agents
transfer_to_hotel = create_handoff_tool(
    agent_name="hotel_assistant",
    description="Transfer to hotel-booking assistant for hotel searches and bookings.",
)

transfer_to_flight = create_handoff_tool(
    agent_name="flight_assistant",
    description="Transfer to flight-booking assistant for flight searches and bookings.",
)

transfer_to_triage = create_handoff_tool(
    agent_name="triage_assistant",
    description="Transfer back to triage for general questions.",
)

# Triage agent routes to specialists
triage_agent = create_react_agent(
    model,
    [transfer_to_hotel, transfer_to_flight],
    prompt="You are a travel assistant. Route users to the appropriate specialist.",
    name="triage_assistant",
)

hotel_agent = create_react_agent(
    model,
    [search_hotels, book_hotel, transfer_to_triage, transfer_to_flight],
    prompt="You are a hotel booking specialist.",
    name="hotel_assistant",
)

flight_agent = create_react_agent(
    model,
    [search_flights, book_flight, transfer_to_triage, transfer_to_hotel],
    prompt="You are a flight booking specialist.",
    name="flight_assistant",
)

# Build swarm
workflow = create_swarm(
    [triage_agent, hotel_agent, flight_agent],
    default_active_agent="triage_assistant"
)
app = workflow.compile(checkpointer=InMemorySaver())

---

Nested Hierarchies

Build multi-level supervisors for complex organizations:

from langgraph_supervisor import create_supervisor
from langgraph.prebuilt import create_react_agent

# Team-level supervisors
research_team = create_supervisor(
    [research_agent, data_agent],
    model=model,
    supervisor_name="research_supervisor"
).compile(name="research_team")

writing_team = create_supervisor(
    [writer_agent, editor_agent],
    model=model,
    supervisor_name="writing_supervisor"
).compile(name="writing_team")

# Top-level supervisor manages teams
top_level = create_supervisor(
    [research_team, writing_team],
    model=model,
    supervisor_name="top_level_supervisor",
    prompt="Route research tasks to research_team, content tasks to writing_team."
).compile(name="top_level_supervisor")

---

Custom Handoff Tools

Create handoff tools with additional context:

from typing import Annotated
from langchain_core.tools import tool, BaseTool, InjectedToolCallId
from langchain_core.messages import ToolMessage
from langgraph.types import Command
from langgraph.prebuilt import InjectedState

def create_custom_handoff(*, agent_name: str, name: str, description: str) -> BaseTool:

    @tool(name, description=description)
    def handoff_to_agent(
        task_description: Annotated[str, "Detailed description for the next agent"],
        state: Annotated[dict, InjectedState],
        tool_call_id: Annotated[str, InjectedToolCallId],
    ):
        tool_message = ToolMessage(
            content=f"Successfully transferred to {agent_name}",
            name=name,
            tool_call_id=tool_call_id,
        )
        return Command(
            goto=agent_name,
            graph=Command.PARENT,
            update={
                "messages": state["messages"] + [tool_message],
                "active_agent": agent_name,
                "task_description": task_description,  # Pass context
            },
        )

    return handoff_to_agent

# Usage
transfer_with_context = create_custom_handoff(
    agent_name="specialist",
    name="transfer_to_specialist",
    description="Transfer with detailed task description"
)

---

Choosing a Pattern

Key Considerations

1. Supervisor: More predictable, central control, easier to debug 2. Swarm: More flexible, agents decide routing, better for dynamic workflows 3. Hybrid: Use swarm within teams, supervisor between teams

---

Official Resources

• langgraph-supervisor - Hierarchical multi-agent
• langgraph-swarm - Swarm-style handoffs
• Multi-agent tutorial - Official notebook

Official LangGraph Resources

Quick reference to official documentation, tutorials, and tools.

Contents

• Documentation Sites
• Getting Started
• Core Concepts
• How-To Guides
• Agent Development
• Platform & Deployment
• Tools & Integrations
• Troubleshooting

---

Documentation Sites

---

Getting Started

---

Core Concepts

---

How-To Guides

Graph API

Human-in-the-Loop

---

Agent Development

---

Platform & Deployment

LangGraph Platform

Deployment Guides

CLI & Studio

---

Tools & Integrations

---

Troubleshooting

---

Tutorials

---

GitHub Repositories

---

Fetching Latest Docs

For always up-to-date documentation, fetch the llms.txt index:

# Using WebFetch or requests
url = "https://langchain-ai.github.io/langgraph/llms.txt"
# Returns structured list of all documentation URLs with descriptions

Persistence and Memory

Enable conversation memory, survive crashes, and debug with time-travel.

Contents

• Why Persistence?
• Checkpointers
• Thread IDs
• Multi-Turn Conversations
• Time Travel
• Durable Execution

---

Why Persistence?

---

Checkpointers

InMemorySaver (Development)

from langgraph.checkpoint.memory import InMemorySaver

chain = graph.compile(checkpointer=InMemorySaver())

PostgresSaver (Production)

from langgraph.checkpoint.postgres import PostgresSaver

DB_URI = "postgresql://user:pass@localhost:5432/langgraph"
checkpointer = PostgresSaver.from_conn_string(DB_URI)
chain = graph.compile(checkpointer=checkpointer)

SQLiteSaver

from langgraph.checkpoint.sqlite import SqliteSaver

checkpointer = SqliteSaver.from_conn_string("checkpoints.db")
chain = graph.compile(checkpointer=checkpointer)

---

Thread IDs

Identify separate conversations:

chain = graph.compile(checkpointer=InMemorySaver())

config = {"configurable": {"thread_id": "user-123"}}

# First message
chain.invoke({"messages": [HumanMessage(content="I'm Alice")]}, config=config)

# Second message - remembers Alice
chain.invoke({"messages": [HumanMessage(content="What's my name?")]}, config=config)

# Different thread - doesn't know Alice
other_config = {"configurable": {"thread_id": "user-456"}}
chain.invoke({"messages": [HumanMessage(content="What's my name?")]}, config=other_config)

Thread ID Strategies

---

Multi-Turn Conversations

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AnyMessage
from typing_extensions import TypedDict, Annotated
import operator

class State(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]

llm = ChatOpenAI(model="gpt-4")

def chat(state: State) -> dict:
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

graph = StateGraph(State)
graph.add_node("chat", chat)
graph.add_edge(START, "chat")
graph.add_edge("chat", END)

chain = graph.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "demo"}}

# Turn 1
chain.invoke({"messages": [HumanMessage(content="My color is blue.")]}, config=config)

# Turn 2 - remembers
result = chain.invoke({"messages": [HumanMessage(content="What's my color?")]}, config=config)
# "Your color is blue."

---

Time Travel

Viewing History

for state in chain.get_state_history(config):
    print(f"Checkpoint: {state.config['configurable']['checkpoint_id']}")
    print(f"Next: {state.next}")

Resuming from Checkpoint

checkpoint_id = "abc123..."

resume_config = {
    "configurable": {
        "thread_id": "my-thread",
        "checkpoint_id": checkpoint_id
    }
}

result = chain.invoke(None, config=resume_config)

Modifying and Resuming

state = chain.get_state(config)
modified = dict(state.values)
modified["messages"].append(HumanMessage(content="Try again."))

chain.update_state(config, modified)
result = chain.invoke(None, config=config)

Forking

fork_config = {
    "configurable": {
        "thread_id": "forked-thread",
        "checkpoint_id": checkpoint_id
    }
}
result = chain.invoke(new_input, config=fork_config)

---

Durable Execution

Survive crashes and resume:

Idempotency

# ⚠️ Problem: re-sends on resume
def send_email(state):
    send(state["draft"])
    return {"sent": True}

# ✓ Fix: check first
def send_email(state):
    if not state.get("sent"):
        send(state["draft"])
    return {"sent": True}

# ✓ Or use @task
from langgraph.func import task

@task
def send_email(state):
    send(state["draft"])
    return {"sent": True}

Long-Running Workflows

chain = graph.compile(checkpointer=PostgresSaver.from_conn_string(DB_URI))

config = {"configurable": {"thread_id": "long-job-1"}}
chain.invoke(initial_state, config=config)

# If server restarts, invoke again - resumes from checkpoint
chain.invoke(None, config=config)

---

Production Setup

import os

ENV = os.getenv("ENVIRONMENT", "development")

if ENV == "production":
    from langgraph.checkpoint.postgres import PostgresSaver
    checkpointer = PostgresSaver.from_conn_string(os.getenv("DATABASE_URL"))
else:
    from langgraph.checkpoint.memory import InMemorySaver
    checkpointer = InMemorySaver()

chain = graph.compile(checkpointer=checkpointer)

Production Deployment

Deploy LangGraph applications to production using LangGraph Platform (cloud/self-hosted) or custom infrastructure.

Contents

• Deployment Options
• LangGraph Platform
• Application Structure
• LangGraph CLI
• RemoteGraph Client
• Self-Hosted Deployment
• Production Checklist

---

Deployment Options

---

LangGraph Platform

LangGraph Platform is a runtime for deploying stateful agent workflows with:

• Execution APIs — Invoke, stream, manage runs
• Persistence — Built-in checkpointing and memory
• Monitoring — Observability via LangSmith integration
• Scaling — Automatic scaling for production loads

---

Application Structure

Required Files

my-langgraph-app/
├── langgraph.json      # Deployment configuration
├── requirements.txt    # Python dependencies
├── agent.py            # Graph definition
└── .env                # Local env vars (not deployed)

langgraph.json

{
  "dependencies": ["."],
  "graphs": {
    "my_agent": "./agent.py:graph"
  },
  "env": ".env"
}

Agent Code (agent.py)

from langgraph.graph import StateGraph, MessagesState, START
from langgraph.checkpoint.memory import InMemorySaver
from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o")

def call_model(state: MessagesState):
    response = model.invoke(state["messages"])
    return {"messages": [response]}

def create_graph():
    builder = StateGraph(MessagesState)
    builder.add_node("agent", call_model)
    builder.add_edge(START, "agent")

    # Use InMemorySaver for local dev; Platform provides production checkpointer
    checkpointer = InMemorySaver()
    return builder.compile(checkpointer=checkpointer)

# Export for LangGraph Platform
graph = create_graph()

Extended Configuration

{
  "dependencies": [
    ".",
    "langchain-openai>=0.1.0",
    "langgraph>=0.2.0"
  ],
  "graphs": {
    "agent": "./agent.py:graph",
    "researcher": "./agents/researcher.py:graph"
  },
  "env": ".env",
  "dockerfile_lines": [
    "RUN apt-get update && apt-get install -y poppler-utils",
    "RUN pip install pdf2image"
  ]
}

---

LangGraph CLI

Installation

pip install -U langgraph-cli

Commands

# Deploy to LangGraph Platform
langgraph deploy --config langgraph.json

# Build Docker image locally
langgraph build --config langgraph.json

# Run locally for development
langgraph dev --config langgraph.json

Development Server

# Start local development server
langgraph dev

# Server runs at http://localhost:8000
# API docs at http://localhost:8000/docs

---

RemoteGraph Client

Interact with deployed graphs using RemoteGraph:

from langgraph.pregel.remote import RemoteGraph
from langgraph_sdk import get_client

# Connect to deployed graph
url = "https://your-deployment.langgraph.com"
remote_graph = RemoteGraph(
    name="agent",
    url=url,
    api_key="your-langsmith-api-key"
)

# Use exactly like local graph
result = remote_graph.invoke({
    "messages": [{"role": "user", "content": "Hello!"}]
})

# Streaming
for chunk in remote_graph.stream({
    "messages": [{"role": "user", "content": "Tell me a story"}]
}, stream_mode="updates"):
    print(chunk)

# Thread-based persistence
client = get_client(url=url)
thread = client.threads.create()

config = {"configurable": {"thread_id": thread["thread_id"]}}
result = remote_graph.invoke(
    {"messages": [{"role": "user", "content": "Remember: code=12345"}]},
    config
)

# Continue conversation
result2 = remote_graph.invoke(
    {"messages": [{"role": "user", "content": "What was the code?"}]},
    config
)
print(result2["messages"][-1].content)  # "The code was 12345"

LangGraph SDK

pip install langgraph-sdk

from langgraph_sdk import get_client

client = get_client(url="https://your-deployment.langgraph.com")

# Create thread
thread = client.threads.create()

# Run agent
run = client.runs.create(
    thread_id=thread["thread_id"],
    graph_id="agent",
    input={"messages": [{"role": "user", "content": "Hello"}]}
)

# Wait for completion
result = client.runs.wait(thread_id=thread["thread_id"], run_id=run["run_id"])

---

Self-Hosted Deployment

Docker Deployment

FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

# Expose the LangGraph server port
EXPOSE 8000

CMD ["python", "-m", "langgraph", "serve", "--host", "0.0.0.0", "--port", "8000"]

FastAPI Wrapper

from fastapi import FastAPI
from pydantic import BaseModel
from agent import graph

app = FastAPI()

class InvokeRequest(BaseModel):
    messages: list
    thread_id: str | None = None

@app.post("/invoke")
async def invoke(request: InvokeRequest):
    config = {}
    if request.thread_id:
        config = {"configurable": {"thread_id": request.thread_id}}

    result = graph.invoke(
        {"messages": request.messages},
        config
    )
    return {"messages": result["messages"]}

@app.post("/stream")
async def stream(request: InvokeRequest):
    from fastapi.responses import StreamingResponse

    def generate():
        config = {}
        if request.thread_id:
            config = {"configurable": {"thread_id": request.thread_id}}

        for chunk in graph.stream(
            {"messages": request.messages},
            config,
            stream_mode="updates"
        ):
            yield f"data: {chunk}\n\n"

    return StreamingResponse(generate(), media_type="text/event-stream")

PostgreSQL Persistence

from langgraph.checkpoint.postgres import PostgresSaver

def create_production_graph():
    connection_string = os.environ["DATABASE_URL"]
    checkpointer = PostgresSaver.from_conn_string(connection_string)

    builder = StateGraph(MessagesState)
    # ... add nodes and edges

    return builder.compile(checkpointer=checkpointer)

---

Production Checklist

Before Deployment

• [ ] All environment variables configured in deployment platform (not in code)
• [ ] langgraph.json has correct graph paths
• [ ] Dependencies pinned to specific versions
• [ ] Production checkpointer configured (PostgreSQL recommended)
• [ ] Error handling in all nodes
• [ ] Timeouts configured for LLM calls

Environment Variables

# Required
OPENAI_API_KEY=sk-...
# or
ANTHROPIC_API_KEY=sk-ant-...

# For LangSmith observability
LANGSMITH_API_KEY=ls-...
LANGSMITH_TRACING=true
LANGSMITH_PROJECT=my-production-agent

# For PostgreSQL persistence
DATABASE_URL=postgresql://user:pass@host:5432/langgraph

Monitoring

import logging
from datetime import datetime

logger = logging.getLogger("langgraph")

def monitored_node(state):
    start = datetime.now()
    try:
        result = process(state)
        duration = (datetime.now() - start).total_seconds()
        logger.info(f"Node completed in {duration:.2f}s")
        return result
    except Exception as e:
        logger.error(f"Node failed: {e}")
        raise

Human-in-the-Loop Interrupts

app = workflow.compile(
    checkpointer=checkpointer,
    interrupt_before=["human_approval"],  # Pause before this node
    interrupt_after=["critical_action"]   # Pause after this node
)

---

Official Resources

• LangGraph Platform docs
• Application structure
• LangGraph CLI
• Deployment options

Tool-Using Agent Pattern

Build agents that call external tools in a loop until the task is complete.

Contents

• Overview
• Defining Tools
• Binding Tools to Model
• Agent State
• The Agent Loop
• Complete Example
• Tool Error Handling
• Advanced Patterns

---

Overview

A tool-using agent follows this cycle:

START → LLM decides → Tool needed? 
                         ↓ Yes: Execute tool → Loop back to LLM
                         ↓ No: Return answer → END

---

Defining Tools

from langchain_core.tools import tool

@tool
def search(query: str) -> str:
    """Search the knowledge base for information.
    
    Args:
        query: The search query string
    """
    return f"Results for: {query}"

@tool
def calculate(expression: str) -> float:
    """Evaluate a mathematical expression.
    
    Args:
        expression: Math expression like '2 + 3 * 4'
    """
    return eval(expression)

Requirements: Clear docstring, type hints, descriptive name.

---

Binding Tools to Model

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4", temperature=0)
tools = [search, calculate]
model_with_tools = llm.bind_tools(tools)
tools_by_name = {t.name: t for t in tools}

---

Agent State

from typing_extensions import TypedDict, Annotated
from langchain_core.messages import AnyMessage
import operator

class AgentState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]

---

The Agent Loop

LLM Node

from langchain_core.messages import SystemMessage

def llm_node(state: AgentState) -> dict:
    system = SystemMessage(content="You are a helpful assistant with tools.")
    response = model_with_tools.invoke([system] + state["messages"])
    return {"messages": [response]}

Tool Node

from langchain_core.messages import ToolMessage

def tool_node(state: AgentState) -> dict:
    results = []
    for tool_call in state["messages"][-1].tool_calls:
        tool = tools_by_name[tool_call["name"]]
        output = tool.invoke(tool_call["args"])
        results.append(ToolMessage(
            content=str(output),
            tool_call_id=tool_call["id"]
        ))
    return {"messages": results}

Router

from typing import Literal

def should_continue(state: AgentState) -> Literal["tool_node", "__end__"]:
    last_message = state["messages"][-1]
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tool_node"
    return "__end__"

Assembly

from langgraph.graph import StateGraph, START, END

graph = StateGraph(AgentState)
graph.add_node("llm", llm_node)
graph.add_node("tool_node", tool_node)
graph.add_edge(START, "llm")
graph.add_conditional_edges("llm", should_continue, ["tool_node", END])
graph.add_edge("tool_node", "llm")

agent = graph.compile()

---

Complete Example

from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, SystemMessage, ToolMessage, AnyMessage
from typing_extensions import TypedDict, Annotated
from typing import Literal
import operator

@tool
def add(a: int, b: int) -> int:
    """Add two numbers."""
    return a + b

@tool
def multiply(a: int, b: int) -> int:
    """Multiply two numbers."""
    return a * b

tools = [add, multiply]
tools_by_name = {t.name: t for t in tools}

llm = ChatOpenAI(model="gpt-4", temperature=0)
model_with_tools = llm.bind_tools(tools)

class State(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]

def llm_node(state: State) -> dict:
    system = SystemMessage(content="You are a calculator assistant.")
    response = model_with_tools.invoke([system] + state["messages"])
    return {"messages": [response]}

def tool_node(state: State) -> dict:
    results = []
    for tc in state["messages"][-1].tool_calls:
        output = tools_by_name[tc["name"]].invoke(tc["args"])
        results.append(ToolMessage(content=str(output), tool_call_id=tc["id"]))
    return {"messages": results}

def should_continue(state: State) -> Literal["tool_node", "__end__"]:
    if state["messages"][-1].tool_calls:
        return "tool_node"
    return "__end__"

graph = StateGraph(State)
graph.add_node("llm", llm_node)
graph.add_node("tool_node", tool_node)
graph.add_edge(START, "llm")
graph.add_conditional_edges("llm", should_continue, ["tool_node", "__end__"])
graph.add_edge("tool_node", "llm")

agent = graph.compile()

result = agent.invoke({
    "messages": [HumanMessage(content="What is 3 + 4 multiplied by 2?")]
})
print(result["messages"][-1].content)

Expected Output:

14

The agent processes the request as follows: 1. LLM receives the question and generates tool calls for add(3, 4) then multiply(7, 2) 2. Tool node executes each tool call and returns ToolMessage results 3. LLM synthesizes final answer from tool outputs

---

Tool Error Handling

def tool_node(state: State) -> dict:
    results = []
    for tc in state["messages"][-1].tool_calls:
        try:
            tool = tools_by_name[tc["name"]]
            output = tool.invoke(tc["args"])
            content = str(output)
        except Exception as e:
            content = f"Error: {type(e).__name__}: {str(e)}"
        
        results.append(ToolMessage(content=content, tool_call_id=tc["id"]))
    return {"messages": results}

---

Advanced Patterns

Maximum Iterations

class State(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    iterations: int

def llm_node(state: State) -> dict:
    response = model_with_tools.invoke(state["messages"])
    return {"messages": [response], "iterations": state.get("iterations", 0) + 1}

def should_continue(state: State) -> str:
    if state.get("iterations", 0) >= 10:
        return "__end__"
    if state["messages"][-1].tool_calls:
        return "tool_node"
    return "__end__"

Dynamic Tool Selection

def llm_node(state: State) -> dict:
    if state.get("mode") == "math":
        active_tools = [add, multiply]
    else:
        active_tools = [search]
    
    model = llm.bind_tools(active_tools)
    response = model.invoke(state["messages"])
    return {"messages": [response]}

Workflow Patterns

Build structured multi-step pipelines with branching, parallel execution, and prompt chaining.

Contents

• Workflows vs Agents
• Sequential Chain
• Conditional Branching
• Parallel Execution
• Prompt Chaining
• Map-Reduce Pattern
• Complete Example

---

Workflows vs Agents

Use workflows when the sequence is known ahead of time.

---

Sequential Chain

A → B → C → END

from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict

class State(TypedDict):
    input: str
    step1_output: str
    step2_output: str

def step1(state: State) -> dict:
    return {"step1_output": f"Processed: {state['input']}"}

def step2(state: State) -> dict:
    return {"step2_output": f"Final: {state['step1_output']}"}

graph = StateGraph(State)
graph.add_node("step1", step1)
graph.add_node("step2", step2)
graph.add_edge(START, "step1")
graph.add_edge("step1", "step2")
graph.add_edge("step2", END)

chain = graph.compile()

---

Conditional Branching

classify → handler_a → END
         → handler_b → END

Router Function

def route_by_category(state: State) -> str:
    category = state["category"]
    if category == "urgent":
        return "urgent_handler"
    elif category == "normal":
        return "normal_handler"
    return "fallback_handler"

Adding Conditional Edges

graph.add_conditional_edges(
    "classify",
    route_by_category,
    ["urgent_handler", "normal_handler", "fallback_handler"]
)

Pass/Fail Loop

def check_quality(state: State) -> str:
    return "accept" if state["score"] >= 0.8 else "revise"

graph.add_conditional_edges("evaluate", check_quality, ["accept", "revise"])
graph.add_edge("accept", END)
graph.add_edge("revise", "improve")
graph.add_edge("improve", "evaluate")  # Loop back

---

Parallel Execution

Fan out to multiple nodes, then join:

from typing_extensions import Annotated
import operator

class State(TypedDict):
    text: str
    sentiment: str
    entities: Annotated[list[str], operator.add]
    summary: str

def analyze_sentiment(state: State) -> dict:
    return {"sentiment": "positive"}

def extract_entities(state: State) -> dict:
    return {"entities": ["entity1", "entity2"]}

def summarize(state: State) -> dict:
    return {"summary": "Brief summary..."}

graph = StateGraph(State)
graph.add_node("sentiment", analyze_sentiment)
graph.add_node("entities", extract_entities)
graph.add_node("summarize", summarize)
graph.add_node("combine", combine_results)

# Fan out
graph.add_edge(START, "sentiment")
graph.add_edge(START, "entities")
graph.add_edge(START, "summarize")

# Fan in
graph.add_edge("sentiment", "combine")
graph.add_edge("entities", "combine")
graph.add_edge("summarize", "combine")
graph.add_edge("combine", END)

LangGraph waits for all parallel branches before executing combine.

---

Prompt Chaining

Sequential LLM calls building on each other:

class State(TypedDict):
    topic: str
    outline: str
    draft: str
    final: str

def generate_outline(state: State) -> dict:
    prompt = f"Create an outline for: {state['topic']}"
    return {"outline": llm.invoke(prompt).content}

def write_draft(state: State) -> dict:
    prompt = f"Write a draft from this outline:\n{state['outline']}"
    return {"draft": llm.invoke(prompt).content}

def polish(state: State) -> dict:
    prompt = f"Polish this draft:\n{state['draft']}"
    return {"final": llm.invoke(prompt).content}

graph = StateGraph(State)
graph.add_node("outline", generate_outline)
graph.add_node("draft", write_draft)
graph.add_node("polish", polish)

graph.add_edge(START, "outline")
graph.add_edge("outline", "draft")
graph.add_edge("draft", "polish")
graph.add_edge("polish", END)

---

Map-Reduce Pattern

class State(TypedDict):
    documents: list[str]
    summaries: Annotated[list[str], operator.add]
    final_summary: str

def map_summarize(state: State) -> dict:
    summaries = []
    for doc in state["documents"]:
        summaries.append(llm.invoke(f"Summarize: {doc}").content)
    return {"summaries": summaries}

def reduce_combine(state: State) -> dict:
    combined = "\n".join(state["summaries"])
    final = llm.invoke(f"Combine:\n{combined}").content
    return {"final_summary": final}

graph = StateGraph(State)
graph.add_node("map", map_summarize)
graph.add_node("reduce", reduce_combine)
graph.add_edge(START, "map")
graph.add_edge("map", "reduce")
graph.add_edge("reduce", END)

---

Complete Example

Joke workflow with quality check:

from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI
from typing_extensions import TypedDict

llm = ChatOpenAI(model="gpt-4", temperature=0.7)

class JokeState(TypedDict):
    topic: str
    joke: str
    improved_joke: str
    final_joke: str

def generate_joke(state: JokeState) -> dict:
    response = llm.invoke(f"Tell a short joke about {state['topic']}.")
    return {"joke": response.content}

def check_punchline(state: JokeState) -> str:
    joke = state["joke"]
    return "pass" if ("?" in joke or "!" in joke) else "fail"

def improve_joke(state: JokeState) -> dict:
    response = llm.invoke(f"Make this funnier:\n{state['joke']}")
    return {"improved_joke": response.content}

def polish_joke(state: JokeState) -> dict:
    source = state.get("improved_joke") or state["joke"]
    response = llm.invoke(f"Add a twist:\n{source}")
    return {"final_joke": response.content}

graph = StateGraph(JokeState)
graph.add_node("generate", generate_joke)
graph.add_node("improve", improve_joke)
graph.add_node("polish", polish_joke)

graph.add_edge(START, "generate")
graph.add_conditional_edges("generate", check_punchline, {"pass": "polish", "fail": "improve"})
graph.add_edge("improve", "polish")
graph.add_edge("polish", END)

chain = graph.compile()
result = chain.invoke({"topic": "programmers"})

Flow: START → generate → [pass] → polish → END Or: START → generate → [fail] → improve → polish → END