Langgraph Python

LangGraph Skill

LangGraph: Framework for stateful, multi-actor AI applications.

For latest API syntax, use context7 MCP.

Core Concepts

from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.prebuilt import create_react_agent, ToolNode, tools_condition
from langgraph.types import Command, interrupt, Send
from langgraph.checkpoint.memory import InMemorySaver

Pattern Selection Guide

Need to use tools?
├─ Yes → ReAct Agent
└─ No → Workflow patterns

Sequential or parallel?
├─ Sequential → Prompt Chaining
├─ Parallel → Parallelization
└─ Both → Orchestrator-Worker

Different handling by input type?
└─ Yes → Routing

Need output validation?
└─ Yes → Evaluator-Optimizer

Multiple specialized agents?
├─ Central coordinator → Supervisor
├─ Team hierarchy → Hierarchical
└─ Peer collaboration → Network

1. ReAct Agent (Tool-Using)

Use case: LLM autonomously decides when to use tools

START → LLM → [Tool calls?] → Yes → Tools → LLM (loop)
                            → No  → END

from langgraph.prebuilt import create_react_agent

agent = create_react_agent(model, tools=[search, calculator])
result = agent.invoke({"messages": [{"role": "user", "content": "..."}]})

Pitfall: Agent stuck in loop → add max_iterations or explicit stop condition

2. Prompt Chaining (Sequential)

Use case: Step-by-step processing

START → Generate → Edit → Review → END

builder = StateGraph(State)
builder.add_edge(START, "generate")
builder.add_edge("generate", "edit")
builder.add_edge("edit", "review")
builder.add_edge("review", END)

Pitfall: Forgetting END edge → graph never terminates

3. Parallelization

Use case: Independent tasks simultaneously

START ──┬── Task A ──┐
        ├── Task B ──┼── Aggregate → END
        └── Task C ──┘

builder.add_edge(START, "task_a")
builder.add_edge(START, "task_b")
builder.add_edge(START, "task_c")
builder.add_edge(["task_a", "task_b", "task_c"], "aggregate")

Pitfall: State conflicts → use reducer (e.g., Annotated[list, operator.add])

4. Routing

Use case: Route to handler based on classification

START → Classifier → [Route] → Handler A/B/C → END

def route_fn(state: State) -> str:
    return state["category"]  # Must match node names exactly

builder.add_conditional_edges("classifier", route_fn, 
    ["handler_a", "handler_b", "handler_c"])

Pitfall: Route function returns value not in edge map → runtime error

5. Orchestrator-Worker (Send)

Use case: Dynamically create parallel subtasks

START → Orchestrator → [Send] → Worker 1..N → Synthesizer → END

from langgraph.types import Send

def assign_workers(state: State):
    return [Send("worker", {"task": t}) for t in state["tasks"]]

builder.add_conditional_edges("orchestrator", assign_workers, ["worker"])

Pitfall: Workers must return same state schema; results need aggregation reducer

6. Evaluator-Optimizer (Reflection)

Use case: Generate → evaluate → improve loop

START → Generate → Evaluate → [Pass?] → Yes → END
                                      → No  → Generate

def should_continue(state: State) -> str:
    if state["score"] == "pass" or state["iterations"] >= 3:
        return "end"
    return "retry"

builder.add_conditional_edges("evaluate", should_continue, 
    {"retry": "generate", "end": END})

Pitfall: No max iterations → infinite loop; always add iteration counter

7. Multi-Agent: Supervisor

Use case: Central coordinator delegates to specialists

START → Supervisor ←→ Agent A/B/C → END

from langgraph.types import Command

def supervisor(state) -> Command[Literal["agent_a", "agent_b", "__end__"]]:
    next_agent = decide_next(state)
    return Command(goto=next_agent)

Pitfall: Supervisor must explicitly return __end__ to terminate

8. Multi-Agent: Hierarchical

Use case: Team leads manage sub-teams

from langgraph_supervisor import create_supervisor

research_team = create_supervisor([search_agent, scraper_agent], model=model)
writing_team = create_supervisor([writer_agent, editor_agent], model=model)
top_supervisor = create_supervisor([research_team, writing_team], model=model)

Pitfall: Deep hierarchies slow down; keep ≤3 levels

9. Human-in-the-Loop

Use case: Human approval, state editing, input requests

from langgraph.types import interrupt, Command

def approval_node(state) -> Command[Literal["proceed", "cancel"]]:
    response = interrupt({"question": "Approve?", "data": state["action"]})
    return Command(goto="proceed" if response["approved"] else "cancel")

# Requires checkpointer
graph = builder.compile(checkpointer=InMemorySaver())
result = graph.invoke(inputs, {"configurable": {"thread_id": "1"}})
# Resume
graph.invoke(Command(resume={"approved": True}), config)

Pitfall: No checkpointer → interrupt fails; thread_id required for resume

Persistence (Memory)

from langgraph.checkpoint.memory import InMemorySaver  # Dev only
from langgraph.checkpoint.postgres import PostgresSaver  # Production

graph = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "user-123"}}

State Definition

from typing import TypedDict, Annotated
import operator

class State(TypedDict):
    messages: Annotated[list, operator.add]  # Reducer: accumulate
    current_step: str                         # No reducer: overwrite

Pitfall: Parallel nodes updating same field without reducer → last write wins (data loss)

Common Errors & Fixes

Graph Never Terminates

Cause: Missing edge to END

# ❌ Forgot END
builder.add_edge("final_node", "somewhere")

# ✅ Add END edge
builder.add_edge("final_node", END)

Infinite Loop

Cause: Conditional edge always returns same route

# ✅ Add counter or max iterations
def should_continue(state):
    if state["iterations"] >= 3:
        return "end"
    return "retry"

"Node X not found"

Cause: Route function returns string not in edge mapping

# ❌ Mismatch
def route(state): return "process"  # But node is "processor"

# ✅ Match exactly
def route(state): return "processor"

State Update Not Reflected

Cause: Node not returning dict, or returning wrong keys

# ❌ Returns None
def my_node(state):
    do_something(state)
    # Missing return!

# ✅ Return state updates
def my_node(state):
    return {"result": do_something(state)}

Interrupt Not Working

Cause: Missing checkpointer or thread_id

# ❌ No persistence
graph = builder.compile()

# ✅ With checkpointer
graph = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "123"}}

Parallel State Conflicts

Cause: Multiple nodes write same field without reducer

# ❌ Overwrites
class State(TypedDict):
    results: list  # Last write wins

# ✅ Use reducer
class State(TypedDict):
    results: Annotated[list, operator.add]  # Accumulates

Debugging

# Visualize graph
from IPython.display import Image
Image(graph.get_graph().draw_mermaid_png())

# Stream with debug info
for chunk in graph.stream(inputs, stream_mode="debug"):
    print(chunk)

# Inspect state
graph.get_state(config).values

Quick Reference