diff --git "a/documents/zh/\346\265\201\345\274\217\350\276\223\345\207\272\350\247\204\346\240\274.md" "b/documents/zh/\346\265\201\345\274\217\350\276\223\345\207\272\350\247\204\346\240\274.md" new file mode 100644 index 000000000..5ff98a96d --- /dev/null +++ "b/documents/zh/\346\265\201\345\274\217\350\276\223\345\207\272\350\247\204\346\240\274.md" @@ -0,0 +1,303 @@ +# 流式输出规格 + +> 适用范围：`agent-core-java` 的 stream 路径（`DeepAgent.stream` / `ReActAgent.invokeForStream`）。本文档定义 stream 中所有 `OutputSchema` chunk 的 type、payload 结构与字段语义，供下游消费者（前端 / SSE 网关 / 测试）统一识别。 + +--- + +## 1. 顶层结构 + +每个流式 chunk 都是一个 `OutputSchema` 对象，三个固定字段： + +```json +{ + "type": "<事件类型>", + "index": , + "payload": { ... } +} +``` + +| 字段 | 类型 | 说明 | +|------|------|------| +| `type` | `string` | 事件类型，决定 `payload` 的结构（见第 2 节） | +| `index` | `int` | 同类型事件的序号，从 0 起；下游按 `(type, index)` 保序重组 | +| `payload` | `object` | 类型专属载荷 | + +--- + +## 2. 事件类型总览 + +### 2.1 任务生命周期 + +#### `task_output`（任务启动） + +**来源**：`TaskScheduler.executeTask`（任务真正调用前） + +```json +{ + "type": "task_output", + "index": 0, + "payload": { + "task_id": "deep_agent_task__", + "task_type": "deep_agent_task", + "description": "搜索:Topic-A" + } +} +``` + +| 字段 | 类型 | 说明 | +|------|------|------| +| `task_id` | `string` | 任务唯一标识，命名规则 `deep_agent_task__`，源于 `DeepAgent.executeCoreLoopRound` | +| `task_type` | `string` | 任务类型标签（如 `deep_agent_task`） | +| `description` | `string` | 任务描述（截断 120 字符 + `...`） | + +> 每个任务只发一次，出现在该任务所有 `tool_output` / `llm_output` 之前。同一会话连续派生多个任务时，`task_id` 的 `` 部分递增。 + +--- + +### 2.2 LLM 输出 + +#### `llm_output`（LLM 内容 / 工具调用决策） + +**来源**：`ReActAgent.writeAssistantStreamChunk` + +```json +// 形态 A：文本内容 +{ + "type": "llm_output", + "index": 0, + "payload": { + "task_id": "deep_agent_task__", + "content": "根据搜索结果，", + "result_type": "answer" + } +} + +// 形态 B：工具调用决策 +{ + "type": "llm_output", + "index": 0, + "payload": { + "task_id": "deep_agent_task__", + "tool_calls": [ + { + "id": "call_1_703f5cfd", + "type": "function", + "name": "web_search", + "arguments": "{\"query\":\"Topic-A\"}", + "index": 0 + } + ] + } +} +``` + +| 字段 | 类型 | 说明 | +|------|------|------| +| `task_id` | `string` | 从 `session.state.task_id` 读取；standalone ReActAgent 调用时为空字符串 | +| `content` | `string` | LLM 文本片段（形态 A） | +| `tool_calls` | `array` | LLM 决策的工具调用列表（形态 B） | +| `result_type` | `string` | 仅形态 A 携带，固定 `"answer"` | + +#### `llm_reasoning`（LLM 推理过程 / 思维链） + +```json +{ + "type": "llm_reasoning", + "index": 0, + "payload": { + "task_id": "...", + "content": "<思维链内容>", + "result_type": "answer" + } +} +``` + +#### `llm_usage`（LLM token 用量统计） + +```json +{ + "type": "llm_usage", + "index": 0, + "payload": { + "task_id": "...", + "usage_metadata": { ... }, + "result_type": "answer" + } +} +``` + +--- + +### 2.3 工具输出 + +#### `tool_output`（工具流式片段 / 非流式结果 / 工具错误） + +**来源**：`AbilityManager.buildToolOutputChunk`，由 `executeStreamingTool` / `streamSingleToolCall` / `executePreparedToolCallWithStreaming` 发出。**三种场景共用同一 type**，下游统一消费 `payload.content`。 + +```json +{ + "type": "tool_output", + "index": 0, + "payload": { + "task_id": "deep_agent_task__", + "tool_name": "web_search", + "tool_call_id": "call_1_703f5cfd", + "content": "正在搜索 'Topic-A'...\n" + } +} +``` + +| 字段 | 类型 | 说明 | +|------|------|------| +| `task_id` | `string` | 从 `session.state.task_id` 读取；无 task 概念时为空字符串 | +| `tool_name` | `string` | 工具名（来自 `ToolCall.name`） | +| `tool_call_id` | `string` | 工具调用 id（来自 `ToolCall.id`），与 `llm_output.tool_calls[].id` 对应 | +| `content` | `any` | 工具返回的片段内容；流式工具逐 chunk 产出，非流式工具一次性产出完整结果，错误场景产出 `"tool error: "` | + +| 场景 | `index` 语义 | `content` 语义 | +|------|--------------|----------------| +| 流式工具 chunk | per-tool chunkIndex，从 0 递增 | 单个片段 | +| 流式工具回退 invoke | 0 | 完整结果 | +| 非流式工具 invoke | 0 | 完整结果 | +| 工具异常 | 0 | `"tool error: "` | + + + +--- + +### 2.4 Agent 终结输出 + +#### `answer`（Agent 最终答案） + +**来源**：`ReActAgent.writeStreamAnswer` / `DeepAgent.collectStreamToResult` / `ReActAgentEvolve` / `LlmEventHandler` + +ReAct 循环或 Deep 任务循环正常结束后发出，payload 含最终答案。该事件是会话/任务级的终结帧，下游收到后即可结束本轮渲染。 + +```json +{ + "type": "answer", + "index": 0, + "payload": { + "output": "<最终答案对象>", + "result_type": "answer" + } +} +``` + +| 字段 | 类型 | 说明 | +|------|------|------| +| `output` | `any` | 最终答案内容（通常为 string，也可能是结构化对象） | +| `result_type` | `string` | 固定 `"answer"`，与 `error` 的 `"error"` 区分 | + +> DeepAgent 多轮场景下，`index` 对应 round 序号；ReActAgent 单轮场景下 `index=0`。 + +#### `error`（Agent 执行错误） + +**来源**：`ReActAgent.writeStreamError` / `DeepAgent.writeStreamError` / `DeepAgent.collectStreamToResult`（轮次错误）/ `ReActAgentEvolve` + +ReAct 循环或 Deep 任务循环异常终止时发出，payload 含错误描述。该事件是会话/任务级的终结帧，下游收到后应停止本轮渲染并展示错误。 + +```json +{ + "type": "error", + "index": 0, + "payload": { + "output": "<错误消息>", + "result_type": "error" + } +} +``` + +| 字段 | 类型 | 说明 | +|------|------|------| +| `output` | `string` | 错误描述；`Throwable.getMessage()` 为 null 时回退到异常类名 | +| `result_type` | `string` | 固定 `"error"` | + +> DeepAgent 轮次错误场景下，`index` 对应出错轮次序号（`rounds.size()`）。 + +> **answer 与 error 的对称性**：两者 payload 结构相同（`output` + `result_type`），`result_type` 区分正常终结与异常终结。下游可用同一渲染逻辑处理 `output`，仅在 `result_type=error` 时切换错误样式。 + +--- + +## 3. task_id 的传播链路 + +``` +DeepAgent.executeCoreLoopRound L1643 + 生成 "deep_agent_task__" + ↓ 写入 InputEvent.metadata.task_id +TaskLoopEventHandler.handleInput L132 + 读取 metadata.task_id（fallback UUID） + ↓ new Task(sessionId, taskId, ...) → taskManager.addTask +TaskScheduler.executeTask L246 + task.getTaskId() + ↓ 透传给 executor.executeAbility(taskId, session) +CoreTaskLoopEventExecutor.buildEffectiveInputs L236 + effective.put("task_id", taskId) + ↓ 透传给 DeepAgent.invokeInnerRoundStreaming(effectiveInputs, ...) +DeepAgent.invokeInnerRoundStreaming L1756 + 从 effectiveInputs.get("task_id") 读取（不重新生成） + ↓ innerSession.updateState(Map.of("task_id", taskId)) +AbilityManager.resolveTaskId / ReActAgent.writeAssistantStreamChunk + 从 session.getState("task_id") 读取 + ↓ 写入 tool_output / llm_output payload.task_id +``` + +**关键约束**： +- `invokeInnerRoundStreaming` **不重新生成** task_id，复用上游注入的值 +- `copySessionState` 前后各注入一次，确保 inner session state 不被外层覆盖 +- standalone ReActAgent 调用（不经 DeepAgent）时 task_id 为空字符串，字段始终存在 + +--- + +## 5. 下游消费建议 + +### 5.1 按 type 分发 + +```python +for chunk in stream: + t = chunk["type"] + if t == "task_output": + on_task_started(chunk["payload"]) + elif t == "llm_output": + p = chunk["payload"] + if "tool_calls" in p: + on_llm_tool_calls(p["tool_calls"]) + else: + on_llm_content(p["content"]) + elif t == "tool_output": + on_tool_output(chunk["payload"]) + elif t == "llm_reasoning": + on_llm_reasoning(chunk["payload"]["content"]) + elif t == "llm_usage": + on_llm_usage(chunk["payload"]["usage_metadata"]) +``` + +### 5.2 按 task_id 聚合 + +同一 `task_id` 下的所有 `tool_output` / `llm_output` 属于同一任务。会话级聚合用 `task_id` 的 `` 部分提取会话标识。 + +### 5.3 工具结果重组 + +`tool_output` 的 `index` 是 per-tool chunkIndex。重组完整工具结果时，按 `(tool_call_id, index)` 排序拼接 `content`： + +```python +from collections import defaultdict +tool_chunks = defaultdict(list) +for chunk in stream: + if chunk["type"] == "tool_output": + p = chunk["payload"] + tool_chunks[p["tool_call_id"]].append((chunk["index"], p["content"])) + +for call_id, chunks in tool_chunks.items(): + chunks.sort(key=lambda x: x[0]) + full_result = "".join(str(c) for _, c in chunks) +``` + +--- + +## 6. 变更历史 + +| 日期 | 变更 | +|------|------| +| 2026-08-12 | 初始版本：定义 `task_output` / `llm_output` / `llm_reasoning` / `llm_usage` / `tool_output` 五种 type；移除 `session_id` 字段；移除 Hermes 风格 replay cursor 字段 | +| 2026-08-13 | 补充 `answer` / `error` 两种 Agent 终结输出 type 规格（payload 含 `output` + `result_type`，结构对称） | diff --git a/examples/deep_agent/DeepAgentToolCallStreamExample.java b/examples/deep_agent/DeepAgentToolCallStreamExample.java new file mode 100644 index 000000000..0ae032c09 --- /dev/null +++ b/examples/deep_agent/DeepAgentToolCallStreamExample.java @@ -0,0 +1,768 @@ +/* + * Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved. + */ + +package examples.deep_agent; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; + +import com.openjiuwen.core.foundation.llm.Model; +import com.openjiuwen.core.foundation.llm.model_clients.BaseModelClient; +import com.openjiuwen.core.foundation.llm.output_parsers.BaseOutputParser; +import com.openjiuwen.core.foundation.llm.schema.AssistantMessage; +import com.openjiuwen.core.foundation.llm.schema.AssistantMessageChunk; +import com.openjiuwen.core.foundation.llm.schema.AudioGenerationResponse; +import com.openjiuwen.core.foundation.llm.schema.ImageGenerationResponse; +import com.openjiuwen.core.foundation.llm.schema.ModelClientConfig; +import com.openjiuwen.core.foundation.llm.schema.ModelRequestConfig; +import com.openjiuwen.core.foundation.llm.schema.ToolCall; +import com.openjiuwen.core.foundation.llm.schema.UserMessage; +import com.openjiuwen.core.foundation.llm.schema.VideoGenerationResponse; +import com.openjiuwen.core.foundation.tool.Tool; +import com.openjiuwen.core.foundation.tool.ToolCard; +import com.openjiuwen.core.foundation.tool.function.LocalFunction; +import com.openjiuwen.core.runner.Runner; +import com.openjiuwen.core.runner.RunnerConfig; +import com.openjiuwen.core.session.checkpointer.CheckpointerFactory; +import com.openjiuwen.core.session.checkpointer.InMemoryCheckpointer; +import com.openjiuwen.core.session.stream.OutputSchema; +import com.openjiuwen.core.singleagent.rail.AgentCallbackContext; +import com.openjiuwen.core.singleagent.rail.AgentRail; +import com.openjiuwen.core.singleagent.rail.ToolCallInputs; +import com.openjiuwen.core.singleagent.schema.AgentCard; +import com.openjiuwen.harness.deep_agent.DeepAgent; +import com.openjiuwen.harness.factory.HarnessFactory; +import com.openjiuwen.harness.schema.config.DeepAgentConfig; +import com.openjiuwen.harness.workspace.Workspace; + +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.Iterator; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.atomic.AtomicInteger; + +/** + * Demonstrates DeepAgent tool-call behavior in non-streaming ({@code invoke}) and + * streaming ({@code stream}) modes. No concurrency — just two sequential calls that + * each drive the agent to call a registered {@code web_search} tool, then inspect + * the result / stream to see whether tool-call information is surfaced to the caller. + * + *

A deterministic fake LLM is used so the example runs without any real + * model API key or network dependency. The fake model follows a simple protocol: + *

    + *
  • user message starting with {@code 搜索:X} → returns a {@code web_search} + * tool call with {@code {"query":"X"}};
    • + *
  • tool result message → returns a final stop answer summarizing the result.
    • + *
+ * + *

The example answers two questions: + *

    + *
  1. Non-streaming invoke: does the returned result map contain tool-call + * information (tool_calls, messages, stream_chunks, etc.)?
    2. + *
  2. Streaming: does the stream contain the tool's output stream — i.e. + * {@code llm_output} chunks carrying {@code tool_calls}?
    3. + *
+ * + *

A {@link ToolCallLoggerRail} is registered to independently confirm that the + * tool was actually called in both modes, even when the result / stream does not + * expose tool-call details. + * + *

Run (from the agent-core-java repo root): + * 

+ * mvn -DskipTests compile
+ * mvn dependency:copy-dependencies "-DoutputDirectory=target/dependency" "-DincludeScope=test" -q
+ * javac -encoding UTF-8 -source 17 -target 17 -cp "target/classes;target/dependency/*" \
+ *   -d examples/deep_agent/build examples/deep_agent/DeepAgentToolCallStreamExample.java
+ * java "-Dfile.encoding=UTF-8" -cp "examples/deep_agent/build;target/classes;target/dependency/*" \
+ *   examples.deep_agent.DeepAgentToolCallStreamExample
+ *
+ * + * @since 0.1.14 + */ +public final class DeepAgentToolCallStreamExample { + private static final ObjectMapper MAPPER = new ObjectMapper(); + + private static final String WEB_SEARCH_TOOL = "web_search"; + + private DeepAgentToolCallStreamExample() { + } + + /** + * Entry point. + * + * @param args unused + * @throws Exception if the example fails + */ + public static void main(String[] args) throws Exception { + System.out.println("=== DeepAgent 流式工具调用示例（同一会话派生 2 个任务，每个任务 2 次工具调用）==="); + + Path workspacePath = Files.createTempDirectory("deep-agent-toolcall-"); + System.out.println("[setup] workspace=" + workspacePath); + + CheckpointerFactory.setDefaultCheckpointer(new InMemoryCheckpointer()); + Runner.setConfig(RunnerConfig.DEFAULT); + Runner.start(); + + FakeToolModelClient.ensureFactoryRegistered(); + Model fakeModel = FakeToolModelClient.newModel(); + + List toolCallLog = new CopyOnWriteArrayList<>(); + DeepAgent agent = buildAgent(workspacePath, fakeModel, toolCallLog); + + // 同一会话 conversation_id，连续派生 2 个任务（每次 stream 调用 -> 新 handlerRound -> 新 task_id） + String conversationId = "session_multi_task"; + try { + // --- 任务 1 --- + System.out.println(); + System.out.println("========== [任务1] stream（query=搜索:Topic-A，预期 2 次工具调用）=========="); + Map task1Inputs = new LinkedHashMap<>(); + task1Inputs.put("query", "搜索:Topic-A"); + task1Inputs.put("conversation_id", conversationId); + int logStart1 = toolCallLog.size(); + Iterator stream1 = agent.stream(task1Inputs); + inspectStreamResult(stream1, toolCallLog, logStart1, "任务1"); + + // --- 任务 2（同一会话，handlerRound 递增 -> 不同 task_id）--- + System.out.println(); + System.out.println("========== [任务2] stream（query=搜索:Topic-B，预期 2 次工具调用）=========="); + Map task2Inputs = new LinkedHashMap<>(); + task2Inputs.put("query", "搜索:Topic-B"); + task2Inputs.put("conversation_id", conversationId); + int logStart2 = toolCallLog.size(); + Iterator stream2 = agent.stream(task2Inputs); + inspectStreamResult(stream2, toolCallLog, logStart2, "任务2"); + } finally { + try { + agent.close(); + } catch (Exception ignored) { + // best-effort + } + try { + CheckpointerFactory.setDefaultCheckpointer(null); + } catch (Exception ignored) { + // best-effort + } + Runner.stop(); + } + + System.out.println(); + System.out.println("=== 示例结束 ==="); + } + + private static DeepAgent buildAgent(Path workspacePath, Model fakeModel, + List toolCallLog) { + Map modelMap = new LinkedHashMap<>(); + modelMap.put("model", "fake-tool-model"); + modelMap.put("temperature", 0.0); + modelMap.put("max_tokens", 128); + + DeepAgentConfig config = DeepAgentConfig.builder() + .enableTaskLoop(true) + .enableTaskPlanning(false) + .enableTenantIsolation(false) + .restrictToWorkDir(false) + .systemPrompt("你是一个搜索助手。根据用户请求调用 web_search 工具，" + + "然后用一句话中文总结结果。") + .maxIterations(8) + .completionTimeout(120.0) + .language("cn") + .model(modelMap) + .workspacePath(workspacePath.toString()) + .build(); + + AgentCard card = AgentCard.builder() + .name("tool_call_stream_agent") + .description("DeepAgent 流式/非流式工具调用示例").build(); + Workspace ws = Workspace.builder().rootPath(workspacePath.toString()).language("cn").build(); + DeepAgent agent = HarnessFactory.createDeepAgent(card, config, ws); + agent.getAgent().setLlm(fakeModel); + + agent.registerHarnessTool(buildWebSearchTool("web_search_tool")); + agent.getAgent().registerRail(new ToolCallLoggerRail(toolCallLog)); + agent.ensureInitialized(); + return agent; + } + + private static Tool buildWebSearchTool(String toolId) { + ToolCard card = ToolCard.builder() + .id(toolId).name(WEB_SEARCH_TOOL) + .description("在互联网上搜索信息（模拟流式 HTTP 调用，返回多片段）").build(); + // 注意：被包装的函数返回 StreamingSearchResult（实现了 Iterable）。 + // 这使得 LocalFunction.stream() 会把它当作流式结果返回多个片段； + // 而 LocalFunction.invoke() 也会把同一个对象返回（toString() 为完整拼接字符串）。 + return new LocalFunction(card, inputs -> { + String query = readString(inputs.get("query")); + System.out.println(" -> [web_search] 正在执行流式搜索 query='" + query + "' ..."); + try { + Thread.sleep(200); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + List chunks = new ArrayList<>(); + chunks.add("正在搜索 '" + query + "'...\n"); + chunks.add("找到 3 条相关结果:\n"); + chunks.add("web_search 结果: 关于 '" + query + "' 检索到 3 条摘要"); + return new StreamingSearchResult(chunks); + }); + } + + // ---------- Non-streaming result inspection ---------- + + private static void inspectInvokeResult(Map result, + List toolCallLog, int logStart) { + System.out.println("--- 非流式 invoke 结果分析 ---"); + + System.out.println("原始结果 (JSON):"); + printJsonIndented(result, " "); + + System.out.println("顶层 keys: " + result.keySet()); + + String[] toolRelatedKeys = {"tool_calls", "messages", "stream_chunks", + "tool_results", "tool_call_ids"}; + System.out.println("工具相关字段检查:"); + for (String key : toolRelatedKeys) { + Object value = result.get(key); + System.out.println(" " + key + ": " + + (value != null ? "存在 (" + truncate(String.valueOf(value), 80) + ")" : "不存在")); + } + + Object output = result.get("output"); + System.out.println("output: " + truncate(String.valueOf(output), 200)); + + Object roundsObj = result.get("rounds"); + if (roundsObj instanceof List rounds) { + System.out.println("rounds 数量: " + rounds.size()); + for (int i = 0; i < rounds.size(); i++) { + if (rounds.get(i) instanceof Map round) { + System.out.println(" round[" + i + "] keys: " + round.keySet()); + System.out.println(" round[" + i + "] output: " + + truncate(String.valueOf(round.get("output")), 120)); + System.out.println(" round[" + i + "] tool_calls: " + + (round.get("tool_calls") != null ? "存在" : "不存在")); + System.out.println(" round[" + i + "] messages: " + + (round.get("messages") != null ? "存在" : "不存在")); + System.out.println(" round[" + i + "] stream_chunks: " + + (round.get("stream_chunks") != null ? "存在" : "不存在")); + } + } + } + + Object finalResult = result.get("final_result"); + if (finalResult instanceof Map fr) { + System.out.println("final_result keys: " + fr.keySet()); + } + + System.out.println("Rail 记录的工具调用:"); + for (int i = logStart; i < toolCallLog.size(); i++) { + ToolCallRecord r = toolCallLog.get(i); + System.out.println(" [" + r.phase() + "] tool=" + r.tool() + " id=" + r.toolCallId() + + " args=" + truncate(r.args(), 60) + + ("after".equals(r.phase()) ? " result=" + truncate(r.result(), 60) : "")); + } + + boolean hasToolCallsInResult = result.get("tool_calls") != null; + boolean hasToolCallsInRounds = false; + if (roundsObj instanceof List rounds) { + for (Object r : rounds) { + if (r instanceof Map round && round.get("tool_calls") != null) { + hasToolCallsInRounds = true; + break; + } + } + } + boolean toolWasCalled = toolCallLog.size() > logStart; + + System.out.println(); + System.out.println("结论:"); + System.out.println(" 工具是否被调用: " + (toolWasCalled ? "是（Rail 记录了调用）" : "否")); + System.out.println(" 结果顶层是否含 tool_calls: " + (hasToolCallsInResult ? "是" : "否")); + System.out.println(" rounds 是否含 tool_calls: " + (hasToolCallsInRounds ? "是" : "否")); + System.out.println(" => 非流式 invoke 结果" + + (hasToolCallsInResult || hasToolCallsInRounds ? "包含" : "不包含") + + "工具调用信息（仅返回最终回答，中间工具调用不外露）"); + } + + // ---------- Streaming result inspection ---------- + + private static void inspectStreamResult(Iterator stream, + List toolCallLog, int logStart, String taskLabel) { + System.out.println("--- [" + taskLabel + "] 流式 stream 结果分析 ---"); + + List items = new ArrayList<>(); + while (stream.hasNext()) { + items.add(stream.next()); + } + + System.out.println("[" + taskLabel + "] 流中 item 数量: " + items.size()); + + boolean hasToolCallChunk = false; + boolean hasContentChunk = false; + boolean hasAnswerChunk = false; + int toolOutputChunkCount = 0; + int toolCallDecisionCount = 0; + String firstSessionId = null; + String firstTaskId = null; + // 每个 tool_call_id 对应一次工具调用；统计不同的 tool_call_id 数量 + java.util.Set distinctToolCallIds = new java.util.LinkedHashSet<>(); + + for (int i = 0; i < items.size(); i++) { + Object item = items.get(i); + System.out.println(" [" + taskLabel + "] chunk[" + i + "] 原始 (JSON):"); + printJsonIndented(item, " "); + + if (!(item instanceof OutputSchema os)) { + continue; + } + String type = os.getType(); + Object payload = os.getPayload(); + + if ("llm_output".equals(type) && payload instanceof Map pm) { + if (pm.get("tool_calls") != null) { + hasToolCallChunk = true; + Object tcs = pm.get("tool_calls"); + if (tcs instanceof List list) { + toolCallDecisionCount += list.size(); + for (Object tc : list) { + if (tc instanceof Map tcMap && tcMap.get("id") != null) { + distinctToolCallIds.add(String.valueOf(tcMap.get("id"))); + } + } + } + } + if (pm.get("content") != null) { + hasContentChunk = true; + } + } else if ("tool_output".equals(type) && payload instanceof Map pm) { + toolOutputChunkCount++; + if (firstSessionId == null && pm.get("session_id") != null) { + firstSessionId = String.valueOf(pm.get("session_id")); + } + if (firstTaskId == null && pm.get("task_id") != null) { + firstTaskId = String.valueOf(pm.get("task_id")); + } + if (pm.get("tool_call_id") != null) { + distinctToolCallIds.add(String.valueOf(pm.get("tool_call_id"))); + } + System.out.println(" -> [" + taskLabel + "][tool_output] session_id=" + pm.get("session_id") + + " task_id=" + pm.get("task_id") + + " tool_name=" + pm.get("tool_name") + + " tool_call_id=" + pm.get("tool_call_id") + + " index=" + os.getIndex() + + " content=" + truncate(String.valueOf(pm.get("content")), 80)); + } + if ("answer".equals(type)) { + hasAnswerChunk = true; + } + } + + System.out.println("[" + taskLabel + "] Rail 记录的工具调用:"); + for (int i = logStart; i < toolCallLog.size(); i++) { + ToolCallRecord r = toolCallLog.get(i); + System.out.println(" [" + r.phase() + "] tool=" + r.tool() + " id=" + r.toolCallId() + + " args=" + truncate(r.args(), 60) + + ("after".equals(r.phase()) ? " result=" + truncate(r.result(), 60) : "")); + } + + boolean toolWasCalled = toolCallLog.size() > logStart; + int railAfterCount = 0; + for (int i = logStart; i < toolCallLog.size(); i++) { + if ("after".equals(toolCallLog.get(i).phase())) { + railAfterCount++; + } + } + + System.out.println(); + System.out.println("[" + taskLabel + "] 结论:"); + System.out.println(" 工具是否被调用: " + (toolWasCalled ? "是（Rail 记录了调用）" : "否")); + System.out.println(" 流中是否含 tool_calls chunk (type=llm_output, LLM 决策): " + + (hasToolCallChunk ? "是" : "否")); + System.out.println(" LLM 决策的 tool_calls 数量: " + toolCallDecisionCount); + System.out.println(" 流中是否含 content chunk (type=llm_output, LLM 回答): " + + (hasContentChunk ? "是" : "否")); + System.out.println(" 流中是否含 answer chunk: " + (hasAnswerChunk ? "是" : "否")); + System.out.println(" tool_output chunk 数量: " + toolOutputChunkCount); + System.out.println(" Rail after(工具完成) 记录数: " + railAfterCount); + System.out.println(" 不同 tool_call_id 数量: " + distinctToolCallIds.size()); + if (!distinctToolCallIds.isEmpty()) { + System.out.println(" tool_call_id 列表: " + distinctToolCallIds); + } + System.out.println(" 首个 session_id: " + firstSessionId); + System.out.println(" 首个 task_id: " + firstTaskId); + System.out.println(" => [" + taskLabel + "] 流式 stream " + + (hasToolCallChunk ? "包含" : "不包含") + " LLM 工具调用决策信息；" + + (toolOutputChunkCount > 0 ? "包含" : "不包含") + " 工具响应流信息" + + (toolOutputChunkCount > 0 + ? "（统一 tool_output 类型，payload 含 " + + "session_id/task_id/tool_name/tool_call_id/content，index=chunkIndex）" + : "（ReActAgent stream 模式调用 tool.invoke()，不转发 tool.stream()）")); + } + + // ---------- Helpers ---------- + + private static String mapToCompactString(Map map) { + if (map == null) { + return "null"; + } + try { + return MAPPER.writeValueAsString(map); + } catch (Exception e) { + return map.toString(); + } + } + + /** + * 把任意对象序列化为 pretty JSON 字符串；无法序列化时回退到 {@code toString()}。 + */ + private static String toJsonString(Object obj) { + if (obj == null) { + return "null"; + } + try { + return MAPPER.writerWithDefaultPrettyPrinter().writeValueAsString(obj); + } catch (Exception e) { + try { + return MAPPER.writeValueAsString(MAPPER.valueToTree(obj)); + } catch (Exception e2) { + return String.valueOf(obj); + } + } + } + + /** + * 按 JSON pretty 格式打印对象，每行加 {@code indent} 前缀。 + */ + private static void printJsonIndented(Object obj, String indent) { + String json = toJsonString(obj); + for (String line : json.split("\n", -1)) { + System.out.println(indent + line); + } + } + + private static String truncate(String s, int max) { + if (s == null) { + return ""; + } + String one = s.replace('\n', ' ').replace('\r', ' '); + return one.length() <= max ? one : one.substring(0, max) + "..."; + } + + private static String readString(Object value) { + return value == null ? "" : String.valueOf(value); + } + + private record ToolCallRecord(String phase, String tool, String toolCallId, + String args, String result) { + } + + /** + * 流式搜索结果：既可作为 {@link Iterable} 供 {@link LocalFunction#stream} 迭代输出多个片段， + * 又重写了 {@code toString()} 以便在非流式 invoke 模式下作为 ToolMessage content 时是可读的完整字符串。 + */ + static final class StreamingSearchResult implements Iterable { + private final List chunks; + + StreamingSearchResult(List chunks) { + this.chunks = chunks; + } + + @Override + public Iterator iterator() { + return chunks.iterator(); + } + + public int size() { + return chunks.size(); + } + + public List chunks() { + return chunks; + } + + @Override + public String toString() { + return String.join("", chunks); + } + } + + /** + * Rail that logs every tool call (before/after) into a thread-safe sink. + */ + private static final class ToolCallLoggerRail extends AgentRail { + private final List sink; + + ToolCallLoggerRail(List sink) { + this.sink = sink; + } + + @Override + public int getPriority() { + return 10; + } + + @Override + public void beforeToolCall(AgentCallbackContext ctx) { + record(ctx, "before"); + } + + @Override + public void afterToolCall(AgentCallbackContext ctx) { + record(ctx, "after"); + } + + private void record(AgentCallbackContext ctx, String phase) { + String tool = "?"; + String args = ""; + String toolCallId = "?"; + String resultSnippet = ""; + String resultKind = ""; + if (ctx.getInputs() instanceof ToolCallInputs inputs) { + tool = inputs.getToolName() == null ? "?" : inputs.getToolName(); + args = String.valueOf(inputs.getToolArgs()); + if (inputs.getToolCall() != null && inputs.getToolCall().getId() != null) { + toolCallId = inputs.getToolCall().getId(); + } + if ("after".equals(phase) && inputs.getToolResult() != null) { + Object res = inputs.getToolResult(); + if (res instanceof StreamingSearchResult ssr) { + resultKind = "[流式结果 " + ssr.size() + " 片段] "; + resultSnippet = truncate(String.valueOf(res), 60); + } else if (res instanceof Iterable iter) { + List list = new ArrayList<>(); + iter.forEach(list::add); + resultKind = "[流式结果 " + list.size() + " 片段] "; + resultSnippet = truncate(String.valueOf(res), 60); + } else { + resultKind = "[普通结果] "; + resultSnippet = truncate(String.valueOf(res), 60); + } + } + } + sink.add(new ToolCallRecord(phase, tool, toolCallId, + truncate(args, 80), resultKind + resultSnippet)); + System.out.println(" [TOOL][" + phase + "] tool=" + tool + " id=" + toolCallId + + ("after".equals(phase) ? " result=" + resultKind + resultSnippet : "") + + " args=" + truncate(args, 80)); + } + } + + /** + * Deterministic fake LLM that needs no API key or network. + *

It inspects the last message role: + *

    + *
  • user message starting with {@code 搜索:X} → returns a {@code web_search} + * tool call with {@code {"query":"X"}};
    • + *
  • tool result message → returns a final stop answer.
    • + *
+ */ + static final class FakeToolModelClient extends BaseModelClient { + private static final String PROVIDER = "fake-tool-stream"; + private static volatile boolean factoryRegistered = false; + private final AtomicInteger callCounter = new AtomicInteger(0); + + FakeToolModelClient(ModelRequestConfig modelConfig, ModelClientConfig modelClientConfig) { + super(modelConfig, modelClientConfig); + } + + static void ensureFactoryRegistered() { + if (!factoryRegistered) { + synchronized (FakeToolModelClient.class) { + if (!factoryRegistered) { + Model.registerFactory(new Model.ModelClientFactory() { + @Override + public String providerName() { + return PROVIDER; + } + + @Override + public BaseModelClient create(ModelRequestConfig mc, ModelClientConfig cc) { + return new FakeToolModelClient(mc, cc); + } + }); + factoryRegistered = true; + } + } + } + } + + static Model newModel() { + ModelClientConfig clientConfig = ModelClientConfig.builder() + .clientProvider(PROVIDER).clientId("fake-tool-stream-client") + .apiKey("fake-key").apiBase("http://fake-base").timeout(60.0).build(); + return new Model(clientConfig, null); + } + + @Override + protected void validateConfig() { + // no-op: this fake client does not require real api_key / api_base + } + + @Override + public AssistantMessage invoke(Object messages, Object tools, Float temperature, Float topP, + String model, Integer maxTokens, String stop, BaseOutputParser outputParser, + Float timeout, Map kwargs) { + return buildResponse(messages); + } + + @Override + public Iterator stream(Object messages, Object tools, Float temperature, + Float topP, String model, Integer maxTokens, String stop, BaseOutputParser outputParser, + Float timeout, Map kwargs) { + AssistantMessage msg = buildResponse(messages); + AssistantMessageChunk chunk = AssistantMessageChunk.builder() + .content(msg.getContent()).toolCalls(msg.getToolCalls()) + .finishReason(msg.getFinishReason()).build(); + return List.of(chunk).iterator(); + } + + @Override + public ImageGenerationResponse generateImage(List messages, String model, String size, + String negativePrompt, int n, boolean promptExtend, boolean watermark, int seed, + Map kwargs) { + throw new UnsupportedOperationException("FakeToolModelClient does not support generateImage"); + } + + @Override + public AudioGenerationResponse generateSpeech(List messages, String model, String voice, + String languageType, Map kwargs) { + throw new UnsupportedOperationException("FakeToolModelClient does not support generateSpeech"); + } + + @Override + public VideoGenerationResponse generateVideo(List messages, String imgUrl, + String audioUrl, String model, String size, String resolution, int duration, + boolean promptExtend, boolean watermark, String negativePrompt, Integer seed, + Map kwargs) { + throw new UnsupportedOperationException("FakeToolModelClient does not support generateVideo"); + } + + private AssistantMessage buildResponse(Object messages) { + JsonNode tree; + try { + tree = MAPPER.valueToTree(messages); + } catch (Exception ignored) { + tree = null; + } + if (tree == null || !tree.isArray() || tree.isEmpty()) { + return AssistantMessage.builder() + .content("请使用 '搜索:X' 格式提出请求。") + .finishReason("stop").build(); + } + // 找到最后一个 user message 作为"本轮"起点，避免跨任务历史污染 + int lastUserIdx = -1; + for (int i = tree.size() - 1; i >= 0; i--) { + if ("user".equals(tree.get(i).path("role").asText(""))) { + lastUserIdx = i; + break; + } + } + if (lastUserIdx < 0) { + return AssistantMessage.builder() + .content("请使用 '搜索:X' 格式提出请求。") + .finishReason("stop").build(); + } + // 统计本轮（lastUserIdx 之后）assistant 发起的 tool_call 次数 + int assistantToolCallCount = 0; + for (int i = lastUserIdx + 1; i < tree.size(); i++) { + JsonNode node = tree.get(i); + if ("assistant".equals(node.path("role").asText(""))) { + JsonNode tcs = node.path("tool_calls"); + if (tcs.isArray() && tcs.size() > 0) { + assistantToolCallCount++; + } + } + } + JsonNode last = tree.get(tree.size() - 1); + String lastRole = last.path("role").asText(""); + String lastContent = readContent(last); + String userQuery = extractQuery(readContent(tree.get(lastUserIdx))); + + if ("user".equals(lastRole)) { + // 本轮开始，发起第 1 次工具调用 + return buildToolCallResponse(userQuery, 1); + } + if ("tool".equals(lastRole)) { + if (assistantToolCallCount < 2) { + // 本轮已发起 1 次工具调用，发起第 2 次（扩展搜索） + return buildToolCallResponse("扩展:" + userQuery, 2); + } + // 本轮已发起 2 次工具调用，给最终答案 + return buildFinalAnswer(lastContent); + } + return AssistantMessage.builder() + .content("请使用 '搜索:X' 格式提出请求。") + .finishReason("stop").build(); + } + + private static int countToolResults(JsonNode tree) { + int count = 0; + for (JsonNode node : tree) { + if ("tool".equals(node.path("role").asText(""))) { + count++; + } + } + return count; + } + + private static String extractQuery(String userContent) { + String query = userContent == null ? "" : userContent.trim(); + if (!query.startsWith("搜索:")) { + return ""; + } + return query.substring("搜索:".length()).trim(); + } + + private static String extractOriginalUserQuery(JsonNode tree) { + for (JsonNode node : tree) { + if ("user".equals(node.path("role").asText(""))) { + return extractQuery(readContent(node)); + } + } + return ""; + } + + private static String readContent(JsonNode msg) { + JsonNode contentNode = msg.path("content"); + if (contentNode.isTextual()) { + return contentNode.asText(""); + } + if (contentNode.isArray() && contentNode.size() > 0) { + return contentNode.get(0).path("text").asText(contentNode.get(0).asText("")); + } + return ""; + } + + private AssistantMessage buildToolCallResponse(String query, int callSeq) { + String q = query == null ? "" : query; + ToolCall toolCall = ToolCall.builder() + .id("call_" + callCounter.incrementAndGet() + "_" + + UUID.randomUUID().toString().substring(0, 8)) + .name(WEB_SEARCH_TOOL) + .arguments(String.format("{\"query\":\"%s\"}", escape(q))) + .index(0).build(); + return AssistantMessage.builder() + .content("").toolCalls(List.of(toolCall)) + .finishReason("tool_calls").build(); + } + + private static AssistantMessage buildFinalAnswer(String toolResult) { + String summary = toolResult == null ? "" : toolResult; + if (summary.length() > 120) { + summary = summary.substring(0, 120) + "..."; + } + return AssistantMessage.builder() + .content("已完成搜索。" + summary) + .finishReason("stop").build(); + } + + private static String escape(String s) { + return s == null ? "" : s.replace("\\", "\\\\") + .replace("\"", "\\\"").replace("\n", "\\n"); + } + } +} diff --git a/src/main/java/com/openjiuwen/core/controller/modules/TaskScheduler.java b/src/main/java/com/openjiuwen/core/controller/modules/TaskScheduler.java index e341fec65..13c11f305 100644 --- a/src/main/java/com/openjiuwen/core/controller/modules/TaskScheduler.java +++ b/src/main/java/com/openjiuwen/core/controller/modules/TaskScheduler.java @@ -21,9 +21,11 @@ import com.openjiuwen.core.controller.schema.TaskStatus; import com.openjiuwen.core.controller.schema.DataFrame; import com.openjiuwen.core.session.AgentSessionApi; +import com.openjiuwen.core.session.stream.OutputSchema; import java.util.ArrayList; import java.util.Iterator; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; @@ -246,6 +248,17 @@ private void executeTask(String taskId, AgentSessionApi session) { Task task = tasks.get(0); Loggers.CONTROLLER.info("Executing task {} (type: {})", taskId, task.getTaskType()); + // 任务调用前，把任务元信息写入输出流（与 tool_output 同 type，下游统一消费） + Map taskStartedPayload = new LinkedHashMap<>(); + taskStartedPayload.put("task_id", task.getTaskId()); + taskStartedPayload.put("task_type", task.getTaskType()); + String description = task.getDescription() == null ? "" : task.getDescription(); + if (description.length() > 120) { + description = description.substring(0, 120) + "..."; + } + taskStartedPayload.put("description", description); + session.writeStream(new OutputSchema("task_output", 0, taskStartedPayload)); + // 2. Create TaskExecutor TaskExecutorDependencies dependencies = new TaskExecutorDependencies(config, abilityManager, contextEngine, taskManager, eventQueue); diff --git a/src/main/java/com/openjiuwen/core/singleagent/AbilityManager.java b/src/main/java/com/openjiuwen/core/singleagent/AbilityManager.java index ebbd41d54..a5a46f0b9 100644 --- a/src/main/java/com/openjiuwen/core/singleagent/AbilityManager.java +++ b/src/main/java/com/openjiuwen/core/singleagent/AbilityManager.java @@ -4,8 +4,8 @@ package com.openjiuwen.core.singleagent; -import com.fasterxml.jackson.core.type.TypeReference; import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.core.type.TypeReference; import com.fasterxml.jackson.databind.ObjectMapper; import com.openjiuwen.core.common.concurrent.OpenJiuwenExecutors; import com.openjiuwen.core.common.exception.BaseError; @@ -13,17 +13,19 @@ import com.openjiuwen.core.common.logging.Loggers; import com.openjiuwen.core.foundation.llm.schema.ToolCall; import com.openjiuwen.core.foundation.llm.schema.ToolMessage; -import com.openjiuwen.core.session.AgentSessionApi; -import com.openjiuwen.core.session.SessionContextHolder; import com.openjiuwen.core.foundation.tool.Tool; -import com.openjiuwen.core.foundation.tool.mcp.McpServerConfig; import com.openjiuwen.core.foundation.tool.ToolCard; +import com.openjiuwen.core.foundation.tool.function.LocalFunction; +import com.openjiuwen.core.foundation.tool.mcp.McpServerConfig; import com.openjiuwen.core.foundation.tool.schema.ToolInfo; import com.openjiuwen.core.operator.tool_call.ToolExecutionResult; import com.openjiuwen.core.operator.tool_call.ToolRegistry; import com.openjiuwen.core.runner.Runner; import com.openjiuwen.core.runner.base.TagMatchStrategy; +import com.openjiuwen.core.session.AgentSessionApi; import com.openjiuwen.core.session.Session; +import com.openjiuwen.core.session.SessionContextHolder; +import com.openjiuwen.core.session.stream.OutputSchema; import com.openjiuwen.core.singleagent.agents.ReActAgentConfig; import com.openjiuwen.core.singleagent.interrupt.ToolInterruptException; import com.openjiuwen.core.singleagent.rail.AgentCallbackContext; @@ -34,12 +36,14 @@ import com.openjiuwen.core.workflow.WorkflowCard; import java.util.ArrayList; +import java.util.Collections; +import java.util.Iterator; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import java.util.concurrent.CancellationException; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletionException; -import java.util.concurrent.CancellationException; import java.util.concurrent.ConcurrentHashMap; /** @@ -1043,6 +1047,393 @@ private Object invokeTool(Tool tool, Map toolArgs, Session sessi } } + // ==================== Streaming tool execution ==================== + + /** + * Streaming-aware version of {@link #execute(AgentCallbackContext, Object, Session, String)}. + *

+ * When {@code agentSession} is non-null AND the tool supports streaming + * (currently {@link LocalFunction}), each chunk yielded by + * {@link Tool#stream(Map, Map)} is forwarded as + * {@code OutputSchema("tool_stream_chunk", ...)}, and the method finally + * emits a {@code "tool_stream_end"} chunk. Non-streaming tools and + * non-tool branches (workflows, agents, MCP) fall back to the synchronous + * {@link #execute(AgentCallbackContext, Object, Session, String)} behaviour. + *

+ * The returned {@code List} is identical to + * {@code execute(...)} so the caller continues to work unchanged. + * + * @param ctx callback context (for rails / force-finish / steering) + * @param toolCall tool call(s) to execute (ToolCall, List, Map, array) + * @param session session + * @param tag optional routing tag + * @param agentSession stream writer; {@code null} disables streaming forwarding + * @return tool execution entries, same as {@link #execute(...)} + * @since 0.1.15 + */ + public List executeStream( + AgentCallbackContext ctx, + Object toolCall, + Session session, + String tag, + AgentSessionApi agentSession + ) { + List toolCalls = normalizeToolCalls(toolCall); + if (toolCalls.isEmpty()) { + return List.of(); + } + if (toolCalls.size() == 1) { + int toolIndex = toolCalls.get(0).getIndex() != null + ? toolCalls.get(0).getIndex() : 0; + return List.of(executeOneToolCallWithStreaming( + ctx, toolCalls.get(0), session, tag, agentSession, toolIndex)); + } + return executeParallelToolCallsWithStreaming(ctx, toolCalls, session, tag, agentSession); + } + + private List executeParallelToolCallsWithStreaming( + AgentCallbackContext ctx, + List toolCalls, + Session session, + String tag, + AgentSessionApi agentSession + ) { + List toolContexts = new ArrayList<>(); + List> futures = new ArrayList<>(); + for (int i = 0; i < toolCalls.size(); i++) { + ToolCall tc = toolCalls.get(i); + final int toolIndex = tc.getIndex() != null ? tc.getIndex() : i; + AgentCallbackContext toolCtx = buildToolCallbackContext(ctx, tc, session); + toolContexts.add(toolCtx); + CompletableFuture future = OpenJiuwenExecutors.withToolCallTimeout( + OpenJiuwenExecutors.supplyToolCallAsync( + () -> executePreparedToolCallWithStreaming( + toolCtx, tc, session, tag, agentSession, toolIndex) + ) + ); + futures.add(future); + } + List finalResults = new ArrayList<>(); + for (int i = 0; i < futures.size(); i++) { + finalResults.add(joinToolExecution(toolCalls.get(i), futures.get(i))); + } + for (AgentCallbackContext toolCtx : toolContexts) { + mergeToolContext(ctx, toolCtx); + } + return finalResults; + } + + private ToolExecutionEntry executeOneToolCallWithStreaming( + AgentCallbackContext ctx, + ToolCall singleToolCall, + Session session, + String tag, + AgentSessionApi agentSession, + int toolIndex + ) { + AgentCallbackContext toolCtx = buildToolCallbackContext(ctx, singleToolCall, session); + try { + return executePreparedToolCallWithStreaming(toolCtx, singleToolCall, session, tag, agentSession, toolIndex); + } finally { + mergeToolContext(ctx, toolCtx); + } + } + + private ToolExecutionEntry executePreparedToolCallWithStreaming( + AgentCallbackContext toolCtx, + ToolCall singleToolCall, + Session session, + String tag, + AgentSessionApi agentSession, + int toolIndex + ) { + Session previousSession = SessionContextHolder.getCurrentSession(); + try { + SessionContextHolder.setCurrentSession(session); + ToolExecutionEntry result; + try { + result = railedExecuteStreamSingleToolCall( + toolCtx, singleToolCall, session, tag, agentSession, toolIndex); + } finally { + toolCtx.getExtra().remove("_skip_tool"); + } + if (toolCtx.getInputs() instanceof ToolCallInputs inputs) { + Object toolResult = inputs.getToolResult() != null + ? inputs.getToolResult() + : (result != null ? result.result() : null); + ToolMessage toolMsg = inputs.getToolMsg() != null + ? inputs.getToolMsg() + : (result != null ? result.toolMessage() : null); + return new ToolExecutionEntry(toolResult, toolMsg); + } + return result; + } catch (ToolInterruptException | AbilityExecutionError e) { + if (agentSession != null) { + agentSession.writeStream(buildToolOutputChunk( + resolveTaskId(session), singleToolCall, + "tool error: " + (e.getMessage() == null ? "" : e.getMessage()), 0)); + } + return handleToolExecutionException(singleToolCall, toolCtx, e); + } catch (RuntimeException re) { + if (agentSession != null) { + agentSession.writeStream(buildToolOutputChunk( + resolveTaskId(session), singleToolCall, + "tool error: " + (re.getMessage() == null ? "" : re.getMessage()), 0)); + } + throw re; + } finally { + SessionContextHolder.restoreCurrentSession(previousSession); + } + } + + private ToolExecutionEntry railedExecuteStreamSingleToolCall( + AgentCallbackContext ctx, + ToolCall toolCall, + Session session, + String tag, + AgentSessionApi agentSession, + int toolIndex + ) { + return RailExecutor.execute(ctx, AgentCallbackEvent.BEFORE_TOOL_CALL, + AgentCallbackEvent.AFTER_TOOL_CALL, + AgentCallbackEvent.ON_TOOL_EXCEPTION, + () -> { + if (Boolean.TRUE.equals(ctx.getExtra().get("_skip_tool"))) { + if (ctx.getInputs() instanceof ToolCallInputs inputs) { + return new ToolExecutionEntry(inputs.getToolResult(), inputs.getToolMsg()); + } + return new ToolExecutionEntry(null, null); + } + + if (ctx.getInputs() instanceof ToolCallInputs inputs) { + if (inputs.getToolName() != null && !inputs.getToolName().isEmpty()) { + toolCall.setName(inputs.getToolName()); + } + if (inputs.getToolArgs() != null) { + toolCall.setArguments(inputs.getToolArgs() instanceof String s + ? s + : MAPPER.writeValueAsString(inputs.getToolArgs())); + } + } + + ToolExecutionEntry result = streamSingleToolCall( + toolCall, session, tag, agentSession, toolIndex); + + if (ctx.getInputs() instanceof ToolCallInputs inputs) { + inputs.setToolCall(toolCall); + inputs.setToolName(toolCall.getName()); + inputs.setToolArgs(toolCall.getArguments()); + inputs.setToolResult(result.result()); + inputs.setToolMsg(result.toolMessage()); + } + return result; + }).orElseGet(() -> new ToolExecutionEntry(null, null)); + } + + /** + * Execute one tool call, choosing the streaming or synchronous path. + * Tool instance path only (tools.containsKey / fallback). For + * workflows / agents / MCP it simply delegates to executeSingleToolCall. + */ + private ToolExecutionEntry streamSingleToolCall( + ToolCall toolCall, + Session session, + String tag, + AgentSessionApi agentSession, + int toolIndex + ) { + String toolName = toolCall.getName(); + + // --- Tool branch --- + Tool tool = null; + if (tools.containsKey(toolName)) { + ToolCard toolCard = tools.get(toolName); + String toolId = toolCard.getId() != null ? toolCard.getId() : toolCard.getName(); + tool = getToolFromResourceMgr(toolId, tag); + } else if (!mcpServers.isEmpty()) { + tool = resolveMcpToolByName(toolName); + if (tool == null && !mcpServers.containsKey(toolName)) { + tool = getToolFromResourceMgr(toolName, tag); + } + } else { + tool = getToolFromResourceMgr(toolName, tag); + } + + if (tool != null) { + Map toolArgs = parseToolArgs(toolCall.getArguments()); + boolean isStreaming = agentSession != null && canStream(tool); + try { + if (isStreaming) { + return executeStreamingTool(tool, toolCall, toolArgs, session, agentSession, toolIndex); + } + // 非流式或 agentSession 为 null：走原 invoke 路径 + Object result = invokeTool(tool, toolArgs, session); + logToolResult(result); + if (agentSession != null) { + agentSession.writeStream(buildToolOutputChunk( + resolveTaskId(session), toolCall, + result == null ? "" : result, 0)); + } + ToolMessage toolMsg = ToolMessage.builder() + .content(result == null ? "" : result.toString()) + .toolCallId(toolCall.getId()) + .build(); + return new ToolExecutionEntry(result, toolMsg); + } catch (BaseError e) { + throw e; + } catch (Exception e) { + String errorMsg = "Tool execution error: " + + (e instanceof BaseError be ? be.toString() : e.getMessage()); + Loggers.AGENT.error(errorMsg); + Loggers.TOOL.info("Tool result: None"); + if (agentSession != null) { + agentSession.writeStream(buildToolOutputChunk( + resolveTaskId(session), toolCall, + "tool error: " + (e.getMessage() == null ? "" : e.getMessage()), 0)); + } + throw buildExecutionError(toolCall, errorMsg); + } + } + + // --- Non-tool branches (workflows / agents / MCP servers): use original path --- + return executeSingleToolCall(toolCall, session, tag); + } + + private ToolExecutionEntry executeStreamingTool( + Tool tool, + ToolCall toolCall, + Map toolArgs, + Session session, + AgentSessionApi agentSession, + int toolIndex + ) throws Exception { + Map kwargs = new LinkedHashMap<>(); + Session previousSession = SessionContextHolder.getCurrentSession(); + if (session != null) { + kwargs.put("session", session); + SessionContextHolder.setCurrentSession(session); + } + String taskId = resolveTaskId(session); + List accumulated = new ArrayList<>(); + int chunkIndex = 0; + try { + Iterator streamIt = tool.stream(toolArgs, kwargs); + while (streamIt != null && streamIt.hasNext()) { + Object chunk = streamIt.next(); + accumulated.add(chunk); + agentSession.writeStream(buildToolOutputChunk( + taskId, toolCall, chunk, chunkIndex)); + chunkIndex++; + } + } catch (BaseError e) { + // If LocalFunction.stream() throws TOOL_LOCAL_FUNCTION_EXECUTION_ERROR, + // it means the underlying func is not streaming — fall back to tool.invoke(). + if (e.getCode() == StatusCode.TOOL_LOCAL_FUNCTION_EXECUTION_ERROR.getCode()) { + Loggers.AGENT.debug("Tool '{}' does not support streaming, falling back to synchronous invoke", + tool.getCard() != null ? tool.getCard().getId() : toolCall.getName()); + Object result = invokeTool(tool, toolArgs, session); + logToolResult(result); + agentSession.writeStream(buildToolOutputChunk( + taskId, toolCall, result == null ? "" : result, 0)); + ToolMessage toolMsg = ToolMessage.builder() + .content(result == null ? "" : result.toString()) + .toolCallId(toolCall.getId()) + .build(); + return new ToolExecutionEntry(result, toolMsg); + } + throw e; + } finally { + SessionContextHolder.restoreCurrentSession(previousSession); + } + + Object merged = mergeStreamChunks(accumulated); + logToolResult(merged); + + ToolMessage toolMsg = ToolMessage.builder() + .content(merged == null ? "" : merged.toString()) + .toolCallId(toolCall.getId()) + .build(); + return new ToolExecutionEntry(merged, toolMsg); + } + + /** + * Determines whether the tool instance is worth trying tool.stream(). + * McpTool and RestfulApi's stream() currently raise an explicit error; + * only LocalFunction truly supports streaming (when it wraps a func + * returning Iterator/Iterable). For other tool types we skip the try + * to avoid an exception-control-flow code path. + */ + private static boolean canStream(Tool tool) { + return tool instanceof LocalFunction; + } + + /** + * Resolve task id from session state, if the upstream agent (e.g. DeepAgent) + * injected {@code task_id} into the session state before invoking tools. + * Returns {@code null} when no task id is bound (standalone ReActAgent + * invocations have no task concept). + */ + private static String resolveTaskId(Session session) { + if (session == null) { + return null; + } + Object value = session.getState("task_id"); + return value == null ? null : String.valueOf(value); + } + + /** + * Build a unified {@code tool_output} stream chunk. + *

+ * All tool streaming events (per-chunk output, non-streaming invoke + * result, tool error) are emitted using the same shape: + * 

+     * {
+     *   "type": "tool_output",
+     *   "index": chunkIndex,
+     *   "payload": {
+     *     "task_id": "...",
+     *     "tool_name": "...",
+     *     "tool_call_id": "...",
+     *     "content": ...
+     *   }
+     * }
+     *
+ * Downstream consumers only need to read {@code payload.content} for + * every {@code tool_output} chunk to render the tool's progressive + * output; no lifecycle events or replay cursors are emitted. + */ + private static OutputSchema buildToolOutputChunk( + String taskId, ToolCall toolCall, Object content, int chunkIndex) { + Map payload = new LinkedHashMap<>(); + payload.put("task_id", taskId == null ? "" : taskId); + payload.put("tool_name", toolCall.getName()); + payload.put("tool_call_id", toolCall.getId()); + payload.put("content", content); + return new OutputSchema("tool_output", chunkIndex, payload); + } + + /** + * Merge a list of tool stream chunks into a single result for ToolMessage. + * All-String chunks are joined as-is (a typical streaming tool case); + * otherwise the list is wrapped so the caller sees every chunk. + */ + private static Object mergeStreamChunks(List chunks) { + if (chunks == null || chunks.isEmpty()) { + return ""; + } + boolean isAllStrings = chunks.stream().allMatch(c -> c instanceof String || c == null); + if (isAllStrings) { + StringBuilder sb = new StringBuilder(); + for (Object c : chunks) { + if (c != null) { + sb.append(c); + } + } + return sb.toString(); + } + return new ArrayList<>(chunks); + } + /** * resolveMcpToolByName. * diff --git a/src/main/java/com/openjiuwen/core/singleagent/agents/ReActAgent.java b/src/main/java/com/openjiuwen/core/singleagent/agents/ReActAgent.java index 090c25a84..f2796c947 100644 --- a/src/main/java/com/openjiuwen/core/singleagent/agents/ReActAgent.java +++ b/src/main/java/com/openjiuwen/core/singleagent/agents/ReActAgent.java @@ -581,6 +581,39 @@ private void executeToolCall(AgentCallbackContext ctx, List toolCalls, Sessio */ private List executeToolCallEntries(AgentCallbackContext ctx, List toolCalls, Session session, ModelContext context) { + return executeToolCallEntries(ctx, toolCalls, session, context, null); + } + + /** + * Streaming-aware overload of {@link #executeToolCallEntries}. + *

+ * When {@code agentSession} is non-null the underlying AbilityManager + * forwards tool stream chunks as unified {@code tool_output} events + * (covering per-chunk output, non-streaming invoke results, and tool + * errors); when {@code null} this behaves identically to the 4-arg + * overload. + *

+ * Only the main ReAct loop invokes this with a session; the + * interrupt-resume branch continues to use the 4-arg (non-streaming) + * variant because the interrupted tool has already been executed + * upstream. + * + * @param ctx callback context + * @param toolCalls tool calls to execute + * @param session session + * @param context model context (to append ToolMessage) + * @param agentSession nullable stream writer; when set, tool stream + * chunks are forwarded to the outer agent stream + * @return same as 4-arg executeToolCallEntries + * @since 0.1.15 + */ + private List executeToolCallEntries( + AgentCallbackContext ctx, + List toolCalls, + Session session, + ModelContext context, + AgentSessionApi agentSession + ) { if (toolCalls == null || toolCalls.isEmpty()) { return List.of(); } @@ -593,7 +626,12 @@ private List executeToolCallEntries(AgentCallbackContext ctx } } - List results = getAbilityManager().execute(ctx, toolCalls, session, null); + List results; + if (agentSession == null) { + results = getAbilityManager().execute(ctx, toolCalls, session, null); + } else { + results = getAbilityManager().executeStream(ctx, toolCalls, session, null, agentSession); + } for (ToolExecutionEntry entry : results) { if (entry.toolMessage() != null) { context.addMessages(entry.toolMessage()); @@ -1376,7 +1414,7 @@ private Map invokeForStream(Object inputs, Session session, Agen if (hasToolCalls) { List results = - executeToolCallEntries(ctx, aiMessage.getToolCalls(), session, context); + executeToolCallEntries(ctx, aiMessage.getToolCalls(), session, context, agentSession); AgentCallbackContext.ForceFinishRequest finishAfterTool = ctx.consumeForceFinish(); if (finishAfterTool != null) { @@ -1548,25 +1586,35 @@ private void writeAssistantStreamChunk(AgentSessionApi agentSession, AssistantMe if (agentSession == null || chunk == null) { return; } + // 复用 AbilityManager.resolveTaskId 同样的取值方式：从 session state 读 task_id， + // 与 tool_output / task_output chunk 保持一致。standalone ReActAgent 调用 + // 时 task_id 为空字符串，不影响下游消费。 + Object taskIdRaw = agentSession.getState("task_id"); + String taskId = taskIdRaw == null ? "" : String.valueOf(taskIdRaw); + if (chunk.getReasoningContent() != null) { Map reasoningPayload = new HashMap(); + reasoningPayload.put("task_id", taskId); reasoningPayload.put("content", chunk.getReasoningContent()); reasoningPayload.put("result_type", "answer"); agentSession.writeStream(new OutputSchema("llm_reasoning", index, reasoningPayload)); } if (chunk.getContent() != null && !(chunk.getContent() instanceof String str && str.isBlank())) { Map contentPayload = new HashMap(); + contentPayload.put("task_id", taskId); contentPayload.put("content", chunk.getContent()); contentPayload.put("result_type", "answer"); agentSession.writeStream(new OutputSchema("llm_output", index, contentPayload)); } if (chunk.getToolCalls() != null && !chunk.getToolCalls().isEmpty()) { Map toolPayload = new HashMap(); + toolPayload.put("task_id", taskId); toolPayload.put("tool_calls", cloneToolCalls(chunk.getToolCalls())); agentSession.writeStream(new OutputSchema("llm_output", index, toolPayload)); } if (chunk.getUsageMetadata() != null) { Map usagePayload = new HashMap(); + usagePayload.put("task_id", taskId); usagePayload.put("usage_metadata", chunk.getUsageMetadata()); usagePayload.put("result_type", "answer"); agentSession.writeStream(new OutputSchema("llm_usage", 0, usagePayload)); diff --git a/src/main/java/com/openjiuwen/core/sysop/local/LocalShellOperation.java b/src/main/java/com/openjiuwen/core/sysop/local/LocalShellOperation.java index cb6513fc4..c6f644a82 100644 --- a/src/main/java/com/openjiuwen/core/sysop/local/LocalShellOperation.java +++ b/src/main/java/com/openjiuwen/core/sysop/local/LocalShellOperation.java @@ -27,10 +27,13 @@ import java.nio.charset.Charset; import java.nio.file.Path; import java.util.ArrayList; +import java.util.Collections; import java.util.Iterator; import java.util.List; import java.util.Locale; import java.util.Map; +import java.util.NoSuchElementException; +import java.util.concurrent.atomic.AtomicInteger; import java.util.regex.Pattern; /** @@ -145,32 +148,30 @@ public ExecuteCmdResult executeCmd(String command, String cwd, int timeout, Map< @Override public Iterator executeCmdStream(String command, String cwd, int timeout, Map environment, Map options) { - String methodName = "executeCmdStream"; - long startTime = System.currentTimeMillis(); - List results = new ArrayList<>(); - Loggers.SYS_OPERATION.info("Start to execute cmd streaming"); - int chunkIndex = 0; + // fail-fast: 参数 / 白名单 / 危险字符 校验同步完成，错误立即返回单元素迭代器 if (command == null || command.isBlank()) { - results.add(buildCmdStreamErrorResult("command can not be empty", - ExecuteCmdChunkData.builder().chunkIndex(chunkIndex).exitCode(-1).build())); - return results.iterator(); + return Collections.singletonList( + buildCmdStreamErrorResult("command can not be empty", + ExecuteCmdChunkData.builder().chunkIndex(0).exitCode(-1).build()) + ).iterator(); } try { int effectiveTimeout = normalizeTimeoutSeconds(timeout); - if (!checkAllowlist(command)) { - results.add(buildCmdStreamErrorResult("command not allowed by allowlist", - ExecuteCmdChunkData.builder().chunkIndex(chunkIndex).exitCode(-1).build())); - return results.iterator(); + return Collections.singletonList( + buildCmdStreamErrorResult("command not allowed by allowlist", + ExecuteCmdChunkData.builder().chunkIndex(0).exitCode(-1).build()) + ).iterator(); } String dangerousReason = checkDangerousPatterns(command); if (dangerousReason != null && !dangerousReason.isBlank()) { - results.add(buildCmdStreamErrorResult(dangerousReason, - ExecuteCmdChunkData.builder().chunkIndex(chunkIndex).exitCode(-1).build())); - return results.iterator(); + return Collections.singletonList( + buildCmdStreamErrorResult(dangerousReason, + ExecuteCmdChunkData.builder().chunkIndex(0).exitCode(-1).build()) + ).iterator(); } Map env = OperationUtils.prepareEnvironment(environment); @@ -189,24 +190,59 @@ public Iterator executeCmdStream(String command, String ProcessHandler handler = new ProcessHandler(process, chunkSize, charset, effectiveTimeout); Iterator eventIterator = handler.stream(); - while (eventIterator.hasNext()) { - StreamEvent event = eventIterator.next(); - ExecuteCmdStreamResult transformed = transformCmdStreamEvent(event, chunkIndex); - if (transformed != null) { - results.add(transformed); - chunkIndex++; + // 真正惰性的迭代器：hasNext/next 时才去拉取下一事件，避免先全量收集 + return new Iterator() { + private final AtomicInteger chunkIndex = new AtomicInteger(0); + private StreamEvent nextEvent; + private boolean hasNext = true; + + @Override + public boolean hasNext() { + advanceIfNeeded(); + return hasNext; } - if (event.getType() == StreamEventType.ERROR || event.getType() == StreamEventType.EXIT) { - break; + + @Override + public ExecuteCmdStreamResult next() { + if (!hasNext()) { + throw new NoSuchElementException("No more streaming events"); + } + StreamEvent event = nextEvent; + nextEvent = null; + int idx = chunkIndex.getAndIncrement(); + // EXIT / ERROR 事件产出本次后不再继续 + if (event.getType() == StreamEventType.EXIT + || event.getType() == StreamEventType.ERROR) { + hasNext = false; + } + return transformCmdStreamEvent(event, idx); } - } - return results.iterator(); + private void advanceIfNeeded() { + if (nextEvent != null || !hasNext) { + return; + } + try { + if (eventIterator.hasNext()) { + nextEvent = eventIterator.next(); + } else { + hasNext = false; + } + } catch (Exception e) { + Loggers.SYS_OPERATION.error("Failed to execute cmd streaming", e); + nextEvent = StreamEvent.builder() + .type(StreamEventType.ERROR) + .data("unexpected streaming error: " + e.getMessage()) + .build(); + } + } + }; } catch (Exception e) { - Loggers.SYS_OPERATION.error("Failed to execute cmd streaming", e); - results.add(buildCmdStreamErrorResult("unexpected error: " + e.getMessage(), - ExecuteCmdChunkData.builder().chunkIndex(chunkIndex).exitCode(-1).build())); - return results.iterator(); + Loggers.SYS_OPERATION.error("Failed to start cmd streaming", e); + return Collections.singletonList( + buildCmdStreamErrorResult("unexpected error: " + e.getMessage(), + ExecuteCmdChunkData.builder().chunkIndex(0).exitCode(-1).build()) + ).iterator(); } } diff --git a/src/main/java/com/openjiuwen/harness/deep_agent/DeepAgent.java b/src/main/java/com/openjiuwen/harness/deep_agent/DeepAgent.java index ff02edd10..a639e5c3c 100644 --- a/src/main/java/com/openjiuwen/harness/deep_agent/DeepAgent.java +++ b/src/main/java/com/openjiuwen/harness/deep_agent/DeepAgent.java @@ -1751,6 +1751,16 @@ private Map invokeInnerRoundStreaming(Map effect AgentSessionApi session) { AgentSessionApi innerSession = new AgentSessionApi(String.valueOf(effectiveInputs.get("conversation_id")), session != null ? session.getEnvs() : null, card, List.of(StreamMode.OUTPUT)); + // 复用上游（CoreTaskLoopEventExecutor.buildEffectiveInputs L236）注入到 effectiveInputs + // 的 task_id，不在此重新生成。该值最终源自 DeepAgent.executeCoreLoopRound L1643 + // 的 "deep_agent_task__"。stream 路径若由用户直接调 + // DeepAgent.stream 测试（effectiveInputs 无 task_id），则 taskId 为 null， + // AbilityManager 会在 tool_output.payload.task_id 写空字符串。 + Object taskIdRaw = effectiveInputs.get("task_id"); + String taskId = taskIdRaw == null ? null : String.valueOf(taskIdRaw); + if (taskId != null) { + innerSession.updateState(java.util.Map.of("task_id", taskId)); + } // 传播租户上下文到 inner session TenantContext ctx = session != null ? session.getTenantContext() : null; if (ctx != null && ctx.isTenantAware()) { @@ -1758,6 +1768,10 @@ private Map invokeInnerRoundStreaming(Map effect } innerSession.preRun(effectiveInputs); copySessionState(session, innerSession); + // copySessionState 可能覆盖 inner session state，重新注入 task_id 以确保下游可见 + if (taskId != null) { + innerSession.updateState(java.util.Map.of("task_id", taskId)); + } // task-loop 独立线程需重新绑定租户上下文 if (ctx != null && ctx.isTenantAware()) { TenantContextHolder.setCurrentTenant(ctx); diff --git a/src/test/java/com/openjiuwen/core/singleagent/AbilityManagerStreamOutputTest.java b/src/test/java/com/openjiuwen/core/singleagent/AbilityManagerStreamOutputTest.java new file mode 100644 index 000000000..74e73feab --- /dev/null +++ b/src/test/java/com/openjiuwen/core/singleagent/AbilityManagerStreamOutputTest.java @@ -0,0 +1,169 @@ +// Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved. + +package com.openjiuwen.core.singleagent; + +import static org.assertj.core.api.Assertions.assertThat; + +import com.openjiuwen.core.foundation.llm.schema.ToolCall; +import com.openjiuwen.core.session.stream.OutputSchema; + +import org.junit.jupiter.api.Test; + +import java.lang.reflect.Method; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Map; + +/** + * Unit tests for the unified {@code tool_output} stream chunk shape produced + * by {@link AbilityManager#executeStream}. Covers {@code buildToolOutputChunk} + * (payload shape, task_id null-safety, index propagation) and + * {@code mergeStreamChunks} (string concatenation, empty list, mixed types). + *

+ * These tests use reflection to invoke the private static helpers directly, + * avoiding the heavier AgentSessionApi / StreamWriter / Runner bootstrap + * needed by integration tests. The contract under test: + * 

+ *   { "type": "tool_output", "index": chunkIndex,
+ *     "payload": { "task_id", "tool_name", "tool_call_id", "content" } }
+ *
+ * No {@code session_id} field is emitted (it was intentionally removed). + */ +class AbilityManagerStreamOutputTest { + + private static final String TOOL_NAME = "web_search"; + private static final String TOOL_CALL_ID = "call_1_abc"; + + private static ToolCall newToolCall() { + return ToolCall.builder() + .id(TOOL_CALL_ID) + .name(TOOL_NAME) + .arguments("{\"query\":\"test\"}") + .index(0) + .build(); + } + + private static OutputSchema invokeBuild(String taskId, Object content, int chunkIndex) throws Exception { + Method m = AbilityManager.class.getDeclaredMethod( + "buildToolOutputChunk", String.class, ToolCall.class, Object.class, int.class); + m.setAccessible(true); + return (OutputSchema) m.invoke(null, taskId, newToolCall(), content, chunkIndex); + } + + @SuppressWarnings("unchecked") + private static Object invokeMerge(List chunks) throws Exception { + Method m = AbilityManager.class.getDeclaredMethod("mergeStreamChunks", List.class); + m.setAccessible(true); + return m.invoke(null, chunks); + } + + // ========== buildToolOutputChunk ========== + + @Test + void buildToolOutputChunk_normalCase_returnsUnifiedToolOutputShape() throws Exception { + OutputSchema chunk = invokeBuild("deep_agent_task_s1_1", "找到 3 条结果", 2); + + assertThat(chunk.getType()).isEqualTo("tool_output"); + assertThat(chunk.getIndex()).isEqualTo(2); + assertThat(chunk.getPayload()).isInstanceOf(Map.class); + Map payload = (Map) chunk.getPayload(); + + // 必须含这 4 个字段 + assertThat(payload).containsOnlyKeys("task_id", "tool_name", "tool_call_id", "content"); + assertThat(payload.get("task_id")).isEqualTo("deep_agent_task_s1_1"); + assertThat(payload.get("tool_name")).isEqualTo(TOOL_NAME); + assertThat(payload.get("tool_call_id")).isEqualTo(TOOL_CALL_ID); + assertThat(payload.get("content")).isEqualTo("找到 3 条结果"); + } + + @Test + void buildToolOutputChunk_noSessionIdField_emitted() throws Exception { + // 回归：session_id 字段已从 payload 中移除，下游不应再期望该字段 + OutputSchema chunk = invokeBuild("t1", "content", 0); + @SuppressWarnings("unchecked") + Map payload = (Map) chunk.getPayload(); + assertThat(payload).doesNotContainKey("session_id"); + } + + @Test + void buildToolOutputChunk_nullTaskId_writesEmptyString() throws Exception { + // standalone ReActAgent 调用（无 task_id 注入）时 task_id 字段为空字符串， + // 保证字段始终存在、类型稳定，下游不需要 null 判断 + OutputSchema chunk = invokeBuild(null, "content", 0); + @SuppressWarnings("unchecked") + Map payload = (Map) chunk.getPayload(); + assertThat(payload.get("task_id")).isEqualTo(""); + } + + @Test + void buildToolOutputChunk_indexPropagated() throws Exception { + // chunkIndex 直接映射到 OutputSchema.index，用于下游按序重组流 + for (int i = 0; i < 5; i++) { + OutputSchema chunk = invokeBuild("t1", "c" + i, i); + assertThat(chunk.getIndex()).isEqualTo(i); + } + } + + @Test + void buildToolOutputChunk_payloadIsInsertionOrdered() throws Exception { + // LinkedHashMap 保证字段顺序：task_id → tool_name → tool_call_id → content + OutputSchema chunk = invokeBuild("t1", "c", 0); + @SuppressWarnings("unchecked") + Map payload = (Map) chunk.getPayload(); + assertThat(new ArrayList<>(payload.keySet())) + .containsExactly("task_id", "tool_name", "tool_call_id", "content"); + } + + @Test + void buildToolOutputChunk_nullContent_preservedAsNull() throws Exception { + // content 字段原样透传，null 时不做转换（工具返回 null 的场景） + OutputSchema chunk = invokeBuild("t1", null, 0); + @SuppressWarnings("unchecked") + Map payload = (Map) chunk.getPayload(); + assertThat(payload.get("content")).isNull(); + } + + // ========== mergeStreamChunks ========== + + @Test + void mergeStreamChunks_allStrings_concatenated() throws Exception { + // 典型流式工具场景：每个 chunk 是 String，合并为完整文本 + Object merged = invokeMerge(Arrays.asList("找到 3 条", "相关结果", ":\n")); + assertThat(merged).isEqualTo("找到 3 条相关结果:\n"); + } + + @Test + void mergeStreamChunks_emptyList_returnsEmptyString() throws Exception { + // 工具 yield 0 个 chunk（如直接 return）时合并为空字符串，避免 null 污染 ToolMessage + assertThat(invokeMerge(new ArrayList<>())).isEqualTo(""); + assertThat(invokeMerge(null)).isEqualTo(""); + } + + @Test + void mergeStreamChunks_mixedTypes_returnsListCopy() throws Exception { + // 非全 String（如结构化对象）时返回 ArrayList 包装，让 caller 看到每个 chunk + Object obj = new Object(); + Object merged = invokeMerge(Arrays.asList("text", obj, 42)); + assertThat(merged).isInstanceOf(ArrayList.class); + @SuppressWarnings("unchecked") + List list = (List) merged; + assertThat(list).containsExactly("text", obj, 42); + // 返回的是副本，修改不影响原入参 + list.add("extra"); + assertThat(list).hasSize(4); + } + + @Test + void mergeStreamChunks_nullStringElements_skipped() throws Exception { + // String 列表里的 null 元素被跳过（不拼成 "null" 字符串） + Object merged = invokeMerge(Arrays.asList("a", null, "b")); + assertThat(merged).isEqualTo("ab"); + } + + @Test + void mergeStreamChunks_singleString_returnedAsIs() throws Exception { + // 单 chunk 的 String 直接返回（StringBuilder 拼接结果） + assertThat(invokeMerge(List.of("only"))).isEqualTo("only"); + } +} diff --git a/src/test/java/com/openjiuwen/core/singleagent/agents/ReActAgentTest.java b/src/test/java/com/openjiuwen/core/singleagent/agents/ReActAgentTest.java index 27d974c8b..5399d7310 100644 --- a/src/test/java/com/openjiuwen/core/singleagent/agents/ReActAgentTest.java +++ b/src/test/java/com/openjiuwen/core/singleagent/agents/ReActAgentTest.java @@ -479,6 +479,11 @@ void testStreamLogsPostRailToolMessageWhenNotSensitive() throws Exception { .execute(any(AgentCallbackContext.class), any(), any(Session.class), isNull())) .thenReturn(List.of( new AbilityManager.ToolExecutionEntry("30.0", new ToolMessage("30.0", "call-log")))); + when(testAgent.getAbilityManager() + .executeStream(any(AgentCallbackContext.class), any(), any(Session.class), isNull(), + any(AgentSessionApi.class))) + .thenReturn(List.of( + new AbilityManager.ToolExecutionEntry("30.0", new ToolMessage("30.0", "call-log")))); testAgent.setLlm(model); AgentSessionApi session = new AgentSessionApi("react-stream-log-session", null, testAgent.getCard(), diff --git a/src/test/java/com/openjiuwen/core/sysop/local/LocalShellOperationTest.java b/src/test/java/com/openjiuwen/core/sysop/local/LocalShellOperationTest.java index 8ac9ba515..1fbd9e4a6 100644 --- a/src/test/java/com/openjiuwen/core/sysop/local/LocalShellOperationTest.java +++ b/src/test/java/com/openjiuwen/core/sysop/local/LocalShellOperationTest.java @@ -420,4 +420,56 @@ void testStreamContinuousOutput() { assertNotNull(exitChunk); assertEquals(0, exitChunk.getData().getExitCode()); } + + @Test + @DisplayName("Stream: hasNext is idempotent before next (lazy iterator)") + void testStreamHasNextIsIdempotentBeforeNext() { + // 修复后 executeCmdStream 返回真正惰性的迭代器：重复调用 hasNext() 不应消费事件， + // 也不应阻塞。EXIT 事件产出后 hasNext() 必须立即返回 false，避免下游死循环。 + String cmd = OsTestSupport.isWindows() ? "echo hello" : "echo hello"; + Iterator it = + shell().executeCmdStream(cmd, null, 10, null, null); + + // 多次 hasNext() 不消费 + assertTrue(it.hasNext()); + assertTrue(it.hasNext()); + assertTrue(it.hasNext()); + ExecuteCmdStreamResult first = it.next(); + assertNotNull(first); + + // 消费完剩余事件 + List rest = new ArrayList<>(); + while (it.hasNext()) { + rest.add(it.next()); + } + + // 终止后再次 hasNext() 必须返回 false（不抛异常、不阻塞） + assertFalse(it.hasNext(), "hasNext must return false after iterator exhausted"); + assertFalse(it.hasNext(), "repeated hasNext on exhausted iterator stays false"); + + // 至少产出一个 EXIT 事件 + List all = new ArrayList<>(); + all.add(first); + all.addAll(rest); + assertTrue(all.stream().anyMatch(r -> r.getData() != null && r.getData().getExitCode() != null), + "should contain an EXIT event"); + } + + @Test + @DisplayName("Stream: empty command iterator terminates after single ERROR chunk") + void testStreamEmptyCommandIteratorTerminatesAfterError() { + // 参数校验失败时返回单元素迭代器：取完一个 ERROR chunk 后 hasNext() 必须返回 false， + // 防止下游 while(it.hasNext()) 死循环（对应 advanceIfNeeded catch 块的 hasNext=false 修复） + Iterator it = + shell().executeCmdStream("", null, 300, null, null); + + assertTrue(it.hasNext()); + ExecuteCmdStreamResult err = it.next(); + assertEquals(StatusCode.SYS_OPERATION_SHELL_EXECUTION_ERROR.getCode(), err.getCode()); + assertTrue(err.getMessage().contains("command can not be empty")); + + // 取完 ERROR 后必须终止 + assertFalse(it.hasNext(), "iterator must terminate after the single ERROR chunk"); + assertFalse(it.hasNext()); + } }