Skip to main content
当与基于模型进行决策的非确定性系统（例如由LLM驱动的智能体）一起工作时，详细检查其决策过程可能很有用：
  1. 理解推理：分析导致成功结果的过程步骤。
  2. 调试错误：确定错误发生的位置和原因。
  3. 探索替代方案：测试不同的路径以发现更好的解决方案。
LangGraph提供了时间旅行功能以支持这些用例。具体来说，您可以从先前的检查点恢复执行——要么重新播放相同的状态，要么修改它以探索替代方案。在所有情况下，恢复过去的执行都会在历史中产生一个新的分支。 要使用 LangGraph 中的 时间旅行：
  1. 使用invoke或stream方法，以初始输入运行图。
  2. 在现有线程中识别一个检查点：使用get_state_history()方法检索特定thread_id的执行历史，并定位所需的checkpoint_id。或者，在希望执行暂停的节点之前设置一个interrupt。然后，您可以找到记录到该中断为止的最新检查点。
  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]


model = init_chat_model(
    "anthropic:claude-sonnet-4-5",
    temperature=0,
)


def generate_topic(state: State):
    """LLM call to generate a topic for the joke"""
    msg = model.invoke("Give me a funny topic for a joke")
    return {"topic": msg.content}


def write_joke(state: State):
    """LLM call to write a joke based on the topic"""
    msg = model.invoke(f"Write a short joke about {state['topic']}")
    return {"joke": msg.content}


# Build workflow
workflow = StateGraph(State)

# Add nodes
workflow.add_node("generate_topic", generate_topic)
workflow.add_node("write_joke", write_joke)

# Add edges to connect nodes
workflow.add_edge(START, "generate_topic")
workflow.add_edge("generate_topic", "write_joke")
workflow.add_edge("write_joke", END)

# Compile
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"])
输出： 
How about "The Secret Life of Socks in the Dryer"? You know, exploring the mysterious phenomenon of how socks go into the laundry as pairs but come out as singles. Where do they go? Are they starting new lives elsewhere? Is there a sock paradise we don't know about? There's a lot of comedic potential in the everyday mystery that unites us all!

# The Secret Life of Socks in the Dryer

I finally discovered where all my missing socks go after the dryer. Turns out they're not missing at all—they've just eloped with someone else's socks from the laundromat to start new lives together.

My blue argyle is now living in Bermuda with a red polka dot, posting vacation photos on Sockstagram and sending me lint as alimony.

​
2. 识别检查点

# The states are returned in reverse chronological order.
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

# This is the state before last (states are listed in chronological order)
selected_state = states[1]
print(selected_state.next)
print(selected_state.values)
输出： 
('write_joke',)
{'topic': 'How about "The Secret Life of Socks in the Dryer"? You know, exploring the mysterious phenomenon of how socks go into the laundry as pairs but come out as singles. Where do they go? Are they starting new lives elsewhere? Is there a sock paradise we don\\'t know about? There\\'s a lot of comedic potential in the everyday mystery that unites us all!'}

​
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': 'Why did the chicken join a band?\n\nBecause it had excellent drumsticks!'}
在GitHub上编辑此页面的源代码。
通过MCP将这些文档编程连接到Claude、VSCode等，以获取实时答案。