Skip to main content
在处理基于模型进行决策的非确定性系统（例如由大语言模型驱动的智能体）时，详细检查其决策过程非常有用：
  1. 理解推理过程：分析导致成功结果的各个步骤。
  2. 调试错误：识别错误发生的位置及原因。
  3. 探索替代方案：测试不同路径以发现更优解决方案。
LangGraph 提供了时间旅行功能来支持这些用例。具体而言，您可以从先前的检查点恢复执行——既可以重放相同的状态，也可以修改状态以探索替代路径。在所有情况下，从过去恢复执行都会在历史记录中产生一个新的分支。 要在 LangGraph 中使用时间旅行功能：
  1. 运行图：使用 invoke 或 stream 方法，以初始输入运行图。
  2. 在现有线程中识别检查点：使用 get_state_history() 方法检索特定 thread_id 的执行历史，并定位所需的 checkpoint_id。 或者，在您希望暂停执行的节点之前设置一个中断点。随后即可找到截至该中断点所记录的最新检查点。
  3. 更新图状态（可选）：使用 update_state 方法修改检查点处的图状态，并从新的替代状态继续执行。
  4. 从检查点恢复执行：使用 invoke 或 stream 方法，输入为 None，配置中包含相应的 thread_id 和 checkpoint_id。
如需了解时间旅行的概念概述，请参阅时间旅行。

​
在工作流中的应用

本示例构建了一个简单的 LangGraph 工作流，用于生成笑话主题并使用 LLM 编写笑话。它演示了如何运行图、检索过去的执行检查点、选择性地修改状态，以及从选定的检查点恢复执行以探索不同的结果。

​
设置

首先，我们需要安装所需的包： 
%%capture --no-stderr
%pip install --quiet -U langgraph langchain_anthropic
接下来，我们需要设置 Anthropic（我们将使用的 LLM）的 API 密钥： 
import getpass
import os


def _set_env(var: str):
    if not os.environ.get(var):
        os.environ[var] = getpass.getpass(f"{var}: ")


_set_env("ANTHROPIC_API_KEY")
注册 LangSmith，快速发现并提升您的 LangGraph 项目的性能。LangSmith 可让您利用追踪数据来调试、测试和监控基于 LangGraph 构建的 LLM 应用程序。
import uuid

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


class State(TypedDict):
    topic: NotRequired[str]
    joke: NotRequired[str]


llm = init_chat_model(
    "anthropic:claude-3-7-sonnet-latest",
    temperature=0,
)


def generate_topic(state: State):
    """调用 LLM 生成笑话主题"""
    msg = llm.invoke("给我一个有趣的笑话主题")
    return {"topic": msg.content}


def write_joke(state: State):
    """调用 LLM 根据主题编写笑话"""
    msg = llm.invoke(f"写一个关于 {state['topic']} 的短笑话")
    return {"joke": msg.content}


# 构建工作流
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)

# 编译
checkpointer = InMemorySaver()
graph = workflow.compile(checkpointer=checkpointer)
graph

​
1. 运行图

config = {
    "configurable": {
        "thread_id": uuid.uuid4(),
    }
}
state = graph.invoke({}, config)

print(state["topic"])
print()
print(state["joke"])
输出： 
“洗衣机里袜子的秘密生活”怎么样？你知道的，探讨一下这个神秘现象：为什么袜子成对进入洗衣机，却总是单只出来？它们去哪儿了？是不是在其他地方开始了新生活？是否存在我们不知道的“袜子天堂”？这个日常谜团蕴含着丰富的喜剧潜力，能引起所有人的共鸣！

# 洗衣机里袜子的秘密生活

我终于发现了所有失踪袜子的去向。原来它们根本没有失踪——只是和洗衣店别人的袜子私奔了，一起开始新生活。

我的蓝色菱形花纹袜现在正和一只红色波点袜在百慕大度假，还在“Sockstagram”上晒照片，并寄给我一些绒毛作为赡养费。

​
2. 识别检查点

# 状态按逆时间顺序返回。
states = list(graph.get_state_history(config))

for state in states:
    print(state.next)
    print(state.config["configurable"]["checkpoint_id"])
    print()
输出： 
()
1f02ac4a-ec9f-6524-8002-8f7b0bbeed0e

('write_joke',)
1f02ac4a-ce2a-6494-8001-cb2e2d651227

('generate_topic',)
1f02ac4a-a4e0-630d-8000-b73c254ba748

('__start__',)
1f02ac4a-a4dd-665e-bfff-e6c8c44315d9

# 这是倒数第二个状态（状态按时间顺序列出）
selected_state = states[1]
print(selected_state.next)
print(selected_state.values)
输出： 
('write_joke',)
{'topic': '“洗衣机里袜子的秘密生活”怎么样？你知道的，探讨一下这个神秘现象：为什么袜子成对进入洗衣机，却总是单只出来？它们去哪儿了？是不是在其他地方开始了新生活？是否存在我们不知道的“袜子天堂”？这个日常谜团蕴含着丰富的喜剧潜力，能引起所有人的共鸣！'}

​
3. 更新状态

update_state 将创建一个新的检查点。新检查点将关联到相同的线程，但拥有新的检查点 ID。 
new_config = graph.update_state(selected_state.config, values={"topic": "chickens"})
print(new_config)
输出： 
{'configurable': {'thread_id': 'c62e2e03-c27b-4cb6-8cea-ea9bfedae006', 'checkpoint_ns': '', 'checkpoint_id': '1f02ac4a-ecee-600b-8002-a1d21df32e4c'}}

​
4. 从检查点恢复执行

graph.invoke(None, new_config)
输出： 
{'topic': 'chickens',
 'joke': '为什么鸡要加入乐队？\n\n因为它有很棒的鼓槌！'}