📚 AI 工程与智能体教程站

第 7 章 · 持久化：Checkpointer 与 Thread ​

本章目标：

  • 理解持久化在 Agent 服务中的重要性
  • 掌握三种 Checkpointer 的实现与适用场景
  • 学会使用 thread_id 管理多会话
  • 实现断点续跑与时间旅行

7.1 为什么需要持久化 ​

Agent 应用通常需要：

  • 会话连续性：多轮对话保持上下文
  • 断点续跑：服务重启后恢复执行
  • Human-in-the-loop：暂停等待人工审批
  • 时间旅行：回溯到任意历史状态

7.2 Checkpointer 类型 ​

python
from langgraph.checkpoint.memory import MemorySaver
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.checkpoint.postgres import PostgresSaver
import psycopg2

# 1. MemorySaver：内存存储（开发调试）
memory_saver = MemorySaver()

# 2. SqliteSaver：文件持久化（中小规模）
sqlite_saver = SqliteSaver.from_conn_string("checkpoints.db")

# 3. PostgresSaver：生产级（大规模）
conn = psycopg2.connect("dbname=agent host=localhost")
postgres_saver = PostgresSaver(conn)

7.3 带 Checkpointer 的 Agent ​

python
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()

def llm_node(state: MessagesState) -> dict:
    # 实际调用 LLM
    return {
        "messages": [{
            "role": "assistant",
            "content": f"这是第 {len(state['messages'])} 轮对话"
        }]
    }

graph = StateGraph(MessagesState)
graph.add_node("llm", llm_node)
graph.add_edge(START, "llm")
graph.add_edge("llm", END)
compiled = graph.compile(checkpointer=memory)

# 不同 thread_id = 不同会话
thread_config = {"configurable": {"thread_id": "user_123"}}
result = compiled.invoke(
    {"messages": [{"role": "user", "content": "你好"}]},
    config=thread_config
)

# 同一 thread 保持上下文
result2 = compiled.invoke(
    {"messages": [{"role": "user", "content": "你刚才说了什么？"}]},
    config=thread_config
)
print(result2["messages"][-1]["content"])

7.4 断点续跑 ​

python
from langgraph.types import Command, interrupt

def human_approval_node(state: MessagesState) -> dict:
    # 中断等待人工审批
    approval = interrupt({
        "question": "是否继续执行？",
        "context": state["messages"]
    })
    
    if approval:
        return {"messages": [{"role": "assistant", "content": "已批准，继续执行"}]}
    else:
        return {"messages": [{"role": "assistant", "content": "已取消"}]}

# 恢复执行
resume_config = {"configurable": {"thread_id": "user_123", "checkpoint_id": "xxx"}}
result = compiled.invoke(Command(resume=True), config=resume_config)

本章小结 ​

  • MemorySaver 用于开发，PostgresSaver 用于生产
  • thread_id 区分不同会话
  • interrupt() 实现人机协作
  • Checkpointer 支持断点续跑和时间旅行

🛠️ 动手实践 ​

  1. 实现带 MemorySaver 的多轮对话 Agent
  2. 使用 SqliteSaver 持久化，验证服务重启后状态保留
  3. 实现中断恢复：在工具调用前暂停，等待用户确认

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. MemorySaver 适用于什么场景？

2. thread_id 的作用是什么？

3. interrupt() 函数的用途是？

4. PostgresSaver 相比 SqliteSaver 的优势是什么？