上一篇我们搭出了一个能调用工具的 ReAct Agent但它有个硬伤没有记忆。程序一重启、或者换一个会话它就完全不记得之前发生过什么更别提「在写数据库前先让人确认一下」这种需求了。这一篇解决三件事持久化用 Checkpointer 把每一步状态存下来重启不丢会话隔离用thread_id区分不同用户 / 不同会话人机协同Human-in-the-loop用interruptCommand实现「暂停等待人工输入 → 恢复执行」。代码可以直接接着上一篇的graph继续跑。一、Checkpointer给图装一个「存档系统」先说清楚它是什么。Checkpointer检查点器会在图的每个超步super-step之后把当前完整状态存一份。有了它你才能多轮对话记忆同一个线程再次调用时自动加载历史状态错误恢复 / 断点续跑崩了从上次存档继续时间旅行Time Travel回到任意历史检查点重新执行人工干预暂停执行等人输入后再恢复下一篇重点。一句话Checkpointer 是 LangGraph 里所有「有状态」能力的底座。三种 Checkpointer 怎么选Checkpointer导入路径存储位置适用场景InMemorySaverlanggraph.checkpoint.memory内存本地调试、Demo进程一退全没SqliteSaverlanggraph.checkpoint.sqlite本地 SQLite 文件单机开发、小规模服务重启不丢PostgresSaverlanggraph.checkpoint.postgresPostgreSQL生产环境多实例共享状态老教程里的MemorySaver在 v1 里已更名为InMemorySaver功能一样只是名字更表意。pipinstall-Ulanggraph-checkpoint-sqlite# SQLite 版需要单独装pipinstall-Ulanggraph-checkpoint-postgres# Postgres 版需要单独装二、动手给 Agent 加上记忆只需要两步编译时传入 checkpointer调用时传入 thread_id。importsqlite3fromlanggraph.checkpoint.sqliteimportSqliteSaver# 用 SQLite 做持久化check_same_threadFalse 让连接可跨线程复用connsqlite3.connect(checkpoints.sqlite,check_same_threadFalse)checkpointerSqliteSaver(conn)# 关键把 checkpointer 传进 compilegraphbuilder.compile(checkpointercheckpointer)# builder 沿用上一篇的图调用时必须给thread_id它相当于这次会话的「档案编号」fromlangchain_core.messagesimportHumanMessage config{configurable:{thread_id:user-123-session-1}}# 第一轮graph.invoke({messages:[HumanMessage(content我叫小明喜欢用 Python 举例。)]},config)# 第二轮同一个 thread_id它会自动带上第一轮的上下文resultgraph.invoke({messages:[HumanMessage(content我刚才说我叫什么名字)]},config)print(result[messages][-1].content)# 你叫小明而且偏好用 Python 举例。划重点config是invoke/stream的第二个位置参数别忘了传而且同一个会话的所有调用必须用同一个thread_id否则图找不到历史状态会当成全新会话。三、会话隔离不同用户互不串台thread_id就是天然的隔离维度换一个thread_id就是一块干净的白板config_a{configurable:{thread_id:user-A}}config_b{configurable:{thread_id:user-B}}graph.invoke({messages:[HumanMessage(content我叫小红)]},config_a)graph.invoke({messages:[HumanMessage(content你知道我叫什么吗)]},config_b)# B 的回答不知道我们刚开始聊。A 的信息不会串到 B常见的划分策略一个用户一段对话 →f{user_id}-{conversation_id}想要「长期记忆跨会话」→ 那不是thread_id的职责需要额外用BaseStore存用户偏好、事实thread_id只负责单会话的短期记忆。四、查看与回溯get_state/get_state_history调试和生产排障时能随时把某个线程的状态捞出来看非常有用。# 1) 看某个线程的当前状态stategraph.get_state(config)print(state.values)# 状态快照例如 {messages: [...]}print(state.next)# 下一步将要执行的节点空元组表示已结束# 2) 看完整的历史检查点实现「时间旅行」forsnapshotingraph.get_state_history(config):print(snapshot.config,snapshot.values)有了历史检查点你甚至可以用graph.update_state(...)手动把状态改回某个点再重跑做「后悔药」式的重规划。五、Human-in-the-loop用interruptCommand插入人工审批这是 LangGraph 最有价值的能力之一让图跑到一半停下来等人给个答复再继续。核心是两个原语interrupt(payload)在节点函数内部调用图会立刻暂停把payload抛给调用方Command(resume...)用人工的答复恢复执行interrupt的返回值就是你 resume 传进去的东西。⚠️前提必须有 checkpointerinterrupt需要把「暂停现场」持久化下来没有 checkpointer 会直接报错。这也是为什么本篇先讲 checkpointer。完整可运行示例发送邮件前人工审批fromtypingimportTypedDictfromlanggraph.checkpoint.memoryimportInMemorySaverfromlanggraph.graphimportStateGraph,START,ENDfromlanggraph.typesimportinterrupt,CommandclassEmailState(TypedDict):recipient:strcontent:strapproved:booldefdraft_email(state:EmailState)-dict:起草一封邮件这里用固定内容示意。return{recipient:bossexample.com,content:老板您好我想申请下周三休假一天谢谢,}defhuman_approval(state:EmailState)-dict:需要人工审批的节点执行到这里会暂停。decisioninterrupt({question:是否批准发送这封邮件,recipient:state[recipient],content:state[content],})# resume 传进来的值就是 interrupt 的返回值return{approved:decisionyes}defsend_email(state:EmailState)-dict:ifstate[approved]:print(f✅ 已发送给{state[recipient]}{state[content]})else:print(❌ 已取消发送)return{}builderStateGraph(EmailState)builder.add_node(draft,draft_email)builder.add_node(approval,human_approval)builder.add_node(send,send_email)builder.add_edge(START,draft)builder.add_edge(draft,approval)builder.add_edge(approval,send)builder.add_edge(send,END)graphbuilder.compile(checkpointerInMemorySaver())config{configurable:{thread_id:email-demo-001}}# ---------- 第一次调用跑到 interrupt 处暂停 ----------resultgraph.invoke({recipient:,content:,approved:False},config)print(result[__interrupt__])# (Interrupt(value{question: 是否批准发送这封邮件, ...}),)# ---------- 人工审批后恢复 ----------# 批准finalgraph.invoke(Command(resumeyes),config)print(final[approved])# True# 控制台输出✅ 已发送给 bossexample.com……# 如果拒绝只要换一句# graph.invoke(Command(resumeno), config)执行流程draft ──▶ approval ──■ 暂停等待人工 ──(Command(resume...))──▶ send ──▶ END ▲ __interrupt__ 抛出待审批内容注意两个点interrupt之后的代码不会重复执行恢复时从interrupt的返回值继续往下走恢复调用时必须用同一个thread_id否则图找不到暂停现场。另一种写法Command(goto...)动态路由除了暂停恢复Command还能在节点内部直接指定下一步去哪相当于把条件边写进节点里fromlanggraph.typesimportCommanddefroute_after_review(state:EmailState)-Command:ifstate[approved]:returnCommand(gotosend)# 跳到发送returnCommand(gotoEND,update{content:})# 结束并顺手清空内容这让「节点既干活又决定流向」的写法更紧凑。六、5 个高频报错与坑附解决1.ValueError: No checkpointer provided to the graph只要 config 里带了thread_idLangGraph 就会去找 checkpointer没有就抛错。→ 解决builder.compile(checkpointer...)别漏。2.interrupt报「没有 checkpointer」或恢复不了interrupt依赖持久化且两次调用必须同一个thread_id。→ 解决加 checkpointer并复用同一个 config。3. 用langgraph dev测试持久化重启后状态却没了langgraph dev会用服务端自己注入的持久化你在图里 compile 的 checkpointer 会被绕过看起来「存了」其实没存。→ 解决验证持久化要用直接的 Python 调用别只信langgraph dev。4.Recursion limit of 25 reached without hitting a stop condition图里有循环比如工具调用一旦模型反复要调工具、路由又不收敛就会撞上默认递归上限 25。→ 解决检查路由逻辑是否可能死循环确需更长链可临时调高graph.invoke(inputs,config{recursion_limit:50,**config})5. 字段莫名被覆盖 / 消息历史不全要累积的字段忘了挂 reducer。→ 解决列表字段用Annotated[list, add_messages]消息或Annotated[list, operator.add]普通列表。七、小结能力关键 API记忆点持久化compile(checkpointer...)编译时挂上存档系统会话隔离config{configurable: {thread_id: ...}}换 id 换白板查看状态get_state/get_state_history排障与时间旅行人工审批interruptCommand(resume...)暂停靠 checkpointer 兜底动态路由Command(goto...)节点内直接决定下一步到这里「入门 → 有状态 → 人机协同」这条主线就走通了。下一步可以往这两个方向深挖多智能体用子图Subgraph和SendAPI 做并行 fan-out / supervisor 编排长期记忆用BaseStore存跨会话的用户偏好和thread_id的短期记忆配合使用。需要的话下一篇我们接着聊多智能体编排。本文是「LangGraph 实战」系列第二篇建议先读第一篇《从 StateGraph 到你的第一个 ReAct Agent》。代码基于 LangGraph v1实测可跑欢迎评论区交流你遇到的坑。