Agent skill

langgraph-python

Design and build AI agents with LangGraph. Use when building ReAct agents, multi-agent systems, workflow orchestration, human-in-the-loop patterns, or state machine workflows.

Stars 163
Forks 31

Install this agent skill to your Project

npx add-skill https://github.com/majiayu000/claude-skill-registry/tree/main/skills/other/langgraph-python

SKILL.md

LangGraph Skill

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

For latest API syntax, use context7 MCP.


Core Concepts

python
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
Concept Description
StateGraph Graph managing state. Updates after each node
MessagesState Built-in state for chat (includes messages list)
Node Function: receives state → returns updates
Edge Transition: static (add_edge) or conditional
Checkpointer Persistence for conversation history, pause/resume
Command Dynamic routing + state update together
interrupt Pause for human-in-the-loop
Send Dynamically spawn parallel workers

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
python
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
python
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 ──┘
python
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
python
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
python
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
python
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
python
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

python
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

python
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)

Type Purpose Implementation
Short-term Within conversation checkpointer + thread_id
Long-term Across conversations store
python
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

python
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

python
# ❌ 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

python
# ✅ 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

python
# ❌ 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

python
# ❌ 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

python
# ❌ 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

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

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

Debugging

python
# 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

Pattern When to Use Key Component
ReAct Tool-using agent create_react_agent
Chaining Sequential steps add_edge
Parallel Independent tasks Multiple edges from START
Routing Input classification add_conditional_edges
Orchestrator Dynamic subtasks Send
Evaluator Quality loop Conditional + counter
Supervisor Agent coordination Command
HITL Human approval interrupt + checkpointer

Expand your agent's capabilities with these related and highly-rated skills.

Didn't find tool you were looking for?

Be as detailed as possible for better results