跳到主要内容

LangGraph 深度解读

LangGraph 是 LangChain 团队推出的 Agent 编排框架，将 Agent 工作流建模为有向图（Graph），节点是计算步骤，边是条件路由。它解决了复杂 Agent 的状态管理、循环控制和人工介入问题，是当前最流行的多 Agent 编排框架之一。

GitHub: https://github.com/langchain-ai/langgraph

前置知识​

项目定位​

LangGraph 解决的核心问题：复杂 Agent 工作流如何可视化、可控制、可调试？为什么不能只用一个 while 循环？

核心概念​

概念说明
State全局状态对象，在节点间传递，类似 reducer 模式
Node计算单元，可以是 LLM 调用、工具执行、条件判断
Edge节点间的连接，分为普通边和条件边
Checkpoint状态快照，支持断点恢复和时间旅行调试
Human-in-the-loop在节点间暂停，等待人工审批后继续

完整项目文件树​

langgraph/
├── langgraph/
│   ├── graph/
│   │   ├── state.py              # StateGraph 定义
│   │   ├── graph.py              # 图执行引擎
│   │   ├── node.py               # 节点定义
│   │   └── edge.py               # 边和条件路由
│   ├── prebuilt/
│   │   ├── react_agent.py        # 预构建的 ReAct Agent
│   │   ├── tool_node.py           # 工具执行节点
│   │   └── chat_agent.py         # 对话式 Agent
│   ├── checkpoints/
│   │   ├── base.py               # Checkpoint 抽象基类
│   │   ├── memory.py             # 内存存储
│   │   └── sqlite.py             # SQLite 持久化
│   ├── prebuilt/
│   │   └── interfaces.py         # 工具接口定义
│   ├── types.py                   # 类型定义
│   └── errors.py                 # 异常类型
└── docs/

关键模块解析​

1. StateGraph — 图的核心​

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]  # 消息累积
    tool_calls: list

# 定义节点
def call_model(state: AgentState):
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

def call_tools(state: AgentState):
    tool_call = state["messages"][-1].tool_calls[0]
    result = tools.execute(tool_call)
    return {"messages": [result]}

# 条件路由
def should_continue(state: AgentState):
    last_message = state["messages"][-1]
    return "tools" if last_message.tool_calls else END

# 构建图
workflow = StateGraph(AgentState)
workflow.add_node("agent", call_model)
workflow.add_node("tools", call_tools)
workflow.set_entry_point("agent")
workflow.add_conditional_edges("agent", should_continue, {
    "tools": "tools",
    END: END
})
workflow.add_edge("tools", "agent")  # 工具执行后回到 agent

app = workflow.compile()

2. 状态管理 — Reducer 模式​

LangGraph 的状态更新采用 reducer 模式：

  • 每个节点返回一个 partial state
  • reducer 函数（如 operator.add）决定如何合并到全局状态
  • 这样多个节点可以并行执行，结果自动合并

3. Checkpoint — 断点恢复​

from langgraph.checkpoint.sqlite import SqliteSaver

checkpointer = SqliteSaver.from_conn_string("checkpoints.db")
app = workflow.compile(checkpointer=checkpointer)

# 执行时传入线程 ID，状态自动持久化
config = {"configurable": {"thread_id": "user-123"}}
result = app.invoke(initial_state, config=config)

# 可以查看历史状态
history = list(app.get_state_history(config))

这是 LangGraph 区别于其他 Agent 框架的核心能力——每个步骤的状态都被记录，支持：

  • 断点恢复：服务重启后从上次状态继续
  • 时间旅行：回退到任意历史状态重新执行
  • Human-in-the-loop：在关键节点暂停，人工审批后继续

4. 条件路由 — Agent 的决策点​

条件边是 LangGraph 实现 Agent 决策循环的关键：

def route(state: AgentState):
    """根据当前状态决定下一步走哪个节点"""
    if error_occurred(state):
        return "error_handler"
    if needs_human_approval(state):
        return "human_approval"
    if task_complete(state):
        return END
    return "continue"

FDE 实战要点​

  1. 复杂工作流用图，简单用循环：单一 Agent 的 ReAct 循环用 Prebuilt ReactAgent 即可；多 Agent 协作、分支逻辑、人工审批才需要 StateGraph
  2. State 设计很关键：定义清晰的状态 schema 和 reducer，是图可维护的基础
  3. Checkpoint 必须配持久化：生产环境用 SQLite/PostgreSQL checkpointer，不要用内存版
  4. 和 MCP 配合：LangGraph 的 ToolNode 可以直接包装 MCP Client，让图中的节点调用 MCP Server 的工具

相关链接​