LangGraph 状态图编排实战
构建生产级 AI Agent 工作流

想象这个场景：你花了一周构建了一个 AI 客服 Agent，它在测试环境运行完美——能准确理解用户意图、调用正确的工具、给出专业回答。但上线第一天就出了问题：

• 服务器重启后，所有进行中的对话全部丢失
• 用户要求"稍等一下，我去查下订单"，Agent 无法等待
• 某个工具调用超时，整个工作流崩溃，用户需要重新描述问题
• 你想看看 Agent 为什么做出了错误决策，但没有任何执行日志

本教程将带你掌握 LangGraph 1.0 的状态图编排能力，这是一个由 LangChain 团队开发的生产级 AI 工作流框架，已被 NVIDIA、IBM 等公司用于构建关键业务系统。学完本教程后，你将能够：

首先创建项目并安装依赖：

# 创建项目目录
mkdir langgraph-tutorial && cd langgraph-tutorial

# 初始化 Python 项目
python -m venv .venv
source .venv/bin/activate  # Windows: .venv\Scripts\activate

# 安装依赖
pip install langgraph langchain-openai langchain-community
pip install psycopg2-binary  # PostgreSQL 驱动

创建 .env 文件配置 API 密钥：

OPENAI_API_KEY=sk-xxx
DATABASE_URL=postgresql://user:pass@localhost:5432/langgraph_demo

状态是 LangGraph 的核心抽象。所有节点共享同一个状态对象，节点接收状态、返回更新。我们使用 Python 的 TypedDict 定义状态结构：

from typing import TypedDict, Annotated, List, Literal
import operator

class ResearchState(TypedDict):
    # 用户输入的研究主题
    query: str
    # 累积消息历史（使用 reducer 确保追加而非覆盖）
    messages: Annotated[List[str], operator.add]
    # 搜索到的来源 URL 列表
    sources: Annotated[List[str], operator.add]
    # LLM 生成的综合内容
    synthesis: str
    # 人类审批结果
    is_approved: bool
    # 错误信息（如果有）
    error: str

节点是状态图的处理单元。每个节点是一个纯函数：接收状态字典，返回要更新的字段。节点应该无副作用——不直接修改外部状态，所有变更通过返回值体现。

from langchain_openai import ChatOpenAI

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

def search_node(state: ResearchState) -> dict:
    """搜索网络获取相关资料"""
    query = state["query"]
    # 实际项目中这里调用搜索 API（如 Tavily、Exa）
    sources = [
        f"https://example.com/result1?q={query}",
        f"https://example.com/result2?q={query}",
    ]
    return {"sources": sources}

def synthesize_node(state: ResearchState) -> dict:
    """使用 LLM 综合搜索结果生成报告"""
    context = "\n".join(state["sources"])
    prompt = f"基于以下资料综合研究报告：{context}"
    response = llm.invoke(prompt)
    return {"synthesis": response.content}

def human_review_node(state: ResearchState) -> dict:
    """等待人类审批（实际项目中通过 API 实现）"""
    # 这里会触发 interrupt_before 暂停
    # 审批后通过 update_state 设置 is_approved
    return {"is_approved": state.get("is_approved", False)}

有了状态和节点，现在使用 StateGraph 构建完整的执行流程。关键点在于条件边——它允许根据状态动态决定下一步。

from langgraph.graph import StateGraph, START, END

# 1. 初始化图
builder = StateGraph(ResearchState)

# 2. 添加节点
builder.add_node("search", search_node)
builder.add_node("synthesize", synthesize_node)
builder.add_node("review", human_review_node)
builder.add_node("publish", publish_node)
builder.add_node("revise", revise_node)

# 3. 定义边（固定流程）
builder.add_edge(START, "search")
builder.add_edge("search", "synthesize")
builder.add_edge("synthesize", "review")

# 4. 条件边：根据审批结果决定走向
def route_after_review(state: ResearchState) -> Literal["publish", "revise"]:
    if state["is_approved"]:
        return "publish"
    else:
        return "revise"

builder.add_conditional_edges(
    "review",
    route_after_review,
    {"publish": "publish", "revise": "revise"}
)

# 5. 循环：修订后回到综合步骤
builder.add_edge("revise", "synthesize")
builder.add_edge("publish", END)

# 6. 编译（添加持久化）
from langgraph.checkpoint.postgres import PostgresSaver

graph = builder.compile(
    checkpointer=PostgresSaver.from_conn_string(
        "postgresql://user:pass@localhost:5432/langgraph_demo"
    ),
    interrupt_before=["review"]  # 在审批节点前暂停
)

现在执行工作流。注意 interrupt_before=["review"] 配置——这会让图在执行到 review 节点前自动暂停，等待人类决策。

# 配置线程（用于持久化会话）
config = {"configurable": {"thread_id": "session-001"}}

# 启动工作流
result = graph.invoke({
    "query": "2026 年 AI Agent 工作流引擎技术趋势",
    "messages": [],
    "sources": [],
    "synthesis": "",
    "is_approved": False,
}, config)

# 检查暂停状态
snapshot = graph.get_state(config)
print(f"下一步执行：{snapshot.next}")  # 输出：['review']

# 查看当前状态
current_state = snapshot.values
print(f"已收集来源：{len(current_state['sources'])} 个")

暂停后，你可以通过外部系统（Web 界面、Slack 机器人等）展示当前结果，等待用户审批：

# 用户批准后更新状态
graph.update_state(config, {
    "is_approved": True,
    "messages": ["用户已批准，继续发布"]
})

# 从断点继续执行
final_result = graph.invoke(None, config)

LangGraph 的检查点机制会在每个 Super-Step 后自动保存状态到数据库。这意味着：

• 服务器重启后，从最近的检查点恢复
• 用户间隔几天继续对话，历史状态完整保留
• 调试时可以回放任意时间点的状态

# 查看历史检查点
from langgraph.checkpoint.postgres import PostgresSaver

saver = PostgresSaver.from_conn_string(DATABASE_URL)

# 获取某个线程的所有检查点
checkpoints = saver.list({"configurable": {"thread_id": "session-001"}})

for checkpoint in checkpoints:
    print(f"时间：{checkpoint['created_at']}")
    print(f"状态：{checkpoint['channel_values']}")
    print("---")

# 回滚到特定检查点
target_checkpoint = checkpoints[-2]  # 倒数第二个
graph.update_state(config, target_checkpoint["channel_values"])

LangGraph Studio 是官方可视化工具，可以实时查看状态图的执行过程：

# 安装 LangGraph Studio
pip install langgraph-cli

# 启动开发服务器
langgraph dev

# 在浏览器打开 http://localhost:8123

Studio 提供以下能力：

• 可视化执行轨迹：看到状态如何在节点间流转
• 状态检查：点击任意节点查看当时的状态值
• 时间旅行调试：回放到历史检查点重新执行
• 人工审批界面：直接在 UI 中批准或拒绝

最后，补充生产环境必备的错误处理和优化策略：

import logging
from tenacity import retry, stop_after_attempt, wait_exponential

logger = logging.getLogger(__name__)

# 重试装饰器
@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=10)
)
def search_with_retry(query: str):
    return search_api(query)

# 在节点内捕获异常
def safe_search_node(state: ResearchState) -> dict:
    try:
        sources = search_with_retry(state["query"])
        return {"sources": sources}
    except Exception as e:
        logger.error(f"搜索失败：{e}")
        return {"error": str(e)}

# 设置最大迭代次数防止无限循环
graph = builder.compile(
    checkpointer=saver,
    interrupt_before=["review"]
)

# 执行时限制最大步数
result = graph.invoke(
    input_data,
    config,
    stream_mode="updates",  # 流式输出中间状态
    recursion_limit=25  # 防止无限循环
)