企业官网建设流程全解析

• 人工智能
• AI Agent
• Agent 框架
• 工具调用
• Agent 工作流
• RAG
• MCP Clients
• 强化学习

导读

GraphInterrupt是 openJiuwen agent-core-java 中 Pregel 图引擎（com.openjiuwen.core.graph.pregel）用于表达"图执行被外部事件打断"的专用异常类型。当智能体工作流在运行过程中需要暂停等待用户输入、人工审批或外部反馈时，图引擎不会把这种暂停当作普通错误处理，而是通过GraphInterrupt携带中断载荷沿节点执行、任务池、超步循环逐层向上传递，最终由顶层调用方捕获并据此恢复执行。读完本文，你将掌握GraphInterrupt的完整 API、它与Interrupt载荷的关系、中断从产生到抛出的完整调用链，以及如何区分"可恢复的中断"与"真正的错误"。

类定位：Pregel 图中的"暂停信号"

GraphInterrupt定义在 core/src/main/java/com/openjiuwen/core/graph/pregel/GraphInterrupt.java，其源码注释明确说明它是对 Python 版openjiuwen/core/graph/pregel/base.py中同名类的镜像移植，类声明如下：

public class GraphInterrupt extends Exception

从源码结构看，Pregel 图引擎中异常大致分两类：

• 普通错误（Error）：节点执行失败、组件调用超时等，会触发重试、错误恢复与检查点保存，属于需要"修复"的失败；
• 图中断（GraphInterrupt）：图执行被有意打断，等待外部输入后可以继续，属于"可恢复的暂停"。

这一区别贯穿整条执行链：TaskExecutorPool会先取消剩余任务并抛出普通错误，而对中断则进行收集、合并后单独抛出；Vertex执行组件时遇到中断会跳过重试逻辑原样上抛，避免把"等待用户"误判为"执行失败"。

完整 API：字段、构造方法与访问器

GraphInterrupt本身 API 非常精简，官方文档将其描述为"图执行被中断时抛出的异常"，包含一个字段、两个构造方法和一个访问器：

需要说明的是，文档表格中value字段标称类型为Interrupt，而实际源码（GraphInterrupt.java）中该字段声明为private final Object value，并通过normalizeWorkflowInterruptPayload(...)做了一次归一化处理。也就是说，载荷的运行时类型可能是Interrupt、任意业务对象，或这些对象的List集合（例如多个节点同时发出中断时会被合并为列表）。构造时value会同时用于异常消息：

public GraphInterrupt(Object value) { super(String.valueOf(normalizeWorkflowInterruptPayload(value))); this.value = normalizeWorkflowInterruptPayload(value); }

中断载荷：Interrupt是什么

与GraphInterrupt配套的载荷类型是 core/src/main/java/com/openjiuwen/core/graph/pregel/Interrupt.java：

public class Interrupt implements Serializable { private final Object value; public Interrupt(Object value) { this.value = value; } public Object getValue() { return value; } }

其源码注释点明了用途："Interrupt value captured when graph execution pauses for user input."（图执行因等待用户输入而暂停时捕获的中断值）。Interrupt实现了Serializable，因此可以被序列化写入检查点（Checkpointer），这正是"中断后恢复执行"的底层前提——暂停状态可以被持久化，随后从检查点重建。

典型用法是：节点函数返回new GraphInterrupt(new Interrupt("需要用户确认的内容"))，将需要呈现给用户或后续恢复时使用的数据包装进Interrupt。

中断的产生：节点与组件如何发出信号

中断主要有两种产生路径：

1. 节点函数直接返回

在 core/src/main/java/com/openjiuwen/core/graph/pregel/NodeTask.java 的run()中：

Object result = node.getFunc().apply(buildInvocation()); if (result instanceof GraphInterrupt graphInterrupt) { return graphInterrupt; // 作为节点返回值直接上抛 }

节点的函数式结果若是GraphInterrupt实例，则不再执行路由器（Router）分发，直接作为任务结果返回。

2. 组件执行中抛出

在 core/src/main/java/com/openjiuwen/core/graph/Vertex.java 中，执行节点能力（ability）时：

} catch (GraphInterrupt interrupt) { LOGGER.info("Interrupt to call node [{}] ability [{}]", nodeId, ability.name()); throw interrupt; // 原样上抛，不进入错误恢复 } catch (BaseError error) { ... } catch (Exception exception) { GraphInterrupt interrupt = findGraphInterrupt(exception); if (interrupt != null) { throw interrupt; } ... }

Vertex对中断采取"识别即放行"策略：findGraphInterrupt（Vertex.java）会沿着Throwable的 cause 链逐层查找GraphInterrupt，即便中断被包装在CompletionException/ExecutionException等异步异常中也能被还原。同时，在带重试与超时的执行入口（Vertex.java）中，GraphInterrupt会被优先于重试逻辑捕获并立即上抛，避免"等待用户输入"被误当成失败而重复执行节点。

中断的收集与合并：TaskExecutorPool 的语义

Pregel 图引擎每个超步（super-step）会并行提交多个节点任务，中断可能同时来自多个节点。这些并发中断由 core/src/main/java/com/openjiuwen/core/graph/pregel/TaskExecutorPool.java 的waitAll()统一处理：

} else if (outcome.result() instanceof GraphInterrupt graphInterrupt) { commitFailure(node, graphInterrupt); if (interrupt == null) { interrupt = graphInterrupt; } else { interrupt = mergeInterrupts(interrupt, graphInterrupt); } }

关键语义：

• 中断不取消兄弟节点：waitAll()只会在出现"非中断错误"（firstError != null）时调用cancelRemaining()取消剩余任务；若只是中断，则等待所有已提交任务完成，已成功节点的消息照常收集；
• 多中断合并：多个节点同时抛出GraphInterrupt时，mergeInterrupts会把各自载荷展开并合并为List，构造一个新的GraphInterrupt（TaskExecutorPool.java）；
• 状态标记：发生中断的节点会被记入failed映射，状态为PregelConstants.TASK_STATUS_INTERRUPT（值为"__interrupt__"），与普通错误的TASK_STATUS_ERROR（"__error__"）区分开。

中断的传播：PregelLoop 与 Pregel.run 的收口

PregelLoop.runStep()（core/src/main/java/com/openjiuwen/core/graph/pregel/PregelLoop.java）会调用publicInterruptValue对中断做"降级"处理：若合并后的中断列表只有一个元素且不含Interrupt实例，则还原为单值GraphInterrupt（PregelLoop.java）。中断同样会触发saveStateOnError，将当前通道值、待处理消息、失败节点与节点版本写入检查点，为后续恢复保留现场。

顶层入口 core/src/main/java/com/openjiuwen/core/graph/pregel/Pregel.java 捕获中断后：

} catch (GraphInterrupt interrupt) { triggerLoopEvent(WorkflowEvents.LOOP_FINISHED, innerConfig, loop.getStep(), true); Loggers.GRAPH.info("Pregel graph engine execution interrupted"); if (isTopLevel) { Map<String, Object> result = new LinkedHashMap<>(); result.put(PregelConstants.TASK_STATUS_INTERRUPT, interrupt.getValue()); return result; } throw interrupt; }

• 对顶层图：中断不再作为异常向外抛，而是被转换为返回结果，键为"__interrupt__"，值即中断载荷；
• 对子图（存在parentNs）：中断继续向上抛给父图处理。

至此，一条完整的"暂停信号"闭环形成：节点抛出 → 任务池合并 → 超步循环传播 → 顶层返回__interrupt__结果，等待外部输入后从检查点恢复。

载荷归一化：与工作流输出 Schema 的适配

GraphInterrupt构造时执行的normalizeWorkflowInterruptPayload（GraphInterrupt.java）承担了与 Session 交互体系对接的职责：

• 当载荷为List<Interrupt>且其中某个Interrupt的值是工作流的OutputSchema时，会将其 payload 中原始的InteractionOutput包装为ComparableInteractionOutput；
• ComparableInteractionOutput（GraphInterrupt.java）基于id、value、metadata三要素重写了equals/hashCode/toString，使交互输出具备可比较性，便于在中断合并、去重与恢复比对时使用。

这一细节表明GraphInterrupt不仅服务于纯图执行层，也与com.openjiuwen.core.session.interaction.InteractionOutput的会话交互体系打通，用于承载"等待用户输入"这类真实交互场景。

测试验证：中断行为有据可查

仓库中的测试用例直接印证了上述行为：

1. 载荷携带（PregelTest.java）

@Test @DisplayName("GraphInterrupt carries Interrupt value") void testGraphInterrupt() { Interrupt interrupt = new Interrupt("test_value"); GraphInterrupt gi = new GraphInterrupt(interrupt); Object value = gi.getValue(); assertInstanceOf(Interrupt.class, value); assertEquals("test_value", ((Interrupt) value).getValue()); }

2. 中断收集与兄弟节点消息保留（TaskExecutorPoolTest.java）

@Test void poolInterruptRaisesAndKeepsCompletedSiblingMessages() { ... pool.submit(new PregelNode("B", invocation -> { sleep(100L); return new GraphInterrupt(new Interrupt("B_Interrupt")); }, ...), 1); ... GraphInterrupt interrupt = assertThrows(GraphInterrupt.class, pool::waitAll); assertInstanceOf(Interrupt.class, interrupt.getValue()); assertEquals(PregelConstants.TASK_STATUS_INTERRUPT, pool.getFailed().get("B").getStatus()); // 已成功节点 A、C 的消息被保留，且 A、C 未进入 failed assertEquals(Set.of("A", "C"), pool.getSucceedMessages().stream().map(Message::getSender).collect(...)); }

该测试验证了核心语义：节点 B 抛出中断时，并行执行且已完成的节点 A、C 的消息不会被丢弃，中断节点被标记为__interrupt__而不是__error__。同文件中waitAllRecordsInterruptAndRaisesIt（TaskExecutorPoolTest.java）则验证了"任务池记录中断并重新抛出"的行为。

此外，GraphInterrupt还被WorkflowInteractionTest、InteractionTest、WorkflowTracerPythonParityTest等测试使用，说明中断机制与交互（Interaction）、追踪（Tracer）体系深度耦合。

实战建议：正确处理图中断

在实际使用 Pregel 图或工作流时，可遵循以下原则：

1. 用中断表达"等待"，用异常表达"失败"：需要暂停等待用户输入、人工审批、外部确认时，返回new GraphInterrupt(new Interrupt(payload))，不要抛普通RuntimeException，否则会触发重试与错误恢复逻辑；
2. 捕获位置区分层级：顶层调用方捕获GraphInterrupt后读取getValue()获取暂停载荷，将载荷呈现给用户并保存现场；子图/嵌套图场景下中断会继续上抛，交由最外层统一处理；
3. 结合检查点恢复：由于Interrupt实现了Serializable且PregelLoop在中断时会调用saveStateOnError落盘现场状态，可配合 Checkpointer 机制在用户返回输入后恢复执行（相关机制可参考 高阶用法中的 Checkpointer 文档 与 Session 中断恢复文档）；
4. 善用合并语义：并行分支同时请求输入时，多个GraphInterrupt会被合并为List载荷，可在恢复时统一处理所有待决输入。

小结

GraphInterrupt虽然只是 Pregel 包中一个 API 精简的异常类，但它是整个图引擎"可中断、可恢复"执行模型的关键枢纽：向上对接会话交互（InteractionOutput）承载用户输入，向下衔接任务池的并发收集与合并，并全程与普通错误隔离。理解它的语义——"暂停而非失败"，是正确编排需要人工参与的工作流的基础。

• 人工智能
• AI Agent
• Agent 框架
• 工具调用
• Agent 工作流
• RAG
• MCP Clients
• 强化学习