LangGraph 之【持久化能力的应用】（记忆、人机交互与时间旅行、interrupt、Command、update_state）

能力	解决的问题	实现基础
记忆（Memory）	保持短期对话上下文，并跨会话记住用户偏好	线程级持久化 + 跨会话存储
人机交互（Human-in-the-loop）	在关键节点暂停流程，等待人工决策	中断机制 + 检查点
时间旅行（Time Travel）	回溯历史状态，调试或探索替代路径	检查点历史 + 状态重放

2. 持久化基础：检查点与存储

具体细节可参考上一期博客，LangGraph 之【持久化能力】

在深入应用层之前，我们先明确 LangGraph 的两个底层概念：

检查点（Checkpoint）：每次节点执行后自动保存的全量状态快照。所有状态（包括消息列表、自定义字段）都被序列化并存入后端
存储（Store）：用于跨会话的持久化数据，例如用户档案、长期偏好。它独立于会话存在

持久化存储后端支持多种实现：

阶段	检查点（Checkpointer）	存储（Store）
开发/测试	`InMemorySaver`	`InMemoryStore`
生产环境	`PostgresSaver`	`PostgresStore`

Checkpointer 负责会话内的状态历史（自动存档），Store 负责跨会话的长期数据（手动读写），记忆能力正是建立在这两者之上

3. 记忆：让智能体“记住”与“总结”

记忆是智能体与用户建立连续对话体验的核心。在 LangGraph 中，记忆分为两个层次：

短期记忆：当前会话中的上下文信息，由线程级检查点自动维护
长期记忆：跨会话保存的用户数据或应用数据，通过 Store 手动存取

3.1 短期记忆的精细化管理：修剪、删除、总结

关于短期记忆和长期记忆如何添加（即持久化部分），这里不再赘述。但在应用层，我们还需掌握面对具体业务场景时，如何动态管理记忆。例如，当消息历史过长时，我们需要进行消息裁剪、摘要总结或选择性删除等操作，以控制上下文长度、节省成本并提升响应质量

3.1.1 修剪消息

使用 trim_messages 按 token 数或消息条数截断历史，保留最近的、最关键的部分

from langchain_core.messages.utils import trim_messages

def call_model(state: MessagesState):
    messages = trim_messages(
        state["messages"],
        strategy="last",                     # 保留最后的部分
        token_counter=model,                 # 用于计数的模型
        max_tokens=128,                      # 最大 token 数
        start_on="human",                    # 从用户消息开始
        end_on=("human", "tool"),            # 结束于用户或工具消息
    )
    response = model.invoke(messages)
    return {"messages": [response]}

trim_messages参数详解

参数	类型	是否必需	描述
`messages`	Iterable[MessageLike]	是	待裁剪的原始聊天消息列表
`max_tokens`	int	是	裁剪后消息所允许的最大Token数量
`strategy`	'first' 或 'last'	否	裁剪策略。`'last'`（默认）表示保留最新的消息，丢弃最旧的；`'first'` 则相反
`token_counter`	Callable 或 'approximate'	是	Token计数器。可以是一个用于精确计数的模型（如 `ChatOpenAI`），或使用内置的 `'approximate'` 进行粗略估算
`include_system`	bool	否	是否始终保留对话开头的系统消息（SystemMessage）。默认为 `False`
`start_on`	消息类型	否	确保裁剪后的消息列表以指定类型开始，例如 `'human'` 以确保对话历史有效
`end_on`	消息类型	否	确保裁剪后的消息列表以指定类型结束
`allow_partial`	bool	否	是否允许在Token限制下部分保留一条消息。默认为 `False`
`text_splitter`	TextSplitter	否	当 `allow_partial=True` 时，用于分割文本的工具

基于 Token 数的修剪

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage, AIMessage, trim_messages

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

messages = [
    SystemMessage(content="you're a good assistant"),
    HumanMessage(content="hi! I'm bob"),
    AIMessage(content="hi!"),
    HumanMessage(content="I like vanilla ice cream"),
    AIMessage(content="nice"),
    HumanMessage(content="whats 2 + 2"),
    AIMessage(content="4"),
    HumanMessage(content="thanks"),
    AIMessage(content="no problem!"),
    HumanMessage(content="having fun?"),
    AIMessage(content="yes!"),
    HumanMessage(content="What's my name?"),
]

trimmer = trim_messages(
    max_tokens=65,                    # 最大令牌数
    strategy="last",                  # "last" 保留最后的消息，"first" 保留最早的消息
    token_counter=model,              # 使用模型进行令牌计数
    include_system=True,              # 始终保留系统消息
    allow_partial=False,              # 是否允许拆分消息内容
    start_on="human",                 # 确保第一条消息（不含系统消息）是 HumanMessage
)

chain = trimmer | model
print(chain.invoke(messages))

修剪后的消息列表（只保留了最后几轮对话）：

SystemMessage("you're a good assistant"),
HumanMessage('whats 2 + 2'),
AIMessage('4'),
HumanMessage('thanks'),
AIMessage('no problem!'),
HumanMessage('having fun?'),
AIMessage('yes!'),
HumanMessage("What's my name?")

基于消息数的修剪

通过设置 token_counter=len 来按消息数量修剪：

trimmer = trim_messages(
    max_tokens=11,        # 最大消息数
    strategy="last",
    token_counter=len,    # 根据消息数裁剪
    include_system=True,
    allow_partial=False,
    start_on="human",
)

修剪的最佳实践

修剪后的聊天记录需要遵循以下原则：

以 HumanMessage 或 SystemMessage 开头
以 HumanMessage 或 ToolMessage 结尾
ToolMessage 只能出现在工具调用相关的 AIMessage 之后
保留 SystemMessage（如果原始消息中存在）

3.1.2 删除消息

使用 RemoveMessage 删除特定消息或清空全部历史

from langgraph.graph.message import RemoveMessage, REMOVE_ALL_MESSAGES

# 删除最早的 6 条消息
def call_model(state: MessagesState):
    if len(state["messages"]) > 6:
        return {"messages": [RemoveMessage(id=m.id) for m in state["messages"][:6]]}
    # ...

# 删除全部消息
def clear_all(state: MessagesState):
    return {"messages": [RemoveMessage(id=REMOVE_ALL_MESSAGES)]}

删除操作不可逆，且需保证剩余消息构成有效的对话上下文

3.1.3 消息总结

修剪和删除会丢失信息，更好的做法是将旧对话压缩成摘要，保留关键内容，同时释放上下文窗口

基本思路：

扩展状态，增加 summary 字段
每次对话后，调用模型生成或更新摘要
删除被摘要覆盖的旧消息（保留最新一条用于展示）

class State(MessagesState):
    summary: str

def summarize_conversation(state: State):
    summary = state.get("summary", "")
    if summary:
        prompt = f"这是到目前为止的对话摘要：{summary}\n\n基于上面的新消息扩展摘要："
    else:
        prompt = "创建上面对话的摘要："
    messages = state["messages"] + [HumanMessage(content=prompt)]
    response = model.invoke(messages)
    # 删除除了最后一条之外的所有消息
    return {
        "summary": response.content,
        "messages": [RemoveMessage(id=m.id) for m in state["messages"][:-1]]
    }

无需每次调用后都进行总结，更合理的做法是设置阈值——例如，当消息数量超过预设上限（如 20 条）或累计 Token 数超过模型窗口限制时，才触发总结与删除操作。此外，

LangChain 官方库 LangMem 提供了 SummarizationNode 等封装，可进一步简化实现

4. 人机交互：中断的艺术

人机交互的核心是中断：在流程中插入一个“暂停点”，等待外部输入后再继续

4.1 中断原理

中断的实现依赖于检查点机制：

当执行到 interrupt() 时，系统抛出一个特殊异常，LangGraph 运行时捕获该异常，立即保存当前状态快照

控制权返回给调用者，调用者可以通过 invoke 返回值中的 __interrupt__ 字段获取中断信息

# invoke 的返回值包含 __interrupt__ 字段
result = graph.invoke({"messages": "hi"}, config)
if "__interrupt__" in result:
    # 中断信息在 result["__interrupt__"] 中
    print(result["__interrupt__"])
    # 输出：[Interrupt(value={'question': '批准吗？'}, ...)]

外部系统（如 UI）获取信息并收集用户决策
再次调用图，传入 Command(resume=...)，系统从保存的检查点恢复，重新执行中断所在的节点（从头开始），但 interrupt() 返回传入的 resume 值

流程图如下：

interrupt() 函数

维度	详细说明
函数签名	`interrupt(value: Any) -> Any`
核心定位	“主动暂停”：在图节点内部调用的内置函数，用于触发中断，将控制权交还给外部调用者
参数 `value`	中断载荷。必须是 JSON 可序列化的数据（`str`、`int`、`dict`、`list`）作用：向外部系统（如 UI）传递信息例如 `{"question": "是否批准转账？", "amount": 100}`
返回值	当外部传入 `Command(resume=...)` 时，`interrupt()` 会直接返回 `resume` 中携带的具体值（注意：第一次执行时该函数不返回，直接抛异常）
执行行为	1. 调用时立即抛出 `GraphInterrupt` 特殊异常 2. 当前节点函数立即终止（后续代码不执行） 3. 运行时捕获异常，自动保存当前状态快照（Checkpoint） 4. 控制权返回给调用者，`invoke()` 结束
程序生命周期	不是阻塞等待（区别于 `input()`）它让程序彻底结束本次调用，等待外部通过新的 `invoke` 请求恢复
严禁事项 1	禁止包裹在 `try/except` 中：通用的 `except Exception` 会捕获特殊异常，导致中断失效正确：将其暴露在顶层，或只捕获明确的业务异常（如 `NetworkError`）
严禁事项 2	禁止中断前执行非幂等操作：节点恢复时会从头重跑。若在 `interrupt()` 前扣款/发消息，恢复时会重复执行正确：将副作用放在 `interrupt()` 之后（确认后再执行）
关键约束	调用顺序必须固定：同一节点内多次调用时，顺序不能变，不能用 `if` 条件跳过某个中断，否则恢复时值会错位（按索引匹配）

Command 对象

维度	详细说明
类定义	`class Command(GraphCommand):`（来自 `langgraph.types`）
核心定位	1. 外部恢复：在 `graph.invoke()` 中传入，恢复中断并传回数据。 2. 内部路由：在节点函数中 `return`，动态控制下一步跳转和状态更新。
字段：`resume`	恢复值（外部用法）类型：JSON 可序列化数据作用：作为 `interrupt()` 的返回值传回节点必须与中断期望的类型匹配（如 `True` / `False` 或编辑后的文本）
字段：`goto`	目标节点（内部用法）类型：`str`（节点名称）或 `END` 常量作用：覆盖原图静态边，强制路由到指定节点
字段：`update`	状态更新（内部用法）类型：`Dict[str, Any]` 作用：在跳转至 `goto` 节点之前，先更新 State 中的指定字段（遵循 Reducer 规则）
字段：`graph`	子图定位（高级用法）类型：`Callable` 作用：在多图嵌套时，指定 `Command` 作用于哪一个子图实例
使用场景 A	外部恢复执行： `graph.invoke(Command(resume="用户输入的值"), config)` -> LangGraph 加载检查点，重跑节点，`interrupt()` 返回 `"用户输入的值"`
使用场景 B	内部动态路由： `return Command(goto="proceed", update={"status": "已批准"})` -> 先更新 `status` 字段，再跳转到 `"proceed"` 节点
严禁事项	`resume` 必须可序列化：和 `interrupt` 一样，`resume` 也必须是 JSON 可序列化的传函数或数据库连接会导致恢复失败
关键规则	数量匹配：每次 `invoke(Command(resume=...))` 只能恢复一个 `interrupt()`。若节点内有 N 个 `interrupt()` 调用，需要连续调用 N 次 `graph.invoke(Command(resume=...))`

4.2 中断的黄金法则

为了让中断稳定工作，必须遵守以下规则：

规则1：只传递可序列化的数据

interrupt() 的参数和 Command.resume 的值都必须是 JSON 可序列化的（字符串、数字、字典、列表等）。禁止传递函数、类实例、数据库连接等

# 正确
interrupt({"question": "你的名字？", "fields": ["name", "age"]})

# 错误
interrupt(validate_input)   # 函数不可序列化

规则2：不要将 interrupt() 包裹在 try/except 中

interrupt() 通过抛出特殊异常实现暂停，如果被通用的 except Exception 捕获，会导致中断失效

# 错误
try:
    interrupt("确认？")
except Exception as e:
    print(e)

# 正确：先中断，再处理可能出错的代码
interrupt("确认？")
try:
    risky_operation()
except SpecificError as e:
    handle(e)

规则3：中断前的操作必须幂等

节点恢复时会从头重新执行，因此中断前的代码可能被执行多次。若这些操作有副作用（如扣款、发送消息、创建记录），必须保证幂等

幂等：一个操作无论执行多少次，它对系统产生的最终效果（副作用）都是一样的

# 错误：每次恢复都会创建重复记录
audit_id = db.create_audit_log(...)
decision = interrupt("批准？")

# 正确：使用 upsert（幂等）
db.upsert_audit_log(user_id=..., status="pending")
decision = interrupt("批准？")

规则4：中断顺序必须固定

在同一个节点中使用多次 interrupt() 调用时，LangGraph 遵循严格的顺序和索引匹配机制：

节点恢复时会从函数开头重新运行，而不是从中断点继续
LangGraph 为每个执行任务维护一个恢复值列表（Resume Value List）。遇到 interrupt() 时，按顺序从这个列表中取对应的值
interrupt() 的调用顺序和数量，在中断触发时和恢复执行时必须完全相同
场景结果
中断前后顺序一致按索引精确匹配，一切正常
中断前后顺序不一致（条件性跳过） 索引错乱 → 值赋给错误变量或抛出异常

场景	结果
中断前后顺序一致	按索引精确匹配，一切正常
中断前后顺序不一致（条件性跳过）	索引错乱 → 值赋给错误变量或抛出异常

# 固定顺序
name = interrupt("姓名？")   # 索引0
age = interrupt("年龄？")    # 索引1

# 错误：可能跳过某个中断
if need_age:
    age = interrupt("年龄？")  # 索引不稳定

4.3 四种典型应用模式

4.3.1 批准/拒绝

在关键操作（转账、删除数据）前暂停，让用户决定是否继续

def approval_node(state):
    decision = interrupt({
        "question": "批准此操作？",
        "details": state["action_details"]
    })
    return Command(goto="proceed" if decision else "cancel")

恢复时传入 Command(resume=True) 或 Command(resume=False)，图自动路由到不同分支

4.3.2 审查与编辑状态

让用户修改当前状态中的内容（如生成的文本）

def review_node(state):
    updated = interrupt({
        "instruction": "请编辑以下内容",
        "content": state["generated_text"]
    })
    return {"generated_text": updated}

恢复时传入修改后的文本，状态即被更新

4.3.3 在工具中中断

将 interrupt() 直接写在工具函数内部，当 LLM 调用该工具时自动触发暂停

@tool
def send_email(to: str, subject: str, body: str):
    response = interrupt({
        "action": "发送邮件",
        "to": to,
        "subject": subject,
        "body": body
    })
    if response.get("action") == "同意":
        # 实际发送
        pass
    else:
        return "已取消"

4.3.4 验证人工输入

使用循环反复中断，直到用户输入符合校验规则

def get_age_node(state):
    prompt = "请输入年龄："
    while True:
        answer = interrupt(prompt)
        if isinstance(answer, int) and answer > 0:
            return {"age": answer}
        prompt = f"'{answer}' 无效，请输入正数。"

5. 时间旅行：重放与调试的利器

时间旅行允许我们查看历史状态、修改状态并重新执行，从而调试或探索不同路径

5.1 原理

每个检查点都有一个唯一的 checkpoint_id
graph.get_state_history(config) 返回按时间倒序的所有检查点
通过graph.update_state(config, values)可以基于某个历史检查点创建新分支(不修改原检查点)
用新配置（包含 checkpoint_id）调用 graph.invoke(None, new_config)，即可从该点继续执行

graph.update_state()

维度	详细说明
功能定位	手动更新指定线程（`thread_id`）的状态，在指定检查点之后创建一个新的检查点分支，实现状态编辑与时间旅行
前置条件	图必须配置了检查点存储（如 `InMemorySaver`、`PostgresSaver`），否则无法调用
核心参数	`config`：包含 `thread_id` 的配置（必填），指定要操作的线程 `values`：要更新的状态字段（必填），支持字典或字典列表 `as_node`：指定本次更新“假装”由哪个节点执行（可选，强烈建议显式指定） `checkpoint_id`：要修改的特定检查点ID（可选，默认为最新状态）
返回值	返回一个新的配置字典 `RunnableConfig`，包含新创建的检查点ID，可用于后续恢复执行
状态合并规则（Reducer）	- 普通字段（无特殊Reducer）：直接覆盖旧值 - 列表字段（如 `messages` 使用 `operator.add`）：新值追加到列表末尾 - 若使用 `RemoveMessage`，可从列表中删除指定消息
典型应用场景	1. 时间旅行调试：修改历史检查点的状态，测试不同路径 2. 人机交互编辑：在断点处允许人工修改状态（如文本、参数）后继续 3. 错误修复：修正错误中间状态，避免重新执行整个工作流
重要注意事项	- `as_node` 建议显式设置：省略时系统自动推断，但在复杂图中可能定位不准，影响后续执行路径 - 新分支独立：原始检查点保留不变，新状态会创建新分支，不影响旧历史 - `messages` 追加行为：若需替换或删除消息，需配合 `RemoveMessage` 使用 - 需保存检查点：`update_state` 本身会写入一个新的检查点，因此需确保存储后端可用

5.2 四步法

执行工作流（确保指定了 checkpointer）
获取历史：states = list(graph.get_state_history(config))
修改状态（可选）：new_config = graph.update_state(selected_state.config, values={...})
从检查点恢复：result = graph.invoke(None, new_config)

5.3 完整示例：笑话生成器调试

假设我们有两个节点：generate_topic 和 write_joke。生成的主题过于宽泛，我们想修改主题后重新生成笑话

from typing_extensions import TypedDict, NotRequired
from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver

# ---------- 状态类型定义 ----------
class State(TypedDict):
    topic: NotRequired[str]   # 笑话主题，可选字段
    joke: NotRequired[str]    # 笑话内容，可选字段

# ---------- 节点函数实现 ----------
model = init_chat_model("gpt-4o-mini")

def generate_topic(state: State):
    """第一个AI调用：生成主题"""
    response = model.invoke("给我一个搞笑的笑话主题")
    return {"topic": response.content}   # 更新状态中的 topic

def write_joke(state: State):
    """第二个AI调用：编写笑话"""
    response = model.invoke(f"写一个关于{state['topic']}的笑话")
    return {"joke": response.content}    # 更新状态中的 joke

# ---------- 工作流构建 ----------
workflow = StateGraph(State)
workflow.add_node("generate_topic", generate_topic)
workflow.add_node("write_joke", write_joke)

workflow.add_edge(START, "generate_topic")
workflow.add_edge("generate_topic", "write_joke")
workflow.add_edge("write_joke", END)

# 编译时启用检查点（持久化），以便支持时间旅行
graph = workflow.compile(checkpointer=InMemorySaver())

# ---------- 执行工作流 ----------
config = {"configurable": {"thread_id": "1"}}
result = graph.invoke({}, config)
print("原始笑话：", result["joke"])

# ---------- 时间旅行调试过程 ----------
# 1. 发现问题：最终笑话主题太宽泛，想修改主题重新生成

# 2. 回溯分析：获取所有历史检查点
states = list(graph.get_state_history(config))

# 查看 generate_topic 节点执行后的检查点（通常索引 1 是该节点后的状态）
topic_state = states[1] if len(states) > 1 else states[0]
print("AI 生成的主题：", topic_state.values.get("topic", "无主题"))

# 3. 修改状态：将主题改为更具体的内容
new_config = graph.update_state(
    topic_state.config,
    values={"topic": "程序员调试代码时的趣事"}
)

# 4. 从修改后的检查点重新执行（输入为 None，因为状态已存在）
new_result = graph.invoke(None, new_config)
print("改进后的笑话：", new_result["joke"])

时间旅行尤其适用于：

分析 AI 的推理链
定位错误发生的精确步骤
比较不同输入对最终结果的影响

6. 小结

LangGraph 的持久化能力赋能了：

记忆：通过线程级检查点保留短期上下文，通过独立存储维持长期用户画像，并借助修剪、删除、总结等工具精细控制上下文长度
人机交互：通过中断机制，在关键节点引入人工决策，支持审批、编辑、工具干预、输入验证等丰富场景，且需遵循序列化、幂等、异常处理等黄金法则
时间旅行：通过检查点历史，开发者可以回溯、修改和重放执行路径，极大提升复杂工作流的调试效率和探索灵活性

掌握这三项能力，便可构建出可回溯、可干预、有记忆的智能体应用

DAMO开发者矩阵

DAMO开发者矩阵，由阿里巴巴达摩院和中国互联网协会联合发起，致力于探讨最前沿的技术趋势与应用成果，搭建高质量的交流与分享平台，推动技术创新与产业应用链接，围绕“人工智能与新型计算”构建开放共享的开发者生态。

更多推荐

110台机器人齐舞被骂“遥控玩具“？规则里藏着答案

人形机器人在地面，要维持躯干平衡、应对地面凹凸，每台几十个关节同步做跳跃下蹲。一百台机器只要一台摔了，就是现场事故——这是量产品控的公开考试。但它测的是三样东西——集群调度、故障容错、量产一致性，这三个能力恰恰是未来几百台机器人同时在工厂/园区上岗的技术底座。智元 110 台机器人齐舞创吉尼斯纪录后，评论区最多的质疑是：这不就是大号遥控玩具吗？也就是说：110 台机器人在没有人工干预的情况下，自主