修复Responses推理项分段串接

按item_id与summary_index隔离累计补充快照替换流式通知测试
2026-09-14 23:12:31 +08:00
parent 9e4f06b3e1
commit cb002896fb
2 changed files with 117 additions and 16 deletions
@@ -19,7 +19,7 @@
 2. `OpenAiChat`、`OpenAiResponses` 与 `Anthropic` 三类 API kind 都支持 JSON 请求、非流式响应和 SSE 流式响应；默认 API kind 仍为 `OpenAiResponses`。
 3. 三类协议都使用统一的 `function_tools` / `tool_choice` 输入和 `LlmRunResponse.tool_calls` 输出。Anthropic 请求使用顶层 `tools[].input_schema` 与对象形态 `tool_choice`；Anthropic URL 默认在 base URL 后拼 `/v1/messages`，如果 base URL 已以 `/v1` 结尾则只拼 `/messages`。
 4. Anthropic 当前仍不支持 `web_search`、图片内容和纯 system 消息；至少需要一条非 system 文本消息。角色动画、图片、视频、资产轮询仍留在其他平台适配和业务模块任务里。
-5. 流式 `on_delta` 发送正文增量、可选的独立 reasoning 增量与完成原因；reasoning 只有在 `LlmRunRequest.capture_reasoning=true` 时才累计，默认关闭。Responses 的 reasoning summary 按 `summary_index` 分段累计，单段 `.done` 只校正对应段，不覆盖其它段；Anthropic 的 `thinking_delta` 同样进入独立 reasoning 通道。工具调用增量在 crate 内按 slot 聚合，完整调用只从最终 `LlmRunResponse.tool_calls` 读取。reasoning 不进入 `delta_text`、`accumulated_text`、正式 assistant message 或工具参数；上下文管理、后台执行和业务状态不写回本 crate。
+5. 流式 `on_delta` 发送正文增量、可选的独立 reasoning 增量与完成原因；reasoning 只有在 `LlmRunRequest.capture_reasoning=true` 时才累计，默认关闭。Responses 的 reasoning summary 按 `item_id + summary_index` 分段累计，单段 `.done` 只校正对应段，不覆盖其它 item；Anthropic 的 `thinking_delta` 同样进入独立 reasoning 通道。分段快照即使改写了原内容也会通知调用方刷新累计 reasoning。工具调用增量在 crate 内按 slot 聚合，完整调用只从最终 `LlmRunResponse.tool_calls` 读取。reasoning 不进入 `delta_text`、`accumulated_text`、正式 assistant message 或工具参数；上下文管理、后台执行和业务状态不写回本 crate。
 6. 支持按 provider 打标签，但不把业务 prompt、SSE 转发和模块状态写回本 crate。
 7. `DashScope` 当前只通过“调用方显式提供兼容文本网关 base url”的方式接入，不复用图像 API。
 8. 角色动画、图片、视频、资产轮询仍留在后续 `platform-llm` / `platform-oss` / 业务模块任务里另行实现。
@@ -713,7 +713,8 @@ struct OpenAiCompatibleSseParser {
 struct ParsedStreamEvent {
    delta_text: Option<String>,
    reasoning_delta: Option<String>,
-    // Responses 的 summary delta / done 事件按 summary_index 分段；Chat 与终态全量快照为 None。
+    // Responses 的 summary part 还必须绑定 item_id；summary_index 只在单个 reasoning item 内唯一。
+    reasoning_summary_item_id: Option<String>,
    reasoning_summary_index: Option<u64>,
    reasoning_snapshot: Option<String>,
    responses_output: Option<Vec<serde_json::Value>>,
@@ -882,8 +883,9 @@ fn normalize_tool_calls(
 struct StreamAccumulation {
    text: String,
    reasoning: String,
-    // Responses summary part 的流式累计。reasoning 保留为对外统一的拼接结果。
-    reasoning_summary_parts: BTreeMap<u64, String>,
+    // Responses summary part 的流式累计，key 必须同时包含 reasoning item 与 part 索引。
+    reasoning_summary_parts: BTreeMap<(String, u64), String>,
+    reasoning_summary_item_order: Vec<String>,
    responses_output: Vec<serde_json::Value>,
    finish_reason: Option<String>,
    usage: Option<LlmTokenUsage>,
@@ -2318,6 +2320,7 @@ where
        let ParsedStreamEvent {
            delta_text,
            reasoning_delta,
+            reasoning_summary_item_id,
            reasoning_summary_index,
            reasoning_snapshot,
            responses_output,
@@ -2346,24 +2349,43 @@ where
        } else {
            String::new()
        };
+        let mut reasoning_snapshot_corrected = false;
        if capture_reasoning {
            if let Some(summary_index) = reasoning_summary_index {
+                let item_id = reasoning_summary_item_id
+                    .unwrap_or_else(|| "__default_reasoning_item__".to_string());
+                if !accumulation
+                    .reasoning_summary_item_order
+                    .iter()
+                    .any(|known| known == &item_id)
+                {
+                    accumulation
+                        .reasoning_summary_item_order
+                        .push(item_id.clone());
+                }
+                let key = (item_id, summary_index);
                if !reasoning_delta.is_empty() {
                    accumulation
                        .reasoning_summary_parts
-                        .entry(summary_index)
+                        .entry(key.clone())
                        .or_default()
                        .push_str(reasoning_delta.as_str());
                    accumulation.reasoning = accumulation
-                        .reasoning_summary_parts
-                        .values()
-                        .map(String::as_str)
+                        .reasoning_summary_item_order
+                        .iter()
+                        .flat_map(|item_id| {
+                            accumulation
+                                .reasoning_summary_parts
+                                .iter()
+                                .filter(move |((known_id, _), _)| known_id == item_id)
+                                .map(|(_, text)| text.as_str())
+                        })
                        .collect::<String>();
                }
                if let Some(snapshot) = reasoning_snapshot.filter(|text| !text.trim().is_empty()) {
                    let current_part = accumulation
                        .reasoning_summary_parts
-                        .get(&summary_index)
+                        .get(&key)
                        .cloned()
                        .unwrap_or_default();
                    if snapshot != current_part {
@@ -2375,14 +2397,19 @@ where
                                .unwrap_or_default()
                                .to_string()
                        };
-                        accumulation
-                            .reasoning_summary_parts
-                            .insert(summary_index, snapshot);
+                        accumulation.reasoning_summary_parts.insert(key, snapshot);
                        accumulation.reasoning = accumulation
-                            .reasoning_summary_parts
-                            .values()
-                            .map(String::as_str)
+                            .reasoning_summary_item_order
+                            .iter()
+                            .flat_map(|item_id| {
+                                accumulation
+                                    .reasoning_summary_parts
+                                    .iter()
+                                    .filter(move |((known_id, _), _)| known_id == item_id)
+                                    .map(|(_, text)| text.as_str())
+                            })
                            .collect::<String>();
+                        reasoning_snapshot_corrected = true;
                    } else {
                        reasoning_delta.clear();
                    }
@@ -2484,6 +2511,7 @@ where
            accumulation.finish_reason = Some(event_finish_reason.clone());
            if has_delta
                || !reasoning_delta.is_empty()
+                || reasoning_snapshot_corrected
                || emit_finish_only_delta
                || snapshot_corrected
            {
@@ -2496,7 +2524,7 @@ where
                };
                on_delta(&update);
            }
-        } else if has_delta || !reasoning_delta.is_empty() {
+        } else if has_delta || !reasoning_delta.is_empty() || reasoning_snapshot_corrected {
            let update = LlmStreamDelta {
                accumulated_text: accumulation.text.clone(),
                delta_text,
@@ -3871,6 +3899,10 @@ fn parse_responses_sse_event(data: &str) -> Result<Option<ParsedStreamEvent>, Ll
                .get("delta")
                .and_then(serde_json::Value::as_str)
                .map(str::to_string),
+            reasoning_summary_item_id: parsed
+                .get("item_id")
+                .and_then(serde_json::Value::as_str)
+                .map(str::to_string),
            reasoning_summary_index: parsed
                .get("summary_index")
                .and_then(serde_json::Value::as_u64),
@@ -3881,6 +3913,10 @@ fn parse_responses_sse_event(data: &str) -> Result<Option<ParsedStreamEvent>, Ll
                .get("text")
                .and_then(serde_json::Value::as_str)
                .map(str::to_string),
+            reasoning_summary_item_id: parsed
+                .get("item_id")
+                .and_then(serde_json::Value::as_str)
+                .map(str::to_string),
            reasoning_summary_index: parsed
                .get("summary_index")
                .and_then(serde_json::Value::as_u64),
@@ -7422,6 +7458,71 @@ mod tests {
        }));
    }

+    #[tokio::test]
+    async fn stream_run_keeps_reasoning_parts_separate_across_items() {
+        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.reasoning_summary_text.delta","item_id":"item-a","summary_index":0,"delta":"前置"}"#, "\n\n",
+                r#"data: {"type":"response.reasoning_summary_text.delta","item_id":"item-b","summary_index":0,"delta":"后置"}"#, "\n\n",
+                r#"data: {"type":"response.completed","response":{"output":[{"type":"message","content":[{"type":"output_text","text":"答案"}]}]}}"#, "\n\n"
+            )
+            .to_string(),
+            extra_headers: Vec::new(),
+        }]);
+
+        let response = build_test_client(server_url, 0)
+            .stream_run(
+                LlmRunRequest::single_turn("系统", "用户")
+                    .with_openai_responses()
+                    .with_reasoning_capture(true),
+                |_| {},
+            )
+            .await
+            .expect("reasoning items should remain separate");
+
+        assert_eq!(response.reasoning, "前置后置");
+    }
+
+    #[tokio::test]
+    async fn stream_run_notifies_when_reasoning_snapshot_replaces_non_prefix_part() {
+        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.reasoning_summary_text.delta","item_id":"item-a","summary_index":0,"delta":"旧内容"}"#, "\n\n",
+                r#"data: {"type":"response.reasoning_summary_text.done","item_id":"item-a","summary_index":0,"text":"新内容"}"#, "\n\n",
+                r#"data: {"type":"response.completed","response":{"output":[{"type":"message","content":[{"type":"output_text","text":"答案"}]}]}}"#, "\n\n"
+            )
+            .to_string(),
+            extra_headers: Vec::new(),
+        }]);
+
+        let mut updates = Vec::new();
+        let response = build_test_client(server_url, 0)
+            .stream_run(
+                LlmRunRequest::single_turn("系统", "用户")
+                    .with_openai_responses()
+                    .with_reasoning_capture(true),
+                |delta| {
+                    updates.push((
+                        delta.accumulated_reasoning.clone(),
+                        delta.reasoning_delta.clone(),
+                    ))
+                },
+            )
+            .await
+            .expect("replacement snapshot should parse");
+
+        assert_eq!(response.reasoning, "新内容");
+        assert!(
+            updates
+                .iter()
+                .any(|(accumulated, delta)| accumulated == "新内容" && delta.is_empty())
+        );
+    }
+
    #[tokio::test]
    async fn stream_run_accumulates_parallel_anthropic_tool_calls() {
        let server_url = spawn_mock_server(vec![MockResponse {