From 8a93d69590f9e0a8d374b99f85fdba5e069c9959 Mon Sep 17 00:00:00 2001 From: Linghong Date: Thu, 10 Sep 2026 12:21:59 +0000 Subject: [PATCH] =?UTF-8?q?=E6=B5=81=E5=BC=8F=20Responses=20=E4=BB=8E?= =?UTF-8?q?=E5=A2=9E=E9=87=8F=E4=BA=8B=E4=BB=B6=E7=B4=AF=E7=A7=AF=E5=8E=9F?= =?UTF-8?q?=E7=94=9F=20output?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit completed 未带 output 时保留 output_item 与工具参数 补空 completed 与顶层 output 的回放测试并记录排障 --- docs/project-memory/shared-memory/pitfalls.md | 8 + server-rs/crates/platform-llm/src/lib.rs | 212 +++++++++++++++--- 2 files changed, 184 insertions(+), 36 deletions(-) diff --git a/docs/project-memory/shared-memory/pitfalls.md b/docs/project-memory/shared-memory/pitfalls.md index f6a3b783b..3e6028243 100644 --- a/docs/project-memory/shared-memory/pitfalls.md +++ b/docs/project-memory/shared-memory/pitfalls.md @@ -2,6 +2,14 @@ > 当前口径：本文件保留可复用的排障经验；历史条目的旧路由、旧版本和已删除文档仅作根因背景，不得据此恢复退役入口。当前命令、路由和 schema 以代码与 `docs/README.md` 为准。 +## 2026-09-10 设计 Agent 需要流式 Responses 的原生 output[]，不能只靠 tool_calls + +- **现象**：新策划报 `Provider 返回工具调用但未提供完整 Responses output`。模型已经在调工具，`.debug/design-agent` 里 `tool_calls` 有值但 `output` 是 `[]`。 +- **原因**：设计 Agent 下一轮要把 Responses `output[]` 原样接回 `input`（`store=false` + 加密 reasoning）。旧 Planning V2 只消费 `tool_calls` 再由 Runtime 重拼 messages。官方流式常在增量里发 `output_item.added` / `function_call_arguments.done`，终态却是不带 `/response/output` 的 `response.completed`。旧解析只在 completed 里取原生数组，所以只有新策划会失败。 +- **处理**：流式按 `output_index` 累积全部 output item，`arguments.done` 写回对应 item；completed 仅在带非空 `output` 时覆盖。设计 Agent 的空 output 守卫保留。 +- **排查顺序**：先看 debug 响应里 `output` 是否为空、是否同时有工具调用；不要当成模型拒调工具或策划提示词错误。 +- **验证**：`platform-llm` 流式夹具覆盖「空 completed + 增量 item」能拿出可回放 `responses_output`，以及 completed 顶层 `output`。 + ## 2026-09-05 Planning V2 审批和续跑必须等过项目锁瞬时争用 - **现象**：策划 V2 在 GDD 审批提交修改意见后提示 `项目正在被其他写操作占用：...\\.agent\\project.lock`，聊天区再出现 `项目总控 Agent 执行失败，请稍后重试`。 diff --git a/server-rs/crates/platform-llm/src/lib.rs b/server-rs/crates/platform-llm/src/lib.rs index 195617a42..80382e778 100644 --- a/server-rs/crates/platform-llm/src/lib.rs +++ b/server-rs/crates/platform-llm/src/lib.rs @@ -701,6 +701,8 @@ struct ParsedStreamEvent { // 不能借它写 finish_reason，否则会覆盖 message_delta 给出的真实 end_turn。 is_completion: bool, tool_fragments: Vec, + // Responses 增量 output item。completed 若未带完整 output[]，靠这些槽位拼出可回放的原生数组。 + output_items: Vec<(u64, serde_json::Value)>, } // 三种协议的工具调用增量归一：slot 是协议各自的索引（Chat/Anthropic 的 index、 @@ -1908,7 +1910,7 @@ impl LlmClient { response_id, usage: accumulation.usage, tool_calls, - responses_output: accumulation.responses_output, + responses_output: compact_responses_output(accumulation.responses_output), }) } @@ -2256,11 +2258,15 @@ where is_terminal, is_completion, tool_fragments, + output_items, } = event; if let Some(output) = responses_output { accumulation.responses_output = output; } + for (slot, item) in output_items { + upsert_responses_output_item(&mut accumulation.responses_output, slot, item); + } if is_completion { accumulation.completion_observed = true; @@ -2289,6 +2295,13 @@ where // 工具调用只累加，不进 on_delta：调用方的流式通道仍然只承载文本。 let ids_repeated_in_event = tool_fragment_ids_repeated_in_event(&tool_fragments); for fragment in tool_fragments { + if let Some(arguments) = fragment.arguments_complete.as_deref() { + patch_responses_output_arguments( + &mut accumulation.responses_output, + fragment.slot, + arguments, + ); + } accumulation.push_tool_fragment(fragment, &ids_repeated_in_event)?; } @@ -3632,41 +3645,16 @@ fn parse_responses_sse_event(data: &str) -> Result, Ll // 工具会让纯文本的 completed-only 响应变成 EmptyResponse，让「正文 + 工具」 // 响应静默丢掉模型的前置说明。 text_snapshot: extract_responses_terminal_text(&parsed), - responses_output: parsed - .pointer("/response/output") - .and_then(serde_json::Value::as_array) - .cloned(), + responses_output: extract_responses_output_array(&parsed), tool_fragments: extract_responses_completed_tool_fragments(&parsed)?, ..Default::default() })), // 工具调用先由 output_item.added 宣告身份，再用 arguments delta 拼参数； // .done 给出权威完整参数，用它覆盖拼接结果。三个事件共用 output_index 作为槽位。 - "response.output_item.added" => { - let item = parsed.get("item"); - if item - .and_then(|item| item.get("type")) - .and_then(serde_json::Value::as_str) - != Some("function_call") - { - return Ok(None); - } - let slot = responses_output_slot(&parsed) - .ok_or_else(|| missing_tool_slot_error("Responses", event_type, "output_index"))?; - Ok(Some(ParsedStreamEvent { - tool_fragments: vec![ToolCallFragment { - slot, - id: item - .and_then(|item| item.get("call_id").or_else(|| item.get("id"))) - .and_then(serde_json::Value::as_str) - .map(str::to_string), - name: item - .and_then(|item| item.get("name")) - .and_then(serde_json::Value::as_str) - .map(str::to_string), - ..Default::default() - }], - ..Default::default() - })) + // output_item.added / done 同时把原生 item 按槽位写入，供 completed 未带 + // output[] 时回放；function_call 仍走 tool_fragments 归并。 + "response.output_item.added" | "response.output_item.done" => { + parse_responses_output_item_event(&parsed, event_type) } "response.function_call_arguments.delta" => { let slot = responses_output_slot(&parsed) @@ -3732,14 +3720,82 @@ fn extract_responses_terminal_text(parsed: &serde_json::Value) -> Option extract_responses_text(&envelope).filter(|text| !text.trim().is_empty()) } +fn extract_responses_output_array(parsed: &serde_json::Value) -> Option> { + parsed + .pointer("/response/output") + .or_else(|| parsed.get("output")) + .and_then(serde_json::Value::as_array) + .filter(|items| !items.is_empty()) + .cloned() +} + +fn upsert_responses_output_item( + output: &mut Vec, + slot: u64, + item: serde_json::Value, +) { + let index = slot as usize; + if index >= output.len() { + output.resize(index + 1, serde_json::Value::Null); + } + output[index] = item; +} + +fn patch_responses_output_arguments(output: &mut [serde_json::Value], slot: u64, arguments: &str) { + let Some(serde_json::Value::Object(map)) = output.get_mut(slot as usize) else { + return; + }; + map.insert( + "arguments".to_string(), + serde_json::Value::String(arguments.to_string()), + ); +} + +fn compact_responses_output(mut output: Vec) -> Vec { + output.retain(|item| !item.is_null()); + output +} + +fn parse_responses_output_item_event( + parsed: &serde_json::Value, + event_type: &str, +) -> Result, LlmError> { + let Some(item) = parsed.get("item").filter(|item| item.is_object()).cloned() else { + return Ok(None); + }; + let slot = responses_output_slot(parsed) + .ok_or_else(|| missing_tool_slot_error("Responses", event_type, "output_index"))?; + let mut event = ParsedStreamEvent { + output_items: vec![(slot, item.clone())], + ..Default::default() + }; + if item.get("type").and_then(serde_json::Value::as_str) == Some("function_call") { + event.tool_fragments = vec![ToolCallFragment { + slot, + id: item + .get("call_id") + .or_else(|| item.get("id")) + .and_then(serde_json::Value::as_str) + .map(str::to_string), + name: item + .get("name") + .and_then(serde_json::Value::as_str) + .map(str::to_string), + arguments_complete: item + .get("arguments") + .and_then(serde_json::Value::as_str) + .filter(|arguments| !arguments.is_empty()) + .map(str::to_string), + ..Default::default() + }]; + } + Ok(Some(event)) +} + fn extract_responses_completed_tool_fragments( parsed: &serde_json::Value, ) -> Result, LlmError> { - let Some(items) = parsed - .get("response") - .and_then(|response| response.get("output")) - .and_then(serde_json::Value::as_array) - else { + let Some(items) = extract_responses_output_array(parsed) else { return Ok(Vec::new()); }; @@ -6320,6 +6376,90 @@ mod tests { arguments: r#"{"city":"杭州"}"#.to_string(), }] ); + assert_eq!( + response.responses_output, + vec![serde_json::json!({ + "id":"fc_0", + "type":"function_call", + "status":"in_progress", + "arguments":"{\"city\":\"杭州\"}", + "call_id":"call_EkOU4", + "name":"get_weather" + })] + ); + } + + #[tokio::test] + async fn stream_run_replays_incremental_native_output_when_completed_omits_it() { + let server_url = spawn_mock_server(vec![MockResponse { + status_line: "200 OK", + content_type: "text/event-stream; charset=utf-8", + body: concat!( + r#"data: {"type":"response.output_item.added","item":{"type":"reasoning","id":"rs_1","summary":[],"encrypted_content":"encrypted-reasoning-payload"},"output_index":0}"#, "\n\n", + r#"data: {"type":"response.output_item.added","item":{"id":"fc_1","type":"function_call","status":"in_progress","arguments":"","call_id":"call_1","name":"get_weather"},"output_index":1}"#, "\n\n", + r#"data: {"type":"response.function_call_arguments.done","item_id":"fc_1","output_index":1,"arguments":"{\"city\":\"杭州\"}"}"#, "\n\n", + r#"data: {"type":"response.output_item.done","item":{"id":"fc_1","type":"function_call","status":"completed","arguments":"{\"city\":\"杭州\"}","call_id":"call_1","name":"get_weather"},"output_index":1}"#, "\n\n", + r#"data: {"type":"response.completed"}"#, "\n\n" + ) + .to_string(), + extra_headers: Vec::new(), + }]); + let response = build_test_client(server_url, 0) + .stream_run( + weather_tool_request(LlmApiKind::OpenAiResponses).with_responses_input(vec![ + serde_json::json!({"role":"user", "content":"查询天气"}), + ]), + |_| {}, + ) + .await + .expect("incremental native output"); + assert_eq!( + response.responses_output, + vec![ + serde_json::json!({ + "type":"reasoning", + "id":"rs_1", + "summary":[], + "encrypted_content":"encrypted-reasoning-payload" + }), + serde_json::json!({ + "id":"fc_1", + "type":"function_call", + "status":"completed", + "arguments":"{\"city\":\"杭州\"}", + "call_id":"call_1", + "name":"get_weather" + }), + ] + ); + assert_eq!(response.tool_calls[0].id, "call_1"); + assert_eq!(response.tool_calls[0].arguments, r#"{"city":"杭州"}"#); + } + + #[tokio::test] + async fn stream_run_uses_completed_top_level_output_when_response_wrapper_missing() { + let output = native_responses_output_fixture(); + let body = format!( + "data: {}\n\n", + serde_json::json!({"type":"response.completed", "output":output}) + ); + let server_url = spawn_mock_server(vec![MockResponse { + status_line: "200 OK", + content_type: "text/event-stream", + body, + extra_headers: Vec::new(), + }]); + let response = build_test_client(server_url, 0) + .stream_run( + weather_tool_request(LlmApiKind::OpenAiResponses).with_responses_input(vec![ + serde_json::json!({"role":"user", "content":"查询天气"}), + ]), + |_| {}, + ) + .await + .expect("top-level completed output"); + assert_eq!(response.responses_output, output); + assert_eq!(response.tool_calls[0].id, "call_1"); } // 同一个终态载荷里两条相同 call_id、相同函数名、不同参数：上游违反了 call id 唯一性。