一、引言单轮对话的天花板在哪里把 LLM 接进业务系统的第一版代码通常长这样拼一个 prompt调一次invoke把字符串返回给前端。这在 Demo 阶段完全够用但一旦进入生产问题会集中爆发状态无处安放。多轮对话的历史、检索到的资料、上一次工具调用的返回值只能靠调用方自己维护一个不断膨胀的messages列表。流程不可中断。凡是需要“人工审批后再继续”的场景发邮件、改数据库、下单单轮调用只能把整个上下文序列化出去等审批完了再重新拼一遍 prompt —— 中间的 LLM 调用被白白重跑。失败即重来。第三步的 API 超时了前两步的 token 已经烧掉你只能从头再来一遍。循环不可控。Agent 自我反思、工具重试这类天然带环的结构在 LCEL 那种有向无环图DAG里写不出来只能塞进一个 while 循环于是可观测性彻底归零。没有审计轨迹。出事之后你无法回答“当时模型看到了什么、在哪一步做了错误决策”。LangChain 的 LCEL 解决的是编排问题它本质上是一张静态 DAG。而真实的 Agent 带环的状态机 持久化 人机协同。LangGraph 正是为这个模型设计的把工作流显式建模成一张图节点是函数边是控制流共享状态由 reducer 驱动合并每一步执行后都落盘成 checkpoint。这篇文章会从原理讲到代码搭一个“调研 → 起草 → 人工审批 → 修订 → 归档”的可中断、可恢复智能体。1.1 先看一段“看起来能跑”的代码为了让后面的对比足够直观我们先把那个 Demo 版本完整写出来。它的典型形态是这样的fromlangchain_openaiimportChatOpenAI llmChatOpenAI(modelgpt-4o-mini)history[]# 调用方自己维护的全局状态defchat(user_input:str)-str:history.append({role:user,content:user_input})respllm.invoke(history)history.append({role:assistant,content:resp.content})returnresp.content这段代码的每一个“缺点”都不是风格问题而是结构问题history是进程内的一个 list进程一重启就没了多副本部署时更是各说各话一旦要加“先检索资料再回答”你就得在chat里塞if加了工具调用就要再塞一个while循环判断模型是否还要调工具想加“下单前需要人工确认”你只能把整个history序列化到数据库审批通过后再反序列化重建 —— 而重建之后模型为了走到同一个位置还得把之前所有的 token 重新烧一遍出问题时你翻遍日志也只能看到input和output中间发生过什么完全黑盒。把这五个问题抽象一下其实是三个能力缺失状态的持久化、控制流的显式化、执行过程的可观测。LangGraph 的三个核心抽象 —— State、Edge、Checkpointer —— 正好一一对应。1.2 LangGraph 到底在解决什么问题可以用一句话概括 LangGraph 的心智模型一张带环的有向图 一个可序列化的共享状态 一个每步落盘的检查点。把这句话和几个相邻概念做个对比定位会更清楚方案控制流状态管理中断/恢复循环支持适用场景手写 while 循环隐式藏在代码里自己维护自己实现支持但难控一次性脚本LCELRunnable 组合静态 DAG靠管道传递不支持不支持线性数据流水线AgentExecutor隐式 ReAct 循环靠 memory弱内置但不可定制通用工具型 AgentLangGraph显式图结构TypedDict reducercheckpoint 原生支持一等公民生产级多步骤工作流注意表格里 LCEL 那一行 —— 它并不是“旧方案”LCEL 依然是 LangChain 表达式组合的基础LangGraph 的节点内部照样可以用 LCEL 写链。二者的关系是嵌套而不是替代LCEL 负责“一段数据怎么流”LangGraph 负责“多个步骤怎么调度、在哪停、从哪续”。二、核心原理StateGraph 的三要素2.1 State 与 Reducer并发写入的合并规则LangGraph 的状态就是一个TypedDict。关键设计在于字段级别的 reducer当两个并行节点同时往同一个 key 写值时用哪个 reducer 决定合并结果。classState(TypedDict):messages:Annotated[list,add_messages]# 追加而非覆盖自动去重/合并消息sources:Annotated[list[str],operator.add]# 并行检索结果自动拼接draft:str# 无注解 后写覆盖先写没有给 list 字段加 reducer是新手最常踩的坑 —— 并行分支一跑就抛InvalidUpdateError。记住一条规则任何可能被多个节点写入的字段都必须显式声明 reducer。理解 reducer 的关键是理解“节点返回的是什么”。节点函数的返回值不是新状态而是状态增量partial update。框架拿到这个增量后对每一个 key 做如下判断有 reducer 的 key调用reducer(旧值, 新值)把结果写回没有 reducer 的 key直接用新值覆盖旧值增量里没出现的 key原封不动保留。所以下面两种写法语义完全不同# 写法 A覆盖。第二次执行后 messages 只剩 1 条defnode_a(state):return{messages:[new_msg]}# 写法 B追加。第二次执行后 messages 有 2 条依赖 add_messagesdefnode_b(state):return{messages:[new_msg]}# 前提是 State 里 messages 带了 add_messages写法 A 和 B 的代码长得一模一样差别完全隐藏在State的注解里。这正是 reducer 设计“隐蔽但强大”的地方也是团队协作时最容易出事故的地方 —— 建议在State定义上方统一写注释标注每个字段的合并语义。几个常用 reducer 一览字段类型推荐 reducer语义list[消息]add_messages按id去重合并支持RemoveMessage删除list[str]/list[dict]operator.add简单拼接int累加operator.add计数、求和bool标记位不加覆盖后写覆盖dict合并自定义函数通常做浅合并{**old, **new}下面是一个自定义 dict 合并 reducer 的例子在“配置逐步补全”的场景里非常有用defmerge_dict(old:dict,new:dict)-dict:浅合并新值覆盖同名 key未出现的 key 保留。return{**old,**new}classState(TypedDict):config:Annotated[dict,merge_dict]还有一个必须知道的细节返回未在 State 中声明的 key会被静默忽略。这不像 Python 字典会报KeyError它不会报错只会让你的“精心构造的字段”永远消失。调试时如果发现某个字段永远是初始值第一件事就是检查它有没有写进TypedDict。2.2 节点与边把控制流画出来节点是纯函数(state) - dict返回的 dict 是状态增量partial update由框架用 reducer 合并回全局状态。边分两种add_edge(a, b)确定性跳转add_conditional_edges(a, router, {x: b, y: c})router 函数读状态返回 key实现分支与循环把条件边指回上游节点即可成环。START和END是两个虚节点标明入口与出口。关于节点函数有三条容易被忽略的约定签名可以不接收全部参数。你可以只写def node(state: State)也可以用Literal[messages, draft]声明只关心某些 keyLangGraph 会把该 key 的最新值传进来。后者在状态很大时能减少无谓的数据传递。可以额外接收config。写成def node(state: State, config: RunnableConfig)就能拿到config[configurable][thread_id]之类的运行时信息做多租户路由、日志打点时必备。返回值必须是 dict 或 None。返回None表示“本节点不修改状态”合法但不推荐 —— 因为它会让图的语义难以阅读。条件边里的router函数写法也有讲究。它只读状态、不做副作用返回值必须是字典里声明过的 key否则运行时会报错。如果分支很多用Literal标注返回值类型能在 IDE 里获得自动补全fromtypingimportLiteraldefrouter(state:State)-Literal[search,answer,give_up]:ifstate[retry]3:returngive_upifstate.get(need_search):returnsearchreturnanswer关于“循环”再强调一遍LangGraph 里没有专门的while原语循环就是“条件边指回上游节点”。这也意味着你必须自己准备熔断条件比如重试计数否则图会一直转到触发recursion_limit抛异常为止。默认recursion_limit是 25 个 super-step可以在config里调整config{recursion_limit:50,configurable:{thread_id:t1}}2.3 Checkpointer可追踪、可恢复的全部秘密编译图时传入 checkpointerLangGraph 会在每个 super-step一次图执行批次结束后把完整状态序列化落盘主键是thread_id。这一个机制同时解决了四件事能力依赖的 API断点续跑thread_id checkpointer重启进程后继续人工介入interrupt()挂起Command(resume...)恢复状态回放/时间旅行get_state_history()列出所有历史 checkpoint故障重试从最后一个成功的 checkpoint 重新调度interrupt()在节点内部调用会抛出特殊信号让图暂停并把 payload 交给人。关键陷阱恢复时该节点会从头重跑所以interrupt()之前的副作用代码必须幂等。生产环境用PostgresSaver本地开发用SqliteSaver纯测试用InMemorySaver。这里把“super-step”这个概念再展开一下因为它直接决定了 checkpoint 的粒度。LangGraph 的执行是按批次推进的一个 super-step 里所有可以并行执行的节点同时跑比如plan之后的两个检索节点等这一批全部执行完状态统一合并、写一次 checkpoint然后才计算下一批要执行哪些节点。所以 checkpoint 的数量不等于节点数量而是等于“批次数”。在多分支并行的图里一次 checkpoint 可能对应好几个节点的产出这是正常的。interrupt()的行为同样值得推敲。它的完整生命周期是节点执行到interrupt(payload)LangGraph 抛出一个内部信号当前节点中断图暂停状态包括中断前的所有变更写入 checkpointpayload暴露给调用方调用方用Command(resumevalue)重新invokeLangGraph 从断点所在的 checkpoint 恢复重新执行该节点函数这一次interrupt()不再抛异常而是直接返回resume传入的值函数继续往下跑。第 4 步是重点“重跑”是真的从头跑一遍函数体。所以下面的写法是危险的defbad_node(state):send_email(state[draft])# ❌ 副作用在 interrupt 之前decisioninterrupt({draft:state[draft]})return{approved:decision[approved]}审批恢复后send_email会被再调用一次。正确做法是把副作用挪到interrupt()之后或者用幂等键去重。关于持久化后端选型建议如下实现包适用场景注意点InMemorySaver内置单元测试、一次性脚本进程结束即丢失SqliteSaverlanggraph-checkpoint-sqlite本地开发、单机部署需要check_same_threadFalsePostgresSaverlanggraph-checkpoint-postgres生产环境首次使用需调.setup()建表Postgres 的接法大致长这样fromlanggraph.checkpoint.postgresimportPostgresSaver DB_URIpostgresql://user:pwdlocalhost:5432/agent?sslmodedisablewithPostgresSaver.from_conn_string(DB_URI)assaver:saver.setup()# 建表只需执行一次graphbuilder.compile(checkpointersaver)setup()会创建checkpoints、checkpoint_blobs、checkpoint_writes等几张表。生产环境务必建索引并按时间分区/清理否则高并发下这张表会迅速膨胀到几十 GB。三、实战可中断、可恢复的调研智能体场景用户提一个技术选型问题Agent 规划检索词 → 检索 → 起草报告 →暂停等待人工审批→ 不通过则按意见修订。3.1 定义状态先想清楚“什么要持久化”状态设计是整个实现里最重要的一步。经验法则是凡是需要在中断后恢复时还能拿到的东西都必须进 State凡是进 State 的东西都必须可序列化。fromtypingimportAnnotated,Literalimportoperatorfromlanggraph.graphimportStateGraph,START,ENDfromlanggraph.graph.messageimportadd_messagesfromlanggraph.checkpoint.memoryimportInMemorySaverfromlanggraph.typesimportinterrupt,Commandfromlangchain_core.messagesimportHumanMessage,SystemMessage MAX_REVISIONS3# 熔断阈值防止无限修订循环classResearchState(TypedDict):# ---- 输入 ----topic:str# ---- 中间产物 ----queries:list[str]# plan 节点产出sources:Annotated[list[str],operator.add]# 两个检索节点并行写入 → 必须加 reducerdraft:str# 每轮修订覆盖# ---- 人机协同 ----approved:boolreview_notes:str# 人工填写的修改意见# ---- 熔断 ----revision_count:Annotated[int,operator.add]# 每次审批 1# ---- 审计 ----messages:Annotated[list,add_messages]# 全链路消息日志几个设计决策值得说明sources用operator.add因为两个检索节点会并行写入必须拼接而非覆盖revision_count也用operator.add每次进入 review 节点就 1这样即使节点重跑语义依然是“审批次数 1”注意由于 review 节点的重跑发生在 interrupt 之后实际只会 1 一次draft不加 reducer因为它是“覆盖式”字段每次修订就是最新版本messages用add_messages它会按消息id去重天然适配流式追加和中断恢复。3.2 定义节点每个函数只做一件事规划节点——把用户的问题拆成检索词defplan_node(state:ResearchState)-dict:把 topic 拆成 2~3 个检索词。纯函数、无副作用可安全重跑。prompt(f你是技术调研助手。请针对下面的问题生成 3 个互补的检索关键词f每行一个不要编号不要解释。\n\n问题{state[topic]})rawllm.invoke([HumanMessage(contentprompt)]).content queries[q.strip()forqinraw.splitlines()ifq.strip()][:3]return{queries:queries,# 用 add_messages 追加形成审计日志messages:[HumanMessage(contentf[plan] 生成的检索词{queries})],}检索节点——两个并行版本分别打内部知识库和公网defsearch_internal(state:ResearchState)-dict:查内部知识库。返回 sources 增量由 operator.add 自动拼接。hits[]forqinstate[queries]:hits.extend(vector_store.similarity_search(q,k2))return{sources:[f[内部]{h.page_content[:200]}forhinhits]}defsearch_web(state:ResearchState)-dict:查公网。失败时返回空增量不阻断主流程。try:results[f[公网]{r[title]}:{r[snippet]}forqinstate[queries]forrinweb_search(q,limit2)]exceptExceptionasexc:# 单点失败降级return{messages:[HumanMessage(contentf[search_web] 失败{exc})]}return{sources:results}起草节点——汇总资料写初稿defdraft_node(state:ResearchState)-dict:根据 sources 和人工意见生成/修订草稿。notesstate.get(review_notes)or首轮无人工意见context\n.join(state[sources])[:8000]# 截断防止超出上下文promptf你是技术分析师请撰写一份关于「{state[topic]}」的调研简报。 要求 1. 结构为结论 / 依据 / 风险 / 建议 2. 只使用下面提供的资料不要编造 3. 如果在修订必须逐条回应人工意见。 【参考资料】{context}【人工修改意见】{notes}draftllm.invoke([SystemMessage(contentprompt)]).contentreturn{draft:draft,messages:[HumanMessage(contentf[draft] 第{state[revision_count]1}稿完成)],}审批节点——整个方案的核心用interrupt()暂停defreview_node(state:ResearchState)-dict:把草稿交给人工审批。interrupt 会让图在此暂停并落盘。# ⚠️ interrupt 之前的代码在恢复时会被重跑这里必须保持无副作用decisioninterrupt({type:approval_request,topic:state[topic],draft:state[draft],instructions:用 Command(resume{approved: bool, notes: str}) 恢复,})return{approved:bool(decision.get(approved)),review_notes:decision.get(notes,),revision_count:1,# 累加到 state 里operator.addmessages:[HumanMessage(contentf[review] approved{decision.get(approved)}, fnotes{decision.get(notes,)[:100]})],}归档节点deffinalize_node(state:ResearchState)-dict:审批通过后归档。这是第一个可以安全做副作用的节点。archive_idsave_to_knowledge_base(topicstate[topic],draftstate[draft],sourcesstate[sources],)return{messages:[HumanMessage(contentf[finalize] 已归档id{archive_id})]}3.3 组装图与条件边defroute_after_review(state:ResearchState)-Literal[revise,approve,abort]:审批后的三岔路由通过 / 打回重写 / 熔断放弃。ifstate[approved]:returnapproveifstate[revision_count]MAX_REVISIONS:returnabort# 防死循环的关键returnrevisebuilderStateGraph(ResearchState)builder.add_node(plan,plan_node)builder.add_node(search_internal,search_internal)builder.add_node(search_web,search_web)builder.add_node(draft,draft_node)builder.add_node(review,review_node)builder.add_node(finalize,finalize_node)builder.add_edge(START,plan)# fan-outplan 之后两个检索节点并行执行同一个 super-stepbuilder.add_edge(plan,search_internal)builder.add_edge(plan,search_web)# fan-in两个检索都完成后才进 draft框架自动等待全部前置边builder.add_edge(search_internal,draft)builder.add_edge(search_web,draft)builder.add_edge(draft,review)# 条件边既能分支也能成环revise → draft 就是一条回边builder.add_conditional_edges(review,route_after_review,{revise:draft,approve:finalize,abort:END},)builder.add_edge(finalize,END)graphbuilder.compile(checkpointerInMemorySaver())这里有一个容易被忽略的语义fan-in 是自动的。你不需要写“等待两个检索节点都完成”的逻辑只要draft的两个上游节点都在LangGraph 就会等它们全部落定后才调度draft。这也正是sources必须带operator.add的原因 —— 两个节点在同一个 super-step 里各自返回增量框架需要 reducer 才知道怎么合并。3.4 第一次运行走到审批就停config{configurable:{thread_id:research-001},recursion_limit:50}resultgraph.invoke({topic:2025 年选型Postgres 还是 MongoDB 作为主库,revision_count:0,sources:[]},config,)print(result[__interrupt__])# [Interrupt(value{type: approval_request, draft: .**更多硬核网安与AI工具包请扫码获取完整源码**..,
一、引言单轮对话的天花板在哪里把 LLM 接进业务系统的第一版代码通常长这样拼一个 prompt调一次invoke把字符串返回给前端。这在 Demo 阶段完全够用但一旦进入生产问题会集中爆发状态无处安放。多轮对话的历史、检索到的资料、上一次工具调用的返回值只能靠调用方自己维护一个不断膨胀的messages列表。流程不可中断。凡是需要“人工审批后再继续”的场景发邮件、改数据库、下单单轮调用只能把整个上下文序列化出去等审批完了再重新拼一遍 prompt —— 中间的 LLM 调用被白白重跑。失败即重来。第三步的 API 超时了前两步的 token 已经烧掉你只能从头再来一遍。循环不可控。Agent 自我反思、工具重试这类天然带环的结构在 LCEL 那种有向无环图DAG里写不出来只能塞进一个 while 循环于是可观测性彻底归零。没有审计轨迹。出事之后你无法回答“当时模型看到了什么、在哪一步做了错误决策”。LangChain 的 LCEL 解决的是编排问题它本质上是一张静态 DAG。而真实的 Agent 带环的状态机 持久化 人机协同。LangGraph 正是为这个模型设计的把工作流显式建模成一张图节点是函数边是控制流共享状态由 reducer 驱动合并每一步执行后都落盘成 checkpoint。这篇文章会从原理讲到代码搭一个“调研 → 起草 → 人工审批 → 修订 → 归档”的可中断、可恢复智能体。1.1 先看一段“看起来能跑”的代码为了让后面的对比足够直观我们先把那个 Demo 版本完整写出来。它的典型形态是这样的fromlangchain_openaiimportChatOpenAI llmChatOpenAI(modelgpt-4o-mini)history[]# 调用方自己维护的全局状态defchat(user_input:str)-str:history.append({role:user,content:user_input})respllm.invoke(history)history.append({role:assistant,content:resp.content})returnresp.content这段代码的每一个“缺点”都不是风格问题而是结构问题history是进程内的一个 list进程一重启就没了多副本部署时更是各说各话一旦要加“先检索资料再回答”你就得在chat里塞if加了工具调用就要再塞一个while循环判断模型是否还要调工具想加“下单前需要人工确认”你只能把整个history序列化到数据库审批通过后再反序列化重建 —— 而重建之后模型为了走到同一个位置还得把之前所有的 token 重新烧一遍出问题时你翻遍日志也只能看到input和output中间发生过什么完全黑盒。把这五个问题抽象一下其实是三个能力缺失状态的持久化、控制流的显式化、执行过程的可观测。LangGraph 的三个核心抽象 —— State、Edge、Checkpointer —— 正好一一对应。1.2 LangGraph 到底在解决什么问题可以用一句话概括 LangGraph 的心智模型一张带环的有向图 一个可序列化的共享状态 一个每步落盘的检查点。把这句话和几个相邻概念做个对比定位会更清楚方案控制流状态管理中断/恢复循环支持适用场景手写 while 循环隐式藏在代码里自己维护自己实现支持但难控一次性脚本LCELRunnable 组合静态 DAG靠管道传递不支持不支持线性数据流水线AgentExecutor隐式 ReAct 循环靠 memory弱内置但不可定制通用工具型 AgentLangGraph显式图结构TypedDict reducercheckpoint 原生支持一等公民生产级多步骤工作流注意表格里 LCEL 那一行 —— 它并不是“旧方案”LCEL 依然是 LangChain 表达式组合的基础LangGraph 的节点内部照样可以用 LCEL 写链。二者的关系是嵌套而不是替代LCEL 负责“一段数据怎么流”LangGraph 负责“多个步骤怎么调度、在哪停、从哪续”。二、核心原理StateGraph 的三要素2.1 State 与 Reducer并发写入的合并规则LangGraph 的状态就是一个TypedDict。关键设计在于字段级别的 reducer当两个并行节点同时往同一个 key 写值时用哪个 reducer 决定合并结果。classState(TypedDict):messages:Annotated[list,add_messages]# 追加而非覆盖自动去重/合并消息sources:Annotated[list[str],operator.add]# 并行检索结果自动拼接draft:str# 无注解 后写覆盖先写没有给 list 字段加 reducer是新手最常踩的坑 —— 并行分支一跑就抛InvalidUpdateError。记住一条规则任何可能被多个节点写入的字段都必须显式声明 reducer。理解 reducer 的关键是理解“节点返回的是什么”。节点函数的返回值不是新状态而是状态增量partial update。框架拿到这个增量后对每一个 key 做如下判断有 reducer 的 key调用reducer(旧值, 新值)把结果写回没有 reducer 的 key直接用新值覆盖旧值增量里没出现的 key原封不动保留。所以下面两种写法语义完全不同# 写法 A覆盖。第二次执行后 messages 只剩 1 条defnode_a(state):return{messages:[new_msg]}# 写法 B追加。第二次执行后 messages 有 2 条依赖 add_messagesdefnode_b(state):return{messages:[new_msg]}# 前提是 State 里 messages 带了 add_messages写法 A 和 B 的代码长得一模一样差别完全隐藏在State的注解里。这正是 reducer 设计“隐蔽但强大”的地方也是团队协作时最容易出事故的地方 —— 建议在State定义上方统一写注释标注每个字段的合并语义。几个常用 reducer 一览字段类型推荐 reducer语义list[消息]add_messages按id去重合并支持RemoveMessage删除list[str]/list[dict]operator.add简单拼接int累加operator.add计数、求和bool标记位不加覆盖后写覆盖dict合并自定义函数通常做浅合并{**old, **new}下面是一个自定义 dict 合并 reducer 的例子在“配置逐步补全”的场景里非常有用defmerge_dict(old:dict,new:dict)-dict:浅合并新值覆盖同名 key未出现的 key 保留。return{**old,**new}classState(TypedDict):config:Annotated[dict,merge_dict]还有一个必须知道的细节返回未在 State 中声明的 key会被静默忽略。这不像 Python 字典会报KeyError它不会报错只会让你的“精心构造的字段”永远消失。调试时如果发现某个字段永远是初始值第一件事就是检查它有没有写进TypedDict。2.2 节点与边把控制流画出来节点是纯函数(state) - dict返回的 dict 是状态增量partial update由框架用 reducer 合并回全局状态。边分两种add_edge(a, b)确定性跳转add_conditional_edges(a, router, {x: b, y: c})router 函数读状态返回 key实现分支与循环把条件边指回上游节点即可成环。START和END是两个虚节点标明入口与出口。关于节点函数有三条容易被忽略的约定签名可以不接收全部参数。你可以只写def node(state: State)也可以用Literal[messages, draft]声明只关心某些 keyLangGraph 会把该 key 的最新值传进来。后者在状态很大时能减少无谓的数据传递。可以额外接收config。写成def node(state: State, config: RunnableConfig)就能拿到config[configurable][thread_id]之类的运行时信息做多租户路由、日志打点时必备。返回值必须是 dict 或 None。返回None表示“本节点不修改状态”合法但不推荐 —— 因为它会让图的语义难以阅读。条件边里的router函数写法也有讲究。它只读状态、不做副作用返回值必须是字典里声明过的 key否则运行时会报错。如果分支很多用Literal标注返回值类型能在 IDE 里获得自动补全fromtypingimportLiteraldefrouter(state:State)-Literal[search,answer,give_up]:ifstate[retry]3:returngive_upifstate.get(need_search):returnsearchreturnanswer关于“循环”再强调一遍LangGraph 里没有专门的while原语循环就是“条件边指回上游节点”。这也意味着你必须自己准备熔断条件比如重试计数否则图会一直转到触发recursion_limit抛异常为止。默认recursion_limit是 25 个 super-step可以在config里调整config{recursion_limit:50,configurable:{thread_id:t1}}2.3 Checkpointer可追踪、可恢复的全部秘密编译图时传入 checkpointerLangGraph 会在每个 super-step一次图执行批次结束后把完整状态序列化落盘主键是thread_id。这一个机制同时解决了四件事能力依赖的 API断点续跑thread_id checkpointer重启进程后继续人工介入interrupt()挂起Command(resume...)恢复状态回放/时间旅行get_state_history()列出所有历史 checkpoint故障重试从最后一个成功的 checkpoint 重新调度interrupt()在节点内部调用会抛出特殊信号让图暂停并把 payload 交给人。关键陷阱恢复时该节点会从头重跑所以interrupt()之前的副作用代码必须幂等。生产环境用PostgresSaver本地开发用SqliteSaver纯测试用InMemorySaver。这里把“super-step”这个概念再展开一下因为它直接决定了 checkpoint 的粒度。LangGraph 的执行是按批次推进的一个 super-step 里所有可以并行执行的节点同时跑比如plan之后的两个检索节点等这一批全部执行完状态统一合并、写一次 checkpoint然后才计算下一批要执行哪些节点。所以 checkpoint 的数量不等于节点数量而是等于“批次数”。在多分支并行的图里一次 checkpoint 可能对应好几个节点的产出这是正常的。interrupt()的行为同样值得推敲。它的完整生命周期是节点执行到interrupt(payload)LangGraph 抛出一个内部信号当前节点中断图暂停状态包括中断前的所有变更写入 checkpointpayload暴露给调用方调用方用Command(resumevalue)重新invokeLangGraph 从断点所在的 checkpoint 恢复重新执行该节点函数这一次interrupt()不再抛异常而是直接返回resume传入的值函数继续往下跑。第 4 步是重点“重跑”是真的从头跑一遍函数体。所以下面的写法是危险的defbad_node(state):send_email(state[draft])# ❌ 副作用在 interrupt 之前decisioninterrupt({draft:state[draft]})return{approved:decision[approved]}审批恢复后send_email会被再调用一次。正确做法是把副作用挪到interrupt()之后或者用幂等键去重。关于持久化后端选型建议如下实现包适用场景注意点InMemorySaver内置单元测试、一次性脚本进程结束即丢失SqliteSaverlanggraph-checkpoint-sqlite本地开发、单机部署需要check_same_threadFalsePostgresSaverlanggraph-checkpoint-postgres生产环境首次使用需调.setup()建表Postgres 的接法大致长这样fromlanggraph.checkpoint.postgresimportPostgresSaver DB_URIpostgresql://user:pwdlocalhost:5432/agent?sslmodedisablewithPostgresSaver.from_conn_string(DB_URI)assaver:saver.setup()# 建表只需执行一次graphbuilder.compile(checkpointersaver)setup()会创建checkpoints、checkpoint_blobs、checkpoint_writes等几张表。生产环境务必建索引并按时间分区/清理否则高并发下这张表会迅速膨胀到几十 GB。三、实战可中断、可恢复的调研智能体场景用户提一个技术选型问题Agent 规划检索词 → 检索 → 起草报告 →暂停等待人工审批→ 不通过则按意见修订。3.1 定义状态先想清楚“什么要持久化”状态设计是整个实现里最重要的一步。经验法则是凡是需要在中断后恢复时还能拿到的东西都必须进 State凡是进 State 的东西都必须可序列化。fromtypingimportAnnotated,Literalimportoperatorfromlanggraph.graphimportStateGraph,START,ENDfromlanggraph.graph.messageimportadd_messagesfromlanggraph.checkpoint.memoryimportInMemorySaverfromlanggraph.typesimportinterrupt,Commandfromlangchain_core.messagesimportHumanMessage,SystemMessage MAX_REVISIONS3# 熔断阈值防止无限修订循环classResearchState(TypedDict):# ---- 输入 ----topic:str# ---- 中间产物 ----queries:list[str]# plan 节点产出sources:Annotated[list[str],operator.add]# 两个检索节点并行写入 → 必须加 reducerdraft:str# 每轮修订覆盖# ---- 人机协同 ----approved:boolreview_notes:str# 人工填写的修改意见# ---- 熔断 ----revision_count:Annotated[int,operator.add]# 每次审批 1# ---- 审计 ----messages:Annotated[list,add_messages]# 全链路消息日志几个设计决策值得说明sources用operator.add因为两个检索节点会并行写入必须拼接而非覆盖revision_count也用operator.add每次进入 review 节点就 1这样即使节点重跑语义依然是“审批次数 1”注意由于 review 节点的重跑发生在 interrupt 之后实际只会 1 一次draft不加 reducer因为它是“覆盖式”字段每次修订就是最新版本messages用add_messages它会按消息id去重天然适配流式追加和中断恢复。3.2 定义节点每个函数只做一件事规划节点——把用户的问题拆成检索词defplan_node(state:ResearchState)-dict:把 topic 拆成 2~3 个检索词。纯函数、无副作用可安全重跑。prompt(f你是技术调研助手。请针对下面的问题生成 3 个互补的检索关键词f每行一个不要编号不要解释。\n\n问题{state[topic]})rawllm.invoke([HumanMessage(contentprompt)]).content queries[q.strip()forqinraw.splitlines()ifq.strip()][:3]return{queries:queries,# 用 add_messages 追加形成审计日志messages:[HumanMessage(contentf[plan] 生成的检索词{queries})],}检索节点——两个并行版本分别打内部知识库和公网defsearch_internal(state:ResearchState)-dict:查内部知识库。返回 sources 增量由 operator.add 自动拼接。hits[]forqinstate[queries]:hits.extend(vector_store.similarity_search(q,k2))return{sources:[f[内部]{h.page_content[:200]}forhinhits]}defsearch_web(state:ResearchState)-dict:查公网。失败时返回空增量不阻断主流程。try:results[f[公网]{r[title]}:{r[snippet]}forqinstate[queries]forrinweb_search(q,limit2)]exceptExceptionasexc:# 单点失败降级return{messages:[HumanMessage(contentf[search_web] 失败{exc})]}return{sources:results}起草节点——汇总资料写初稿defdraft_node(state:ResearchState)-dict:根据 sources 和人工意见生成/修订草稿。notesstate.get(review_notes)or首轮无人工意见context\n.join(state[sources])[:8000]# 截断防止超出上下文promptf你是技术分析师请撰写一份关于「{state[topic]}」的调研简报。 要求 1. 结构为结论 / 依据 / 风险 / 建议 2. 只使用下面提供的资料不要编造 3. 如果在修订必须逐条回应人工意见。 【参考资料】{context}【人工修改意见】{notes}draftllm.invoke([SystemMessage(contentprompt)]).contentreturn{draft:draft,messages:[HumanMessage(contentf[draft] 第{state[revision_count]1}稿完成)],}审批节点——整个方案的核心用interrupt()暂停defreview_node(state:ResearchState)-dict:把草稿交给人工审批。interrupt 会让图在此暂停并落盘。# ⚠️ interrupt 之前的代码在恢复时会被重跑这里必须保持无副作用decisioninterrupt({type:approval_request,topic:state[topic],draft:state[draft],instructions:用 Command(resume{approved: bool, notes: str}) 恢复,})return{approved:bool(decision.get(approved)),review_notes:decision.get(notes,),revision_count:1,# 累加到 state 里operator.addmessages:[HumanMessage(contentf[review] approved{decision.get(approved)}, fnotes{decision.get(notes,)[:100]})],}归档节点deffinalize_node(state:ResearchState)-dict:审批通过后归档。这是第一个可以安全做副作用的节点。archive_idsave_to_knowledge_base(topicstate[topic],draftstate[draft],sourcesstate[sources],)return{messages:[HumanMessage(contentf[finalize] 已归档id{archive_id})]}3.3 组装图与条件边defroute_after_review(state:ResearchState)-Literal[revise,approve,abort]:审批后的三岔路由通过 / 打回重写 / 熔断放弃。ifstate[approved]:returnapproveifstate[revision_count]MAX_REVISIONS:returnabort# 防死循环的关键returnrevisebuilderStateGraph(ResearchState)builder.add_node(plan,plan_node)builder.add_node(search_internal,search_internal)builder.add_node(search_web,search_web)builder.add_node(draft,draft_node)builder.add_node(review,review_node)builder.add_node(finalize,finalize_node)builder.add_edge(START,plan)# fan-outplan 之后两个检索节点并行执行同一个 super-stepbuilder.add_edge(plan,search_internal)builder.add_edge(plan,search_web)# fan-in两个检索都完成后才进 draft框架自动等待全部前置边builder.add_edge(search_internal,draft)builder.add_edge(search_web,draft)builder.add_edge(draft,review)# 条件边既能分支也能成环revise → draft 就是一条回边builder.add_conditional_edges(review,route_after_review,{revise:draft,approve:finalize,abort:END},)builder.add_edge(finalize,END)graphbuilder.compile(checkpointerInMemorySaver())这里有一个容易被忽略的语义fan-in 是自动的。你不需要写“等待两个检索节点都完成”的逻辑只要draft的两个上游节点都在LangGraph 就会等它们全部落定后才调度draft。这也正是sources必须带operator.add的原因 —— 两个节点在同一个 super-step 里各自返回增量框架需要 reducer 才知道怎么合并。3.4 第一次运行走到审批就停config{configurable:{thread_id:research-001},recursion_limit:50}resultgraph.invoke({topic:2025 年选型Postgres 还是 MongoDB 作为主库,revision_count:0,sources:[]},config,)print(result[__interrupt__])# [Interrupt(value{type: approval_request, draft: .**更多硬核网安与AI工具包请扫码获取完整源码**..,
