LangGraph 中，如何让Agent 针对过去的某个错误决策重新执行（即从历史的某个特定节点重新分叉运行） -...

知识点图片

在 LangGraph 中，让 Agent 针对过去的某个错误决策重新执行（即从历史节点“分叉”或“时间旅行”）是其核心特性之一。这个功能被称为 Time Travel（时间旅行）。

要实现这一点，必须启用检查点机制（Checkpointer）（如 MemorySaver），因为只有保存了图的状态历史，才能回到过去。

完整的流程分为四个步骤：保存状态 -> 获取历史记录 -> 修改过去的状态（分叉） -> 重新执行。

以下是具体的实现步骤和代码示例：

1. 基础准备：启用 Checkpointer

在编译图时，必须传入一个 checkpointer 来持久化状态。

python

from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()
graph = builder.compile(checkpointer=memory)

# 定义一个 Thread ID，用于追踪这次对话
config = {"configurable": {"thread_id": "thread_1"}}

2. 获取历史状态记录

假设 Agent 已经运行了一段距离，并做出了错误决策。你需要查看这个 Thread 的历史状态（Checkpoints），找到出错前的那一步。

python

# 获取状态历史（按时间倒序排列，最新的在最前面）
history = list(graph.get_state_history(config))

for state in history:
    print(f"Checkpoint ID: {state.config['configurable']['checkpoint_id']}")
    print(f"Next node to execute: {state.next}")
    print(f"State values: {state.values}")
    print("-" * 20)

3. 从特定历史节点“分叉”并修改状态

找到你想回到的那个状态的 checkpoint_id。为了纠正 Agent 的错误决策，你可以修改此时的状态（State），然后让图从这里重新运行。

假设历史记录中，Agent 在 checkpoint_id="1234-5678" 的状态后做出了错误决策。你可以获取这个配置：

python

# 提取你想要回到过去的那个特定的 config
past_config = {
    "configurable": {
        "thread_id": "thread_1",
        "checkpoint_id": "1234-5678" # 你想要回退到的特定历史节点
    }
}

接下来，使用 graph.update_state() 来修改状态并创建分叉。这一步非常关键：

values: 你想要强制覆盖的新状态值。
as_node: 你伪装成哪个节点更新了状态。这决定了图接下来会走向哪条边。

python

# 假设我们要修改 agent 内部思考的某条消息，或者修正某个工具的返回结果
new_state_values = {
    "messages": [
        # 添加一条新的消息，或者覆盖之前的消息来纠正它的思路
        ("user", "刚才的思路不对，请尝试使用B方案而不是A方案。")
    ]
}

# 更新状态，这会在那个历史节点处创建一个新的分叉（Fork）
graph.update_state(
    config=past_config, 
    values=new_state_values, 
    as_node="agent"  # 假设这个状态更新是由 "agent" 节点产生的
)

注意：当你传入了包含 checkpoint_id 的 past_config 时，LangGraph 不会覆盖原来的历史，而是会从那个时刻拉出一条新的分支，将当前 Thread 的“头指针（Head）”指向这个新分支。

4. 重新执行

现在，状态已经被纠正，并且图的指针已经停留在分叉点。你只需要传入 None 作为输入，使用原本的 thread_id 继续 invoke 即可：

python

# 使用基础的 config（不需要带 checkpoint_id，它会自动从最新的分叉处继续）
base_config = {"configurable": {"thread_id": "thread_1"}}

# 传入 None，让图直接评估当前状态并继续往下运行
for event in graph.stream(None, base_config):
    for node, values in event.items():
        print(f"Node '{node}' executed.")

典型应用场景示例

场景 A：工具调用失败，手动修正工具结果

Agent 决定调用一个API，但是参数填错了导致API报错。
做法：

回到工具调用后的那个 checkpoint。
使用 update_state，将 values 设为正确的API返回结果，as_node 设为你的 "tool_node"。
重新 invoke，Agent 就会带着正确的结果继续往下走。

场景 B：Agent 陷入死循环，强行扭转它的思考

Agent 在某个节点一直在原地打转。
做法：

回到它开始打转前的 checkpoint。
使用 update_state，往 messages 列表里追加一条 System/User 提示词（例如：“注意：不要再尝试方法A，直接输出最终结果”），as_node 设为输入该提示的节点。
重新 invoke。

场景 C：仅重放（Replay），不修改状态

如果错误决策是因为 LLM 的随机性（Temperature > 0）导致的，你只想让它原样再试一次，不需要调用 update_state。直接用历史的 config 调用 invoke 即可：

python

past_config = {"configurable": {"thread_id": "1", "checkpoint_id": "xxx"}}
# 直接从历史点重放
graph.invoke(None, past_config)

总结

在 LangGraph 中纠错分叉的核心 API 就是：
graph.update_state(past_config, new_values, as_node)
它巧妙地结合了 Git 的分支思想：找到过去的 Commit (checkpoint) -> 提交新的修改 (update_state) -> 从新分支继续运行 (invoke)。

右滑查看面试常问

播面