修复Responses多段推理流式累计

按summary_index独立累计reasoning 补充多段summary回归测试与说明
2026-09-14 20:45:58 +08:00
parent 51dc5f2ad7
commit 59289199ed
2 changed files with 112 additions and 18 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` 时才累计，默认关闭。工具调用增量在 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 按 `summary_index` 分段累计，单段 `.done` 只校正对应段，不覆盖其它段。工具调用增量在 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` / 业务模块任务里另行实现。
@@ -1,4 +1,5 @@
 use std::{
+    collections::BTreeMap,
    env,
    error::Error,
    fmt, fs,
@@ -710,6 +711,8 @@ struct OpenAiCompatibleSseParser {
 struct ParsedStreamEvent {
    delta_text: Option<String>,
    reasoning_delta: Option<String>,
+    // Responses 的 summary delta / done 事件按 summary_index 分段；Chat 与终态全量快照为 None。
+    reasoning_summary_index: Option<u64>,
    reasoning_snapshot: Option<String>,
    responses_output: Option<Vec<serde_json::Value>>,
    // 终态事件携带的完整正文快照。必须与 delta_text 分开：它不是增量，按增量累加会让
@@ -877,6 +880,8 @@ fn normalize_tool_calls(
 struct StreamAccumulation {
    text: String,
    reasoning: String,
+    // Responses summary part 的流式累计。reasoning 保留为对外统一的拼接结果。
+    reasoning_summary_parts: BTreeMap<u64, String>,
    responses_output: Vec<serde_json::Value>,
    finish_reason: Option<String>,
    usage: Option<LlmTokenUsage>,
@@ -2311,6 +2316,7 @@ where
        let ParsedStreamEvent {
            delta_text,
            reasoning_delta,
+            reasoning_summary_index,
            reasoning_snapshot,
            responses_output,
            text_snapshot,
@@ -2338,24 +2344,66 @@ where
        } else {
            String::new()
        };
-        let mut reasoning_snapshot_applied = false;
        if capture_reasoning {
-            if let Some(snapshot) = reasoning_snapshot.filter(|text| !text.trim().is_empty()) {
-                if snapshot != accumulation.reasoning {
-                    reasoning_delta = if accumulation.reasoning.is_empty() {
-                        snapshot.clone()
-                    } else {
-                        snapshot
-                            .strip_prefix(accumulation.reasoning.as_str())
-                            .unwrap_or_default()
-                            .to_string()
-                    };
-                    accumulation.reasoning = snapshot;
-                    reasoning_snapshot_applied = true;
+            if let Some(summary_index) = reasoning_summary_index {
+                if !reasoning_delta.is_empty() {
+                    accumulation
+                        .reasoning_summary_parts
+                        .entry(summary_index)
+                        .or_default()
+                        .push_str(reasoning_delta.as_str());
+                    accumulation.reasoning = accumulation
+                        .reasoning_summary_parts
+                        .values()
+                        .map(String::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)
+                        .cloned()
+                        .unwrap_or_default();
+                    if snapshot != current_part {
+                        reasoning_delta = if current_part.is_empty() {
+                            snapshot.clone()
+                        } else {
+                            snapshot
+                                .strip_prefix(current_part.as_str())
+                                .unwrap_or_default()
+                                .to_string()
+                        };
+                        accumulation
+                            .reasoning_summary_parts
+                            .insert(summary_index, snapshot);
+                        accumulation.reasoning = accumulation
+                            .reasoning_summary_parts
+                            .values()
+                            .map(String::as_str)
+                            .collect::<String>();
+                    } else {
+                        reasoning_delta.clear();
+                    }
+                }
+            } else {
+                let mut reasoning_snapshot_applied = false;
+                if let Some(snapshot) = reasoning_snapshot.filter(|text| !text.trim().is_empty()) {
+                    if snapshot != accumulation.reasoning {
+                        reasoning_delta = if accumulation.reasoning.is_empty() {
+                            snapshot.clone()
+                        } else {
+                            snapshot
+                                .strip_prefix(accumulation.reasoning.as_str())
+                                .unwrap_or_default()
+                                .to_string()
+                        };
+                        accumulation.reasoning = snapshot;
+                        reasoning_snapshot_applied = true;
+                    }
+                }
+                if !reasoning_snapshot_applied && !reasoning_delta.is_empty() {
+                    accumulation.reasoning.push_str(reasoning_delta.as_str());
                }
-            }
-            if !reasoning_snapshot_applied && !reasoning_delta.is_empty() {
-                accumulation.reasoning.push_str(reasoning_delta.as_str());
            }
        }

@@ -3819,6 +3867,9 @@ 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_index: parsed
+                .get("summary_index")
+                .and_then(serde_json::Value::as_u64),
            ..Default::default()
        })),
        "response.reasoning_summary_text.done" => Ok(Some(ParsedStreamEvent {
@@ -3826,6 +3877,9 @@ 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_index: parsed
+                .get("summary_index")
+                .and_then(serde_json::Value::as_u64),
            ..Default::default()
        })),
        // completed 事件携带完整 output；有的网关只发它而不发增量事件，这里再取一遍，
@@ -5463,11 +5517,12 @@ mod tests {

        let responses = parse_sse_event_block(
            LlmApiKind::OpenAiResponses,
-            r#"data: {"type":"response.reasoning_summary_text.delta","delta":"推理"}"#,
+            r#"data: {"type":"response.reasoning_summary_text.delta","summary_index":2,"delta":"推理"}"#,
        )
        .expect("Responses SSE should parse")
        .expect("Responses event should exist");
        assert_eq!(responses.reasoning_delta.as_deref(), Some("推理"));
+        assert_eq!(responses.reasoning_summary_index, Some(2));

        let terminal = parse_sse_event_block(
            LlmApiKind::OpenAiResponses,
@@ -7242,6 +7297,45 @@ mod tests {
        );
    }

+    #[tokio::test]
+    async fn stream_run_keeps_multiple_responses_reasoning_summary_parts() {
+        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","summary_index":0,"delta":"第一段"}"#, "\n\n",
+                r#"data: {"type":"response.reasoning_summary_text.done","summary_index":0,"text":"第一段"}"#, "\n\n",
+                r#"data: {"type":"response.reasoning_summary_text.delta","summary_index":1,"delta":"第二段"}"#, "\n\n",
+                r#"data: {"type":"response.reasoning_summary_text.done","summary_index":1,"text":"第二段"}"#, "\n\n",
+                // 终态故意不带 reasoning，验证不能依赖 response.completed 恢复前面的 summary part。
+                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("multiple reasoning summary parts should parse");
+
+        assert_eq!(response.reasoning, "第一段第二段");
+        assert!(updates.iter().any(|(accumulated, delta)| {
+            accumulated == "第一段第二段" && delta == "第二段"
+        }));
+    }
+
    #[tokio::test]
    async fn stream_run_accumulates_parallel_anthropic_tool_calls() {
        let server_url = spawn_mock_server(vec![MockResponse {