---
title: "harness - 子代理流式输出适配"
---


当主代理通过 `task` / `multitask` 委派子任务时，子代理内部产生的推理事件（思考、工具调用、结果）默认**不会**出现在父流中。本文从原理到实践，完整介绍如何让子代理的实时输出透传给调用方。


### 1、为什么子代理事件需要特殊处理

父代理的流来自 `engine.prompt(prompt).session(session).stream()`，对外是一条 `Flux<AgentEvent>`（`HarnessEngine` 本身没有独立的 `stream()` 入口）。当主代理调用 `task` 或 `multitask` 工具时，`TaskTalent` 会为每个子任务构建独立的 `ReActAgent` 实例与 `InMemoryAgentSession`，并开启各自的子流。

需要说明的是：**子代理自身始终是以 `.stream().blockLast()` 执行完毕的**（`TaskTalent` 的两条分支都是如此），区别只在于子事件是否被包装转发给父流的 sink。所以"子流默认是封闭的"指的不是子代理没开流，而是没有把事件透传出去。若要让子代理的事件出现在父流中，需要满足两个条件：

1. **父请求处于流式模式**：即父级 `ReActTrace.getOptions().getStreamSink()` 不为 null——只有在父请求调用了 `.stream()`、由 `ReActRequest` 注入 `FluxSink` 时才成立。父请求使用 `.call()` 时没有可供外部消费的 sink，子代理事件不会传播（父级只拿到 `task` / `multitask` 的工具结果）。
2. **消费者能识别 `TaskWrapEvent`**：子代理事件被包装成 `TaskWrapEvent` 注入父流，消费者必须识别并解包这一包装层，才能还原出真实的子代理事件。

另外，父 sink 被取消时，子流会通过 `takeUntil(r -> sink.isCancelled())` 主动停止，且每次转发前都会再判一次 `sink.isCancelled()`，避免取消后继续输出。


### 2、原理：TaskWrapEvent 与 FluxSink

流式模式下，子代理每产生一个 `AgentEvent`，`TaskTalent` 都会将它包裹在 `TaskWrapEvent` 中，通过父流的 `FluxSink` 注入主流：

```
主代理流（Flux<AgentEvent>）
  ├── RunStartEvent
  ├── ReasonDeltaEvent        ← 主代理思考
  ├── ToolCallStartEvent      → 触发 task / multitask 工具
  │
  │      TaskTalent 内部，子代理流运行：
  │
  │      ├── TaskWrapEvent(ReasonDeltaEvent)    ← 子代理思考
  │      ├── TaskWrapEvent(ToolCallStartEvent)  ← 子代理工具开始
  │      ├── TaskWrapEvent(ToolCallEndEvent)    ← 子代理工具结束
  │      └── TaskWrapEvent(RunEndEvent)         ← 子代理推理完成
  │
  ├── ToolCallEndEvent        ← task/multitask 工具返回汇总
  └── RunEndEvent             ← 主代理最终结果
```

**`TaskWrapEvent` 携带的信息：**

| API | 说明 |
|------|------|
| `getRealEvent()` | 被包装的真实子代理事件（必须解包后再判断类型） |
| `getParentRunId()` | 父 ReAct 的 runId |
| `getRunId()` | 子代理事件自己的 runId，不是父 runId |
| `getTaskId()` | 子任务唯一 ID |
| `getTaskIndex()` | multitask 的任务序号（取自任务入参 index，默认 1）；单个 task 恒为 1 |
| `getTaskAgentName()` | 子代理名称（如 `explore`、`bash`） |
| `getTaskDescription()` | 子任务摘要描述 |
| `isMultitask()` | 是否来自 `multitask` 并行任务 |
| `getText()` | 委托给被包装事件的文本投影 |

**并发安全：** `multitask` 会并行运行多个子代理，它们共享同一个 `FluxSink`。`TaskTalent` 通过 `synchronized(sink)` 保证多线程并发写入时的串行化：

```java
private void emitTaskChunk(FluxSink<AgentEvent> sink, TaskWrapEvent wrapChunk) {
    synchronized (sink) {
        if (!sink.isCancelled()) {
            sink.next(wrapChunk);
        }
    }
}
```

### 3、如何在消费端适配

消费 `.stream()` 时，流中会混入 `TaskWrapEvent`。适配的核心逻辑是：**遇到 `TaskWrapEvent` 就解包，取出真实事件后与主代理事件按统一方式处理，同时保存子任务元信息（agentName、taskId 等）用于来源标识**。

#### 3.1 基本适配模式

```java
import org.noear.solon.ai.agent.AgentEvent;
import org.noear.solon.ai.agent.react.RunEndEvent;
import org.noear.solon.ai.agent.react.task.ReasonDeltaEvent;
import org.noear.solon.ai.agent.react.task.ToolCallEndEvent;
import org.noear.solon.ai.agent.react.task.ToolCallStartEvent;
import org.noear.solon.ai.chat.event.ChatEvent;
import org.noear.solon.ai.chat.event.ChatEventType;
import org.noear.solon.ai.harness.agent.TaskWrapEvent;

engine.prompt(prompt)
    .session(session)
    .stream()
    .flatMap(outerEvent -> {
        // ① 来源元信息（仅子代理事件才有）
        String taskAgentName = null;
        String taskId = null;
        String taskDescription = null;
        int taskIndex = 0;

        // ② 解包 TaskWrapEvent
        AgentEvent event = outerEvent;
        if (outerEvent instanceof TaskWrapEvent) {
            TaskWrapEvent wrap = (TaskWrapEvent) outerEvent;
            event = wrap.getRealEvent();

            // 只透传感兴趣的事件类型，其余忽略
            if (!(event instanceof ReasonDeltaEvent)
                    && !(event instanceof ToolCallStartEvent)
                    && !(event instanceof ToolCallEndEvent)
                    && !(event instanceof RunEndEvent)) {
                return Flux.empty();
            }

            taskAgentName   = wrap.getTaskAgentName();
            taskId          = wrap.getTaskId();
            taskDescription = wrap.getTaskDescription();
            taskIndex       = wrap.getTaskIndex();

            // 子代理流结束：可在此做一次"子任务完成"通知
            if (event instanceof RunEndEvent) {
                RunEndEvent runEnd = (RunEndEvent) event;
                notifyTaskDone(taskAgentName, taskId, taskDescription, runEnd.isAbnormal());
                return Flux.empty();
            }
        }

        // ③ 统一处理主代理 + 子代理的真实事件
        MyChunk chunk = null;

        if (event instanceof ReasonDeltaEvent) {
            ReasonDeltaEvent delta = (ReasonDeltaEvent) event;
            ChatEvent chatEvent = delta.getChatEvent();

            // 正文只取底层 TEXT_DELTA；思考只取底层 THINKING_DELTA；边界与媒体不混入
            if (chatEvent != null && delta.hasText()) {
                if (chatEvent.is(ChatEventType.THINKING_DELTA)) {
                    chunk = MyChunk.ofThinking(delta.getText());
                } else if (chatEvent.is(ChatEventType.TEXT_DELTA)) {
                    chunk = MyChunk.ofText(delta.getText());
                }
            }
        } else if (event instanceof ToolCallStartEvent) {
            ToolCallStartEvent e = (ToolCallStartEvent) event;
            chunk = MyChunk.ofToolStart(e.getToolName(), e.getArgs());
        } else if (event instanceof ToolCallEndEvent) {
            ToolCallEndEvent e = (ToolCallEndEvent) event;
            chunk = MyChunk.ofToolEnd(e.getToolName(), e.getText());
        } else if (event instanceof RunEndEvent) {
            chunk = MyChunk.ofDone(((RunEndEvent) event).getText());
        }

        if (chunk == null) return Flux.empty();

        // ④ 附加来源标记，供下游区分主代理/子代理
        if (taskAgentName != null) {
            chunk.setAgentName(taskAgentName);
            chunk.setTaskId(taskId);
            chunk.setTaskIndex(taskIndex);
            chunk.setTaskDescription(taskDescription);
        }

        return Flux.just(chunk);
    })
    .subscribe(chunk -> sseEmitter.send(chunk));
```

几个关键点：

- `task` / `multitask` 工具本身的 `ToolCallStartEvent` / `ToolCallEndEvent` 来自**主代理**，不会被包在 `TaskWrapEvent` 里——这两个工具事件通常应在 UI 上过滤掉（它们是调度指令，不是用户可见的操作）。
- 子代理的 `RunEndEvent` 包在 `TaskWrapEvent` 里，可借此在该子任务完成时立刻通知前端（不必等待整条主流结束），在并行任务场景下尤其有用。
- `agentName`（子代理名，如 `explore`）加上 `taskId` 是区分多个并行子任务输出来源的关键字段，前端可据此将不同子任务的 chunk 分组展示；`agentName` 可能重复，不能单独作为标识。
- 正文/思考必须按底层 `ChatEventType` 过滤：仅用 `ReasonDeltaEvent.isThinking()` 只能区分思考分组，无法排除 `THINKING_START` / `THINKING_END` 边界。

#### 3.2 同步模式下的行为差异

若父请求不是流式模式（直接 `.call()`），父链路没有 `FluxSink`，`TaskTalent` 就退回到"不包装、不转发"的分支：

```java
// 同步模式：子代理流仍被完整消费，但事件不外发，只取终态文本
RunEndEvent agentEnd = (RunEndEvent) agent.prompt(prompt)
        .session(session)
        .stream()
        .blockLast();

String result = agentEnd.getText();
```

同步模式下，子代理完整执行后只把最终文本回传给主代理，不产生任何包装的中间事件。主代理的调用方只能看到最终汇总结果，看不到子代理的工具调用过程。

**选择依据：**
- 需要实时展示子代理操作过程 → 必须使用父级 `.stream()` 流式模式
- 只关心最终结果、不关心过程 → `.call()` 同步模式即可，更简单


### 4、参考示例：WebStreamBuilder

SolonCode 中的 `WebStreamBuilder` 是一个生产级的完整适配实现，将 `AgentEvent`（含 `TaskWrapEvent`）映射为 `WebChunk` 序列，通过 SSE 推送给 Web 前端。其对子代理流的处理逻辑与上文描述的模式一致，并在此基础上增加了：

- IM 通道同步转发（微信、飞书等）
- HITL（人机交互循环）挂起卡片推送
- 工具事件的来源前缀规则（`agentName/toolName`）
- 子代理 `RunEndEvent` 时发出"任务完成"信号让前端立即结算 task-group

可直接参考其 `buildStreamFlux` 方法中对 `TaskWrapEvent` 的解包与分发逻辑作为完整示例。