🔬 LangGraph

🔬 DEEP ⭐ 20k+ Python Graph / Pregel

LangGraph — 基于 Pregel 超步模型的有状态图执行框架 · 深度源码分析 · 2026-05-17 18:59

1. 🏗️ 核心架构：Pregel 执行模型

LangGraph 的执行引擎灵感来自 Google 的 Pregel 图计算框架，采用 BSP（Bulk Synchronous Parallel） 超步模型：

2. 🔧 StateGraph 构建器模式

StateGraph 是图的定义层——用户通过声明式 API 描述节点、边和状态结构：

class StateGraph(Generic[StateT, ContextT, InputT, OutputT]):
    """A graph whose nodes communicate by reading and writing to a shared state."""

    def add_node(self, name, func, ...)
    def add_edge(self, start, end)          # 固定转移
    def add_conditional_edges(self, start, path, ...)  # 条件路由
    def compile(self, checkpointer=..., ...) -> CompiledStateGraph

状态定义与 Reducer

状态通过 Annotated 类型 + reducer 函数定义——reducer 决定多个并行节点写入同一 key 时的合并策略：

from typing import Annotated
from typing_extensions import TypedDict
import operator

class State(TypedDict):
    messages: Annotated[list, add_messages]  # reducer: 追加消息
    count: Annotated[int, operator.add]      # reducer: 求和
    name: str                                # 无 reducer: 最后写入覆盖

条件边

def route_by_intent(state):
    if "search" in state["messages"][-1].content:
        return "search_node"
    return "chat_node"

graph.add_conditional_edges("router", route_by_intent)

3. ⚡ 编译与执行流程

编译过程

graph = StateGraph(State)
graph.add_node("think", think_fn)
graph.add_node("act", act_fn)
graph.add_edge("think", "act")
graph.add_edge("act", END)

# 编译：将声明式图定义转换为可执行的 Pregel 实例
compiled = graph.compile(checkpointer=SqliteSaver("db.sqlite"))

# 执行
result = compiled.invoke({"messages": [...]})

流式输出

# 事件流
for event in compiled.astream({"messages": [...]}, stream_mode="values"):
    print(event)

# 支持的 stream_mode:
# "values"    — 每步后的完整状态
# "updates"   — 每步的状态增量
# "messages"  — LLM token 流
# "custom"    — 自定义事件

4. 💾 检查点系统（独有）

LangGraph 拥有所有框架中最复杂的检查点系统——这是其最大的实用优势：

检查点内容

时间旅行（Time Travel）

# 获取线程的所有检查点历史
history = list(graph.get_state_history(config))

# 从特定检查点回放
graph.invoke(
    None,
    {**config, "configurable": {"checkpoint_id": checkpoint_id}}
)

# 查看特定检查点的状态
state = graph.get_state(config)
print(state.values)      # 当前状态
print(state.next)        # 下一步要执行的节点
print(state.tasks)       # 待执行任务

5. 🗂️ 状态管理与通道类型

通道类型体系

消息通道（Message Channel）

from langgraph.graph.message import add_messages

def add_messages(left: list, right: list) -> list:
    """智能消息合并：
    - 追加新消息
    - 按 ID 更新已有消息
    - 通过 RemoveMessage 删除消息
    """
    ...

Context 系统（新特性）

class Context(TypedDict):
    user_id: str
    db_conn: Any

graph = StateGraph(state_schema=State, context_schema=Context)

def node(state: State, runtime: Runtime[Context]) -> dict:
    user_id = runtime.context["user_id"]  # 只读，运行作用域
    return {"result": process(state, user_id)}

6. 🎯 Command 控制流原语

Command 是 LangGraph 最强大的控制流机制——将多种操作统一为单一对象：

from langgraph.types import Command

# 状态更新 + 路由
Command(update={"key": "value"}, goto="node_name")

# 中断恢复
Command(resume="user_input")

# 上报父图
Command(update={"result": x}, graph=Command.PARENT)

# 组合使用
Command(
    update={"approved": True},
    goto="execute",
    resume=None,
    graph=None
)

7. 🤖 多 Agent 模式

LangGraph 支持多种多 Agent 协作模式：

graph.add_conditional_edges(
    "supervisor",
    lambda s: s["next_agent"]
)

def transfer_to_b(state):
    return Command(goto="agent_b")

parent.add_node("team", sub_graph)

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

Send API（动态并行扇出）

from langgraph.types import Send

def route_to_workers(state):
    """动态决定并行执行路径——运行时才知道有多少任务"""
    return [
        Send("worker_node", {"task": task})
        for task in state["tasks"]
    ]

graph.add_conditional_edges("planner", route_to_workers)

# 所有 worker 完成后，结果通过 reducer 合并
# 然后继续执行后续节点

8. 🧑‍💻 人机协作（HITL）

LangGraph 提供了所有框架中最健壮的人机协作实现——基于检查点中断：

graph = builder.compile(
    checkpointer=SqliteSaver("db.sqlite"),
    interrupt_before=["human_review"],  # 在节点执行前暂停
    interrupt_after=["ai_response"],    # 在节点执行后暂停
)

# 第一步：执行到中断点
result = graph.invoke({"messages": [...]}, config)

# 查看当前状态
state = graph.get_state(config)
print(state.values)  # 当前状态
print(state.next)    # 下一步: ["human_review"]

# 第二步：人工审查/修改后继续
# 方式1：直接继续
graph.invoke(None, config)

# 方式2：修改状态后继续
graph.invoke(
    Command(update={"approved": True}),
    config
)

# 方式3：人工修改消息后继续
state = graph.get_state(config)
state.values["messages"][-1].content = "修正后的内容"
graph.invoke(None, config)

9. 📝 提示工程

LangGraph 是低级图框架——不强制提示模板。图结构本身就是控制流，提示只是节点的实现之一。

# 提示完全由节点函数控制
def my_agent_node(state):
    system_prompt = """你是一个专业的分析师。
    请根据上下文分析用户的问题。"""
    
    response = llm.invoke([
        SystemMessage(content=system_prompt),
        *state["messages"]
    ])
    return {"messages": [response]}

graph.add_node("agent", my_agent_node)

10. 🛡️ 错误处理与重试

节点级错误处理器

def error_handler(state, error):
    """节点级错误回调——可以返回状态更新"""
    return {"error_message": str(error), "status": "failed"}

graph.add_node(
    "risky_node",
    risky_func,
    error_handler=error_handler  # 捕获该节点所有异常
)

重试策略

from langgraph.pregel import RetryPolicy

graph.add_node(
    "api_call",
    call_external_api,
    retry=RetryPolicy(
        max_attempts=3,          # 最大重试次数
        initial_interval=1.0,    # 初始间隔（秒）
        backoff_factor=2.0,      # 退避因子
        max_interval=10.0,       # 最大间隔
    )
)

检查点恢复

# 出错后从上一个检查点恢复
state = graph.get_state(config)
# 修复状态或跳过错误节点
graph.invoke(
    Command(goto="fallback_node"),
    config
)

11. ✨ 独有特性总结

12. 📂 关键源码文件

13. 🧱 核心类

14. 🧠 深度洞察

15. ⚖️ 优势与劣势

🎯 适合场景