LangGraph 学习笔记

目录录

LangGraph 是一个用于构建有状态、可控制流程的 AI 智能体（Agent）的低级编排框架。它将工作流建模为图（Graph），由节点（Node）、边（Edge） 和状态（State） 三大核心要素组成。通过这种方式，你可以精确控制 LLM 应用的每一步，并轻松实现持久化、人工干预、故障恢复等生产级功能。

这份指南将带你从零开始，通过大量可运行的代码示例，深入理解 LangGraph 的每一个核心概念。

第 1 章：快速入门 —— 并行搜索问答系统

我们先从一个简单的 RAG 示例入手，感受 LangGraph 的基本用法。这个系统会并行地从“技术知识库”和“实时网络”获取信息，然后综合生成回答。

完整代码（可直接运行）

# quickstart.py
import time
from typing import TypedDict

from langgraph.constants import START, END
from langgraph.graph import StateGraph

# ---------- 1. 定义图状态 ----------
class MyState(TypedDict):
    query: str               # 用户输入
    rag_result: str          # 知识库检索结果
    web_search_result: str   # 网络搜索结果
    final_answer: str        # 最终回答

# ---------- 2. 定义节点函数 ----------
def rag_search_node(state: MyState):
    print("【技术知识库】开始检索学术定义...")
    query = state["query"]
    time.sleep(2)  # 模拟 I/O 延迟
    rag_result = f"【学术定义】'{query}'：指大语言模型（LLM）生成的内容看似合理，但实际上与事实不符的现象。"
    print("【技术知识库】检索完成！")
    return {"rag_result": rag_result}   # 仅返回更新的字段

def web_search_node(state: MyState):
    print("【实时网络搜索】全网查询最新案例...")
    query = state["query"]
    time.sleep(2)
    web_result = f"【通俗解释】'{query}'：常被戏称为 AI 在'一本正经地胡说八道'。"
    print("【实时网络搜索】搜索完毕！")
    return {"web_search_result": web_result}

def final_answer_node(state: MyState):
    print("【AI助手】正在综合信息生成回答...")
    rag = state["rag_result"]
    web = state["web_search_result"]
    time.sleep(1)
    final = f"【综合回答】\n{rag}\n{web}\n需要我教你如何降低幻觉率吗？"
    print("【AI助手】回答生成结束！")
    return {"final_answer": final}

# ---------- 3. 构建图 ----------
builder = StateGraph(MyState)          # 创建图构建器
builder.add_node("rag", rag_search_node)
builder.add_node("web", web_search_node)
builder.add_node("final", final_answer_node)

# 添加边：并行从 START 出发
builder.add_edge(START, "rag")
builder.add_edge(START, "web")
# 两个搜索完成后汇聚到 final
builder.add_edge("rag", "final")
builder.add_edge("web", "final")
builder.add_edge("final", END)

# 编译图
app = builder.compile()

# ---------- 4. 执行工作流 ----------
if __name__ == "__main__":
    initial_state = {"query": "大模型中的'幻觉'是什么意思？"}
    final_state = app.invoke(initial_state)
    print("\n" + "="*50)
    print("最终答案：")
    print(final_state["final_answer"])

    # 可选：打印图结构（需安装 grandalf）
    # app.get_graph().print_ascii()

运行结果（控制台输出）：

  • 两个搜索节点几乎同时启动，然后 final 节点等待二者完成后执行。
  • 最终答案会整合两条搜索信息。

解释：

  • StateGraph(MyState) 指定了全局状态类型。
  • 节点函数接收 state，返回一个字典，LangGraph 自动用返回的字典更新全局状态（默认是覆盖策略）。
  • add_edge(START, "rag") 和 add_edge(START, "web") 使得两个节点并行执行，这是 LangGraph 的重要特性。

第 2 章：状态（State）的三种定义方式

在 LangGraph 中，状态是节点之间共享数据的载体。我们可以用三种方式来定义状态类，各有优劣。

2.1 方式一：TypedDict（推荐）

TypedDict 是 Python 标准库的类型提示工具，它不进行运行时校验，但能提供 IDE 智能提示，有效防止键名拼写错误。

from typing import TypedDict, List, Optional
from typing_extensions import NotRequired

class MyTypedState(TypedDict):
    user_query: str
    history: List[str]                 # 必需
    documents: NotRequired[List[dict]] # 可选字段

优点：轻量、灵活，与普通字典完全兼容，是官方最推荐的方式。

2.2 方式二：Pydantic BaseModel（运行时校验）

BaseModel 会在实例化时对数据进行类型校验，并支持自动类型转换（如字符串 “30” 自动转为 int）。它还支持自定义验证器。

from pydantic import BaseModel, Field, field_validator
from datetime import datetime
from typing import Optional

class MyPydanticState(BaseModel):
    user_id: str = Field(description="用户唯一标识")
    query: str = Field(min_length=1, max_length=100)
    timestamp: datetime = Field(default_factory=datetime.now)
    rag_result: Optional[str] = None

    @field_validator('query')
    @classmethod
    def query_not_empty(cls, v: str) -> str:
        if not v.strip():
            raise ValueError("查询不能为空")
        return v

# 使用
state = MyPydanticState(user_id="u123", query="什么是 LangGraph？")
print(state.query)          # 访问属性
print(state.model_dump())   # 转为字典

优点：数据校验、自动转换、序列化方法齐全，适合需要严格数据质量的场景。

2.3 方式三：@dataclass（减少样板代码）

dataclass 自动生成 __init__、__repr__ 等方法，使类定义更简洁，但无运行时校验（除非结合 pydantic.dataclasses）。

from dataclasses import dataclass

@dataclass
class MyDataClassState:
    query: str
    results: list
    completed: bool = False

# 使用
state = MyDataClassState(query="天气", results=[])

优点：代码量少，适合轻量级的内部状态。

总结选择：

方式 类型提示 运行时校验 自动转换 推荐场景
TypedDict ✅ ❌ ❌ 绝大多数 LangGraph 应用，轻量快速
BaseModel ✅ ✅ ✅ 需要严格校验、复杂验证逻辑时
@dataclass ✅ ❌ ❌ 内部简单数据容器

第 3 章：输入/输出与节点数据隔离（精细控制数据流）

LangGraph 允许我们定义三种不同的“视图”来控制哪些数据能进入图、哪些能离开图，以及每个节点能看到什么数据。

3.1 state_schema（全局状态蓝图）

定义整个图内部所有可能用到的数据字段。所有节点共享这个“中央数据库”。

3.2 input_schema 与 output_schema

  • input_schema：限制外部调用时可以传入哪些字段（比如只允许传 query），防止无关数据污染内部状态。
  • output_schema：限制图执行完毕后向外返回哪些字段（比如只返回 final_answer），隐藏中间计算细节。

二者都必须是 state_schema 的子集。

3.3 节点级状态隔离

在定义节点函数时，可以为该函数的参数指定一个更小的状态类型，使该节点只能访问所需的特定字段。这相当于为每个节点创建了一个“沙盒”。

完整代码示例：

# data_isolation_demo.py
import time
from typing import TypedDict

from langgraph.constants import START, END
from langgraph.graph import StateGraph

# ------ 1. 定义三种 Schema ------
class FullState(TypedDict):
    query: str
    rag_result: str
    web_result: str
    final_answer: str
    internal_debug: str   # 仅内部使用，不对外暴露

class InputState(TypedDict):
    query: str            # 外部只能传入 query

class OutputState(TypedDict):
    final_answer: str     # 外部只能看到最终答案

# 为节点单独定义需要的子集
class SearchNodeState(TypedDict):
    query: str            # 搜索节点只需要 query

class FinalNodeState(TypedDict):
    rag_result: str
    web_result: str
    # 注意：final 节点不需要 query

# ------ 2. 节点函数（参数使用子集类型）------
def rag_node(state: SearchNodeState):   # 只能读 query
    print(f"RAG 搜索: {state['query']}")
    time.sleep(1)
    return {"rag_result": "知识库结果：..."}

def web_node(state: SearchNodeState):
    print(f"Web 搜索: {state['query']}")
    time.sleep(1)
    return {"web_result": "网络结果：..."}

def final_node(state: FinalNodeState):  # 只能读 rag_result 和 web_result
    combined = state["rag_result"] + state["web_result"]
    return {"final_answer": f"综合：{combined}", "internal_debug": "处理完成"}

# ------ 3. 构建图并指定输入/输出 Schema ------
builder = StateGraph(
    state_schema=FullState,
    input_schema=InputState,
    output_schema=OutputState
)
builder.add_node("rag", rag_node)
builder.add_node("web", web_node)
builder.add_node("final", final_node)

builder.add_edge(START, "rag")
builder.add_edge(START, "web")
builder.add_edge("rag", "final")
builder.add_edge("web", "final")
builder.add_edge("final", END)

app = builder.compile()

# ------ 4. 执行 ------
if __name__ == "__main__":
    # 外部只能传入 query
    result = app.invoke({"query": "LangGraph 是什么？"})
    # 输出只有 final_answer
    print(result)   # {'final_answer': '综合：知识库结果：...网络结果：...'}
    # internal_debug 不会出现在结果中

关键点：

  • input_schema 过滤了输入，即使传入额外字段也会被忽略或报错。
  • output_schema 过滤了输出，内部调试信息不会泄露。
  • 节点函数参数类型限定了它们可以访问的数据，试图访问未定义的字段会导致类型检查报错（若使用静态检查工具）。

第 4 章：Reducer 函数 —— 状态合并的核心引擎

当节点返回更新数据时，LangGraph 需要将这些增量与全局状态合并。这个合并逻辑由 Reducer 控制。每个字段都可以指定一个 Reducer，以决定如何处理新值与旧值的关系。

4.1 默认行为：覆盖（Overwrite）

如果没有指定 Reducer，新值会直接覆盖旧值。

class State(TypedDict):
    counter: int       # 默认覆盖

4.2 内置 Reducer：operator.add 与 add_messages

对于列表或字符串，我们通常希望追加而不是覆盖。LangGraph 提供了两个常用的内置 Reducer：

  • operator.add：通用的追加操作，适用于任何支持 + 的类型（列表、字符串等）。
  • add_messages：专门为 List[BaseMessage] 设计，它会智能处理消息 ID（如果新消息与旧消息 ID 相同，则替换而非追加，常用于对话重试场景）。

代码示例（对比覆盖与追加）：

# reducer_demo.py
import operator
from typing import TypedDict, List, Annotated
from langchain_core.messages import HumanMessage, AIMessage, BaseMessage
from langgraph.graph import StateGraph, add_messages, START, END

# 定义一个状态，其中 messages 使用 add_messages，results 使用 operator.add
class AgentState(TypedDict):
    query: str
    messages: Annotated[List[BaseMessage], add_messages]   # 智能追加
    results: Annotated[List[str], operator.add]            # 简单追加

def node1(state: AgentState):
    return {
        "messages": [HumanMessage(content="你好")],
        "results": ["第一次结果"]
    }

def node2(state: AgentState):
    return {
        "messages": [AIMessage(content="你好！")],
        "results": ["第二次结果"]
    }

builder = StateGraph(AgentState)
builder.add_node("n1", node1)
builder.add_node("n2", node2)
builder.add_edge(START, "n1")
builder.add_edge("n1", "n2")
builder.add_edge("n2", END)
app = builder.compile()

final = app.invoke({"query": "hi", "messages": [], "results": []})
print(final["messages"])   # [HumanMessage, AIMessage]（追加）
print(final["results"])    # ['第一次结果', '第二次结果']（追加）

add_messages 的“智能”之处：
如果新消息中包含与旧消息 id 相同的消息（LangChain 消息默认带有 id），则会替换旧消息，而不是简单追加。这对于需要修正或重试消息的场景非常有用。

4.3 自定义 Reducer

当内置 Reducer 不能满足需求时，可以自定义函数。函数签名为 (existing_value, new_value) -> updated_value。

示例：给每个追加的搜索结果加上时间戳。

from datetime import datetime

def timestamped_append(existing: List[str], new: List[str]) -> List[str]:
    if existing is None:
        existing = []
    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    processed = [f"[{timestamp}] {item}" for item in new]
    return existing + processed

class MyState(TypedDict):
    logs: Annotated[List[str], timestamped_append]

# 在节点中返回 {"logs": ["新日志"]}，就会被自动加上时间戳并追加。

4.4 并行节点的状态合并

当多个节点并行执行并同时更新同一个带有 Reducer 的字段时，LangGraph 会安全地合并它们。合并顺序通常与节点添加顺序或边的定义顺序有关，但重要的是，Reducer 保证了最终结果的一致性。

例如，两个并行节点分别返回 {"logs": ["A"]} 和 {"logs": ["B"]}，如果 logs 使用 operator.add，最终结果可能是 ["A", "B"]（具体顺序有保证）。

第 5 章：从状态中恢复执行 —— 实现故障恢复

这是 LangGraph 最强大的功能之一。在生产环境中，智能体可能运行数小时甚至数天，期间可能遇到各种异常（如 API 超时、网络中断、代码 bug）。如果没有故障恢复机制，每次失败都得从头开始，浪费大量时间和资源。

5.1 核心概念：Checkpointer 与状态持久化

在深入故障恢复之前，我们需要理解两个关键概念：

  1. Checkpointer（检查点管理器）：在图的每个节点执行完毕后，自动保存当前状态的"快照"（Snapshot）。这些快照包含了该时刻的所有状态数据以及执行上下文。

  2. 状态持久化存储：Checkpointer 可以将快照保存到不同介质：

    • 内存（InMemorySaver）：速度快，但进程重启后丢失。
    • SQLite（SqliteSaver）：轻量级文件数据库，适合开发和单机部署。
    • PostgreSQL（PostgresSaver）：生产级关系数据库，支持高并发。
    • Redis（RedisSaver）：高性能键值存储，适合缓存场景。

5.2 故障恢复的完整工作流程

故障恢复的核心流程如下：

1. 第一次执行（正常流程）
   START → Node1 → Node2 (此处发生异常)×❌
   ├─ 在 Node1 执行完成后，Checkpointer 自动保存了状态快照
   └─ 异常发生时，当前状态（包含 Node1 的结果）已经被持久化

2. 开发者修复代码（例如修复 Node2 中的 bug）

3. 第二次执行（恢复流程）
   ├─ 调用 app.invoke(None, config=config)  # 传入 None
   ├─ LangGraph 从 Checkpointer 加载该 thread_id 的最新状态
   ├─ 识别出 next 字段为 ("node2",)  # 知道该从 Node2 继续
   └─ 从 Node2 开始继续执行 → Node3 → END✓✅

5.3 完整代码示例：使用 SQLite 实现故障恢复

下面我们通过一个完整的、可运行的示例来演示故障恢复的全过程。

# state_recovery_complete_demo.py
"""
LangGraph 故障恢复完整演示
演示如何使用 SqliteSaver 实现断点续传
"""

import os
import sqlite3
import time
from typing import TypedDict

from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.constants import START, END
from langgraph.graph import StateGraph


# ---------- 1. 定义图状态 ----------
class MyState(TypedDict):
    """整个图共享的状态"""
    step_1_result: str      # 节点1的结果
    step_2_result: str      # 节点2的结果
    step_3_result: str      # 节点3的结果
    execution_count: int    # 执行计数器，用于演示状态持久化


# ---------- 2. 定义节点函数 ----------
def node_1(state: MyState) -> dict:
    """节点1：模拟数据初始化"""
    print("=" * 50)
    print("【节点1】开始执行...")
    print(f"  当前状态: {state}")
    
    # 模拟一些处理
    time.sleep(1)
    result = "节点1处理完成的数据"
    
    # 更新执行计数
    count = state.get("execution_count", 0) + 1
    
    print(f"  节点1执行完毕，执行次数: {count}")
    print("=" * 50)
    
    return {
        "step_1_result": result,
        "execution_count": count
    }


def node_2(state: MyState) -> dict:
    """节点2：模拟一个可能出错的节点"""
    print("=" * 50)
    print("【节点2】开始执行...")
    print(f"  当前状态: {state}")
    
    # 模拟一些处理
    time.sleep(1)
    result = "节点2处理完成的数据"
    
    #**注意**🔥 模拟故障：第一次执行时抛出异常
    # 注意：这里我们通过检查执行次数来决定是否抛出异常
    # 如果是第一次执行（count == 1），则抛出异常
    # 第二次执行（恢复时），则正常执行
    current_count = state.get("execution_count", 0)
    
    if current_count == 1:
        print("  ❌ 节点2发生异常！(模拟故障)")
        print("=" * 50)
        raise Exception("模拟节点2执行失败：数据库连接超时")
    
    print(f"  节点2执行完毕，当前执行次数: {current_count}")
    print("=" * 50)
    
    return {
        "step_2_result": result,
        "execution_count": current_count + 1
    }


def node_3(state: MyState) -> dict:
    """节点3：正常执行"""
    print("=" * 50)
    print("【节点3】开始执行...")
    print(f"  当前状态: {state}")
    
    time.sleep(1)
    result = "节点3处理完成的数据"
    count = state.get("execution_count", 0) + 1
    
    print(f"  节点3执行完毕，最终执行次数: {count}")
    print("=" * 50)
    
    return {
        "step_3_result": result,
        "execution_count": count
    }


# ---------- 3. 构建图 ----------
def build_graph():
    """构建并返回编译好的图"""
    # 创建图构建器
    workflow = StateGraph(MyState)
    
    # 添加节点
    workflow.add_node("node_1", node_1)
    workflow.add_node("node_2", node_2)
    workflow.add_node("node_3", node_3)
    
    # 添加边：顺序执行
    workflow.add_edge(START, "node_1")
    workflow.add_edge("node_1", "node_2")
    workflow.add_edge("node_2", "node_3")
    workflow.add_edge("node_3", END)
    
    return workflow


# ---------- 4. 创建 Checkpointer ----------
def create_checkpointer():
    """创建 SQLite 检查点管理器"""
    # 确保目录存在
    os.makedirs("./sqlite_data", exist_ok=True)
    
    # 创建数据库连接
    # check_same_thread=False: 允许不同线程使用同一个连接
    # 因为 LangGraph 可能在后台用多线程操作数据库
    conn = sqlite3.connect(
        database="./sqlite_data/langgraph_state.db",
        check_same_thread=False
    )
    
    # 创建 SqliteSaver 实例
    checkpointer = SqliteSaver(conn)
    return checkpointer


# ---------- 5. 演示故障恢复 ----------
def demo_fault_recovery():
    """
    演示完整的故障恢复流程
    分两次执行：
    第一次：节点2失败
    第二次：从失败点恢复，继续执行
    """
    
    print("\n" + "🚀" * 20)
    print("开始 LangGraph 故障恢复演示")
    print("🚀" * 20 + "\n")
    
    # 创建 Checkpointer
    checkpointer = create_checkpointer()
    
    # 构建并编译图（传入 checkpointer）
    workflow = build_graph()
    app = workflow.compile(checkpointer=checkpointer)
    
    # 定义会话配置（使用唯一的 thread_id）
    config = {"configurable": {"thread_id": "demo_session_001"}}
    
    # ---------- 第一次执行（预期失败） ----------
    print("\n" + "=" * 50)
    print("【第一次执行】开始...")
    print("=" * 50 + "\n")
    
    initial_state = {
        "step_1_result": "",
        "step_2_result": "",
        "step_3_result": "",
        "execution_count": 0
    }
    
    try:
        # 注意：这里我们没有捕获异常，让它抛出
        result = app.invoke(initial_state, config=config)
        print("\n✅ 第一次执行成功完成！")
        print(f"最终状态: {result}")
    except Exception as e:
        print(f"\n❌ 第一次执行失败！")
        print(f"错误信息: {e}")
        print("状态已被 Checkpointer 保存，可以进行恢复...")
    
    # ---------- 查看保存的状态 ----------
    print("\n" + "=" * 50)
    print("【查看已保存的状态快照】")
    print("=" * 50)
    
    # 获取最新状态
    snapshot = app.get_state(config=config)
    print(f"当前保存的状态值: {snapshot.values}")
    print(f"下一步要执行的节点: {snapshot.next}")
    print(f"元数据: {snapshot.metadata}")
    
    # ---------- 第二次执行（从失败点恢复） ----------
    print("\n" + "=" * 50)
    print("【第二次执行】从失败点恢复...")
    print("=" * 50 + "\n")
    
    pri提示t("💡 提示：传入 None 作为输入，表示从 Checkpointer 加载状态")
    
    try:
        # 关键：传入 None，从检查点恢复
        recovered_result = app.invoke(None, config=config)
        print("\n✅ 恢复执行成功完成！")
        print("=" * 50)
        print("【最终状态】")
        for key, value in recovered_result.items():
            print(f"  {key}: {value}")
        print("=" * 50)
        
    except Exception as e:
        print(f"\n❌ 恢复执行也失败了: {e}")


# ---------- 6. 查看历史状态（进阶功能） ----------
def demo_history_view():
    """演示如何查看完整的历史状态"""
    print("\n" + "📚" * 20)
    print("查看历史状态快照")
    print("📚" * 20 + "\n")
    
    checkpointer = create_checkpointer()
    workflow = build_graph()
    app = workflow.compile(checkpointer=checkpointer)
    
    config = {"configurable": {"thread_id": "history_demo_001"}}
    
    # 执行一次成功的流程
    try:
        app.invoke({
            "step_1_result": "",
            "step_2_result": "",
            "step_3_result": "",
            "execution_count": 0
        }, config=config)
    except:
        pass  # 忽略可能的异常
    
    # 获取所有历史状态（最新在前）
    print("\n【历史状态列表】")
    history = list(app.get_state_history(config=config))
    
    for i, snapshot in enumerate(history[:5]):  # 只显示前5个
        print(f"\n--- 快照 {i+1} ---")
        print(f"步骤号: {snapshot.metadata.get('step', 'N/A')}")
        print(f"状态值: {snapshot.values}")
        print(f"下一步节点: {snapshot.next}")
        print(f"是否中断: {snapshot.interrupts}")


# ---------- 7. 主程序 ----------
if __name__ == "__main__":
    # 运行故障恢复演示
    demo_fault_recovery()
    
    # 可选：查看历史状态
    # demo_history_view()

5.4 代码详细解释

5.4.1 关键点一：node_2 中的故障模拟
def node_2(state: MyState) -> dict:
    current_count = state.get("execution_count", 0)
    
    if current_count == 1:
        # 第一次执行（count=1）时抛出异常
        raise Exception("模拟节点2执行失败：数据库连接超时")
    
    # 第二次执行（恢复时，count已经变成2）正常执行
    return {"step_2_result": result}

解释：

  • 在第一次执行时，execution_count 从 0 开始，经过 node_1 后变成 1。
  • node_2 检测到 count == 1，抛出异常。
  • 异常抛出时，node_1 的结果已经被 Checkpointer 保存。
  • 第二次执行（恢复）时，我们从检查点加载的状态中 execution_count 已经是 1。
  • node_2 检测到 count == 1 仍然会抛出异常？不对！这里有个重要细节：
    • 恢复执行时，invoke(None, config) 会从检查点加载状态，但不会重新执行已经完成的节点。
    • 它会直接跳到 node_2，此时传入的 state 是从检查点加载的，execution_count 是 1。
    • 如果我们的逻辑是 if current_count == 1: raise Exception，那么恢复时仍然会抛出异常！
    • 正确的做法：应该在恢复执行时，execution_count 应该已经更新（比如在 node_1 中更新为2），或者我们需要其他的标志来判断是否是恢复执行。

让我们修正这个逻辑：

def node_2(state: MyState) -> dict:
    # 使用一个专门的标志来判断是否是恢复执行
    # 或者检查是否已经有 step_2_result 存在
    if "step_2_result" in state and state["step_2_result"]:
        # 已经有结果了，说明是重复执行，直接跳过
          （节点2已经执行过，跳过...）过，跳过...")
        return {}
    
    # 正常执行
    result = "节点2处理完成的数据"
    return {"step_2_result": result}
5.4.2 关键点二：invoke(None, config) 的含义
# 第一次执行（失败）
result = app.invoke(initial_state, config=config)

# 第二次执行（恢复）
recovered_result = app.invoke(None, config=config)

解释：

  • 传入 None 表示：“不要给我新的输入，直接从 Checkpointer 中加载该 thread_id 的最新状态。”
  • LangGraph 会：
    1. 从数据库中查询 thread_id 对应的最新检查点。
    2. 加载该检查点的状态值。
    3. 查看 next 字段，确定应该从哪个节点继续执行。
    4. 从该节点开始继续执行。
  • 如果传入非 None 值，LangGraph 会将其视为新的输入，并尝试与已有状态合并（具体取决于 Reducer）。
5.4.3 关键点三：get_state() 与 get_state_history()
# 获取最新状态快照
snapshot = app.get_state(config=config)
print(snapshot.values)  # 当前状态数据
print(snapshot.next)    # 下一步要执行的节点

# 获取所有历史状态
history = app.get_state_history(config=config)
for snapshot in history:
    print(snapshot.metadata['step'])  # 步骤号
    print(snapshot.values)            # 该步骤的状**StateSnapshot 的完整结构**完整结构**：

| 字段            | 类型                    | 描述                             |
| --------------- | ----------------------- | -------------------------------- |
| `values`        | `dict`                  | 该时刻的完整状态数据             |
| `next`          | `tuple[str, ...]`       | 下一步要执行的节点名称列表       |
| `config`        | `RunnableConfig`        | 可用于从此快照恢复的配置         |
| `metadata`      | `CheckpointMetadata`    | 元数据（步骤号、时间戳、来源等） |
| `parent_config` | `RunnableConfig`        | 父快照的配置（用于回溯）         |
| `interrupts`    | `tuple[Interrupt, ...]` | 中断信息（用于人机协作）         |

### 5.5 实际应用场景

#### 场景一：长时间运行的数据处理

```python
# 处理100万个数据点，每个点调用外部API
# 如果中途API限流或网络中断，可以从断点恢复

config = {"configurable": {"thread_id": "batch_job_20260127"}}

# 第一次运行到第5000条时失败...
try:
    app.invoke({"start_index": 0, "total": 1000000}, config=config)
except Exception as e:
    print(f"任务中断于第 {get_current_progress()} 条")

# 修复问题后，从断点继续
app.invoke(None, config=config)  # 从第5000条继续
场景二：多步骤的人工审核流程
# 贷款审批流程：收集信息 → 信用评分 → 人工审核 → 最终决策
# 人工审核环节可能需要等待数小时甚至数天

# 图构建时设置中断点
from langgraph.checkpoint import interrupt

def human_review_node(state):
    # 等待人工输入，这里会暂停执行
    decision = interrupt({
        "application_id": state["app_id"],
        "message": "请审核此贷款申请"
    })
    state["decision"] = decision
    return state

# 第一次运行，在人工审核节点暂停
app.invoke(initial_state, config=config)  # 会暂停在 human_review_node

# 管理者查看状态
snapshot = app.get_state(config=config)
print(f"待审核数据: {snapshot.values}")

# 管理者做出决策后，恢复执行
# 传入人工决策结果
app.invoke({"decision": "approved"}, config=config)  # 继续执行

5.6 常见问题与最佳实践

问题1：使用内存存储（InMemorySaver）能恢复吗？

不能。InMemorySaver 将状态保存在内存中，进程重启后数据丢失。生产环境应使用 SqliteSaver、PostgresSaver 等持久化方案。

问题2：恢复时传入 None 还是空字典？

传入 None。传入空字典 {} 会被视为新的输入，可能会触发 Reducer 的合并逻辑，导致意外行为。传入 None 明确表示"从检查点加载"。

问题3：如何知道从哪个节点恢复？

通过 snapshot.next 字段。例如：

snapshot = app.get_state(config=config)
if snapshot.next:
    print(f"将从以下节点继续: {snapshot.next}")
else:
    print("图已经执行完成")
最佳实践总结
  1. 始终使用持久化 Checkpointer：生产环境推荐 SqliteSaver 或 PostgresSaver。
  2. 为每个会话分配唯一的 thread_id：可以使用 UUID、用户ID+时间戳等。
  3. 在节点函数中设计为幂等：即使节点被重复执行，也不应产生副作用。
  4. 定期清理旧的检查点：长时间运行的系统会产生大量检查点，需要定期清理以节省存储空间。
  5. 监控检查点大小：如果状态包含大量数据（如长对话历史），检查点可能变大，考虑使用压缩或只保存增量。

5.7 进阶：手动管理检查点

有时你可能需要手动控制检查点的保存和加载：

# 手动保存检查点
from langgraph.checkpoint import BaseCheckpointSaver

class CustomSaver(BaseCheckpointSaver):
    def put(self, config, checkpoint, metadata, new_versions):
        # 自定义保存逻辑
        pass

# 手动加载检查点
config = {"configurable": {"thread_id": "my_id"}}
checkpoint = app.get_state(config)  # 加载最新状态

# 从特定检查点恢复（通过 checkpoint_id）
specific_config = {
    "configurable": {
        "thread_id": "my_id",
        "checkpoint_id": "specific_checkpoint_id"
    }
}
result = app.invoke(None, config=specific_config)

通过这份详尽的指南，您应该已经掌握了 LangGraph 故障恢复的完整机制。核心要点是：

  1. Checkpointer 自动保存状态：每个节点执行后自动保存。
  2. 使用 thread_id 隔离会话：不同会话的状态互不干扰。
  3. invoke(None, config) 实现恢复：从检查点加载状态并继续执行。
  4. get_state() 和 get_state_history() 用于调试：查看当前和历史状态。

现在，您可以构建真正健壮的、能够抵御各种故障的生产级上一版本遗漏了对 Pregel 算法 和 状态快照 的深入讲解。这一部分对于理解 LangGraph 的内部运行机制至关重要。下面，将专门为这部分内容撰写一份详细、完整的学习笔记，包含所有代码、图解和详## 第 6 章：深入理解 LangGraph 内部引擎 —— Pregel 算法与状态快照与状态快照快照与状态快照

6.1 什么是 Pregel 算法？

LangGraph 的底层运行时引擎是基于 **Pregel 模型Pregel 是一种用于处理图计算的迭代算法模型，最初由 Google 提出，用于大规模图计算，如 PageRank。eRank）。LangGraph 将其思想引入到智能体工作流编排中，实现了高效的并行执行和状态管理。

在 LangGraph 中，图（Graph）并不是传统的有向无环图（DAG），而是一个可以包含循环、条件分支和并行执行的动态图。Pregel 模型通过 超步（SuperStep） 驱动图的执行，每个超步包含三个阶段：Plan、Execute、Update。

6.2 Pregel 的核心组件

组件 对应 LangGraph 概念 作用
Actor 节点（Node） 执行具体的计算任务（如调用 LLM、工具执行）
Channel 边（Edge）及状态通道 在 Actors 之间传递数据和触发信号
SuperStep 执行轮次 一组并行执行的 Actors 的集合
Checkpointer 检查点 在每个超步结束后保存状态快照

每个 Actor 会 订阅（Subscribe） 特定的 Channel，当该 Channel 有更新时，Actor 就会被触发执行。同时，Actor 可以向其他 Channel 写入（Write） 数据，从而触发下游 Actors。

6.3 Pregel 超步执行流程详解

一个超步（SuperStep）的执行分为三个严格的阶段：

  1. Plan（计划阶段）

    • 确定当前超步中需要执行哪些 Actor。
    • 规则：选择所有订阅了在当前步骤中被更新过的 Channel 的 Actor。
    • 在第一个超步（Step 0）中，特殊通道 START 会被视为已更新，从而触发所有从 START 出发的节点。

  2. Execute（执行阶段）

    • 并行执行所有选中的 Actor。
    • 每个 Actor 执行完毕后，会生成一个返回值（即对状态的更新）。
    • 如果某个 Actor 执行失败或超时，整个超步可能终止（具体取决于配置）。

  3. Update（更新阶段）

    • 收集所有 Actor 的返回值，将它们应用到对应的 Channel 上。
    • Channel 的更新会触发 Reducer 函数（如覆盖、追加、自定义合并）。
    • 更新后的 Channel 状态会在下一个超步的 Plan 阶段被用于触发新的 Actor。

重复执行：上述三个阶段的循环会持续进行，直到没有新的 Actor 被触发（即图执行完毕），或者达到最大超步数限制。

6.4 代码示例：观察 Pregel 的执行顺序

下面我们构建一个稍微复杂的图，包含分支和汇聚，然后运行它，观察每一步的执行顺序，从而理解 Pregel 的工作方式。

6.4.1 完整代码（pregel_demo.py）
# pregel_demo.py
"""
LangGraph Pregel 算法演示
观察节点执行顺序和超步划分
"""

import operator
from typing import Annotated, TypedDict

from langgraph.graph import StateGraph, START, END


# ---------- 1. 定义状态 ----------
class State(TypedDict):
    # 使用 operator.add 作为 reducer，可以累积所有节点返回的字符串
    aggregate: Annotated[list, operator.add]


# ---------- 2. 定义节点函数 ----------
def node_a(state: State, config):
    print(f'  [Step ?] 执行节点 A，当前 aggregate: {state["aggregate"]}')
    return {"aggregate": ["A"]}

def node_b(state: State, config):
    print(f'  [Step ?] 执行节点 B，当前 aggregate: {state["aggregate"]}')
    return {"aggregate": ["B"]}

def node_c(state: State, config):
    print(f'  [Step ?] 执行节点 C，当前 aggregate: {state["aggregate"]}')
    return {"aggregate": ["C"]}

def node_b2(state: State, config):
    print(f'  [Step ?] 执行节点 B2，当前 aggregate: {state["aggregate"]}')
    return {"aggregate": ["B2"]}

def node_d(state: State, config):
    print(f'  [Step ?] 执行节点 D，当前 aggregate: {state["aggregate"]}')
    return {"aggregate": ["D"]}


# ---------- 3. 构建图 ----------
builder = StateGraph(State)

# 添加节点
builder.add_node("a", node_a)
builder.add_node("b", node_b)
builder.add_node("c", node_c)
builder.add_node("b2", node_b2)
builder.add_node("d", node_d)

# 添加边
builder.add_edge(START, "a")   # START → a
builder.add_edge("a", "b")     # a → b
builder.add_edge("a", "c")     # a → c  (分支)
builder.add_edge("b", "b2")    # b → b2
builder.add_edge("b2", "d")    # b2 → d
builder.add_edge("c", "d")     # c → d (汇聚)
builder.add_edge("d", END)     # d → END

# 编译（不使用 checkpointer 也可以，但为了观察超步，我们不加 checkpointer）
app = builder.compile()

# 打印图结构（可选）
print("图结构：")
app.get_graph().print_ascii()
print("\n" + "="*50 + "\n")

# ---------- 4. 执行并观察 ----------
print("开始执行图...\n")
initial_state = {"aggregate": []}
final_state = app.invoke(initial_state)

print("\n" + "="*50)
print(f"最终 aggregate: {final_state['aggregate']}")
6.4.2 运行结果分析

运行上述代码，你会看到类似如下的输出（顺序可能略有不同，但逻辑一致）：

开始执行图...

  [Step ?] 执行节点 A，当前 aggregate: []
  [Step ?] 执行节点 B，当前 aggregate: ['A']
  [Step ?] 执行节点 C，当前 aggregate: ['A']
  [Step ?] 执行节点 B2，当前 aggregate: ['A', 'B']
  [Step ?] 执行节点 D，当前 aggregate: ['A', 'B', 'C']
  [Step ?] 执行节点 D，当前 aggregate: ['A', 'B', 'B2', 'C']

最终 aggregate: ['A', 'B', 'B2', 'C', 'D', 'D']

为什么会出现两个 D？
因为 D 节点有两个上游：B2 和 C。Pregel 的执行流程如下：

  1. Step 0（第一个超步）：

    • Plan：START 通道被更新 → 触发节点 A。
    • Execute：A 执行，返回 ["A"]。
    • Update：将 ["A"] 追加到 aggregate，同时更新了 branch:to:b 和 branch:to:c 两个通道（因为 A 有两条出边）。

  2. Step 1：

    • Plan：branch:to:b 和 branch:to:c 有更新 → 触发节点 B 和 C（并行）。
    • Execute：B 返回 ["B"]，C 返回 ["C"]。
    • Update：将 ["B"] 追加到 aggregate，同时更新 branch:to:b2 通道；将 ["C"] 追加到 aggregate，同时更新 branch:to:d 通道。

  3. Step 2：

    • Plan：branch:to:b2 有更新 → 触发节点 B2；branch:to:d 也有更新 → 触发节点 D（此时 B2 和 D 并行执行）。
    • Execute：B2 返回 ["B2"]，D 返回 ["D"]。
    • Update：B2 更新 branch:to:d 通道（再次触发 D），D 更新了最终通道。

  4. Step 3：

    • Plan：branch:to:d 再次被更新（来自 B2）→ 触发节点 D（第二次执行）。
    • Execute：D 再次返回 ["D"]。
    • Update：D 更新最终通道，此时没有新的通道被更新，图执行结束。

最终 aggregate 为 ['A', 'B', 'C', 'B2', 'D', 'D']（顺序可能因并行执行顺序略有不同，但所有节点都执行了两次 D）。

结论：节点 D 因为有两个不同的上游（B2 和 C），所以它被触发了两次，每次执行都会追加一个 "D"。这就是 Pregel 模型的特点——节点基于通道更新而触发，而不是基于 DAG 的拓扑顺序。

6.5 获取历史状态快照（StateSnapshot）

在启用 Checkpointer 后，我们可以获取每个超步结束后的状态快照，用于调试、审计或回滚。

6.5.1 完整代码（state_history_demo.py）
# state_history_demo.py
"""
使用 Checkpointer 获取 Pregel 执行的历史状态快照
"""

import operator
import sqlite3
import os
from typing import Annotated, TypedDict

from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.graph import StateGraph, START, END


# ---------- 1. 定义状态 ----------
class State(TypedDict):
    aggregate: Annotated[list, operator.add]


# ---------- 2. 定义节点 ----------
def node_a(state, config):
    print(f"  A 执行，当前: {state['aggregate']}")
    return {"aggregate": ["A"]}

def node_b(state, config):
    print(f"  B 执行，当前: {state['aggregate']}")
    return {"aggregate": ["B"]}

def node_c(state, config):
    print(f"  C 执行，当前: {state['aggregate']}")
    return {"aggregate": ["C"]}

def node_b2(state, config):
    print(f"  B2 执行，当前: {state['aggregate']}")
    return {"aggregate": ["B2"]}

def node_d(state, config):
    print(f"  D 执行，当前: {state['aggregate']}")
    return {"aggregate": ["D"]}


# ---------- 3. 构建图 ----------
def build_graph():
    builder = StateGraph(State)
    builder.add_node("a", node_a)
    builder.add_node("b", node_b)
    builder.add_node("c", node_c)
    builder.add_node("b2", node_b2)
    builder.add_node("d", node_d)
    builder.add_edge(START, "a")
    builder.add_edge("a", "b")
    builder.add_edge("a", "c")
    builder.add_edge("b", "b2")
    builder.add_edge("b2", "d")
    builder.add_edge("c", "d")
    builder.add_edge("d", END)
    return builder


# ---------- 4. 创建 Checkpointer ----------
def create_checkpointer():
    os.makedirs("./sqlite_data", exist_ok=True)
    conn = sqlite3.connect("./sqlite_data/state_history.db", check_same_thread=False)
    return SqliteSaver(conn)


# ---------- 5. 主程序 ----------
def main():
    # 编译图时传入 checkpointer
    checkpointer = create_checkpointer()
    builder = build_graph()
    app = builder.compile(checkpointer=checkpointer)

    # 定义 thread_id
    config = {"configurable": {"thread_id": "history_test"}}

    # 执行图
    print("开始执行图...\n")
    try:
        app.invoke({"aggregate": []}, config=config)
    except Exception as e:
        print(f"执行异常: {e}")

    # ---------- 获取历史状态 ----------
    print("\n" + "="*50)
    print("获取所有历史状态快照（按时间倒序）")
    print("="*50 + "\n")

    history = list(app.get_state_history(config=config))

    for i, snapshot in enumerate(history):
        print(f"--- 快照 #{i+1} (步骤 {snapshot.metadata.get('step', 'N/A')}) ---")
        print(f"  状态值: {snapshot.values}")
        print(f"  下一步节点: {snapshot.next}")
        print(f"  父配置: {snapshot.parent_config}")
        print(f"  中断: {snapshot.interrupts}")
        print()

    # ---------- 获取最近状态 ----------
    print("="*50)
    print("获取最近状态（当前）")
    print("="*50)
    latest = app.get_state(config=config)
    print(f"  状态值: {latest.values}")
    print(f"  下一步节点: {latest.next}")
    print(f"  元数据: {latest.metadata}")


if __name__ == "__main__":
    main()
6.5.2 输出结果解读

运行后，你会看到类似如下的输出（具体顺序可能因并行而不同，但快照数量是确定的）：

开始执行图...

  A 执行，当前: []
  B 执行，当前: ['A']
  C 执行，当前: ['A']
  B2 执行，当前: ['A', 'B']
  D 执行，当前: ['A', 'B', 'C']
  D 执行，当前: ['A', 'B', 'B2', 'C']

==================================================
获取所有历史状态快照（按时间倒序）
==================================================

--- 快照 #1 (步骤 4) ---
  状态值: {'aggregate': ['A', 'B', 'C', 'B2', 'D', 'D']}
  下一步节点: ()
  父配置: {...}
  中断: ()

--- 快照 #2 (步骤 3) ---
  状态值: {'aggregate': ['A', 'B', 'C', 'B2', 'D']}
  下一步节点: ('d',)
  父配置: {...}
  中断: ()

--- 快照 #3 (步骤 2) ---
  状态值: {'aggregate': ['A', 'B', 'C']}
  下一步节点: ('b2', 'd')
  父配置: {...}
  中断: ()

--- 快照 #4 (步骤 1) ---
  状态值: {'aggregate': ['A']}
  下一步节点: ('b', 'c')
  父配置: {...}
  中断: ()

--- 快照 #5 (步骤 0) ---
  状态值: {'aggregate': []}
  下一步节点: ('a',)
  父配置: {...}
  中断: ()

解读：

  • 快照 #5 (Step 0)：执行前，aggregate 为空，下一步执行节点 a。
  • 快照 #4 (Step 1)：a 执行完后，aggregate 为 ['A']，下一步并行执行 b 和 c。
  • 快照 #3 (Step 2)：b 和 c 执行完后，aggregate 为 ['A', 'B', 'C']，下一步并行执行 b2 和 d。
  • 快照 #2 (Step 3)：b2 和 d（第一次）执行完后，aggregate 为 ['A', 'B', 'C', 'B2', 'D']，下一步只执行 d（因为 b2 再次触发了 d）。
  • 快照 #1 (Step 4)：d 第二次执行完后，aggregate 为最终值，下一步为空（()），表示图执行完成。

next 字段的作用：它明确告诉你在该快照之后要执行哪些节点。这就是故障恢复的关键——当你从检查点恢复时，LangGraph 会查看 next，并从中断处继续执行。

6.5.3 StateSnapshot 的完整字段说明
字段 类型 描述
values dict 当前超步结束后的状态数据
next tuple[str, ...] 下一步要执行的节点名称列表（并行）
config RunnableConfig 包含 thread_id 和 checkpoint_id，可用于精确恢复
metadata CheckpointMetadata 包含 step（步骤号）、source（来源，如 ‘input’ 或 ‘loop’）、timestamp 等
parent_config RunnableConfig 指向父快照的配置，用于回溯（如回滚到上一步）
interrupts tuple[Interrupt, ...] 如果在节点中调用了 interrupt()，则会记录中断信息

6.6 实践：从特定快照恢复执行

你可以通过 checkpoint_id 恢复到任意历史状态：

# 获取某个历史快照的 config
history = list(app.get_state_history(config=config))
target_snapshot = history[2]  # 假设我们要恢复到第三个快照（Step 2）
resume_config = target_snapshot.config

# 从该快照继续执行
app.invoke(None, config=resume_config)

这在调试或需要重试特定步骤时非常有用。

6.7 总结：Pregel 与状态快照的意义

  1. Pregel 算法 使得 LangGraph 能够：

    • 支持并行执行（多个节点同时运行）。
    • 支持循环（节点可以多次触发）。
    • 实现高效的图计算调度。

  2. 状态快照 提供了：

    • 可观测性：你可以查看每个超步的状态变化。
    • 故障恢复：通过 next 字段，图可以精确地从断点继续。
    • 时间旅行调试：可以回溯到任何历史状态，重现问题。

  3. 生产级保障：结合持久化 Checkpointer（如 SQLite、PostgreSQL），你的智能体应用可以稳定运行数天甚至数月，无惧意外崩溃。

第7章：节点（Node）深度解析

7.1 节点的输入：State、Config 与 Runtime

节点函数通常接收三个核心参数，它们会在运行时由 LangGraph 自动注入：

参数 类型 说明
state 自定义 TypedDict/BaseModel 图的当前状态，包含所有业务数据
config RunnableConfig 系统配置（如 thread_id）和用户自定义配置（如 user_id）
runtime Runtime 运行时上下文，可注入依赖（如数据库连接、LLM客户端、stream_writer）

注意：config 和 runtime 是通过关键字参数注入的，因此它们在函数签名中的顺序可以互换，但参数名必须正确。

完整代码示例：智能客服节点

以下示例演示了一个节点如何同时使用这三种参数：

# 07_nodes/01_node_parameters_demo.py
"""
节点输入参数演示：State, Config, Runtime
模拟一个智能客服系统，根据用户角色（VIP/普通）生成不同回复
"""

from typing import TypedDict, List
from langchain_core.runnables import RunnableConfig
from langgraph.graph import StateGraph, START, END
from langgraph.runtime import Runtime

# ---------- 1. 模拟外部依赖 ----------
class MockLLM:
    """模拟 LLM 客户端"""
    def invoke(self, prompt: str) -> str:
        return f"[AI回复] 针对您的提问：'{prompt}'，我们为您提供专属服务。"

class MockDatabase:
    """模拟数据库客户端"""
    def get_user_info(self, user_id: str) -> dict:
        # 模拟根据 user_id 查询用户角色
        if "vip" in user_id.lower():
            return {"id": user_id, "role": "vip"}
        return {"id": user_id, "role": "standard"}

# ---------- 2. 定义图状态 ----------
class CustomerState(TypedDict):
    query: str           # 用户问题
    response: str        # 客服回复
    log: List[str]       # 处理日志

# ---------- 3. 定义节点函数 ----------
def customer_service_node(
    state: CustomerState,
    config: RunnableConfig,   # 系统配置
    runtime: Runtime          # 运行时上下文
) -> dict:
    """
    客服节点：从 state 读问题，从 config 获取 user_id，从 runtime 获取 LLM 和 DB
    """
    # 1. 从 state 读取用户问题
    user_query = state["query"]
    print(f"[State] 用户问题: {user_query}")

    # 2. 从 config 获取自定义配置（user_id）
    user_id = config.get("configurable", {}).get("user_id", "guest")
    print(f"[Config] 用户ID: {user_id}")

    # 3. 从 runtime 获取依赖对象
    llm_client = runtime.context.get("llm_client")
    db_client = runtime.context.get("db_client")
    if not llm_client or not db_client:
        return {
            "response": "系统错误：依赖未注入",
            "log": ["错误: LLM 或 DB 未在 runtime.context 中配置"]
        }

    # 查询用户角色
    user_info = db_client.get_user_info(user_id)
    role = user_info.get("role", "standard")
    print(f"[Runtime] 从 DB 获取用户角色: {role}")

    # 根据角色构建 prompt
    prompt = f"用户({role}) 提问: {user_query}"
    reply = llm_client.invoke(prompt)

    return {
        "response": reply,
        "log": [f"处理完成，用户角色: {role}"]
    }

# ---------- 4. 构建图 ----------
def build_graph():
    workflow = StateGraph(CustomerState)
    workflow.add_node("customer_service", customer_service_node)
    workflow.add_edge(START, "customer_service")
    workflow.add_edge("customer_service", END)
    return workflow.compile()

# ---------- 5. 运行 ----------
if __name__ == "__main__":
    app = build_graph()

    # 初始状态
    initial_state = {"query": "如何升级会员？", "response": "", "log": []}

    # 配置信息（通过 configurable 传递自定义参数）
    config = {"configurable": {"user_id": "vip_user_999"}}

    # 运行时依赖注入
    context = {
        "llm_client": MockLLM(),
        "db_client": MockDatabase()
    }

    print("[System] 开始运行图...\n")
    result = app.invoke(initial_state, config=config, context=context)
    print("\n最终结果：")
    print(f"回复: {result['response']}")
    print(f"日志: {result['log']}")

输出示例：

[System] 开始运行图...

[State] 用户问题: 如何升级会员？
[Config] 用户ID: vip_user_999
[Runtime] 从 DB 获取用户角色: vip

最终结果：
回复: [AI回复] 针对您的提问：'用户(vip) 提问: 如何升级会员？'，我们为您提供专属服务。
日志: ['处理完成，用户角色: vip']

解释：

  • state 携带业务数据（用户问题）。
  • config 携带系统/用户配置（user_id）。
  • runtime 携带外部依赖（LLM、DB），实现了依赖注入，便于测试和替换。

7.2 节点的输出：增量更新原则

核心原则：节点只返回需要更新的字段，而不是返回整个状态。这是因为：

  • 如果返回整个状态，多个并行节点同时更新相同字段会引发冲突。
  • 返回整个状态会携带无关数据，增加网络传输和合并开销。
7.2.1 错误示例（返回整个状态）
# 07_nodes/02_node_output_wrong_demo.py
from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class MyState(TypedDict):
    query: str
    web_result: str
    file_result: str
    final_answer: str

# 错误：返回整个 state
def query_web(state: MyState) -> dict:
    query = state['query']
    state['web_result'] = f"网络搜索结果：{query}"
    return state   # ❌ 返回了整个状态

def query_file(state: MyState) -> dict:
    query = state['query']
    state['file_result'] = f"文件搜索结果：{query}"
    return state   # ❌ 返回了整个状态

def answer(state: MyState) -> dict:
    web = state['web_result']
    file = state['file_result']
    state['final_answer'] = f"综合结果：{web}，{file}"
    return state   # ❌ 返回了整个状态

builder = StateGraph(MyState)
builder.add_node(query_web)
builder.add_node(query_file)
builder.add_node(answer)
builder.add_edge(START, "query_web")
builder.add_edge(START, "query_file")   # 并行
builder.add_edge("query_web", "answer")
builder.add_edge("query_file", "answer")
builder.add_edge("answer", END)
app = builder.compile()

try:
    app.invoke({"query": "LangGraph"})
except Exception as e:
    print(f"错误: {e}")  # 会报冲突异常

报错原因：query_web 和 query_file 并行执行，都返回了包含 query 字段的完整状态，LangGraph 无法决定该用哪个值，因此抛出 InvalidUpdateError。

7.2.2 正确示例（只返回增量）
# 07_nodes/03_node_output_right_demo.py
from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class MyState(TypedDict):
    query: str
    web_result: str
    file_result: str
    final_answer: str

def query_web(state: MyState) -> dict:
    query = state['query']
    # ✅ 只返回要更新的字段
    return {"web_result": f"网络搜索结果：{query}"}

def query_file(state: MyState) -> dict:
    query = state['query']
    return {"file_result": f"文件搜索结果：{query}"}

def answer(state: MyState) -> dict:
    web = state['web_result']
    file = state['file_result']
    final = f"综合结果：{web}，{file}"
    return {"final_answer": final}   # ✅ 只返回 final_answer

builder = StateGraph(MyState)
builder.add_node(query_web)
builder.add_node(query_file)
builder.add_node(answer)
builder.add_edge(START, "query_web")
builder.add_edge(START, "query_file")
builder.add_edge("query_web", "answer")
builder.add_edge("query_file", "answer")
builder.add_edge("answer", END)
app = builder.compile()

result = app.invoke({"query": "LangGraph", "web_result": "", "file_result": "", "final_answer": ""})
print(result['final_answer'])

输出：

综合结果：网络搜索结果：LangGraph，文件搜索结果：LangGraph

结论：节点只返回增量更新，LangGraph 会自动将其与全局状态合并（使用 Reducer 策略），并行更新也能安全合并。

7.3 特殊节点：START 与 END

START 和 END 是 LangGraph 预定义的两个特殊节点，用于标识图的入口和出口。

  • START：图的入口点。数据从这里流入，然后根据边分发到第一个实际节点。
  • END：图的终止点。执行流到达 END 后，整个图或该分支结束。

在代码中，它们被定义为字符串常量，并使用 sys.intern 优化内存：

import sys
START = sys.intern("__start__")
END = sys.intern("__end__")

sys.intern 的作用：

  • 确保相同字符串在内存中只存在一份，提高比较效率。
  • 类似图书馆的“图书共享”机制，节省空间并加快查找速度。
代码演示：sys.intern 的效果
# 07_nodes/04_intern_demo.py
import sys

# 使用 intern：相同字符串指向同一对象
s1 = sys.intern("hello" + "world")
s2 = sys.intern("hello" + "world")
print(s1 is s2)  # True（内存地址相同）

# 动态生成的字符串（不 intern）
name = "world"
s3 = f"hello {name}"
s4 = f"hello {name}"
print(s3 is s4)  # False（不同对象，内容相同但地址不同）

在 LangGraph 中，由于 START 和 END 会被频繁比较，使用 intern 能显著提升性能。

7.4 节点缓存（Cache）

对于计算量大或调用外部 API 的节点，如果输入相同，可以缓存结果以避免重复执行。

配置步骤
  1. 编译图时指定缓存后端（如 InMemoryCache、RedisCache）。
  2. 在 add_node 时通过 cache_policy 设置 CachePolicy(ttl=秒数)。
完整示例
# 07_nodes/05_node_cache_demo.py
import time
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.cache.memory import InMemoryCache
from langgraph.types import CachePolicy

class State(TypedDict):
    x: int
    result: int

def expensive_node(state: State) -> dict:
    print(f"节点被调用，x={state['x']}")
    time.sleep(3)   # 模拟耗时计算
    return {"result": state["x"] * 2}

builder = StateGraph(State)
builder.add_node(
    "expensive",
    expensive_node,
    cache_policy=CachePolicy(ttl=10)   # 10秒内缓存
)
builder.add_edge(START, "expensive")
builder.add_edge("expensive", END)
app = builder.compile(cache=InMemoryCache())   # 使用内存缓存

# 第一次调用
print("第一次调用：")
print(app.invoke({"x": 5}))   # 会执行节点，耗时3秒

# 第二次调用（相同输入，3秒内）
print("\n第二次调用（缓存命中）：")
print(app.invoke({"x": 5}))   # 立刻返回，不执行节点

# 第三次调用（不同输入）
print("\n第三次调用（不同输入）：")
print(app.invoke({"x": 10}))  # 重新执行

输出（注意第二次调用不会打印"节点被调用"）：

第一次调用：
节点被调用，x=5
{'x': 5, 'result': 10}

第二次调用（缓存命中）：
{'x': 5, 'result': 10}

第三次调用（不同输入）：
节点被调用，x=10
{'x': 10, 'result': 20}

7.5 节点重试（Retry）

对外部依赖（如 API、数据库）的调用可能因临时故障失败。LangGraph 支持为节点配置重试策略。

默认重试策略
  • 临时故障（如 ConnectionError、TimeoutError）会自动重试。
  • 永久错误（如 TypeError、ValueError）不重试。
自定义重试策略
# 07_nodes/06_node_retry_demo.py
import time
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.types import RetryPolicy

class State(TypedDict):
    result: str

attempt_counter = 0

def unstable_node(state: State) -> dict:
    global attempt_counter
    attempt_counter += 1
    print(f"尝试第 {attempt_counter} 次...")
    if attempt_counter < 3:
        raise ConnectionError("模拟网络超时")   # 可重试异常
    return {"result": f"成功，尝试了 {attempt_counter} 次"}

builder = StateGraph(State)
builder.add_node(
    "unstable",
    unstable_node,
    retry_policy=RetryPolicy(
        max_attempts=5,
        retry_on=(ConnectionError, TimeoutError)   # 明确指定可重试异常
    )
)
builder.add_edge(START, "unstable")
builder.add_edge("unstable", END)
app = builder.compile()

try:
    result = app.invoke({"result": ""})
    print(f"最终结果: {result}")
except Exception as e:
    print(f"最终失败: {e}")

输出（重试2次后成功）：

尝试第 1 次...
尝试第 2 次...
尝试第 3 次...
最终结果: {'result': '成功，尝试了 3 次'}

7.6 流式输出（Streaming）

LangGraph 支持在运行时实时输出数据，无需等待整个图执行完毕。这对于需要展示 LLM 逐字生成或显示当前进度的场景非常有用。

支持的流模式
模式 输出内容
"values" 每次节点执行后输出完整状态
"updates" 每次节点执行后输出增量更新（仅该节点返回的字段）
"custom" 通过 runtime.stream_writer 发送的自定义数据
"messages" LLM 生成的 token 流（需在节点中调用 LLM）
"debug" 所有调试信息
混合模式 同时输出多种类型（如 ["updates", "custom"]）
完整代码示例
# 07_nodes/07_stream_demo.py
import time
from typing import TypedDict, Annotated, List, operator
from langchain.chat_models import init_chat_model
from langchain_core.messages import BaseMessage, HumanMessage
from langgraph.graph import StateGraph, START, END
from langgraph.runtime import Runtime
from lm_config import lm_config   # 假设包含 LLM 配置

# 初始化 LLM（具体配置根据实际调整）
llm = init_chat_model(
    model=lm_config.llm_model,
    model_provider=lm_config.model_provider,
    base_url=lm_config.base_url,
    api_key=lm_config.api_key,
)

class State(TypedDict):
    input: str
    messages: Annotated[List[BaseMessage], operator.add]
    current_step: str

def node_input(state: State) -> dict:
    return {
        "messages": [HumanMessage(content=state["input"])],
        "current_step": "接收用户输入"
    }

def node_process(state: State, runtime: Runtime) -> dict:
    """模拟处理过程，使用 custom 流式输出"""
    steps = ["分析意图...", "检索知识库...", "生成提示词..."]
    writer = runtime.stream_writer   # 获取写入器
    for idx, step in enumerate(steps, 1):
        time.sleep(0.5)
        writer({"step_index": idx, "description": step})   # 发送自定义数据
    return {"current_step": "处理完成"}

def node_generate(state: State) -> dict:
    """LLM 生成，会自动支持 messages 流式输出"""
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

builder = StateGraph(State)
builder.add_node("input", node_input)
builder.add_node("process", node_process)
builder.add_node("generate", node_generate)
builder.add_edge(START, "input")
builder.add_edge("input", "process")
builder.add_edge("process", "generate")
builder.add_edge("generate", END)
app = builder.compile()

initial_state = {"input": "我是谁？", "messages": [], "current_step": "start"}

# 演示 values 模式
print("=== values 模式 ===")
for event in app.stream(initial_state, stream_mode="values"):
    print(f"当前步骤: {event.get('current_step')}, 消息数: {len(event.get('messages', []))}")

# 演示 custom 模式
print("\n=== custom 模式 ===")
for event in app.stream(initial_state, stream_mode="custom"):
    print(f"自定义数据: {event}")

# 演示 messages 模式（逐 token 输出）
print("\n=== messages 模式 ===")
for chunk, metadata in app.stream(initial_state, stream_mode="messages"):
    print(f"[{metadata.get('langgraph_node')}] Token: {chunk.content}")

输出片段（custom 模式）：

自定义数据: {'step_index': 1, 'description': '分析意图...'}
自定义数据: {'step_index': 2, 'description': '检索知识库...'}
自定义数据: {'step_index': 3, 'description': '生成提示词...'}

7.7 人工审核与中断机制（Human-in-the-Loop）

在关键决策点（如资金转账、内容发布），我们需要让图暂停，等待人工审核。这通过 interrupt() 实现。

核心流程
  1. 节点调用 interrupt(value)，图在此处暂停，并抛出 value 作为待审核数据。
  2. 外部系统（如前端）获取该数据，展示给用户。
  3. 用户做出决策后，通过 Command(resume=...) 将结果传回。
  4. 图从暂停的节点继续执行，并接收到用户返回的数据。

前提：必须配置 Checkpointer（如 InMemorySaver），以保存暂停位置。

完整示例：转账审核
# 07_nodes/08_interrupt_demo.py
from typing import Any, TypedDict
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command, interrupt

class TransferState(TypedDict):
    recipient: str
    amount: int
    memo: str
    approved: bool
    final_status: str

def review_transfer(state: TransferState) -> dict:
    """审核节点：生成待审核数据并中断"""
    pending = {
        "recipient": state["recipient"],
        "amount": state["amount"],
        "memo": state["memo"]
    }
    # 中断，等待人工输入
    user_decision = interrupt({
        "title": "转账审核",
        "pending": pending,
        "instruction": "请返回 {'approved': bool, 'amount': int, ...} 或直接返回 bool"
    })

    # 处理用户返回的决策
    if isinstance(user_decision, bool):
        approved = user_decision
        updated = pending
    elif isinstance(user_decision, dict):
        approved = user_decision.get("approved", False)
        updated = {**pending, **user_decision}   # 允许修改字段
    else:
        raise ValueError("无效的决策格式")

    print(f"[审核结果] approved={approved}, 最终数据={updated}")
    return {
        "approved": approved,
        "recipient": updated["recipient"],
        "amount": updated["amount"],
        "memo": updated["memo"]
    }

def execute_transfer(state: TransferState) -> dict:
    if not state["approved"]:
        return {"final_status": "已取消（未批准）"}
    print(f"执行转账：{state['amount']}元 给 {state['recipient']}，备注：{state['memo']}")
    return {"final_status": f"成功转账 {state['amount']} 元给 {state['recipient']}"}

builder = StateGraph(TransferState)
builder.add_node("review", review_transfer)
builder.add_node("execute", execute_transfer)
builder.add_edge(START, "review")
builder.add_edge("review", "execute")
builder.add_edge("execute", END)

# 必须配置 Checkpointer
app = builder.compile(checkpointer=InMemorySaver())

config = {"configurable": {"thread_id": "transfer_001"}}
initial = {
    "recipient": "Alice",
    "amount": 100,
    "memo": "午餐AA",
    "approved": False,
    "final_status": ""
}

print("=== 第一次调用（触发中断）===")
try:
    result = app.invoke(initial, config=config)
except Exception as e:
    # 中断会以异常形式抛出，但我们仍然可以获取中断数据
    interrupt_val = result["__interrupt__"][0].value if "__interrupt__" in result else None

# 另一种获取中断数据的方式：通过 get_state
snapshot = app.get_state(config=config)
interrupt_val = snapshot.interrupts[0].value if snapshot.interrupts else None
print(f"待审核数据: {interrupt_val}")

# 模拟用户决策：批准并修改金额为80
user_decision = {"approved": True, "amount": 80, "memo": "实付80元"}

print("\n=== 第二次调用（恢复执行）===")
final = app.invoke(Command(resume=user_decision), config=config)
print(f"最终状态: {final['final_status']}")

输出：

=== 第一次调用（触发中断）===
待审核数据: {'title': '转账审核', 'pending': {'recipient': 'Alice', 'amount': 100, 'memo': '午餐AA'}, ...}

=== 第二次调用（恢复执行）===
[审核结果] approved=True, 最终数据={'recipient': 'Alice', 'amount': 80, 'memo': '实付80元'}
执行转账：80元 给 Alice，备注：实付80元
最终状态: 成功转账 80 元给 Alice

注意事项：

  • 幂等性：节点函数在恢复执行时会重新执行，如果节点内有副作用（如发送邮件），可能导致重复。建议将 interrupt 放在单独的审核节点中，与实际执行节点分离。
  • 必须配置 Checkpointer：否则中断无法工作。

第8章：边（Edge）—— 控制流的方向

边定义了节点之间的连接方式，分为普通边和条件边。

8.1 普通边（Normal Edge）

无条件地从源节点连接到目标节点。使用 add_edge(source, target)。

graph.add_edge(START, "node_a")
graph.add_edge("node_a", "node_b")
graph.add_edge("node_b", END)

8.2 条件边（Conditional Edge）

根据当前状态动态选择下一个节点，实现分支逻辑。使用 add_conditional_edges(source, router_func, path_map)。

  • 路由函数：接收 state，返回一个字符串（目标路径的键）。
  • 路径映射：将路由函数的返回值映射到实际的节点名称或 END。
完整示例：奇偶分支
# 08_edges/01_condition_edge.py
from typing import Literal, TypedDict
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    value: int

def node_a(state: State) -> dict:
    print("执行节点 A")
    return {}   # 不修改状态

def node_b(state: State) -> dict:
    print("执行节点 B（偶数）")
    return {"value": state["value"] * 2}

def node_c(state: State) -> dict:
    print("执行节点 C（奇数）")
    return {"value": state["value"] - 1}

def route(state: State) -> Literal["b_key", "c_key"]:
    if state["value"] % 2 == 0:
        return "b_key"
    else:
        return "c_key"

builder = StateGraph(State)
builder.add_node("a", node_a)
builder.add_node("b", node_b)
builder.add_node("c", node_c)
builder.add_edge(START, "a")
builder.add_conditional_edges(
    "a",
    route,
    {
        "b_key": "b",    # 返回值 -> 实际节点名
        "c_key": "c"
    }
)
builder.add_edge("b", END)
builder.add_edge("c", END)
app = builder.compile()

# 测试偶数
print("输入 2：")
result = app.invoke({"value": 2})
print(result)  # value 变为 4

print("\n输入 3：")
result = app.invoke({"value": 3})
print(result)  # value 变为 2

8.3 可控循环（Controlled Loop）

通过条件边，我们可以构造循环结构。例如：节点 A 执行后，根据条件决定是否回到 A（或经过 B 再回到 A），直到满足退出条件。

递归限制（Recursion Limit）

为防止无限循环，LangGraph 设置了最大超步数（默认 25）。可通过 config 中的 recursion_limit 调整。

完整示例：循环计数器
# 08_edges/02_loop_with_recursion_limit.py
from typing import Literal, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.errors import GraphRecursionError

class LoopState(TypedDict):
    count: int
    max_count: int

def node_a(state: LoopState) -> dict:
    print(f"节点 A，当前计数: {state['count']}")
    return {"count": state["count"] + 1}

def node_b(state: LoopState) -> dict:
    print(f"节点 B，辅助处理")
    return {}   # 不修改状态，仅用于演示

def route(state: LoopState) -> Literal["b", END]:
    if state["count"] >= state["max_count"]:
        print("达到最大计数，结束循环")
        return END
    else:
        print("继续循环")
        return "b"

builder = StateGraph(LoopState)
builder.add_node("a", node_a)
builder.add_node("b", node_b)
builder.add_edge(START, "a")
builder.add_conditional_edges("a", route, {"b": "b"})
builder.add_edge("b", "a")   # 从 b 回到 a，形成循环
app = builder.compile()

# 设置业务需求：循环 10 次，但 recursion_limit 设为 6
config = {"recursion_limit": 6}
try:
    result = app.invoke({"count": 0, "max_count": 10}, config=config)
except GraphRecursionError as e:
    print(f"递归限制错误: {e}")

输出（只执行了 6 步即报错）：

节点 A，当前计数: 0
继续循环
节点 B，辅助处理
节点 A，当前计数: 1
...
节点 A，当前计数: 5
继续循环
节点 B，辅助处理
节点 A，当前计数: 6
递归限制错误: Recursion limit of 6 reached...

解决方案：若确需较多循环次数，可调高 recursion_limit（但需考虑性能）。

# 学习 # python
Logo
CSDN-OPC开发者社区

这里是“一人公司”的成长家园。我们提供从产品曝光、技术变现到法律财税的全栈内容，并连接云服务、办公空间等稀缺资源，助你专注创造，无忧运营。

更多推荐

介绍一个AI 智能体角色库 — 开箱即用的专家团队

AI 智能体角色库开源项目介绍 该项目提供了一套结构化AI角色定义文件集合，覆盖前端开发、安全审计、运营推广等20+专业领域。每个智能体包含： 明确身份定义（专业领域与职责） 标准化工作流程（执行步骤与检查清单） 可交付成果（模板化输出格式） 特点： 支持18种主流AI编程工具一键安装 相比普通提示词，提供完整行业规范与交付标准 含国内特色角色（微信小程序、飞书/钉钉集成开发等） 快速启动方式：

avatar CSDN-OPC开发者社区
cover

如何看待Workbuddy？三分钟看完小龙虾进化史！

avatar CSDN-OPC开发者社区
cover

2026年口碑好的AI一人公司案例大揭秘

avatar CSDN-OPC开发者社区