--- title: "agent - 同步与流式响应（call 与 stream）" --- Solon AI Agent 同时提供同步、异步和流式三种结果路径。`call()` / `callAsync()` 直接返回完整响应；`stream()` 返回 `Flux`，用于展示正文增量、思考、工具执行、HITL 和团队协作过程。 ### 1、调用方式 | API | 返回值 | 说明 | |---|---|---| | `call()` | `SimpleResponse` / `ReActResponse` / `TeamResponse` | 同步取得完整结果，不产生面向调用方的顶层 End 事件 | | `callAsync()` | `CompletableFuture<...>` | 异步取得完整结果，内部委托 `call()` | | `stream()` | `Flux` | 观察运行过程；正常完成时由 Request 包装器追加顶层 End 事件 | ```java ReActResponse response = agent.prompt("北京今天天气怎么样？").call(); System.out.println(response.getContent()); ``` ```java agent.prompt("北京今天天气怎么样？").stream().subscribe( event -> handleEvent(event), error -> log.warn("Agent 流失败", error)); ``` 不同 Agent 的顶层边界如下： | Agent | 开始事件 | 结束事件 | 完整响应 | |---|---|---|---| | `SimpleAgent` | `SimpleStartEvent` | `SimpleEndEvent` | `SimpleEndEvent.getResponse()` | | `ReActAgent` | `RunStartEvent` | `RunEndEvent` | `RunEndEvent.getResponse()` | | `TeamAgent` | `TeamStartEvent` | `TeamEndEvent` | `TeamEndEvent.getResponse()` | `ReasonEndEvent`、`ActionEndEvent`、`ToolCallEndEvent` 和 `NodeEndEvent` 只是局部阶段结束，不能作为整个请求的终态。 ### 2、AgentEvent 公共接口 所有事件实现 `AgentEvent`。消费事件时应按具体事件类使用 `instanceof` 或 Reactor 的 `ofType(...)`。 | 方法 | 描述 | |---|---| | `getRunId()` | 当前逻辑运行标识 | | `getAgentName()` | 事件实际产生者 | | `getSession()` | 当前 `AgentSession` | | `getMeta()` / `hasMeta(name)` | 事件元数据 | | `hasText()` | 是否包含非空文本 | | `getText()` | 事件的文本投影；无文本事件默认返回空字符串 | Agent 层没有统一的 `AgentEventType` 或事件分组。需要精确识别正文、思考和媒体时，应读取 Delta 事件包装的 `ChatEvent`。 ### 3、正文与思考增量 `SimpleDeltaEvent`、`ReasonDeltaEvent` 和 `SupervisorDeltaEvent` 分别包装 Simple、ReAct Reason 和 Team Supervisor 的模型输出。 ```java agent.prompt(query).stream() .ofType(ReasonDeltaEvent.class) .subscribe(event -> { ChatEvent chatEvent = event.getChatEvent(); if (chatEvent == null) { return; } if (chatEvent.is(ChatEventType.TEXT_DELTA) && event.hasText()) { ui.appendText(event.getText()); } else if (chatEvent.is(ChatEventType.THINKING_DELTA) && event.hasText()) { ui.appendThinking(event.getText()); } else if (chatEvent.is(ChatEventType.MEDIA_DONE)) { ui.appendMedia(chatEvent.getBlock()); } }); ``` 内置 Agent 当前只将以下 Chat 事件包装为 Delta： ```text TEXT_START / TEXT_DELTA / TEXT_END THINKING_START / THINKING_DELTA / THINKING_END MEDIA_DONE ``` 边界事件通常没有文本。不要把所有 `ReasonDeltaEvent` 都当作正文，也不要把 Agent Delta 当成完整的底层 Chat 协议回放。 ### 4、ReAct 事件与工具执行 典型时序： ```text RunStartEvent -> [ContextSizeEvent] -> ReasonStartEvent -> ReasonDeltaEvent* -> ReasonEndEvent -> [ActionStartEvent -> ToolCallStartEvent -> ToolCallEndEvent -> ActionEndEvent] -> ...下一轮 Reason -> RunEndEvent -> onComplete ``` ```java agent.prompt(query).stream().subscribe( event -> { if (event instanceof ReasonDeltaEvent) { ReasonDeltaEvent delta = (ReasonDeltaEvent) event; if (delta.hasText()) { if (delta.isThinking()) { ui.appendThinking(delta.getText()); } else { ui.appendText(delta.getText()); } } } else if (event instanceof ToolCallStartEvent) { ToolCallStartEvent start = (ToolCallStartEvent) event; ui.toolStarted(start.getCallId(), start.getToolName(), start.getArgs()); } else if (event instanceof ToolCallEndEvent) { ToolCallEndEvent end = (ToolCallEndEvent) event; ui.toolFinished(end.getCallId(), end.getText(), end.getError()); } else if (event instanceof RunEndEvent) { RunEndEvent end = (RunEndEvent) event; ui.finish(end.getResponse(), end.isAbnormal()); } }, error -> ui.fail(error)); ``` Agent 层当前没有工具参数增量事件。模型响应中的 `TOOL_CALL_ARGS_DELTA` 属于 Chat 层；Agent 的 `ToolCallStartEvent` / `ToolCallEndEvent` 表示本地工具调用的处理边界。 普通工具参数错误或执行异常通常会被转换为 observation 文本，此时 `ToolCallEndEvent.getError()` 仍可能为 `null`。审计端应综合检查 `getText()`、`getError()`、`RunEndEvent.isAbnormal()` 和 Reactor `onError`。 ### 5、计划、上下文和 HITL 规划模式使用 `PlanEvent`： - `getPlanStage()`：`CREATE`、`PROGRESS`、`REVISE`； - `getPlans()`：当前计划列表； - `getPlanIndex()`、`getReasonId()`：计划进度与 Reason 关联。 `ContextSizeEvent` 由上下文压缩拦截器在 Reason 开始前产生，并非每个流都有。主要读取 `getContextLength()`、`getMessageCount()`、`getTokenCount()`、`isCompressed()` 以及压缩前后统计。Token 数是估算值，不是供应商 Usage。 首次 HITL 挂起的典型时序是： ```text ReasonEndEvent -> HITLPendingEvent -> 不产生 ActionStartEvent / ToolCallStartEvent / ToolCallEndEvent / ActionEndEvent -> RunEndEvent（当前实现 isAbnormal=true，Session 仍为 pending） -> onComplete ``` 挂起不是 Reactor 错误。UI 应结合 `HITLPendingEvent` 或 `session.isPending()` 判断“等待审批”，不要仅凭 `isAbnormal()` 将它显示为不可恢复故障。 使用同一 Session 和空 Prompt 恢复后，通常会出现： ```text RunStartEvent（复用原 runId） -> HITLDecidedEvent+ -> ActionStartEvent -> ToolCallStartEvent -> ToolCallEndEvent -> ActionEndEvent -> ... -> RunEndEvent ``` 单个敏感调用被拒绝时可能直接结束，因此收到 `HITLDecidedEvent` 后不能强制期待 Action 或工具事件。 ### 6、TeamAgent 事件 ```text TeamStartEvent（父 Team runId） -> SupervisorDeltaEvent* -> NodeStartEvent（父 Team runId，携带 Node） -> 成员 Agent 的 Start、Reason、Action、Tool 等过程事件*（成员 runId） -> 不产生成员自己的顶层 End 事件 -> NodeEndEvent（父 Team runId，携带 Node） -> ... -> TeamEndEvent（父 Team runId） -> onComplete ``` Team Flow 直接调用成员的 `call(...)`，不会经过成员 Request 的 `stream()` 包装器。因此成员可能产生开始和过程事件，但不会产生自己的 `SimpleEndEvent`、`RunEndEvent` 或 `TeamEndEvent`。成员边界使用父 Team 的 `NodeStartEvent` / `NodeEndEvent`，团队最终结果从 `TeamEndEvent` 读取。 并行节点下，多个节点及成员事件可能交错。成员过程事件不携带父 Node 或 `parentRunId`，不能用“最近收到的 NodeStartEvent”推断归属；需要严格审计时，应在外围上下文或传输 DTO 中补充关联标识。 ### 7、取得完整终态响应 ```java ReActResponse response = agent.prompt(query).stream() .ofType(RunEndEvent.class) .map(RunEndEvent::getResponse) .blockFirst(); ``` 异步代码可以保留为 `Mono`： ```java Mono response = agent.prompt(query).stream() .ofType(RunEndEvent.class) .map(RunEndEvent::getResponse) .next(); ``` 不要对未过滤的事件流直接调用 `blockFirst()`，否则得到的是开始事件。主动取消或流以 `onError` 结束时不会补发顶层 End 事件。 ### 8、失败与取消 Agent 当前没有统一的 `ErrorEvent` 或 `CancelEvent`： - 未处理异常通过 Reactor `onError` 结束，不补发顶层 End 事件； - 部分 ReAct 故障会转换为 `RunEndEvent(isAbnormal=true)` 后正常完成； - 主动取消属于 Reactor cancel，不是 `onError`，也不会补发 End 事件； - `doOnError` 只观察错误；需要恢复时使用 `onErrorResume`，或在 `subscribe` 中提供错误消费者。 事件、Session 和 Trace 都是运行态对象。通过 SSE、WebSocket 或消息队列传输时，应投影为自己的稳定 DTO，不要直接序列化整个事件对象。