langgraph-patterns-expert（langgraph 模式专家）

LangGraph Patterns Expert Skill

Purpose

Master LangGraph for building production-ready AI agents with fine-grained control, checkpointing, streaming, and complex state management.

Core Philosophy

LangGraph is: An orchestration framework with both declarative and imperative APIs focused on control and durability for production agents.

Not: High-level abstractions that hide complexity - instead provides building blocks for full control.

Migration: LangGraph replaces legacy AgentExecutor - migrate all old code.

The Six Production Features

1. Parallelization - Run multiple nodes concurrently
2. Streaming - Real-time partial outputs
3. Checkpointing - Pause/resume execution
4. Human-in-the-Loop - Approval/correction workflows
5. Tracing - Observability and debugging
6. Task Queue - Asynchronous job processing

Graph-Based Architecture

from langgraph.graph import StateGraph, END

# Define state
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    next_action: str

# Create graph
graph = StateGraph(AgentState)

# Add nodes
graph.add_node("analyze", analyze_node)
graph.add_node("execute", execute_node)
graph.add_node("verify", verify_node)

# Define edges
graph.add_edge("analyze", "execute")
graph.add_conditional_edges(
    "execute",
    should_verify,
    {"yes": "verify", "no": END}
)

# Compile
app = graph.compile()

Core Patterns

Pattern 1: Agent with Tools

from langgraph.prebuilt import create_react_agent

tools = [search_tool, calculator_tool, db_query_tool]

agent = create_react_agent(
    model=llm,
    tools=tools,
    checkpointer=MemorySaver()
)

# Run with streaming
for chunk in agent.stream({"messages": [("user", "Analyze sales data")]}):
    print(chunk)

Pattern 2: Multi-Agent Collaboration

# Supervisor coordinates specialist agents
supervisor_graph = StateGraph(SupervisorState)

supervisor_graph.add_node("supervisor", supervisor_node)
supervisor_graph.add_node("researcher", researcher_agent)
supervisor_graph.add_node("analyst", analyst_agent)
supervisor_graph.add_node("writer", writer_agent)

# Supervisor routes to specialists
supervisor_graph.add_conditional_edges(
    "supervisor",
    route_to_agent,
    {
        "research": "researcher",
        "analyze": "analyst",
        "write": "writer",
        "finish": END
    }
)

Pattern 3: Human-in-the-Loop

from langgraph.checkpoint.sqlite import SqliteSaver

checkpointer = SqliteSaver.from_conn_string("checkpoints.db")

graph = StateGraph(State)
graph.add_node("propose_action", propose)
graph.add_node("human_approval", interrupt())  # Pauses here
graph.add_node("execute_action", execute)

app = graph.compile(checkpointer=checkpointer)

# Run until human input needed
result = app.invoke(input, config={"configurable": {"thread_id": "123"}})

# Human reviews, then resume
app.invoke(None, config={"configurable": {"thread_id": "123"}})

State Management

Short-Term Memory (Session)

class ConversationState(TypedDict):
    messages: Annotated[list, add_messages]
    context: dict

checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)

# Maintains context across turns
config = {"configurable": {"thread_id": "user_123"}}
app.invoke({"messages": [("user", "Hello")]}, config)
app.invoke({"messages": [("user", "What did I just say?")]}, config)

Long-Term Memory (Persistent)

from langgraph.checkpoint.postgres import PostgresSaver

checkpointer = PostgresSaver.from_conn_string(db_url)

# Persists across sessions
app = graph.compile(checkpointer=checkpointer)

Advanced Control Flow

Conditional Routing

def route_next(state):
    if state["confidence"] > 0.9:
        return "approve"
    elif state["confidence"] > 0.5:
        return "review"
    else:
        return "reject"

graph.add_conditional_edges(
    "classifier",
    route_next,
    {
        "approve": "auto_approve",
        "review": "human_review",
        "reject": "reject_node"
    }
)

Cycles and Loops

def should_continue(state):
    if state["iterations"] < 3 and not state["success"]:
        return "retry"
    return "finish"

graph.add_conditional_edges(
    "process",
    should_continue,
    {"retry": "process", "finish": END}
)

Parallel Execution

from langgraph.graph import START

# Fan out to parallel nodes
graph.add_edge(START, ["agent_a", "agent_b", "agent_c"])

# Fan in to aggregator
graph.add_edge(["agent_a", "agent_b", "agent_c"], "synthesize")

Production Deployment

Streaming for UX

async for event in app.astream_events(input, version="v2"):
    if event["event"] == "on_chat_model_stream":
        print(event["data"]["chunk"].content, end="")

Error Handling

def error_handler(state):
    try:
        return execute_risky_operation(state)
    except Exception as e:
        return {"error": str(e), "next": "fallback"}

graph.add_node("risky_op", error_handler)
graph.add_conditional_edges(
    "risky_op",
    lambda s: "fallback" if "error" in s else "success"
)

Monitoring with LangSmith

import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "..."

# All agent actions automatically logged to LangSmith
app.invoke(input)

Best Practices

DO: ✅ Use checkpointing for long-running tasks ✅ Stream outputs for better UX ✅ Implement human approval for critical actions ✅ Use conditional edges for complex routing ✅ Leverage parallel execution when possible ✅ Monitor with LangSmith in production

DON'T: ❌ Use AgentExecutor (deprecated) ❌ Skip error handling on nodes ❌ Forget to set thread_id for stateful conversations ❌ Over-complicate graphs unnecessarily ❌ Ignore memory management for long conversations

Integration Examples

With Claude

from langchain_anthropic import ChatAnthropic

llm = ChatAnthropic(model="claude-sonnet-4-5")
agent = create_react_agent(llm, tools)

With OpenAI

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o")
agent = create_react_agent(llm, tools)

With MCP Servers

from langchain_mcp import MCPTool

github_tool = MCPTool.from_server("github-mcp")
tools = [github_tool, ...]
agent = create_react_agent(llm, tools)

Decision Framework

Use LangGraph when:

• Need fine-grained control over agent execution
• Building complex state machines
• Require human-in-the-loop workflows
• Want production-grade durability (checkpointing)
• Need to support multiple LLM providers

Use alternatives when:

• Want managed platform (use OpenAI AgentKit)
• Need visual builder (use AgentKit)
• Want simpler API (use Claude SDK directly)
• Building on Oracle Cloud only (use Oracle ADK)

Resources

• Docs: https://langchain-ai.github.io/langgraph/
• GitHub: https://github.com/langchain-ai/langgraph
• Tutorials: https://langchain-ai.github.io/langgraph/tutorials/

*LangGraph is the production-grade choice for complex agentic workflows requiring maximum control.*