LangGraph循环边人机协同

可视化演示

展示LangGraph循环边(Cycle)实现的人机协同工作流：Agent→人工审批→执行→反馈→循环，含暂停恢复与状态持久化。

循环边原理

在媒体公司的制度里，“打回修改”不会让人慌乱，因为它是流程设计的一部分，而不是事故。编辑把初稿送到总编室，总编写上“结构太散，重点不突出”打回去，编辑修改后再次送审——这个闭环天天在发生。LangGraph 里的循环边（Cycle Edge），正是程序世界里的这条“打回”箭头。

大多数刚接触流程建模的开发者，习惯把工作流画成一条从上到下的直线：开始 → 起草 → 校对 → 发布。这种图叫有向无环图（DAG），它假设每个步骤只走一次。但真实业务几乎都带着“复查、返工、重试”的迭代属性，DAG 表达不了“再来一遍”。LangGraph 的 StateGraph 允许一条边的终点指向更早的节点甚至自身，于是整个图可以被反复执行，直到某个条件被满足。

在图中，条件边（Conditional Edge）负责决定下一跳：“如果状态里人工审批意见为通过，走向结束；否则回到起草节点。”每次循环都会把节点返回值合并进全局状态，下一次迭代能看到上一轮的产出物和批语。这就让 Agent 能够“根据反馈迭代产出”，和编辑照着批语改稿是同一件事。

所以“循环边”并非不受控的死循环。它像一条有护栏的返工通道：只要你把“出口条件”定义清楚——总编签字即放行——那这条通道就能让 Agent 多次修改，直到达到人类认定的质量线。对任何一个需要人工把关的生产流程，这条打回通道都是不可或缺的骨架。

人工审批节点

总编办公室是整张送审图里唯一一个“非 AI”的节点。没有哪个部门会允许 AI 替总编签字——审批这件事，必须由一个真实的人来做。在 LangGraph 中，用来承接人工决策的这个特殊节点，就叫人工审批节点（Human Approval Node）。它的职责有两件事：把待审批的材料呈现给外部的人，把人的意见接收回来写进状态。

技术上，人工审批节点通常是一个普通的 Python 函数，但函数内部调用了 interrupt()。调用 interrupt(payload) 时，图会停下并把 payload（比如草稿全文和审批问题）作为中断结果返回给调用方。调用方在外部拿到这个 __interrupt__ 后，把内容展示给人类用户；用户给出审批意见后，外部再调用 Command(resume=意见) 把意见投递回图中。此时 interrupt() 会带着这个返回值继续执行，节点把这个值存入状态字段。

这个设计把人从“流程实现”中解耦了出来。审批人不需要理解 StateGraph、不需要写代码，只看一眼材料点个按钮即可。而 Agent 开发者要做的工作，就是定义好状态里存意见的字段，并让审批节点读写它。审批意见写入状态后，下游的条件边才有依据决定是放行还是打回。

在真实代码里，审批节点可能只有几行：把 draft 放进 interrupt()，把返回值 action 存进 state["feedback"]。但就是这枚“人的手印”，让整个 Agent 流程从“无人驾驶”变成了“有主心骨”。总编办公室的意义不在于它有多复杂，而在于签字权交给了人。

暂停等待输入

还记得稿件被送到总编办公室之后的样子吗？它安静地躺在总编桌上，办公流程不会自己往下推进。没有人会以为流程出了故障——大家知道，审批中的稿件就是该这样“悬停”着，等总编回来处理。LangGraph 的暂停机制把这种“悬停”精确复刻进了程序里。

当图执行到调用 interrupt() 的那一行，执行器会立刻暂停，不再执行后续任何节点。这个暂停不是抛异常、不是崩溃，而是 human-in-the-loop 模式刻意设计的“等待点”。在外部调用端，这次 invoke 或 stream 会正常返回，返回结果里带着 __interrupt__ 字段，里面包含 interrupt() 抛出的 payload。开发者的代码拿到这个字段，就知道：图走到审批点了，该请人拍板了。

值得注意的是，暂停期间整个会话仍然存在于后台状态中，并没有被销毁。LangGraph 的检查点机制在每次节点执行后都会保存一份快照，暂停时的快照就是“稿件躺在桌上”那一刻的完整桌面。即使进程重启、换一台服务器，只要快照还在，这张审批单就不会丢。这也是它和普通 while True: input() 之间最大的区别。

从这个角度看，“暂停等待输入”其实是一种非常优雅的异步协作模式：AI 先干它擅长的前期工作，到了必须由人拍板的关口就主动停下来，既不擅作主张，也不催促打扰。等总编批完了，流程再把后续的活儿接起来——这正是人与人协作中最自然的节奏。

中断恢复机制

总编出差一周后回到办公室，看到桌上压着的稿件，提笔签字。流程继续往下走时，没有人会要求编辑把稿子从头再写一遍——总编在意的只是签完字之后往后的环节。LangGraph 的中断恢复机制，刻意设计成“从断点续跑”，而不是“从头重跑”。

恢复的关键在于一个被称为 Checkpointer（检查点器）的组件。它会在图的每次节点执行后，把当时的完整状态存到内存或持久化存储中。当外部用 Command(resume=审批意见) 并携带同一个 thread_id（线程标识）发起恢复时，LangGraph 会先从检查点器里读出该线程最后保存的状态，再从这个状态开始继续执行剩余节点。先前已经完成的节点不会重跑，这即是“断点续跑”。

thread_id 在这里相当于纸质流程单上的流水号。同一份稿件对应同一个流水号，恢复操作拿着流水号就能找到正确的“桌面状态”。如果换成一个新的 thread_id，LangGraph 就认为你在发起一个全新的流程，跟原来的审批单没有任何关系。所以在工程实现中，维护好 thread_id 与业务单据的对应关系，是整个审批系统稳定运转的前提。

正因为有了检查点和 thread_id，“打回重改”才能多次成功落地：第一次打回，Agent 修改后再送审；第二次打回，Agent 再修改；直到某次审批通过走到终点。每一次循环都是基于同一个线程的持续演进，而不是推倒重来。总编不必担心签了字流程还要重跑，开发者也不必为“恢复”写一堆复杂的幂等逻辑。

人机协同最佳实践

再健全的制度也要防呆。一家公司如果审批流程没有时限、没有记录、任何人都能冒充总编签字，那这套制度很快会形同虚设。LangGraph 的人机协同也一样：仅把循环边和中断节点连起来，还远远不够，工程上需要再套上几层“制度护栏”。

第一层是超时兜底。人工审批可能迟迟不来，若业务要求两小时内必须响应，可在外部编排层加一个定时任务，超时后自动按策略处理（如自动打回、升级通知或默认通过）。第二层是决策留痕。在状态里增加 reviewer 和 reviewed_at 字段，记录审批人身份与时间，让每次人工干预都可追溯。第三层是输入校验。从 interrupt() 拿到的 feedback 是外部传入的自由文本，写入状态前应做合法性校验（如白名单、长度限制），防止下游 Agent 被恶意或异常输入带偏。

这三条“护栏”对应的其实就是工程中常说的一致性、可观测性和健壮性。超时兜底避免流程无限挂起，保证系统在真实业务里“不至于卡死”；决策留痕让每一条被中断修改过的路径都有账可查，方便排查问题和事后审计；输入校验则保证从人这个环节流入的数据是可信的。三者缺了任何一条，循环边人机协同都会在生产环境里埋下隐患。

回到我们的比喻：给媒体公司加制度，不是为了给总编添麻烦，而是为了保障流程在真实现场不失控、不扯皮、不背锅。LangGraph 的工程实践也是同样的道理——循环边把人带进了 AI 工作流，而超时、留痕与校验保证了这个人机协作循环能长期健康地转下去。

应用场景案例

掌握了一套机制之后，你会发现它的适用面非常广。不只是媒体公司的稿件审批，任何“AI 先行产出、人来拍板把关”的流程，都可以照搬这套循环边人机协同模式。它最迷人的地方在于：业务差异只体现在节点内部怎么处理文档，而“打回—重写—再送审”的骨架几乎原封不动。

第一个典型场景是审批流程。在政务和企业的 OA 系统中，用 Agent 自动起草通知、报销单据、合同初稿，紧接着接入一个审批节点让负责人审批；审批不通过就自动打回给 Agent 改，通过后推送档案系统存档。第二个典型场景是内容审核。AIGC 批量生成营销文案或播报脚本时，通过人工终审节点把“生成—发布”的链路切断，保证不合格内容不会未经人眼就流向公众。第三个场景是代码审查。开发助手 Agent 生成提测代码后，先进入评审节点由资深工程师审查，不通过则带着评审意见打回去继续修改，通过才进入合并流程。

这些场景共用同一个模式：一个 Agent 生产节点、一个 interrupt() 审批节点、一条条件边路由“通过/打回”。从产品视角看，这就是“AI 干粗活、人做决策”的人机分工；从技术视角看，它就是一张带循环和断点的状态图。无论业务领域是公文、内容还是代码，图的拓扑结构都能复用，真正需要定制的，只是每个节点内部的领域逻辑。

你会发现，这套公式几乎刻画了所有“审批类 Agent”的共同形态。媒体总编的签字、OA 里的一键驳回、内容平台的合规终审，本质上都是同一个 pass(h(x)) 函数在不同领域的实例。理解了循环边和人工审批节点的组合，你就握住了一张能把“人机协同”落到任何业务里的通用图纸。

AI 应用场景

Python 演示

# -*- coding: utf-8 -*-"""LangGraph 循环边 + 人工审批节点 的完整可运行示例流程: agent 起草 -> review 人工审批 -> 打回收回 / 通过归档"""from typing import TypedDictfrom langgraph.graph import StateGraph, START, ENDfrom langgraph.checkpoint.memory import InMemorySaverfrom langgraph.types import Command, interrupt  class AgentState(TypedDict):    task: str          # 任务说明    draft: str         # 当前草稿    feedback: str      # 人工审批意见  def agent_node(state: AgentState) -> dict:    """Agent 节点: 第一次生成初稿, 之后按意见修订"""    if not state["draft"]:        draft = "[初稿] 关于「%s」的新闻稿已拟好。" % state["task"]    else:        draft = "[二稿] 已按意见「%s」完成修订。" % state["feedback"]    return {"draft": draft, "feedback": ""}  def review_node(state: AgentState) -> dict:    """人工审批节点: 展示草稿并暂停等待人工输入"""    action = interrupt({"draft": state["draft"], "question": "请审批: 通过 / 打回"})    return {"feedback": action}  def route_after_review(state: AgentState) -> str:    """条件边: 通过则走向 END, 否则循环回 agent"""    if state["feedback"] == "通过":        return "end"    return "agent"  # 构建状态图: 含一条「审批打回」的循环边builder = StateGraph(AgentState)builder.add_node("agent", agent_node)builder.add_node("review", review_node)builder.add_edge(START, "agent")builder.add_edge("agent", "review")builder.add_conditional_edges(    "review", route_after_review, {"agent": "agent", "end": END})app = builder.compile(checkpointer=InMemorySaver()) thread = {"configurable": {"thread_id": "review-demo-001"}} if __name__ == "__main__":    # 第一轮: 起草后进入审批节点, 被 interrupt 暂停    snap = app.invoke({"task": "新品发布会", "draft": "", "feedback": ""}, thread)    print("等待审批:", snap["__interrupt__"][0].value)     # 人工打回: 触发循环边, 流程回到 agent 节点重写    snap = app.invoke(Command(resume="打回: 缺少时间地点"), thread)    print("等待二审:", snap["__interrupt__"][0].value)     # 人工通过: 沿条件边走向 END, 流程结束    result = app.invoke(Command(resume="通过"), thread)    print("最终定稿:", result["draft"])

1. builder.add_conditional_edges("review", route_after_review, {"agent": "agent", "end": END})给审批节点接上分叉路：意见非“通过”就送回 agent，等于在图上画出一条「打回」循环边。
2. action = interrupt({"draft": state["draft"], "question": "请审批: 通过 / 打回"})把草稿作为中断载荷抛出，外部拿到 __interrupt__ 后能向用户展示审批材料。
3. snap["__interrupt__"][0].value读取中断载荷的标准入口，代码据此判断当前处在哪个待审点上。
4. app.invoke(Command(resume="打回: 缺少时间地点"), thread)用 Command(resume=...) 把人工意见注入图，流程从暂停节点恢复。
5. builder.compile(checkpointer=InMemorySaver())挂载检查点器，让中断恢复拥有状态快照，是恢复机制的必要底座。
6. thread = {"configurable": {"thread_id": "review-demo-001"}}线程 ID 相当于流程单编号，同一线程内多次恢复共享同一份快照。

常见误区

面试题