企业官网建设流程全解析

LangGraph Checkpointer：持久化、断点续跑与多轮记忆

图跑完状态就没了——这在 demo 里没问题，在生产里是硬伤：
用户刷新页面对话断了、人工审批要等一天、服务重启任务丢了。

Checkpointer 解决的就是"让图的状态活下来"。
这篇讲清楚它怎么用，以及thread_id这个关键概念。

一、Checkpointer 做了什么

简单说：每执行完一个节点，就把当前 State 快照存一份。

节点 A 执行完 → 存快照 1 节点 B 执行完 → 存快照 2 节点 C 执行完 → 存快照 3

有了这些快照，你就能：

• 继续：同一个会话的下一次输入，从最新快照继续；
• 回溯：拿到历史上任意一个快照；
• 恢复：interrupt暂停后，隔多久都能接着跑；
• 分叉：从某个历史快照复制出一条新分支（做"如果当时……"的实验）。

二、最小可用：内存版

fromlanggraph.checkpoint.memoryimportInMemorySaverfromlanggraph.prebuiltimportcreate_react_agent checkpointer=InMemorySaver()agent=create_react_agent(llm,tools,checkpointer=checkpointer)config={"configurable":{"thread_id":"user-123"}}agent.invoke({"messages":[("user","我叫小明")]},config=config)agent.invoke({"messages":[("user","我叫什么？")]},config=config)# → "你叫小明"

关键就在thread_id。

• 同一个thread_id→ 共享同一份状态，多轮对话有记忆；
• 不同thread_id→ 完全独立的会话。

InMemorySaver只存在进程内存里，重启即丢。
只适合开发调试。生产必须换成持久化的实现。

三、生产：SQLite / Postgres

3.1 SQLite（单机、轻量）

pipinstalllanggraph-checkpoint-sqlite

fromlanggraph.checkpoint.sqliteimportSqliteSaver# 方式一：文件路径，需手动管理连接saver=SqliteSaver.from_conn_string("checkpoints.db")# 方式二：配合 sqlite3 连接（推荐，可设 check_same_thread=False）importsqlite3 conn=sqlite3.connect("checkpoints.db",check_same_thread=False)saver=SqliteSaver(conn)agent=create_react_agent(llm,tools,checkpointer=saver)

SQLite 适合单机部署、中小规模。优点是不需要额外服务，
缺点是写操作串行，并发高了会争锁。

3.2 Postgres（生产推荐）

pipinstalllanggraph-checkpoint-postgres

fromlanggraph.checkpoint.postgresimportPostgresSaver DB_URI="postgresql://user:pwd@localhost:5432/agentdb"withPostgresSaver.from_conn_string(DB_URI)assaver:saver.setup()# 第一次要建表agent=create_react_agent(llm,tools,checkpointer=saver)...

生产选 Postgres 的理由和向量库那篇一样：
备份、监控、扩容都复用现有体系，不用再维护一个新组件。

四、断点续跑：interrupt + resume

这是 Checkpointer 最有价值的用法。结合上一篇讲的人工介入：

fromlanggraph.typesimportinterrupt,Commanddefsensitive_node(state):# 暂停，等待外部决策decision=interrupt({"action":"refund","amount":299.0})ifdecision=="approve":return{"result":do_refund()}return{"result":"用户已拒绝"}

调用流程：

config={"configurable":{"thread_id":"order-888"}}# 第一次调用：跑到 interrupt 就停result=agent.invoke({"messages":[("user","我要退款")]},config=config)print(result.get("__interrupt__"))# [Interrupt(value={'action': 'refund', 'amount': 299.0})]# —— 此时可以关掉服务、过一天再回来 ——# 恢复执行result=agent.invoke(Command(resume="approve"),config=config)

注意第二次invoke传的不是新的用户输入，而是Command(resume=...)。
只要thread_id一致，LangGraph 就能从快照里恢复出暂停时的状态继续跑。

这个能力支撑了非常实际的场景：
退款审批、合同确认、大额下单——图可以挂起几天，
人工在后台点完"同意"，流程自动往下走。

五、查看和管理历史

5.1 拿到当前状态

snapshot=agent.get_state(config)print(snapshot.values)# 当前 State 内容print(snapshot.next)# 下一步会执行哪些节点print(snapshot.tasks)# 待办（含 interrupt 信息）

5.2 遍历历史

forsnapinagent.get_state_history(config):print(snap.config["configurable"]["checkpoint_id"],snap.next)

5.3 从历史快照分叉（时间旅行）

history=list(agent.get_state_history(config))old=history[2]# 回到第 3 个快照# 用旧的 checkpoint_id 构造新 configforked_config={"configurable":{"thread_id":"user-123","checkpoint_id":old.config["configurable"]["checkpoint_id"],}}# 在新分支上继续跑，不影响原分支agent.invoke({"messages":[("user","换个思路重新分析")]},config=forked_config)

"分叉"在调试时特别有用：跑到一半发现走错了，
退回去改个输入重新跑，原分支还留着可以对比。

5.4 手动改状态

agent.update_state(config,{"answer":"手动修正后的答案"})

直接覆写某个字段，然后继续跑。适合人工修正 Agent 的错误输出。

六、Checkpointer 与消息裁剪

有个实际问题：状态存久了，消息列表会越来越长，
每次加载都要把所有历史消息塞进 prompt，成本和延迟都会涨。

解法是定期裁剪，只保留最近 N 条：

fromlangchain_core.messagesimportAnyMessagefromlanggraph.graph.messageimportadd_messages,RemoveMessage MAX_MESSAGES=20deftrim_messages(state):msgs=state["messages"]iflen(msgs)<=MAX_MESSAGES:return{}# 删掉最老的，保留最近 MAX_MESSAGES 条to_remove=[RemoveMessage(id=m.id)forminmsgs[:-MAX_MESSAGES]]return{"messages":to_remove}

在图里加一个节点做这件事：

graph.add_node("trim",trim_messages)graph.add_edge(START,"trim")graph.add_edge("trim","agent")

注意别把系统消息和用户的最新输入裁掉，否则会出奇怪的行为。

另一个思路是摘要：历史太长时用 LLM 压成一段摘要，
替换掉原始消息。LangChain 有现成的summarize_messages。

七、常见坑

八、小结

• Checkpointer每节点存一份 State 快照，支撑继续 / 回溯 / 恢复 / 分叉；
• thread_id是唯一关键：相同 id 共享会话，不同 id 完全隔离；
• InMemorySaver只用于调试，生产用 SqliteSaver 或 PostgresSaver；
• 断点续跑=interrupt()暂停 +Command(resume=...)恢复，
  配合持久化可以挂起几天；
• get_state/get_state_history/update_state三个 API 用于查看和管理；
• 消息会无限累积，务必做裁剪或摘要，否则成本和延迟会失控。

下一篇讲部署上线——把链或 Agent 变成一个 HTTP 服务。