解锁 LangGraph 时间旅行功能：从决策溯源到路径探索的全流程实践

作者：佑瞻
在开发由大语言模型驱动的智能体时，我们常常会遇到这样的困惑：当智能体做出一个非预期的决策时，很难追溯其背后的推理过程。比如，一个对话系统突然给出了偏离主题的回答，或者一个自动化决策模型选择了次优方案。这时候，我们需要一种能够 "时光回溯" 的工具，帮助我们深入理解决策过程，定位问题根源。今天，我们就来探索 LangGraph 中一项强大的功能 —— 时间旅行（Time Travel），它就像为智能体系统配备了一台 "时光机"，让我们能够在历史执行轨迹中自由穿梭，解开非确定性系统的决策黑箱。
一、时间旅行功能的核心价值与应用场景

1.1 三大核心应用场景

在处理基于模型决策的非确定性系统时，时间旅行功能能够解决三大核心问题：
1. 理解推理过程

当智能体产生一个成功的结果时，我们往往需要拆解其背后的决策链条。例如，一个生成式 AI 成功创作了一篇精彩的故事，我们可以通过时间旅行功能，一步步查看每个生成步骤的输入输出，分析是什么样的中间状态促成了最终的成功。这就像考古学家逐层挖掘历史遗迹，还原事件的全貌。
2. 调试错误决策

当智能体出现错误时，时间旅行功能可以帮助我们精准定位问题节点。比如，一个问答系统给出了错误答案，我们可以从最终结果回溯，检查每个处理环节的状态，找出是哪个步骤的输入数据异常，或者哪个模型调用产生了错误输出。这种方式比传统的日志打印更高效，能够直接定位到问题发生的具体上下文。
3. 探索替代路径

在复杂的决策场景中，智能体可能有多种可行路径。通过时间旅行，我们可以在历史检查点处修改状态，探索不同的决策路径，从而找到更优的解决方案。这就像在地图上尝试不同的路线，寻找最短路径或风景最美的路线。
1.2 技术实现原理

LangGraph 的时间旅行功能基于检查点（checkpoint）机制实现。当图执行时，系统会在特定节点保存状态快照，这些快照构成了时间旅行的基础。值得注意的是，恢复过往执行并不会覆盖历史记录，而是会在历史中产生一个新分支，这种设计确保了我们可以在不影响原始流程的情况下进行探索。
二、时间旅行功能的使用流程

要在 LangGraph 中使用时间旅行功能，我们需要遵循以下四个核心步骤，每个步骤都有其技术要点和最佳实践：
2.1 第一步：运行图并生成检查点

首先，我们需要使用invoke或stream API 运行图，这一步的关键是正确配置检查点保存器。检查点保存器负责存储图的状态历史，是时间旅行的基础。以下是一个完整的示例：
python

1. import uuid
2. from langgraph.graph import StateGraph, START, END
3. from langchain.chat_models import init_chat_model
4. from langgraph.checkpoint.memory import InMemorySaver
6. # 定义状态类型
7. class State(TypedDict):
8.     topic: NotRequired[str]
9.     joke: NotRequired[str]
11. # 初始化LLM模型
12. llm = init_chat_model(
13.     "anthropic:claude-3-7-sonnet-latest",
14.     temperature=0,
15. )
17. # 定义生成主题的节点
18. def generate_topic(state: State):
19.     msg = llm.invoke("Give me a funny topic for a joke")
20.     return {"topic": msg.content}
22. # 定义生成笑话的节点
23. def write_joke(state: State):
24.     msg = llm.invoke(f"Write a short joke about {state['topic']}")
25.     return {"joke": msg.content}
27. # 构建工作流
28. workflow = StateGraph(State)
29. workflow.add_node("generate_topic", generate_topic)
30. workflow.add_node("write_joke", write_joke)
31. workflow.add_edge(START, "generate_topic")
32. workflow.add_edge("generate_topic", "write_joke")
33. workflow.add_edge("write_joke", END)
35. # 编译图并指定检查点保存器
36. checkpointer = InMemorySaver()
37. graph = workflow.compile(checkpointer=checkpointer)
39. # 运行图
40. config = {
41.     "configurable": {
42.         "thread_id": uuid.uuid4(),
43.     }
44. }
45. state = graph.invoke({}, config)
46. print(state["topic"])
47. print(state["joke"])

复制代码

在这个示例中，我们构建了一个简单的工作流：首先生成一个笑话主题，然后根据主题生成笑话。通过InMemorySaver检查点保存器，系统会在执行过程中自动保存状态快照。
2.2 第二步：识别检查点

运行图之后，我们需要获取执行历史并识别合适的检查点。LangGraph 提供了get_state_history()方法，该方法返回的状态按时间倒序排列（最新的状态在前）。每个状态包含next（下一个要执行的节点）和checkpoint_id（检查点 ID）等关键信息。
python

1. # 获取状态历史
2. states = list(graph.get_state_history(config))
4. # 遍历状态历史
5. for state in states:
6.     print(f"Next node: {state.next}")
7.     print(f"Checkpoint ID: {state.config['configurable']['checkpoint_id']}")
8.     print()
10. # 选择倒数第二个状态（按时间顺序的上一个状态）
11. selected_state = states[1]
12. print(f"Next node: {selected_state.next}")
13. print(f"State values: {selected_state.values}")

复制代码

输出结果中，我们可以看到每个检查点对应的节点和状态值。例如，在生成笑话的示例中，我们可以看到在执行write_joke节点之前的状态，其中包含已经生成的主题内容。
2.3 第三步：更新状态（可选）

如果我们想探索不同的决策路径，可以选择更新检查点的状态。update_state方法会创建一个新的检查点，该检查点与原线程关联，但具有新的检查点 ID。这一步允许我们修改任何节点的输入，从而改变后续的执行流程。
python

1. # 更新状态，将主题改为"chickens"
2. new_config = graph.update_state(selected_state.config, values={"topic": "chickens"})
3. print(new_config)

复制代码

更新状态后，我们得到一个新的配置，其中包含新的检查点 ID。这个新配置将作为恢复执行的起点。
2.4 第四步：从检查点恢复执行

最后一步是使用新的配置从检查点恢复执行。此时，我们需要将输入设为None，并在配置中指定正确的线程 ID 和检查点 ID。
python

1. # 从检查点恢复执行
2. new_state = graph.invoke(None, new_config)
3. print(new_state)

复制代码

执行结果显示，系统从修改后的状态开始继续执行，生成了关于 "chickens" 的笑话。这种方式让我们能够在不重新运行整个流程的情况下，探索不同的决策路径。
三、实战案例：深入理解笑话生成过程

为了更好地理解时间旅行功能的实际应用，我们以一个完整的笑话生成案例为例，逐步演示如何使用时间旅行功能优化智能体的决策过程。
3.1 初始执行与结果分析

首先，我们运行原始工作流，得到一个关于 "袜子在烘干机中的秘密生活" 的笑话主题和内容：
plaintext

1. 主题："The Secret Life of Socks in the Dryer"
2. 笑话：我终于发现了烘干机里丢失的袜子去哪儿了。原来它们根本没有丢失——它们只是和自助洗衣店别人的袜子私奔了，开始了新的生活。我的蓝色菱形花纹袜子现在和一只红色圆点袜子住在百慕大，在Sockstagram上发布度假照片，还给我寄 lint 作为赡养费。

复制代码

这个笑话虽然有趣，但我们可能想探索是否有更贴近生活的主题。这时候，时间旅行功能就派上用场了。
3.2 回溯到主题生成节点

通过get_state_history()方法，我们获取到执行历史，并选择在write_joke节点执行前的检查点。这个检查点包含了已经生成的主题，但还未生成笑话内容。此时，我们可以修改主题内容，探索不同的方向。
3.3 修改主题并探索新路径

我们将主题修改为 "chickens"（鸡），然后从检查点恢复执行。系统会基于新的主题生成笑话：
plaintext

1. 主题："chickens"
2. 笑话：为什么鸡要加入乐队？因为它有出色的鼓槌！

复制代码

这个新笑话更加简洁幽默，可能更符合我们的需求。通过时间旅行，我们无需重新运行整个流程，就可以快速探索不同的主题选择，大大提高了开发效率。
四、时间旅行功能的高级应用技巧

4.1 结合断点使用时间旅行

在实际开发中，我们可以将时间旅行与断点功能结合使用，更加精准地控制检查点的生成。例如，在关键节点前设置断点，当程序执行到断点时，我们可以手动触发检查点保存，然后进行时间旅行。
python

1. # 在generate_topic节点后设置断点
2. graph = workflow.compile(
3.     interrupt_after=["generate_topic"],
4.     checkpointer=checkpointer,
5. )
7. # 运行图，程序会在generate_topic节点后暂停
8. state = graph.invoke({}, config)
10. # 此时可以获取检查点并进行时间旅行

复制代码

这种方式适合在调试过程中，针对特定节点进行深入分析。
4.2 分布式环境下的时间旅行

在分布式环境中使用时间旅行功能时，需要注意检查点的存储和共享。我们可以将检查点保存到分布式存储系统中，如 Redis 或 S3，确保所有节点都能访问到历史状态。
python

1. from langgraph.checkpoint.redis import RedisCheckpointer
3. # 使用Redis作为检查点存储
4. checkpointer = RedisCheckpointer(
5.     host="redis-server",
6.     port=6379,
7.     db=0,
8. )

复制代码

这种配置允许我们在分布式集群中无缝使用时间旅行功能，大大提高了系统的可调试性。
4.3 自动化测试中的时间旅行

时间旅行功能还可以用于自动化测试，帮助我们验证不同路径下的系统行为。例如，我们可以在测试用例中设置多个检查点，然后依次恢复执行，验证每个阶段的状态是否符合预期。
python

1. def test_graph_time_travel():
2.     # 运行图并获取检查点
3.     state1 = graph.invoke(inputs1, config)
4.     checkpoint1 = get_checkpoint_id(state1)
5.    
6.     # 从检查点恢复并验证
7.     state2 = graph.invoke(None, {"thread_id": config["thread_id"], "checkpoint_id": checkpoint1})
8.     assert state2["joke"] == expected_joke

复制代码

这种测试方式能够覆盖更多的执行路径，提高测试的全面性。
结语

LangGraph 的时间旅行功能为我们提供了一把打开非确定性系统决策黑箱的钥匙。通过精准的状态回溯和路径探索，我们能够更深入地理解智能体的决策过程，更高效地调试问题，更灵活地探索不同的解决方案。
如果本文对你有帮助，别忘了点赞收藏，关注我，一起探索更高效的开发方式～

原文地址：https://blog.csdn.net/The_Thieves/article/details/148829605