diff --git a/server-rs/crates/platform-llm/README.md b/server-rs/crates/platform-llm/README.md index 576aa0306..dc6f5a0b7 100644 --- a/server-rs/crates/platform-llm/README.md +++ b/server-rs/crates/platform-llm/README.md @@ -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` / 业务模块任务里另行实现。 diff --git a/server-rs/crates/platform-llm/src/lib.rs b/server-rs/crates/platform-llm/src/lib.rs index 8c8e246ef..b3a6d00a3 100644 --- a/server-rs/crates/platform-llm/src/lib.rs +++ b/server-rs/crates/platform-llm/src/lib.rs @@ -1,4 +1,5 @@ use std::{ + collections::BTreeMap, env, error::Error, fmt, fs, @@ -710,6 +711,8 @@ struct OpenAiCompatibleSseParser { struct ParsedStreamEvent { delta_text: Option, reasoning_delta: Option, + // Responses 的 summary delta / done 事件按 summary_index 分段；Chat 与终态全量快照为 None。 + reasoning_summary_index: Option, reasoning_snapshot: Option, responses_output: Option>, // 终态事件携带的完整正文快照。必须与 delta_text 分开：它不是增量，按增量累加会让 @@ -877,6 +880,8 @@ fn normalize_tool_calls( struct StreamAccumulation { text: String, reasoning: String, + // Responses summary part 的流式累计。reasoning 保留为对外统一的拼接结果。 + reasoning_summary_parts: BTreeMap, responses_output: Vec, finish_reason: Option, usage: Option, @@ -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::(); + } + 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::(); + } 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, 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, 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 {