第4章状态与记忆¶

学习目标¶

理解 StateGraph 中状态的核心作用与 reducer 机制
掌握检查点（Checkpoint）的原理与多种持久化后端
能够用 thread_id 实现多轮会话的记忆
实现人在回路（Human-in-the-loop）：暂停、审批、修改后继续
构建可断点续跑的长运行 Agent

问题导入¶

前几章的图每次 invoke 都是无状态的：传一个输入，得一个输出，结束。但真实 Agent 需要记忆——用户上一句说了什么、Agent 已经做了哪些操作。更进一步，有些任务长达数小时（等待审批、等待人工补充信息），进程随时可能被重启。如何让 Agent "记住历史""暂停后恢复""回放调试"？这就需要 LangGraph 的状态与检查点机制。

核心概念¶

1. StateGraph 中的状态¶

第2章已介绍：State 是 TypedDict，字段可标注 reducer。这里补充几个要点：

状态是图的唯一记忆：节点函数本身应是无状态的，所有跨步骤数据都放 State。
reducer 决定合并语义：add_messages 追加、operator.add 列表拼接、不标注则覆盖。设计 State 时务必想清楚每个字段的合并方式。
状态快照即检查点：每经过一个节点，LangGraph 都会把"当前 State + 当前位于哪个节点"打包成一个快照。

2. 检查点 Checkpoint¶

检查点是 LangGraph 实现"记忆"与"可恢复"的基石。工作原理：

图编译时注入一个 Checkpointer 实例。
每个节点执行前后，运行时自动把 State 序列化存入 checkpointer。
下次用相同 thread_id 调用时，运行时先从 checkpointer 读取最新快照，从断点继续。

thread_id 是会话标识，相当于"对话 ID"。同一 thread_id 共享记忆，不同 thread_id 互不干扰。这一设计让多用户会话天然隔离。

3. 持久化后端¶

LangGraph 提供多种 checkpointer 实现：

实现	包	适用场景
`MemorySaver`	`langgraph.checkpoint.memory`	开发调试、单进程
`SqliteSaver`	`langgraph-checkpoint-sqlite`	单机持久化、轻量生产
`PostgresSaver`	`langgraph-checkpoint-postgres`	多实例生产、高并发
自定义	实现 `BaseCheckpointSaver`	接 Redis、MongoDB 等

生产环境务必用持久化后端（Sqlite/Postgres），MemorySaver 进程退出即丢失。

4. 数据捕获与调试¶

检查点不仅是恢复手段，也是调试利器。相关 API：

graph.get_state(config)：取当前快照，含 values（State 数据）、next（下一步将执行的节点）、tasks（待执行任务）。
graph.get_state_history(config)：列出该 thread 的所有历史快照，可回放"每一步状态如何演变"。
graph.update_state(config, values)：手动修改状态，用于纠错或注入人工输入。

这套 API 让"Agent 第 3 步在想什么"这类问题变得可观测。

5. 人在回路 Human-in-the-loop¶

人在回路的核心是让图在指定节点暂停，等待外部输入后继续。两种实现：

interrupt_before：在某节点执行前暂停。常用于"执行前审批"。
interrupt_after：在某节点执行后暂停。常用于"执行后审查"。

编译时指定：

graph = builder.compile(
    checkpointer=checkpointer,
    interrupt_before=["dangerous_action"],
)

调用 invoke 时图会在 dangerous_action 前暂停并返回当前状态。外部确认后，用相同 thread_id 再次 invoke(None, config) 即可从断点继续；若需修改状态，先 update_state 再 invoke(None)。

6. 时间旅行 Time Travel¶

get_state_history 返回的每个历史快照都有一个 config，把该 config 传给 invoke(None, config) 即可从那个历史点"分叉"重跑。这让你能回答"如果当时走另一条路会怎样"，是 A/B 测试与回归分析利器。

原理图¶

sequenceDiagram
    participant U as 用户
    participant G as Graph 运行时
    participant C as Checkpointer<br/>(Sqlite/Postgres)
    participant N as 节点

    U->>G: invoke(input, thread_id=t1)
    G->>C: 读取 t1 最新快照
    C-->>G: 空（新会话）
    G->>N: 执行 node_a
    N-->>G: 返回更新
    G->>C: 写入快照 #1
    G->>N: 执行 node_b（被 interrupt_before 暂停）
    G-->>U: 返回当前状态（next=node_b）
    Note over U: 人工审批/补充输入
    U->>G: update_state(t1, {approved: True})
    G->>C: 更新快照
    U->>G: invoke(None, thread_id=t1)
    G->>C: 读取 t1 快照
    C-->>G: 快照（含 approved）
    G->>N: 从 node_b 继续
    N-->>G: 完成
    G->>C: 写入快照 #2
    G-->>U: 最终结果

图：检查点驱动的会话记忆与人在回路。thread_id 贯穿始终，每次节点执行后写快照，暂停时返回当前状态，恢复时从快照续跑。

示例代码¶

下面构建一个"带审批的高风险操作 Agent"：执行删除前暂停，人工确认后继续。

import os
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import MemorySaver  # 演示用内存，生产换 Sqlite/Postgres
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

class State(TypedDict):
    messages: Annotated[list, add_messages]
    approved: bool   # 人工审批标记

llm = ChatOpenAI(model="gpt-4o-mini", api_key=os.getenv("OPENAI_API_KEY"))

@tool
def delete_record(record_id: str) -> str:
    """删除指定记录（高风险操作）。"""
    # 实际执行删除
    return f"已删除记录 {record_id}"

@tool
def query_record(record_id: str) -> str:
    """查询记录详情。"""
    return f"记录 {record_id}: 状态=活跃, 金额=1000"

tools = [delete_record, query_record]
llm_with_tools = llm.bind_tools(tools)

def agent_node(state: State) -> dict:
    return {"messages": [llm_with_tools.invoke(state["messages"])]}

# 模拟 ToolNode（简化版）
def tool_node(state: State) -> dict:
    from langchain_core.messages import ToolMessage
    last = state["messages"][-1]
    results = []
    for call in last.tool_calls:
        fn = {t.name: t for t in tools}[call["name"]]
        results.append(ToolMessage(content=fn.invoke(call["args"]), tool_call_id=call["id"]))
    return {"messages": results}

builder = StateGraph(State)
builder.add_node("agent", agent_node)
builder.add_node("tools", tool_node)
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent",
    lambda s: "tools" if s["messages"][-1].tool_calls else END,
    {"tools": "tools", END: END})
builder.add_edge("tools", "agent")

# 关键：编译时注入 checkpointer 和中断点
checkpointer = MemorySaver()
graph = builder.compile(
    checkpointer=checkpointer,
    interrupt_before=["tools"],   # 执行工具前暂停，等待人工确认
)

config = {"configurable": {"thread_id": "session-001"}}

# 第一次调用：图会在 tools 节点前暂停
print("=== 第一步：用户请求删除 ===")
result = graph.invoke(
    {"messages": [{"role": "user", "content": "帮我删除记录 R-123"}], "approved": False},
    config,
)
state = graph.get_state(config)
print("下一步将执行：", state.next)  # ('tools',) 表示暂停在 tools 前
print("LLM 想调的工具：", state.values["messages"][-1].tool_calls)

# 人工审批：检查是否真的是 delete_record
last_msg = state.values["messages"][-1]
is_dangerous = any(c["name"] == "delete_record" for c in last_msg.tool_calls)
if is_dangerous:
    user_confirm = input("检测到删除操作，确认执行？(y/n): ")
    if user_confirm.lower() == "y":
        graph.update_state(config, {"approved": True})
    else:
        # 拒绝：注入一条用户消息让 Agent 重新考虑
        graph.update_state(config, {"messages": [{"role": "user", "content": "取消删除操作。"}]})
        # 把 tools 节点跳过——这里直接继续，agent 会读到取消指令
        graph.update_state(config, None, as_node="tools")

# 第二次调用：传 None 表示从断点继续
print("=== 第二步：从断点恢复 ===")
final = graph.invoke(None, config)
print("最终回复：", final["messages"][-1].content)
print("是否已审批：", final.get("approved"))

运行后，图在执行 delete_record 前自动暂停，等待人工输入 y 才继续。这就是人在回路的核心模式：用 interrupt_before 暂停、用 update_state 注入决策、用 invoke(None, config) 续跑。

企业案例¶

案例 1：长运行数据分析 Agent¶

某数据团队构建的 Agent 需要执行"取数→清洗→建模→生成报告"四步，全程约 40 分钟，且中途需要数据工程师确认清洗规则。团队用 PostgresSaver 作 checkpointer，在"清洗"节点设置 interrupt_before。Agent 暂停时工程师审查取数结果、调整清洗参数，然后 invoke(None) 继续。即使容器重启，从 Postgres 读取快照即可恢复，不必从头跑。

案例 2：客服会话的多轮记忆¶

客服系统用 thread_id = 用户ID + 会话ID，配合 SqliteSaver 实现跨请求记忆。用户上午问"我的订单到哪了"，下午再问"那它预计什么时候到"——Agent 从检查点读取上午的上下文，无需用户重复。会话窗口由检查点保留策略控制，超期自动归档。

案例 3：故障回放与时间旅行¶

一次 Agent 误操作删错数据，团队用 get_state_history 拉出该 thread 的全部快照，定位到第 4 步 Agent 错误调用了 delete_record。把第 3 步快照的 config 传入 invoke，模拟"如果当时走 query 会怎样"，验证修复方案后部署。

常见问题¶

Q1：MemorySaver 能用于生产吗？ 不能。它只存内存，进程退出即丢失。生产用 SqliteSaver（单机）或 PostgresSaver（多实例）。它们接口一致，切换成本低。

Q2：thread_id 必须传吗？ 配置了 checkpointer 后，每次 invoke 必须在 config["configurable"]["thread_id"] 中提供，否则报错。一个 thread 即一个会话。

Q3：invoke(None, config) 中的 None 是什么意思？ None 表示"不提供新输入，从检查点断点继续"。常用于人在回路恢复、流式续跑。

Q4：interrupt_before 与 interrupt_after 怎么选？ 执行前要审批用 interrupt_before（如删除前确认）；执行后要审查用 interrupt_after（如生成后人工校对再发出）。

Q5：检查点数据会无限增长吗？ 会。每个节点一步一个快照，长会话可能积累大量数据。SqliteSaver/PostgresSaver 支持配置保留策略或定期清理；生产应监控存储量。

面试题¶

1. 检查点解决了哪三个问题？ 答：会话记忆（跨 invoke 保留状态）、断点续跑（进程重启后恢复）、调试回放（逐步查看状态演变）。

2. thread_id 的作用？ 答：会话唯一标识。同一 thread_id 共享状态与记忆，不同 thread_id 互隔离，天然支持多用户。

3. 人在回流的三个关键 API 是什么？ 答：interrupt_before（编译时设暂停点）、update_state（暂停时注入人工决策）、invoke(None, config)（从断点续跑）。

4. MemorySaver 与 PostgresSaver 的本质区别？ 答：存储介质——前者内存（易失、单进程），后者数据库（持久、多实例共享）。接口一致，生产必须用持久化后端。

5. 什么是"时间旅行"？有什么用？ 答：用 get_state_history 取历史快照，把历史 config 传给 invoke 从该点分叉重跑。用于故障回放、A/B 测试、回归分析。

练习题¶

把示例的 MemorySaver 换成 SqliteSaver（需 pip install langgraph-checkpoint-sqlite），验证重启脚本后会话记忆依然存在。
增加一个"双签"机制：删除操作需要两个不同用户分别确认才执行。提示：用 approved_count 字段 + 多次 update_state。
用 get_state_history 打印一次完整调用的所有快照，观察 next 字段如何随执行推进变化。

小结¶

本章是 LangGraph "记忆"的核心。State 承载共享数据，reducer 决定合并语义，检查点把每一步状态持久化。thread_id 隔离会话，interrupt_before/after 实现人在回路，get_state_history 支持时间旅行。掌握这些后，你能构建可暂停、可恢复、可回放的长运行 Agent。下一章进入多智能体编排，把单个图扩展为多个 Agent 协作的图。

第4章 状态与记忆¶