第 16 章 子 Agent 协作
本章导读:主 agent 的上下文是稀缺资源,而一块分析、一次评审、一份调研往往可以独立完成——硬塞进主线程只会让上下文膨胀、任务变慢。本章讲解主 agent 如何把"可独立处理、上下文重、可并行"的任务委派出去——三类声明方式(内置 general-purpose、工作区 spec、编程式)、ISOLATED 与 SHARED 工作区、同步与后台调用及自动反向通知、持久会话、向用户暴露子 Agent、远程子 Agent(Agent Protocol)与流式转发,以及权限继承、Plan Mode 传播等行为细节。读完本章,你可以用一套声明文件把主 agent 从重上下文任务中解放出来。
一块分析一次评审可独立完成
别挤占主 agent 上下文"] --> B["子 Agent 协作
把任务委派给独立运行的子实例"] B --> C["三类声明方式
内置 general-purpose · 工作区 spec · 编程式"] B --> D["工作区模式
ISOLATED 独享 · SHARED 与父共享"] B --> E["同步与后台
后台完成自动反向通知 · 可持久会话"] B --> F["远程与行为细节
远程子 Agent · 权限继承 · Plan Mode 传播"] C --> G["收益
一套声明文件让主 agent
从重上下文任务中解放"] D --> G E --> G F --> G
本章导览图
作用
让主 agent 把"可独立处理、上下文重、可并行"的任务委派出去,避免主线程膨胀。每个子 agent 都是一个临时实例(本地的 HarnessAgent 或远程 stub),跑自己的会话,结果通过工具返回给父 agent。
一个最小例子
最简单的用法:把子 agent 的 spec 写到工作区里就行。文件名就是 agent_id:
workspace/subagents/reviewer.md:
---
description: 代码审查专家。当用户需要 review PR、找代码问题、检查代码规范时使用。
---
你是一个专注代码评审的子 agent。请按以下流程工作:
1. 先 read_file / grep_files 收集上下文
2. 给出按文件 / 行号的具体建议
3. 末尾给一个 1-5 的总体评分
然后主 agent 就能在推理时调用:
agent_spawn agent_id="reviewer" task="review 这次 PR 的所有改动"
不需要做任何注册。
几种声明方式
支持下面三类来源,构建时合并:
| 方式 | 适用 | 怎么配 |
|---|---|---|
内置 general-purpose |
通用兜底(镜像主 agent 能力) | 总是有,不需要配 |
| 工作区 spec 文件 | 项目特有的、能版本控制的 | workspace/subagents/<id>.md |
| 编程式声明 | 跑时才能确定(远程、动态参数) | builder.subagent(SubagentDeclaration.builder()...) |
工作区 spec 文件
非递归扫 workspace/subagents/*.md,文件名(去掉 .md)就是 agent_id,不要在 front matter 里再写 name。
---
description: 代码评审专家 # 必填,agent 选择是否委派的关键依据
workspace:
mode: isolated # 默认 isolated;shared 表示和父共享工作区
path: ./defs/reviewer # 可选;不写就用默认子目录
model: openai:gpt-4o-mini # 可选;不写就继承父 agent
steps: 8 # 可选;这个子 agent 单次最多迭代次数
temperature: 0.2 # 可选;覆盖父的 GenerateOptions
top_p: 0.95 # 可选
hidden: false # true 时不出现在 agent 可见列表(仍可程序化 spawn)
mode: subagent # primary / subagent / all,默认 all;primary 不允许被 spawn
expose_to_user: true # 可选三态;强制/禁止向用户暴露(不写表示不表态)
enable_pending_tool_recovery: true # 可选;不写则继承父 agent 的恢复配置
tools: [read_file, grep_files] # 可选;继承工具的白名单
---
你是一个专注代码评审的子 agent。
编程式声明
HarnessAgent.builder()
.name("orchestrator")
.model(model)
.workspace(workspace)
.subagent(SubagentDeclaration.builder()
.name("reviewer")
.description("代码审查专家")
.workspace(Path.of("./defs/reviewer"))
.workspaceMode(WorkspaceMode.ISOLATED)
.model("qwen3-max")
.steps(8)
.tools(List.of("read_file", "grep_files"))
.build())
.subagent(SubagentDeclaration.builder()
.name("remote-researcher")
.description("远端调研子 agent")
.url("http://agent-task-server:8080") // 远程子 agent
.headers(Map.of("Authorization", "Bearer xxx"))
.build())
.build();
三种来源互斥:workspace(...)、inlineAgentsBody(...)、url(...) 三选一。
内置 general-purpose
不需要写声明文件,总是可用。它的角色是"通用兜底"——能力和主 agent 一致(同样的模型、工具、技能),共享主工作区。适合"主 agent 想隔离上下文跑一个子任务但又懒得专门写 spec"。
ISOLATED vs SHARED
回看上一节的最小例子:声明里没有配任何工作区字段,那 reviewer 读写文件时用的是哪份目录?workspaceMode 决定子 agent 的工作区怎么算:
- ISOLATED(默认):子 agent 有自己独立的工作区(如果声明里
workspace.path没写,框架会自动开一个子目录)。子 agent 的运行时状态按"父 sessionId × 用户"分桶——同一用户在不同对话里 spawn 同名子 agent 也互不污染。 - SHARED:子 agent 直接用主工作区。适合子 agent 的输出会被父立即读到的情况(例如
general-purpose)。
同步还是后台?
委派发出之后,主 agent 是原地等结果,还是先去推进别的步骤?主 agent 通过 agent_spawn 创建子 agent,关键是 timeout_seconds:
timeout_seconds > 0(默认 30,最大 600)—— 同步调用,主 agent 在这一步 block 等待结果,结果作为工具结果返回。默认超时后会 promote 成后台任务(status: timeout_promoted+task_id),子 agent 继续跑。timeout_seconds = 0—— 后台调用,立即返回一个task_id,子 agent 在后台跑。
通过 RuntimeContext 强制同步。 应用侧可在当前调用的 RuntimeContext 里放入 AgentSpawnTool.CTX_FORCE_SYNC = true,覆盖 LLM 的异步选择;可选再放 CTX_FORCE_SYNC_TIMEOUT_SECONDS 指定硬超时(秒):
RuntimeContext ctx = RuntimeContext.builder()
.sessionId("s-1")
.put(AgentSpawnTool.CTX_FORCE_SYNC, true)
.put(AgentSpawnTool.CTX_FORCE_SYNC_TIMEOUT_SECONDS, 120) // 可选;覆盖 LLM 的 timeout_seconds
.build();
开启后:
- 若设置了
CTX_FORCE_SYNC_TIMEOUT_SECONDS,它会完全覆盖 LLM 的timeout_seconds(<=0回退到 30s,上限 600s)。 - 未设置时,LLM 传的
timeout_seconds=0会被改写成默认同步超时(30s),不会提交后台任务;LLM 传的正数超时仍生效。 - 同步等待超时后返回
status: timeout并中断子 agent,不会 promote 成后台task_id。
agent_send 同样遵守该开关。同一轮里多个强制同步的 agent_spawn 仍可按 Toolkit 默认并行推进。
如果一个目标可以拆成多个互不依赖、资源不冲突的子任务,主 agent 可以在同一轮 reasoning 里发起多个同步子 agent 调用。Toolkit 默认启用工具并行(ToolkitConfig.parallel=true),因此在 ReActAgent 与 HarnessAgent 上这些同步调用都会并行推进;主 agent 会等这一批工具结果都返回后再进入下一轮推理,相当于一次同步 fan-out / fan-in。若需串行执行工具,可传入 ToolkitConfig.builder().parallel(false).build() 构建的自定义 Toolkit。
任务拆解时先画清楚独立性和依赖图:没有依赖边的节点适合交给多个子 agent 并行;有依赖关系的节点要等上游结果后再派发或合并。短任务、关键路径任务适合同步等待或先用 barrier 等齐,用于继续推理;长任务可以用后台模式先跑,主 agent 继续处理其他工作,后续再取结果合并。
后台任务自动反向通知
后台任务跑完了,主 agent 不需要轮询——下一次推理开始前,框架会把已完成的任务结果作为系统提醒注入对话末尾:
<system-reminder>
后台任务已交付:
- task_id=xxx,agent=research-analyst,status=COMPLETED
结果摘要:...
</system-reminder>
主 agent 看到这条 reminder 自然地回应或继续行动。这意味着你不需要在 prompt 里写"记得调 task_output 轮询"——那是旧版本的做法。
后台任务工具
子 agent 的生命周期背后由两组工具配合完成:
| 工具 | 职责 |
|---|---|
agent_spawn |
创建子 agent,可选地执行任务(同步或后台) |
agent_send |
向已存在的子 agent 追加消息 |
agent_list |
列出当前活跃的子 agent 实例 |
task_output |
通过 task_id 获取后台任务结果(阻塞或非阻塞) |
wait_async_results |
等待后台结果到达;可按 task_ids 等指定任务全部完成,或用 wait_all=true 等待当前 session 未完成任务快照全部完成 |
task_cancel |
取消正在运行的后台任务 |
task_list |
列出所有后台任务及其当前状态 |
agent_spawn / agent_send 管理子 agent 实例(创建、复用、通信);task_output / wait_async_results / task_cancel / task_list 管理后台任务结果(查状态、取结果、等待、取消)。两者的桥梁是 task_id——在 agent_spawn 或 agent_send 使用 timeout_seconds=0 时返回。
提示:大多数情况下自动反向通知机制会把结果推回来,不需要显式调用任务工具。它们主要用作逃生口:在反向通知触发前主动检查进度、等待一组必须同时拿齐的结果、取消不再需要的任务、或者在对话压缩后恢复任务状态。
异步结果有三种常用收集方式:
- 主动通知:不阻塞等待时的默认路径。子任务完成后,下一轮 reasoning 前通过
<system-reminder>注入。 - 指定任务检查:用
task_output(task_id, block=false)主动查看某个任务的当前状态或终态结果。 - 等待 barrier(必须等齐时优先):用
wait_async_results(task_ids="id1,id2")或wait_async_results(wait_all=true)。barrier 模式会等到集合终态,并把各任务结果直接写进本次工具返回,主 agent 可立刻继续推理。wait_all=true以调用开始时的未完成任务快照为准,等待期间新创建的任务不会加入 wait set。
注意(遗留 inbox-any):不传
task_ids且不传wait_all时,wait_async_results只等到 inbox 中任意一条消息到达就返回,这不是 wait-all。需要一组任务全部完成时,请用task_ids或wait_all=true。
给已存在的子 agent 补一条消息
agent_spawn 返回值里有一个 agent_key(运行时实例句柄),用它或 label 就能后续追加消息:
agent_send agent_key="agent:reviewer:abc-123" message="顺便也看下 schema 变更"
如果 spawn 时设了 label,也可以用 label 来寻址:
agent_spawn agent_id="reviewer" task="review 这次 PR" label="pr-reviewer"
agent_send label="pr-reviewer" message="顺便也看下 schema 变更"
要列当前活跃的子 agent:agent_list。
持久会话
默认每次 agent_spawn 都创建新的子 agent 实例和会话——不保留之前调用的上下文。在声明里设 persistSession(true) 可以让同一子 agent 在多次 spawn 之间复用:
.subagent(SubagentDeclaration.builder()
.name("note-taker")
.description("跨对话轮次积累笔记")
.persistSession(true)
.build())
开启后,框架会根据 (parentSessionId, agentId, label) 生成确定性的 key。如果再次 spawn 同样的组合,就会复用已存在的 agent 实例——对话历史和状态都保留。
向用户暴露子 Agent
通常子 agent 对用户是不可见的——它们在幕后作为父 agent 的内部工具运行。通过 expose_to_user=true,父 agent 可以把子 agent 暴露为用户可直接交互的入口:
agent_spawn agent_id="researcher" task="调研 AI 趋势" expose_to_user=true
这做了两件事:
- 在 Gateway 里注册子 agent,使其成为用户可寻址的入口
- 发出一个
SubagentExposedEvent到流式事件流中,携带subagentId句柄
用户客户端收到 SubagentExposedEvent 后,就可以直接向子 agent 发消息——完全绕过父 agent:
// 客户端:在事件流中监听暴露的子 agent
chat.sendStream(SendOptions.userId("user-1"), "派一个研究员调查 AI 趋势")
.doOnNext(event -> {
if (event instanceof SubagentExposedEvent se) {
// se.getSubagentId() → 用来直接和子 agent 对话
// se.getAgentId() → 子 agent 类型(如 "researcher")
// se.getLabel() → 可选的人类可读名称
}
})
.blockLast();
// 直接向暴露的子 agent 发消息
chat.sendToSubagent(subagentId, "重点关注 LLM agent").block();
适合"分支对话"场景:父 agent spawn 一个专家,用户独立地和那个专家继续交流。完整的 Channel 侧 API 见第 20 章(渠道)"与暴露的子 Agent 对话"一节。
怎么开启
用 agent.channel(...) —— bridge 自动接好,零配置:
HarnessAgent agent = HarnessAgent.builder()
.name("orchestrator")
.model("dashscope:qwen-plus")
.build();
// channel() 创建内部 gateway 并自动接好 bridge——expose_to_user 直接可用。
ChatUiChannel chat = agent.channel(ChatUiChannel.create());
没有绑定 Channel 时,agent_spawn 里的 expose_to_user=true 会被静默忽略——子 agent 照常工作,只是不会暴露给用户。多 agent 场景用 GatewayBootstrap 的接法见第 20 章"GatewayBootstrap 下暴露子 Agent"一节。
用代码控制是否暴露
完全依赖 LLM 传 expose_to_user=true 有时不够灵活。你可以从应用代码侧覆盖这个决策,有两种方式,最终生效值按以下优先级解析(从高到低):
RuntimeContext按调用覆盖 —— 作用于当前这次调用里的所有agent_spawnSubagentDeclaration按类型策略 —— 该子 agent 类型的静态默认值- LLM 传入的
expose_to_user工具参数 - 以上都没有表态时,默认为
false
通过 RuntimeContext 按调用覆盖。 在 AgentSpawnTool.CTX_EXPOSE_TO_USER 这个 key 下放一个 Boolean(或其字符串形式):
RuntimeContext ctx = RuntimeContext.builder()
.userId("user-1")
.put(AgentSpawnTool.CTX_EXPOSE_TO_USER, true) // 强制开启;传 false 则禁止暴露
.build();
通过声明设置按类型策略。 使用三态的 exposeToUser —— TRUE 总是暴露,FALSE 永不暴露(即使 LLM 传了 expose_to_user=true 也会被覆盖),null(默认)则交给 context 覆盖、再交给 LLM 参数决定:
SubagentDeclaration decl = SubagentDeclaration.builder()
.name("researcher")
.description("调研主题并返回汇总报告。")
.exposeToUser(true) // 这个子 agent 类型始终对用户可直接寻址
.build();
或在 Markdown 子 agent spec 的 front matter 里(同样是三态——不写这个 key 表示"不表态"):
---
name: researcher
description: 调研主题并返回汇总报告。
expose_to_user: true
---
这样你就能不管模型怎么决定,都能强制或禁止暴露;同时在代码两侧都不表态时,仍然让 LLM 自行选择。
跨重启与多副本
默认情况下,暴露只存在于创建它的进程里:subagentId 只在那个节点有效,重启即失效。要让暴露的子 agent 在任意副本、重启之后都能解析,给 agent 配上 distributedStore(...) 即可——和配 state、filesystem 是同一行:
HarnessAgent agent = HarnessAgent.builder()
.name("orchestrator")
.model("dashscope:qwen-plus")
.distributedStore(RedisDistributedStore.fromJedis(jedis))
.build();
ChatUiChannel chat = agent.channel(ChatUiChannel.create()); // 恢复能力自动接好
subagentId 会持久化到后端,子 agent 自己的对话会按 session 从分布式 AgentStateStore 重新加载——即使后续消息落到不同节点,用户面对的仍是同一个子 agent。多 agent 的 GatewayBootstrap 传 .distributedStore(...)(不传则继承 main agent 的)。部署建议——包括把某个 subagentId 路由回它的活实例所在节点(粘性路由)——见第 42 章(走向生产)。
让 agent 自己写新的子 agent spec
agent_generate 工具(默认关闭)可以让 LLM 起草一份新的子 agent spec 并直接写到 workspace/subagents/<name>.md:
// 开启方法(构建期):
// 拿到 builder 内部的 SubagentsMiddleware 引用,调一下 enableAgentGenerateTool
适合"agent 跑到一半发现自己需要一类新的助手"。生产环境慎用——通常先让 agent 把方案写出来人工 review 再写文件。
一些行为细节
description要写好:这是模型决定要不要委派的关键依据。"代码评审"远不如"当用户要 review PR、找代码风格问题时使用"有效。- 递归保护:子 agent 不能再 spawn 子 agent(被强制标为"叶子");同时还有一个硬上限 3 层。
- userId 透传:父的
RuntimeContext.userId会自动透到子,所以多租户隔离链不会断。 - 权限继承:父的所有 DENY 权限规则会自动传给子。如果父被禁用了某个工具,子也一样被禁——安全边界不会因为委派被绕过。在声明里设
inheritParentPermissions(false)可以关闭这个行为。 - 流式转发:父 agent
stream()时,同步子 agent 的中间事件会实时流回父的Flux(带来源标记),见下文"子 Agent 流式"一节。
还有一条与声明写法相关的恢复配置,单独说明:框架自动构建的本地子 agent(包括 general-purpose)会继承父级 HarnessAgent.Builder.enablePendingToolRecovery(...) 配置,默认关闭。声明中可用 .enablePendingToolRecovery(true) 或 .enablePendingToolRecovery(false) 显式覆盖,null 表示继承。工作区 spec 支持 enable_pending_tool_recovery,也兼容 enablePendingToolRecovery 写法。开启后,新的普通消息会为悬空工具调用补充错误结果,也适用于从失败会话中重新加载的工具调用。等待人工审批的工具仍须提供审批结果;空输入恢复执行、调用方补交工具结果的行为保持原有语义。远端 agent 和自定义工厂需自行配置恢复策略。
远程子 agent
声明里只填 url + 可选 headers,子 agent 就走远程 HTTP 服务(Agent Protocol,协议概览见第 26 章)执行:
.subagent(SubagentDeclaration.builder()
.name("remote-researcher")
.description("远端调研子 agent")
.url("http://agent-task-server:8080")
.headers(Map.of("Authorization", "Bearer xxx"))
.remoteStreaming(true) // 未设置时默认 true
.remoteStreamDetail(RemoteStreamDetail.VERBOSE) // 未设置时为 FULL
.remoteAskPolicy(RemoteAskPolicy.DENY) // 默认
.remoteContextAttributes(Map.of("region", "cn"))
.build())
同样支持同步(timeout_seconds>0)和后台(timeout_seconds=0)。
远程模式专用声明字段:
| 字段 | 默认 | 说明 |
|---|---|---|
remoteStreaming |
true(未设置时) |
父代理使用 streamEvents() 时,把远程任务的 SSE 事件转发进父流,并带 source 标记与 metadata.taskId / metadata.parentSessionId(与 harness TaskRecord / 父 session 一致) |
remoteStreamDetail |
FULL |
回传多少远程事件——见下文"远程流式详细度" |
remoteAskPolicy |
DENY |
如何处理远程工具确认(HITL)请求——见下文"远程授权" |
remoteContextAttributes |
无 | 每次提交都携带的静态调用方属性,写入 context.attributes。按次追加时,在父代理的 RuntimeContext 上用 AgentSpawnTool.CTX_REMOTE_CONTEXT_ATTRIBUTES 放一个 map;见第 26 章(Agent 协议)"上下文属性"一节 |
远程流式详细度
本地子 agent 会把孩子的事件原样转发给父流。远程子 agent 的事件要过一趟网络,过多少由 remoteStreamDetail 决定,以 context.detail 发送:
| 档位 | 回传内容 |
|---|---|
STATUS |
运行生命周期、工具调用起止、工具结果、确认请求 |
FULL(默认) |
STATUS 之外再加文本与思考增量 |
VERBOSE |
远程 agent 发出的每一个事件——块边界、工具入参增量、工具输出增量、带 token usage 的模型调用、hint、agent 结果、自定义事件 |
如果希望父流在子 agent 是本地还是远程时表现一致,选 VERBOSE——这是唯一能让远程子 agent 的工具输出内容和 token 用量到达父代理的档位。它不作为默认,是因为对只渲染文本的调用方来说这些事件纯粹是额外流量。
没有专属 wire 类型的事件以 AGENT_EVENT 传输,原始事件完整序列化在 payload 字段里,父代理解出来的就是本地场景下同一个类,id、时间戳和 metadata 都在。不认识该字段的旧客户端仍读扁平字段,只是看不到这些透传事件。
远程授权
父代理的 DENY 权限规则会随远程提交的 context.deny_rules 一并转发(与本地子 agent 的权限继承一致;可用 inheritParentPermissions(false) 关闭)。
远程 agent 因工具确认而暂停(awaiting_confirm)时:
- 父代理流式 +
remoteAskPolicy=PROPAGATE:向父的streamEvents()转发带非空source标记的RequireUserConfirmEvent。通过 Agent Protocol 的POST /tasks/{id}/resume恢复(接口见第 26 章),请求体为decisions[{toolCallId, approved, reason}]。 - 父代理非流式(
call)或remoteAskPolicy=DENY(默认):自动拒绝待确认项。工具结果中会附注:remote tool confirmation(s) were auto-denied。
等待确认期间任务状态保持 RUNNING(awaitingConfirm=true)。因此 wait_async_results 等 barrier 会继续等待,直到任务被 resume 并进入终态。
异步任务的存储位置
后台任务的状态默认写到 workspace/agents/<parentAgentId>/tasks/<sessionId>.json。这意味着:
- 在共享存储模式(多副本)下,任意节点都能读到任务状态;
- 任务执行粘在创建节点,但完成结果会被任意节点读到、并能正常推送回父 agent;
- 想取消可以从任意节点调
task_cancel——执行节点轮询取消标记后中止。
在 Plan Mode 下委派子 agent
父 agent 在 Plan Mode 时 spawn 的子 agent 会自动继承只读限制——子 agent 在 spawn 时就会被置入 Plan Mode,无法执行写操作,安全边界在委派链上不会断。
子 Agent 流式
提示:新代码请用
streamEvents()(返回Flux<AgentEvent>)。旧stream()系列(Flux<Event>)在 2.0.0 起@Deprecated(forRemoval = true)——事件体系见第 5 章(消息与事件),stream()→streamEvents()的弃用时间线见官方文档的 V1 迁移指南。
父 agent 通过 agent_spawn / agent_send 同步调用子 agent 时,子 agent 的中间事件会实时转发到父的 streamEvents() 流中。每个子事件都带一个 source 字段(/ 分隔的路径,如 "main/researcher"),父事件的 source 为 null。远程 Agent Protocol 子 agent 还会写入 metadata.taskId(AgentEvent.METADATA_TASK_ID,harness 侧任务 id)与 metadata.parentSessionId(AgentEvent.METADATA_PARENT_SESSION_ID,父 session),因此同一轮里对同一远程 agent 的多次调用即使 source 相同也能区分,并能回溯到发起方会话。
caller
└─ parent.streamEvents(msg, ctx)
│
├─ AGENT_START ← 父 agent 启动
├─ TEXT_BLOCK_DELTA … ← 父推理
├─ TOOL_CALL_START "agent_spawn"
│
│ [子 agent 创建]
├─ AGENT_START (source="main/researcher") ← 子启动
├─ TEXT_BLOCK_DELTA … (source="main/researcher") ← 子推理
├─ TOOL_CALL_START … (source="main/researcher")
├─ TOOL_RESULT_END … (source="main/researcher")
├─ AGENT_END (source="main/researcher") ← 子结束
│ [agent_spawn 返回,子结果作为 TOOL_RESULT 传给父]
│
├─ TOOL_RESULT_END ← 父收到工具结果
├─ TEXT_BLOCK_DELTA … ← 父第二轮推理
└─ AGENT_END ← 父结束
使用 streamEvents()(推荐)
parent.streamEvents(new UserMessage(message), ctx)
.doOnNext(event -> {
String src = event.getSource();
String prefix = (src != null) ? "[" + src + "] " : "";
if (event.getType() == AgentEventType.TEXT_BLOCK_DELTA) {
System.out.print(prefix + ((TextBlockDeltaEvent) event).getDelta());
} else if (event.getType() == AgentEventType.TOOL_CALL_START) {
System.out.println(prefix + "[tool] " + ((ToolCallStartEvent) event).getToolCallName());
} else if (event.getType() == AgentEventType.AGENT_START) {
if (src != null) System.out.println("── 子 agent 启动: " + src);
} else if (event.getType() == AgentEventType.AGENT_END) {
if (src != null) System.out.println("── 子 agent 结束: " + src);
}
})
.blockLast();
区分父子事件:
// 只看父事件
events.filter(e -> e.getSource() == null).subscribe(…);
// 只看子事件
events.filter(e -> e.getSource() != null).subscribe(…);
// 只看特定子 agent 的事件
events.filter(e -> e.getSource() != null && e.getSource().contains("researcher")).subscribe(…);
SSE 转发
@GetMapping(value = "/chat", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<ServerSentEvent<String>> chat(@RequestParam String message,
@RequestParam String sessionId) {
RuntimeContext ctx = RuntimeContext.builder().sessionId(sessionId).build();
return agent.streamEvents(new UserMessage(message), ctx)
.map(event -> {
Map<String, Object> payload = new LinkedHashMap<>();
payload.put("type", event.getType().name());
payload.put("id", event.getId());
if (event.getSource() != null) {
payload.put("source", event.getSource());
}
if (event instanceof TextBlockDeltaEvent delta) {
payload.put("delta", delta.getDelta());
} else if (event instanceof ToolCallStartEvent start) {
payload.put("toolName", start.getToolCallName());
}
return ServerSentEvent.<String>builder()
.data(objectMapper.writeValueAsString(payload))
.build();
});
}
行为边界
| 场景 | 是否实时流转发? |
|---|---|
streamEvents() + 同步本地子 agent(timeout_seconds > 0) |
是 |
call() 模式(非流式) |
否(子结果以 tool_result 字符串返回) |
timeout_seconds = 0 后台任务 |
否(终态会通过反向通知给父 agent 下一轮) |
远程子 agent(Agent Protocol)+ 父 streamEvents() + remoteStreaming=true(默认) |
是 |
远程子 agent + 父 call() 或 remoteStreaming=false |
否 |
错误处理
子 agent 内部出错时,框架会把错误捕获并写成一条 TOOL_RESULT 给父,不会把 onError 传播到父流——父流不会被子 agent 的失败打断。如果父流本身出错(比如模型调用失败),按标准 Reactor 语义处理(onErrorResume 等)。
本章小结
这一章回答一个问题:主 agent 上下文放不下的工作,怎么交出去、还能收回来:
- 声明:三类来源构建时合并——内置
general-purpose通用兜底、工作区 spec(workspace/subagents/<id>.md,文件名即 agent_id)、编程式SubagentDeclaration(远程、动态参数等跑时才能确定的场景); - 工作区:ISOLATED(默认,独立工作区,运行状态按父 sessionId × 用户分桶)与 SHARED(直接用主工作区);
- 调用:
timeout_seconds > 0同步等待(默认 30,最大 600,超时 promote 成后台任务);= 0直接后台; - 收结果:后台完成自动反向通知(无需轮询);
task_output主动查单个任务;wait_async_results用task_ids或wait_all=true等齐一组——多数时候不需要显式调任务工具; - 暴露与远程:
expose_to_user把子 agent 变成用户可交互入口(需绑定 Channel);声明url即走 Agent Protocol 远程执行,remoteAskPolicy决定远程工具确认转发还是自动拒绝; - 边界:递归保护最多 3 层、父 DENY 权限规则自动继承、Plan Mode 只读限制随委派传播、userId 透传不断隔离链。
验收清单——读到这里,你应该能:
- ☐ 用一份
workspace/subagents/reviewer.md声明子 agent,并说出 description 在委派决策中的作用; - ☐ 说出 ISOLATED 与 SHARED 各自适合的场景,以及同一用户在不同对话里 spawn 同名子 agent 是否互不污染;
- ☐ 解释
timeout_seconds默认值超时后发生什么,以及= 0与CTX_FORCE_SYNC强制同步的区别; - ☐ 用
wait_async_results等齐一组后台任务,并解释为什么多数时候不需要显式调任务工具; - ☐ 用
expose_to_user把子 agent 暴露给用户,并知道没绑定 Channel 时会发生什么; - ☐ 声明一个远程
url子 agent,并说出remoteAskPolicy=DENY时待确认项的下场; - ☐ (选做)为跨重启可寻址的暴露子 agent 配置
distributedStore(...)。
相关文档
- 第 20 章(渠道)——
expose_to_user、SendOptions、用户直接与子 agent 交互 - 第 12 章(工作区与文件系统)——
subagents/与agents/<id>/tasks/的目录布局 - 第 17 章(计划模式)—— plan 阶段对子 agent 的限制
- 第 11 章(Harness 架构深入)—— 主/子 agent 怎么协作
- 第 26 章(Agent 协议)—— 远程任务端点(SSE + HITL resume)
- 第 5 章(消息与事件)——
AgentEvent体系(推荐)以及已弃用的Event/EventType/StreamOptions