人工智能AI AgentAgent 框架工具调用Agent 工作流RAGMCP Clients强化学习【免费下载链接】agent-core-javaopenJiuwen agent core Java是 openJiuwen Core的 Java版本提供AI Agent开发、运行、调优与演进相关的全套SDK能力。项目地址https://gitcode.com/openJiuwen/agent-core-java点击查看免费下载com.openjiuwen.core.graph.stream_actor是 openJiuwen agent-core-java 中支撑工作流/图节点间流式数据传递的核心模块它负责把具备STREAM/TRANSFORM能力producer的节点输出逐帧转发给具备COLLECT/TRANSFORM能力consumer的节点并提供基于 schema 的阻塞迭代器式消费接口。读完本文你将掌握ActorManager、StreamActor、StreamProcessor等核心类型的职责划分、produce → send → processor → generator → consume的完整消息调用链、END 帧收敛协议以及STREAM_INPUT_GEN_TIMEOUT等关键配置的底层行为可直接用于理解或扩展自己的工作流流式组件。一、模块定位图节点之间的流式消息协调层在 openJiuwen 的 Workflow / Graph 执行模型中节点通过**组件能力ComponentAbility**声明其输入输出形态。定义位于 ComponentAbility.java能力名称语义INVOKEinvokebatch in, batch out批量进、批量出STREAMstreambatch in, stream out批量进、流式出COLLECTcollectstream in, batch out流式进、批量出TRANSFORMtransformstream in, stream out流式进、流式出graph.stream_actor包正是为「流式进」的节点COLLECT/TRANSFORM与「流式出」的节点STREAM/TRANSFORM之间的帧级消息搬运而设计。它统一处理了 actor 生命周期、payload 协议、consumer 注册、schema 驱动的迭代器生成与输入转换共提供 7 个核心类型类型职责ActorManager根据 producer/consumer 边与组件能力构建StreamActor统一负责生产、消费与关闭StreamActor管理单个消费节点的流式调用生命周期为COLLECT/TRANSFORM能力维护处理器StreamConsumer由图节点实现的流式消费接口定义启动、可处理判断与完成状态StreamGraph保存节点 ID 到StreamConsumer的注册关系StreamPayloadproducer 与 consumer 之间传递的消息载体包含原始消息与来源能力StreamProcessor将消息路由到 schema 叶子路径对应的阻塞迭代器并负责 END 帧收敛StreamTransform提供自定义 transformer 与 schema 两种输入转换方式其中ActorManager、StreamActor、StreamConsumer、StreamGraph、StreamPayload、StreamProcessor、StreamTransform的实现分别位于 core/src/main/java/com/openjiuwen/core/graph/stream_actor/源码注释明确标注其镜像自 Python 版 openjiuwen 的graph/stream_actor/manager.py与base.py。二、ActorManager流式协调的中央调度器ActorManager.java 是模块入口负责从工作流规格中解析流式边、构建 actor、并对外提供统一的生产与消费 API。2.1 关键字段字段类型说明streamEdgesMapString, ListStringproducer 节点到下游 consumer 节点列表的映射来源于workflowSpec.getStreamEdges()consumerDictMapString, ListString由buildReverseGraph(...)反向展开得到的 consumer → producer 关系streamsMapString, StreamActor已注册 consumer 的StreamActor映射LinkedHashMapstreamsTransformStreamTransform提供 schema / 自定义函数两种流式输入转换能力subGraphboolean当前 manager 是否服务于子图subWorkflowStreamAsyncStreamQueue子图模式下暴露给子工作流的共享输出队列容量10 * 1024activeProducerIdsMapString, SetComponentAbility记录本次运行中实际产生过帧的 producer 及其能力用于「未激活源清理」2.2 构造反向展开边 能力过滤构造器ActorManager(WorkflowSpec workflowSpec, StreamGraph graph, boolean subGraph, ActorManagerSession session)的核心逻辑分三步反向建图调用buildReverseGraph把streamEdgesproducer → consumers翻转成consumerDictconsumer → producers这正是原文档说明中「把streamEdges反向展开为 consumer 到 producer 的关系」的实现。能力筛选遍历每个 consumer 的 producer把具备STREAM/TRANSFORM能力的 producer 以producerId-ABILITY如producer-stream的形式收进 source 集合随后调用streamAbilitiesForConsumer仅为具备COLLECT/TRANSFORM能力的 consumer 创建StreamActorINVOKE、STREAM节点不会成为 actor。来源分组优先读取workflowSpec.getStreamSourceGroups().get(consumerId)的分组配置未配置时按字典序把每个 source 拆成独立单元素组。分组语义在收敛逻辑中起关键作用见第四节。从源码还可以看到两个细节producer 的 source 键统一形如producerId - ability.getAbilityName()sourceKey方法后续StreamProcessor解析来源时通过lastIndexOf(-)切分并反向匹配ComponentAbility枚举超时从session.config().getEnv(SessionConstants.STREAM_INPUT_GEN_TIMEOUT_KEY)读取解析失败时回退到默认值1.0d源码常量DEFAULT_STREAM_GENERATOR_TIMEOUT_SECONDS。2.3 对外 API方法行为subWorkflowStream()返回子图输出队列当subGraph为false时抛出GRAPH_STREAM_ACTOR_EXECUTION_ERROR提示 only sub graph has sub_workflow_streamstreamTransform()返回内部复用的StreamTransform实例produce(producerId, messageContent, ability, firstFrame)将消息包装成{producerId: value}单键 Map 并分发到全部下游 consumer没有 consumer 时仅记录 warning 日志 Discard chunk send from [...] to none consumerendMessage(producerId, ability)发送END_ producerId形式的结束帧通知下游处理器收敛consume(consumerId, ability, schema, streamCallback)为指定 consumer 生成与schema结构一致的迭代器映射找不到StreamActor时返回空映射markProducerDone(producerId)/shouldSanitizeStreamSource(...)维护工作流状态中的finished_stream_nodes并在存在「同组替代源已激活」时清理未激活源的队列shutdown()逐个关闭全部StreamActor用于图执行结束或异常中断后的清理produce的实现值得注意它在分发前调用activeProduceAbility记录该 producer 的能力已激活若某个 producer 没有配置任何下游 consumer消息会被直接丢弃并打 warning——这是「无消费方时不阻塞 producer」的设计取舍。三、StreamActor单个消费节点的流式调用生命周期StreamActor.java 对应一个流式消费节点。它为节点上每个可消费能力COLLECT/TRANSFORM维护一个独立的StreamProcessor并在收到首帧时负责拉起vertex.streamCall(...)虚拟线程与全部处理器线程。3.1 关键字段字段类型说明processorsMapComponentAbility, StreamProcessor每个可消费能力对应一个StreamProcessorHashMaptaskFuture?当前vertex.streamCall(...)所在的线程池任务taskCompletion/taskErrorCompletableFutureVoid主 stream 调用的完成 / 异常通知vertexStreamConsumer实际执行流式消费逻辑的图节点对象nodeIdString当前 actor 对应的节点 IDrunningTasksListRunningTask已启动的处理器任务及其完成句柄producerLocksStripedKeyLocks按 producer 分片的锁保证同一 producer 的帧按序送达3.2 生命周期核心方法方法行为send(message, sourceAbility, firstFrame, producerId)发送一帧消息当首帧满足启动条件时会先启动vertex.streamCall(...)与全部处理器任务再把StreamPayload广播给每个StreamProcessorgenerator(ability, schema, streamCallback)为指定能力返回与schema对齐的阻塞迭代器映射该能力不存在时返回空映射awaitCompletion()等待主 stream 调用与全部处理器任务完成单个等待窗口为 5000msSHUTDOWN_TIMEOUT_MSshutdown()取消主任务、异常 future 与全部处理器任务等待清理完成后重置内部状态send内部的启动判定源码sendOnProducer体现了生命周期细节先检查vertex.shouldHandleMessage()为false时直接丢弃该帧在startGuard同步块内检查主任务是否已结束或已异常若taskError已完成且异常则丢弃后续帧若处理器任务仍在运行则继续向处理器分发支持多批次的连续流式输入仅当「是首帧」且vertex.isDone()为true时才通过beginStreamCall()启动主 stream 调用——CountDownLatch等待窗口 30 秒START_LATCH_TIMEOUT_SECONDS随后用startProcessorTasks()启动全部StreamProcessor的run循环。所有帧处理与处理器线程都提交到有界模块线程池STREAM_EXECUTOROpenJiuwenExecutors.newBoundedModulePool(stream-actor, false)且runStreamTask会把Error、BaseError、CompletionException等统一写入对应CompletableFuture以驱动异常通知。四、StreamProcessorschema 驱动的队列路由与 END 帧收敛StreamProcessor.java 是数据面核心它消费StreamActor广播的 payload 队列把每条消息按 schema 引用路径路由到对应的阻塞迭代器队列并在所有来源都送达结束帧后收敛退出。4.1 关键字段字段类型说明END_SENTINEL/TIMEOUT_SENTINELObject流结束哨兵 / 超时哨兵源码中实际使用SessionUtils.EndFrame与静态TIMEOUT_SENTINELnodeIdString当前 processor 归属的 consumer 节点 IDqueueBlockingQueueStreamPayload接收来自StreamActor的原始 payload 队列LinkedBlockingQueueprocessorQueuesMapString, ListBlockingQueueObjectschema 引用路径含$的引用串到下游阻塞队列列表的映射ConcurrentHashMapCopyOnWriteArrayListsourceGroups/sourceIdsListSetString/SetString来源分组与来源 ID 集合来源键格式为producerId-ABILITYhasTimeout/timeoutMillisboolean/long是否启用生成器取流超时streamGeneratorTimeoutSeconds 0时启用否则无限等待超时配置为0或负值即禁用4.2 主循环 run路由与收敛run(ComponentAbility ability)的 while 主循环逻辑源码第 68-110 行取帧pollPayload()以BLOCKING_QUEUE_MS为窗口轮询内部队列识别结束帧isEndMessage判定消息是否为单键 Map 且值是以END_开头的字符串若是则把producerId-ABILITY记入handledSources并关闭该来源对应的迭代器队列向队列投放SessionUtils.EndFrame路由普通帧routeMessageValue遍历processorQueues通过SessionUtils.getValueByNestedPath从消息中按「原始键路径」提取值写入所有对应引用路径的阻塞队列收敛判定allSourceGroupsFinished检查每个 source group 是否至少有一个来源已结束——只要每个组内有一个来源结束即视为该组完成对应测试sourceGroupsFinishWhenAnySourceInEachGroupEnds替代源清理closeInactiveGroupSources在某个组内已有活跃源产生帧时提前关闭同组的其他不活跃来源队列避免消费者空等对应测试activeSourceInAlternativeGroupClosesInactiveSourceQueues。4.3 generator把 schema 变成阻塞迭代器generator(MapString, Object schema, ConsumerObject streamCallback)的核心行为对空 schema 返回空 Map通过DictUtils.extractLeafNodes提取 schema 全部叶子路径叶子的值若为包含$的引用串如${left.answer}则该叶子被替换为阻塞IteratorcreateIterator否则保留静态值阻塞迭代器的hasNext()会从专属BlockingQueue取消息遇到EndFrame返回false流正常结束遇到TIMEOUT_SENTINEL抛出GraphError状态码STREAM_PROCESSOR_QUEUE_TIMEOUT遇到超时针对未声明来源的路径useTimeout hasTimeout !pathHasDeclaredSource(...)也返回falsenext()每次弹出值后调用streamCallback.accept(Map.of(keyPath, value))这就是Vertex.preStream中 trace 流式输入回调的挂载点。由此consumer 节点拿到的是一个结构与 schema 完全一致的嵌套 Map其中流式引用位置是可阻塞等待的迭代器静态叶子保持不变。五、StreamConsumer 与 StreamGraph消费侧注册表5.1 StreamConsumer 接口StreamConsumer.java 由图节点实现定义三个方法方法说明streamCall(CountDownLatch latch, ConsumerException errorCallback)启动当前节点的流式消费逻辑实现方需在准备完成后释放latch通过errorCallback上报异常shouldHandleMessage()当前节点是否应接收流式消息通常由组件能力状态决定isDone()当前节点是否已完成一次执行周期false时会阻止新的首帧启动 stream call5.2 StreamGraph 注册表StreamGraph.java 保存节点 ID 到StreamConsumer的注册关系streamNodesMapString, StreamConsumerLinkedHashMap维护插入顺序addStreamConsumer(consumer, nodeId)同一节点仅首次注册生效putIfAbsent语义getNode(nodeId)按节点 ID 取 consumer不存在返回null。该注册表在ActorManager构造时被查询graph.getNode(consumerId)用于把消费者节点与其流式实现绑定。六、StreamPayload 与 StreamTransform消息协议与输入转换6.1 StreamPayload 消息载体StreamPayload.java 是 producer 与 consumer 之间传递的最小消息单元字段类型说明messageObject原始消息内容通常是单键 Map键为 producerIdsourceAbilityComponentAbility产生该消息的能力类型如STREAM或TRANSFORM构造器StreamPayload(Object message, ComponentAbility sourceAbility)访问器getMessage()/getSourceAbility()。注意StreamProcessor对消息合法性有强校验——必须是单键 Map否则抛出IllegalArgumentException(message is invalid)见singleMessageEntry。6.2 StreamTransform 两种转换方式StreamTransform.java 提供两种输入转换方法行为getByDefinedTransformer(originMessage, transformer)当transformer是FunctionValueTransformer时执行自定义转换函数否则返回原始消息getByDefaultTransformer(originMessage, streamInputsSchema)当原始消息是 Map 且 schema 非空时调用SessionUtils.getBySchema(streamInputsSchema, originMessage)按 schema 提取结构否则返回原始消息在 Vertex.java 的postStream中可以看到其实际用途consumer 节点消费完迭代器后每个 chunk 会先经outputSchema转换优先自定义 transformer否则默认 schema 提取再processChunk输出preStream则通过actorManager.consume(nodeId, ability, finalInputsSchema, chunk - {...})为组件注入 schema 形态的流式输入。七、一次流式消息的完整旅程结合 Vertex.java 与各 actor 实现可梳理出完整的端到端调用链producer 侧输出STREAM节点在postStream中迭代输出结果processChunk调用actorManager.produce(nodeId, message, ability, firstFrame)若节点是 end 节点endNode流数据直接写入StreamWriter非子图或subWorkflowStream().emit(...)子图不再走 actor 分发帧包装与广播ActorManager.produce把消息包装为{producerId: value}单键 MapStreamActor.send按 producer 分片加锁后广播给该 consumer 的全部StreamProcessor首帧启动首个有效帧触发vertex.streamCall(latch, errorCallback)等待 latch 就绪30s 超时随后启动所有 processor 的run主循环路由入队StreamProcessor.run从 payload 中按引用路径提取值写入 schema 叶子对应的阻塞队列consumer 侧消费组件通过actorManager.consume(...)拿到 schema 对齐的嵌套 Map含阻塞迭代器逐块hasNext()/next()消费streamCallback用于 trace结束帧收敛producer 全部输出完后sendStreamEndFrame调用actorManager.endMessage(nodeId, ability)发送END_ producerId帧processor 收到各 source group 的结束帧后向迭代器队列投放EndFrame并退出主循环收尾清理awaitCompletion()等待主任务与处理器线程结束单次 5000ms 窗口shutdown()取消所有任务并重置 actor 状态。其中sendStreamEndFrame还有「延迟结束帧」的优化当节点在同一批次内还存在同类 sibling 能力batch-in 的INVOKE/STREAM或 stream-in 的COLLECT/TRANSFORM未执行时END 帧会被推迟到该批次末尾统一冲刷避免下游 consumer 在 sibling 输出尚未产生时就收到结束帧。八、关键配置STREAM_INPUT_GEN_TIMEOUT迭代器取流的超时通过会话环境变量控制相关键定义在 SessionConstants.java内部键_stream_input_generator_timeoutSTREAM_INPUT_GEN_TIMEOUT_KEY环境变量名STREAM_INPUT_GEN_TIMEOUTSTREAM_INPUT_GEN_TIMEOUT_ENV_KEY。在 Config.java 中该配置注册为float类型内置默认值为-1.0d。结合 StreamProcessor.java 的构造逻辑可以得出取值语义配置值 0启用生成器超时超时时间 配置值秒取流超过该时长且引用路径未声明来源时hasNext()返回false配置值 0含默认 -1.0 与显式 0无限等待不启用超时配置值为非数字字符串或解析失败ActorManager回退到1.0d秒兜底值。ActorManager构造时会从session.config().getEnv(STREAM_INPUT_GEN_TIMEOUT_KEY)读取该值并传递给每个StreamActor/StreamProcessor因此该配置影响的是「consumer 生成器等待 producer 帧」的耐心上限。九、子图模式与共享输出队列ActorManager支持subGraph标志对应嵌套工作流子工作流场景相关创建逻辑在 Workflow.java主工作流buildActorManager(workflowSession, false)subGraph false此时subWorkflowStream为null调用subWorkflowStream()会抛出GRAPH_STREAM_ACTOR_EXECUTION_ERROR子工作流buildActorManager(subSession, true)subGraph true创建容量10 * 1024的AsyncStreamQueue作为对外共享输出队列子工作流 end 节点的流式输出经emit(...)写入该队列父级通过subWorkflowStream()读取队列关闭时发送StreamEmitter.END_FRAME值为 all streaming outputs finish。LoopGroupcore/src/main/java/com/openjiuwen/core/workflow/component/loop/LoopGroup.java同样以subGraph true重建ActorManager说明循环体复用同一套子图流式机制。十、测试佐证与可靠性设计模块配套的单元测试集中在 StreamProcessorTest.java覆盖了原文档说明中提到的三类关键行为schema 叶子路由activeSourceInAlternativeGroupClosesInactiveSourceQueues验证{left: ${left.answer}, right: ${right.answer}}这类 schema 下left来源的帧被路由到对应迭代器且同组未激活的right队列被提前关闭回调触发同一测试断言消费left迭代器时streamCallback收到Map.of(left, L)即回调按「路径 → 值」形式触发结束帧处理sourceGroupsFinishWhenAnySourceInEachGroupEnds验证多来源组a-stream/b-stream一组、c-transform一组在各自收到 END 帧后处理器线程正常退出另有两个超时相关测试未声明来源路径在启用超时时hasNext()直接返回falseundeclaredSourcePathsUseGeneratorTimeoutcloseAllQueuesWithTimeout投放TIMEOUT_SENTINEL后迭代器抛出GraphError状态码STREAM_PROCESSOR_QUEUE_TIMEOUT消息合法性校验messageValidationMirrorsPythonProducerIdChecks验证 producer ID 提取、END 帧识别及非法消息空 Map、非 Map抛IllegalArgumentException。这些测试同时印证了「producerId-ABILITY」来源键协议与「单键 Map」消息格式的严格约束任何不符合{producerId: value}形状的消息都会被拒绝从而保证路由与收敛的正确性。小结graph.stream_actor是 openJiuwen 工作流流式能力的地基StreamGraph负责注册、ActorManager负责调度、StreamActor负责生命周期、StreamProcessor负责路由与收敛、StreamPayload承载帧、StreamTransform负责转换。理解这一层的设计后你既能解释「为什么COLLECT节点拿到的是带阻塞迭代器的 schema 对齐输入」也能在自研流式组件时正确地实现StreamConsumer三方法、遵守单键 Map 帧协议与END_结束帧约定并利用STREAM_INPUT_GEN_TIMEOUT精细控制消费者等待行为。赞分享人工智能AI AgentAgent 框架工具调用Agent 工作流RAGMCP Clients强化学习【免费下载链接】agent-core-javaopenJiuwen agent core Java是 openJiuwen Core的 Java版本提供AI Agent开发、运行、调优与演进相关的全套SDK能力。项目地址https://gitcode.com/openJiuwen/agent-core-java点击查看免费下载相关推荐openJiuwen Agent Core JavaAssistantMessageChunk 流式消息片段与合并机制深度解析openJiuwen Agent Core JavaAssistantMessageChunk 流式消息片段与合并机制深度解析 AssistantMessag人工智能AI AgentAgent 框架工具调用Agent 工作流RAGMCP Clients强化学习openJiuwen agent-core-java 消息片段体系详解BaseMessageChunk 与流式消息合并机制openJiuwen agent core java 消息片段体系详解BaseMessageChunk 与流式消息合并机制 BaseMessageChunk人工智能AI AgentAgent 框架工具调用Agent 工作流RAGMCP Clients强化学习openJiuwen Agent Core Java 控制器流式输出块 ControllerOutputChunk 深入解析openJiuwen Agent Core Java 控制器流式输出块 ControllerOutputChunk 深入解析 ControllerOutputC人工智能AI AgentAgent 框架工具调用Agent 工作流RAGMCP Clients强化学习创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 
长河君 · 2026/10/14 17:11:44 · 阅读 1.2k openJiuwen Agent Core：graph.stream_actor 图流式消息协调机制深入解析
人工智能AI AgentAgent 框架工具调用Agent 工作流RAGMCP Clients强化学习【免费下载链接】agent-core-javaopenJiuwen agent core Java是 openJiuwen Core的 Java版本提供AI Agent开发、运行、调优与演进相关的全套SDK能力。项目地址https://gitcode.com/openJiuwen/agent-core-java点击查看免费下载com.openjiuwen.core.graph.stream_actor是 openJiuwen agent-core-java 中支撑工作流/图节点间流式数据传递的核心模块它负责把具备STREAM/TRANSFORM能力producer的节点输出逐帧转发给具备COLLECT/TRANSFORM能力consumer的节点并提供基于 schema 的阻塞迭代器式消费接口。读完本文你将掌握ActorManager、StreamActor、StreamProcessor等核心类型的职责划分、produce → send → processor → generator → consume的完整消息调用链、END 帧收敛协议以及STREAM_INPUT_GEN_TIMEOUT等关键配置的底层行为可直接用于理解或扩展自己的工作流流式组件。一、模块定位图节点之间的流式消息协调层在 openJiuwen 的 Workflow / Graph 执行模型中节点通过**组件能力ComponentAbility**声明其输入输出形态。定义位于 ComponentAbility.java能力名称语义INVOKEinvokebatch in, batch out批量进、批量出STREAMstreambatch in, stream out批量进、流式出COLLECTcollectstream in, batch out流式进、批量出TRANSFORMtransformstream in, stream out流式进、流式出graph.stream_actor包正是为「流式进」的节点COLLECT/TRANSFORM与「流式出」的节点STREAM/TRANSFORM之间的帧级消息搬运而设计。它统一处理了 actor 生命周期、payload 协议、consumer 注册、schema 驱动的迭代器生成与输入转换共提供 7 个核心类型类型职责ActorManager根据 producer/consumer 边与组件能力构建StreamActor统一负责生产、消费与关闭StreamActor管理单个消费节点的流式调用生命周期为COLLECT/TRANSFORM能力维护处理器StreamConsumer由图节点实现的流式消费接口定义启动、可处理判断与完成状态StreamGraph保存节点 ID 到StreamConsumer的注册关系StreamPayloadproducer 与 consumer 之间传递的消息载体包含原始消息与来源能力StreamProcessor将消息路由到 schema 叶子路径对应的阻塞迭代器并负责 END 帧收敛StreamTransform提供自定义 transformer 与 schema 两种输入转换方式其中ActorManager、StreamActor、StreamConsumer、StreamGraph、StreamPayload、StreamProcessor、StreamTransform的实现分别位于 core/src/main/java/com/openjiuwen/core/graph/stream_actor/源码注释明确标注其镜像自 Python 版 openjiuwen 的graph/stream_actor/manager.py与base.py。二、ActorManager流式协调的中央调度器ActorManager.java 是模块入口负责从工作流规格中解析流式边、构建 actor、并对外提供统一的生产与消费 API。2.1 关键字段字段类型说明streamEdgesMapString, ListStringproducer 节点到下游 consumer 节点列表的映射来源于workflowSpec.getStreamEdges()consumerDictMapString, ListString由buildReverseGraph(...)反向展开得到的 consumer → producer 关系streamsMapString, StreamActor已注册 consumer 的StreamActor映射LinkedHashMapstreamsTransformStreamTransform提供 schema / 自定义函数两种流式输入转换能力subGraphboolean当前 manager 是否服务于子图subWorkflowStreamAsyncStreamQueue子图模式下暴露给子工作流的共享输出队列容量10 * 1024activeProducerIdsMapString, SetComponentAbility记录本次运行中实际产生过帧的 producer 及其能力用于「未激活源清理」2.2 构造反向展开边 能力过滤构造器ActorManager(WorkflowSpec workflowSpec, StreamGraph graph, boolean subGraph, ActorManagerSession session)的核心逻辑分三步反向建图调用buildReverseGraph把streamEdgesproducer → consumers翻转成consumerDictconsumer → producers这正是原文档说明中「把streamEdges反向展开为 consumer 到 producer 的关系」的实现。能力筛选遍历每个 consumer 的 producer把具备STREAM/TRANSFORM能力的 producer 以producerId-ABILITY如producer-stream的形式收进 source 集合随后调用streamAbilitiesForConsumer仅为具备COLLECT/TRANSFORM能力的 consumer 创建StreamActorINVOKE、STREAM节点不会成为 actor。来源分组优先读取workflowSpec.getStreamSourceGroups().get(consumerId)的分组配置未配置时按字典序把每个 source 拆成独立单元素组。分组语义在收敛逻辑中起关键作用见第四节。从源码还可以看到两个细节producer 的 source 键统一形如producerId - ability.getAbilityName()sourceKey方法后续StreamProcessor解析来源时通过lastIndexOf(-)切分并反向匹配ComponentAbility枚举超时从session.config().getEnv(SessionConstants.STREAM_INPUT_GEN_TIMEOUT_KEY)读取解析失败时回退到默认值1.0d源码常量DEFAULT_STREAM_GENERATOR_TIMEOUT_SECONDS。2.3 对外 API方法行为subWorkflowStream()返回子图输出队列当subGraph为false时抛出GRAPH_STREAM_ACTOR_EXECUTION_ERROR提示 only sub graph has sub_workflow_streamstreamTransform()返回内部复用的StreamTransform实例produce(producerId, messageContent, ability, firstFrame)将消息包装成{producerId: value}单键 Map 并分发到全部下游 consumer没有 consumer 时仅记录 warning 日志 Discard chunk send from [...] to none consumerendMessage(producerId, ability)发送END_ producerId形式的结束帧通知下游处理器收敛consume(consumerId, ability, schema, streamCallback)为指定 consumer 生成与schema结构一致的迭代器映射找不到StreamActor时返回空映射markProducerDone(producerId)/shouldSanitizeStreamSource(...)维护工作流状态中的finished_stream_nodes并在存在「同组替代源已激活」时清理未激活源的队列shutdown()逐个关闭全部StreamActor用于图执行结束或异常中断后的清理produce的实现值得注意它在分发前调用activeProduceAbility记录该 producer 的能力已激活若某个 producer 没有配置任何下游 consumer消息会被直接丢弃并打 warning——这是「无消费方时不阻塞 producer」的设计取舍。三、StreamActor单个消费节点的流式调用生命周期StreamActor.java 对应一个流式消费节点。它为节点上每个可消费能力COLLECT/TRANSFORM维护一个独立的StreamProcessor并在收到首帧时负责拉起vertex.streamCall(...)虚拟线程与全部处理器线程。3.1 关键字段字段类型说明processorsMapComponentAbility, StreamProcessor每个可消费能力对应一个StreamProcessorHashMaptaskFuture?当前vertex.streamCall(...)所在的线程池任务taskCompletion/taskErrorCompletableFutureVoid主 stream 调用的完成 / 异常通知vertexStreamConsumer实际执行流式消费逻辑的图节点对象nodeIdString当前 actor 对应的节点 IDrunningTasksListRunningTask已启动的处理器任务及其完成句柄producerLocksStripedKeyLocks按 producer 分片的锁保证同一 producer 的帧按序送达3.2 生命周期核心方法方法行为send(message, sourceAbility, firstFrame, producerId)发送一帧消息当首帧满足启动条件时会先启动vertex.streamCall(...)与全部处理器任务再把StreamPayload广播给每个StreamProcessorgenerator(ability, schema, streamCallback)为指定能力返回与schema对齐的阻塞迭代器映射该能力不存在时返回空映射awaitCompletion()等待主 stream 调用与全部处理器任务完成单个等待窗口为 5000msSHUTDOWN_TIMEOUT_MSshutdown()取消主任务、异常 future 与全部处理器任务等待清理完成后重置内部状态send内部的启动判定源码sendOnProducer体现了生命周期细节先检查vertex.shouldHandleMessage()为false时直接丢弃该帧在startGuard同步块内检查主任务是否已结束或已异常若taskError已完成且异常则丢弃后续帧若处理器任务仍在运行则继续向处理器分发支持多批次的连续流式输入仅当「是首帧」且vertex.isDone()为true时才通过beginStreamCall()启动主 stream 调用——CountDownLatch等待窗口 30 秒START_LATCH_TIMEOUT_SECONDS随后用startProcessorTasks()启动全部StreamProcessor的run循环。所有帧处理与处理器线程都提交到有界模块线程池STREAM_EXECUTOROpenJiuwenExecutors.newBoundedModulePool(stream-actor, false)且runStreamTask会把Error、BaseError、CompletionException等统一写入对应CompletableFuture以驱动异常通知。四、StreamProcessorschema 驱动的队列路由与 END 帧收敛StreamProcessor.java 是数据面核心它消费StreamActor广播的 payload 队列把每条消息按 schema 引用路径路由到对应的阻塞迭代器队列并在所有来源都送达结束帧后收敛退出。4.1 关键字段字段类型说明END_SENTINEL/TIMEOUT_SENTINELObject流结束哨兵 / 超时哨兵源码中实际使用SessionUtils.EndFrame与静态TIMEOUT_SENTINELnodeIdString当前 processor 归属的 consumer 节点 IDqueueBlockingQueueStreamPayload接收来自StreamActor的原始 payload 队列LinkedBlockingQueueprocessorQueuesMapString, ListBlockingQueueObjectschema 引用路径含$的引用串到下游阻塞队列列表的映射ConcurrentHashMapCopyOnWriteArrayListsourceGroups/sourceIdsListSetString/SetString来源分组与来源 ID 集合来源键格式为producerId-ABILITYhasTimeout/timeoutMillisboolean/long是否启用生成器取流超时streamGeneratorTimeoutSeconds 0时启用否则无限等待超时配置为0或负值即禁用4.2 主循环 run路由与收敛run(ComponentAbility ability)的 while 主循环逻辑源码第 68-110 行取帧pollPayload()以BLOCKING_QUEUE_MS为窗口轮询内部队列识别结束帧isEndMessage判定消息是否为单键 Map 且值是以END_开头的字符串若是则把producerId-ABILITY记入handledSources并关闭该来源对应的迭代器队列向队列投放SessionUtils.EndFrame路由普通帧routeMessageValue遍历processorQueues通过SessionUtils.getValueByNestedPath从消息中按「原始键路径」提取值写入所有对应引用路径的阻塞队列收敛判定allSourceGroupsFinished检查每个 source group 是否至少有一个来源已结束——只要每个组内有一个来源结束即视为该组完成对应测试sourceGroupsFinishWhenAnySourceInEachGroupEnds替代源清理closeInactiveGroupSources在某个组内已有活跃源产生帧时提前关闭同组的其他不活跃来源队列避免消费者空等对应测试activeSourceInAlternativeGroupClosesInactiveSourceQueues。4.3 generator把 schema 变成阻塞迭代器generator(MapString, Object schema, ConsumerObject streamCallback)的核心行为对空 schema 返回空 Map通过DictUtils.extractLeafNodes提取 schema 全部叶子路径叶子的值若为包含$的引用串如${left.answer}则该叶子被替换为阻塞IteratorcreateIterator否则保留静态值阻塞迭代器的hasNext()会从专属BlockingQueue取消息遇到EndFrame返回false流正常结束遇到TIMEOUT_SENTINEL抛出GraphError状态码STREAM_PROCESSOR_QUEUE_TIMEOUT遇到超时针对未声明来源的路径useTimeout hasTimeout !pathHasDeclaredSource(...)也返回falsenext()每次弹出值后调用streamCallback.accept(Map.of(keyPath, value))这就是Vertex.preStream中 trace 流式输入回调的挂载点。由此consumer 节点拿到的是一个结构与 schema 完全一致的嵌套 Map其中流式引用位置是可阻塞等待的迭代器静态叶子保持不变。五、StreamConsumer 与 StreamGraph消费侧注册表5.1 StreamConsumer 接口StreamConsumer.java 由图节点实现定义三个方法方法说明streamCall(CountDownLatch latch, ConsumerException errorCallback)启动当前节点的流式消费逻辑实现方需在准备完成后释放latch通过errorCallback上报异常shouldHandleMessage()当前节点是否应接收流式消息通常由组件能力状态决定isDone()当前节点是否已完成一次执行周期false时会阻止新的首帧启动 stream call5.2 StreamGraph 注册表StreamGraph.java 保存节点 ID 到StreamConsumer的注册关系streamNodesMapString, StreamConsumerLinkedHashMap维护插入顺序addStreamConsumer(consumer, nodeId)同一节点仅首次注册生效putIfAbsent语义getNode(nodeId)按节点 ID 取 consumer不存在返回null。该注册表在ActorManager构造时被查询graph.getNode(consumerId)用于把消费者节点与其流式实现绑定。六、StreamPayload 与 StreamTransform消息协议与输入转换6.1 StreamPayload 消息载体StreamPayload.java 是 producer 与 consumer 之间传递的最小消息单元字段类型说明messageObject原始消息内容通常是单键 Map键为 producerIdsourceAbilityComponentAbility产生该消息的能力类型如STREAM或TRANSFORM构造器StreamPayload(Object message, ComponentAbility sourceAbility)访问器getMessage()/getSourceAbility()。注意StreamProcessor对消息合法性有强校验——必须是单键 Map否则抛出IllegalArgumentException(message is invalid)见singleMessageEntry。6.2 StreamTransform 两种转换方式StreamTransform.java 提供两种输入转换方法行为getByDefinedTransformer(originMessage, transformer)当transformer是FunctionValueTransformer时执行自定义转换函数否则返回原始消息getByDefaultTransformer(originMessage, streamInputsSchema)当原始消息是 Map 且 schema 非空时调用SessionUtils.getBySchema(streamInputsSchema, originMessage)按 schema 提取结构否则返回原始消息在 Vertex.java 的postStream中可以看到其实际用途consumer 节点消费完迭代器后每个 chunk 会先经outputSchema转换优先自定义 transformer否则默认 schema 提取再processChunk输出preStream则通过actorManager.consume(nodeId, ability, finalInputsSchema, chunk - {...})为组件注入 schema 形态的流式输入。七、一次流式消息的完整旅程结合 Vertex.java 与各 actor 实现可梳理出完整的端到端调用链producer 侧输出STREAM节点在postStream中迭代输出结果processChunk调用actorManager.produce(nodeId, message, ability, firstFrame)若节点是 end 节点endNode流数据直接写入StreamWriter非子图或subWorkflowStream().emit(...)子图不再走 actor 分发帧包装与广播ActorManager.produce把消息包装为{producerId: value}单键 MapStreamActor.send按 producer 分片加锁后广播给该 consumer 的全部StreamProcessor首帧启动首个有效帧触发vertex.streamCall(latch, errorCallback)等待 latch 就绪30s 超时随后启动所有 processor 的run主循环路由入队StreamProcessor.run从 payload 中按引用路径提取值写入 schema 叶子对应的阻塞队列consumer 侧消费组件通过actorManager.consume(...)拿到 schema 对齐的嵌套 Map含阻塞迭代器逐块hasNext()/next()消费streamCallback用于 trace结束帧收敛producer 全部输出完后sendStreamEndFrame调用actorManager.endMessage(nodeId, ability)发送END_ producerId帧processor 收到各 source group 的结束帧后向迭代器队列投放EndFrame并退出主循环收尾清理awaitCompletion()等待主任务与处理器线程结束单次 5000ms 窗口shutdown()取消所有任务并重置 actor 状态。其中sendStreamEndFrame还有「延迟结束帧」的优化当节点在同一批次内还存在同类 sibling 能力batch-in 的INVOKE/STREAM或 stream-in 的COLLECT/TRANSFORM未执行时END 帧会被推迟到该批次末尾统一冲刷避免下游 consumer 在 sibling 输出尚未产生时就收到结束帧。八、关键配置STREAM_INPUT_GEN_TIMEOUT迭代器取流的超时通过会话环境变量控制相关键定义在 SessionConstants.java内部键_stream_input_generator_timeoutSTREAM_INPUT_GEN_TIMEOUT_KEY环境变量名STREAM_INPUT_GEN_TIMEOUTSTREAM_INPUT_GEN_TIMEOUT_ENV_KEY。在 Config.java 中该配置注册为float类型内置默认值为-1.0d。结合 StreamProcessor.java 的构造逻辑可以得出取值语义配置值 0启用生成器超时超时时间 配置值秒取流超过该时长且引用路径未声明来源时hasNext()返回false配置值 0含默认 -1.0 与显式 0无限等待不启用超时配置值为非数字字符串或解析失败ActorManager回退到1.0d秒兜底值。ActorManager构造时会从session.config().getEnv(STREAM_INPUT_GEN_TIMEOUT_KEY)读取该值并传递给每个StreamActor/StreamProcessor因此该配置影响的是「consumer 生成器等待 producer 帧」的耐心上限。九、子图模式与共享输出队列ActorManager支持subGraph标志对应嵌套工作流子工作流场景相关创建逻辑在 Workflow.java主工作流buildActorManager(workflowSession, false)subGraph false此时subWorkflowStream为null调用subWorkflowStream()会抛出GRAPH_STREAM_ACTOR_EXECUTION_ERROR子工作流buildActorManager(subSession, true)subGraph true创建容量10 * 1024的AsyncStreamQueue作为对外共享输出队列子工作流 end 节点的流式输出经emit(...)写入该队列父级通过subWorkflowStream()读取队列关闭时发送StreamEmitter.END_FRAME值为 all streaming outputs finish。LoopGroupcore/src/main/java/com/openjiuwen/core/workflow/component/loop/LoopGroup.java同样以subGraph true重建ActorManager说明循环体复用同一套子图流式机制。十、测试佐证与可靠性设计模块配套的单元测试集中在 StreamProcessorTest.java覆盖了原文档说明中提到的三类关键行为schema 叶子路由activeSourceInAlternativeGroupClosesInactiveSourceQueues验证{left: ${left.answer}, right: ${right.answer}}这类 schema 下left来源的帧被路由到对应迭代器且同组未激活的right队列被提前关闭回调触发同一测试断言消费left迭代器时streamCallback收到Map.of(left, L)即回调按「路径 → 值」形式触发结束帧处理sourceGroupsFinishWhenAnySourceInEachGroupEnds验证多来源组a-stream/b-stream一组、c-transform一组在各自收到 END 帧后处理器线程正常退出另有两个超时相关测试未声明来源路径在启用超时时hasNext()直接返回falseundeclaredSourcePathsUseGeneratorTimeoutcloseAllQueuesWithTimeout投放TIMEOUT_SENTINEL后迭代器抛出GraphError状态码STREAM_PROCESSOR_QUEUE_TIMEOUT消息合法性校验messageValidationMirrorsPythonProducerIdChecks验证 producer ID 提取、END 帧识别及非法消息空 Map、非 Map抛IllegalArgumentException。这些测试同时印证了「producerId-ABILITY」来源键协议与「单键 Map」消息格式的严格约束任何不符合{producerId: value}形状的消息都会被拒绝从而保证路由与收敛的正确性。小结graph.stream_actor是 openJiuwen 工作流流式能力的地基StreamGraph负责注册、ActorManager负责调度、StreamActor负责生命周期、StreamProcessor负责路由与收敛、StreamPayload承载帧、StreamTransform负责转换。理解这一层的设计后你既能解释「为什么COLLECT节点拿到的是带阻塞迭代器的 schema 对齐输入」也能在自研流式组件时正确地实现StreamConsumer三方法、遵守单键 Map 帧协议与END_结束帧约定并利用STREAM_INPUT_GEN_TIMEOUT精细控制消费者等待行为。赞分享人工智能AI AgentAgent 框架工具调用Agent 工作流RAGMCP Clients强化学习【免费下载链接】agent-core-javaopenJiuwen agent core Java是 openJiuwen Core的 Java版本提供AI Agent开发、运行、调优与演进相关的全套SDK能力。项目地址https://gitcode.com/openJiuwen/agent-core-java点击查看免费下载相关推荐openJiuwen Agent Core JavaAssistantMessageChunk 流式消息片段与合并机制深度解析openJiuwen Agent Core JavaAssistantMessageChunk 流式消息片段与合并机制深度解析 AssistantMessag人工智能AI AgentAgent 框架工具调用Agent 工作流RAGMCP Clients强化学习openJiuwen agent-core-java 消息片段体系详解BaseMessageChunk 与流式消息合并机制openJiuwen agent core java 消息片段体系详解BaseMessageChunk 与流式消息合并机制 BaseMessageChunk人工智能AI AgentAgent 框架工具调用Agent 工作流RAGMCP Clients强化学习openJiuwen Agent Core Java 控制器流式输出块 ControllerOutputChunk 深入解析openJiuwen Agent Core Java 控制器流式输出块 ControllerOutputChunk 深入解析 ControllerOutputC人工智能AI AgentAgent 框架工具调用Agent 工作流RAGMCP Clients强化学习创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考