07：时间旅行

大家好，我是大志。

在上一篇文章中，我们详细介绍了如何使用LangGraph的中断和恢复，以及使用中断和恢复的一些注意点，本文将会介绍LangGraph时间旅行功能。

文中所有示例代码：https://github.com/wzycoding/langchain-study

一、什么是时间旅行

在开发AI Agent系统时，LLM的输出是不确定性的，我们经常可能会对Agent执行的每一步进行调试，比如：

分析每一步产生的结果
调试和解决bug，定位错误原因
通过调试和分析来寻找更好的问题解决方案
修改某个节点的状态数据，然后重新执行整个 Agent。

LangGraph提供了这种调试的能力，支持从任意一个检查点恢复执行，并且会产生一个新的分支，在恢复执行过程中生成新的检查点，这种调试能力就是时间旅行。

除此之外，还可以修改指定检查点的图状态数据，LangGraph会创建一个修改数据后的新检查点，并且从该新的检查点恢复执行，同样也会生成一个新的分支。

二、时间旅行使用方法

2.1 使用方法

时间旅行的使用方法如下：

创建、编译、运行图
通过图对象获取历史的检查点数据，找到想要操作的检查点，并找到指定的检查点id。
根据实际需要决定是否对图状态数据进行更新，如果需要更新，使用update_state方法修改图中检查点的状态。
从之前找到的检查点id对应的检查点恢复执行，使用invoke或者stream方法，并且输入数据为None，并传入指定的thread_id和checkpoint_id.

2.2 使用案例

假设有以下业务场景：

用户输入主题，LLM根据主题生成相关的问题，再让LLM根据问题生成问题对应答案，接下来，我们使用LangGraph来完成这个需求。

首先定义LLM对象和图状态数据，在图状态数据中包括：用户输入的主题、生成的问题、生成的回答。

python

llm = init_chat_model(
    "gpt-4o-mini",
    temperature=0
)


class State(TypedDict):
    # 主题
    topic: str
    # 问题
    question: str
    # 回答
    answer: str

定义生成问题节点和生成答案节点

python

def generate_question_node(state: State):
    """生成问题节点"""
    ai_message = llm.invoke(f"根据{state['topic']}为主题生成一道选择题，难度适中，只需要生成题目，不要输出额外内容。")

    return {"question": ai_message.content}

def generate_answer_node(state: State):
    """生成答案节点"""
    ai_message = llm.invoke(f"生成这个问题的答案：{state['question']}，并且给出清晰易懂的解释。")

    return {"answer": ai_message.content}

构建图

python

graph = StateGraph(State)

graph.add_node("generate_question", generate_question_node)
graph.add_node("generate_answer", generate_answer_node)

graph.add_edge(START, "generate_question")
graph.add_edge("generate_question", "generate_answer")
graph.add_edge("generate_answer", END)

创建检查点管理器

python

conn = connect("postgres://postgres:postgres@localhost:5432/langgraph", autocommit=True)
checkpointer = PostgresSaver(conn)
checkpointer.setup()

编译运行图，编译时指定checkpointer，执行时指定线程id。

python

config = {"configurable": {"thread_id": "100"}}
agent = graph.compile(checkpointer=checkpointer)
state = agent.invoke({"topic": "英语可数名词与不可数名词"}, config=config)

输出图状态数据中生成的问题和答案，通过get_state_history方法获取到当前线程id的所有检查点数据，将这些检查点信息进行打印，因为我们要找到想要恢复执行的检查点id。

python

print(f"主题：{state['topic']}\n问题：{state['question']}\n答案：{state['answer']}\n")

states = list(agent.get_state_history(config))

for state in states:
    print(f"下一个节点：{state.next}")
    print(state.config["configurable"]["checkpoint_id"])
    print()

执行结果如下，正确的输出了问题和答案，并且打印出了检查点信息。

python

主题：英语可数名词与不可数名词
问题：Which of the following sentences uses the noun correctly?  

A) I have much friends in this city.  
B) She bought a few milk at the store.  
C) There is some water in the bottle.  
D) He needs many furniture for his new house.
 答案：答案是 **C) There is some water in the bottle.**

### 解释：
- **A) I have much friends in this city.**
  - 错误：英语中“friends”是可数名词，应该使用“many”而不是“much”。“Much”通常用于不可数名词。
  - 正确句子应该是：**I have many friends in this city.**

- **B) She bought a few milk at the store.**
  - 错误：“milk”是不可数名词，所以不应该用“a few”来修饰。对于不可数名词，我们使用“some”或者其他合适的量词。
  - 正确句子应该是：**She bought some milk at the store.**

- **C) There is some water in the bottle.**
  - 正确：这句话使用了正确的不可数名词“water”和合适的量词“some”。“Some”用于不可数名词表示某些、一些的意思。

- **D) He needs many furniture for his new house.**
  - 错误：“furniture”是不可数名词，所以不能用“many”来修饰。应该使用“much”。
  - 正确句子应该是：**He needs much furniture for his new house.**

总的来说，选项C使用了不可数名词“water”并正确地使用了“some”来表示某些量。

下一个节点：()
1f0eb753-7bde-69be-8002-70121dbd1943

下一个节点：('generate_answer',)
1f0eb753-494e-64ea-8001-d5851d91fb48

下一个节点：('generate_question',)
1f0eb753-25a1-69ac-8000-19b986b7f081

下一个节点：('__start__',)
1f0eb753-259c-6aa6-bfff-73b0709d8e73

下面是即将执行生成回答之前的检查点。

python

下一个节点：('generate_answer',)
1f0eb753-494e-64ea-8001-d5851d91fb48

现在我们想进行调试，在执行生成答案节点之前，改变主题和问题内容，让LLM重新为我们生成新问题的答案，首先，使用update_state方法更新这个检查点的图状态信息，将topic改为地理知识，将question改为“中国有几个直辖市?”，该方法会创建一个新的检查点。

python

config = {"configurable": {"thread_id": "100", "checkpoint_id": "1f0eb753-494e-64ea-8001-d5851d91fb48"}}
agent = graph.compile(checkpointer=checkpointer)
new_config = agent.update_state(config, values={"question": "中国有几个直辖市?", "topic": "地理知识"})

之后，使用update_state方法返回的new_config作为配置数据，传入invoke方法，表示从update_state创建的新检查点开始恢复继续向下执行。

python

state = agent.invoke(None, config=new_config)

print(f"主题：{state['topic']}\n问题：{state['question']}\n答案：{state['answer']}\n")

执行结果如下，根据问题重新生成了新的问题答案。

python

主题：地理知识
问题：中国有几个直辖市?
答案：中国有 **4 个直辖市**，分别是：

1. **北京**：是中国的首都，也是政治、文化、国际交流中心。
2. **上海**：中国的经济中心之一，是全球重要的金融和航运中心。
3. **天津**：位于华北地区，是中国重要的港口城市之一，也是工业和商业中心。
4. **重庆**：位于西南部，是中国内陆最大的直辖市，也是重要的交通枢纽和工业基地。

三、总结

本文介绍了什么是LangGraph 的时间旅行机制。接着，通过示例展示了时间旅行在真实业务中的应用方式。

通过本文，相信你已经对 LangGraph 的时间旅行机制 有了完整、清晰的理解。它不仅能解决调试和优化的问题，更重要的是，让 Agent 执行过程变得可控、可追踪。

一、什么是时间旅行 ​

二、时间旅行使用方法 ​

2.1 使用方法 ​

2.2 使用案例 ​

三、总结 ​

一、什么是时间旅行

二、时间旅行使用方法

2.1 使用方法

2.2 使用案例

三、总结