From 9c2add9a09e3270a3723f69b45aa9039078a9dc4 Mon Sep 17 00:00:00 2001 From: zhizhuowq <60698291+zhizhuowq@users.noreply.github.com> Date: Wed, 27 May 2026 10:49:41 +0800 Subject: [PATCH] =?UTF-8?q?fix(proxy):=20=E4=BF=AE=E5=A4=8D=20Claude=20?= =?UTF-8?q?=E5=85=BC=E5=AE=B9=E6=A8=A1=E5=BC=8F=E4=B8=8B=E6=B5=81=E5=BC=8F?= =?UTF-8?q?=E5=93=8D=E5=BA=94=E5=9B=A0=E7=A9=BA=20tool=5Fcalls=20=E6=95=B0?= =?UTF-8?q?=E7=BB=84=E5=AF=BC=E8=87=B4=20block=20=E7=8A=B6=E6=80=81?= =?UTF-8?q?=E9=87=8D=E7=BD=AE=20(#2915)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 问题：OpenAI 兼容 Provider（如 MiniMax）在发送 reasoning_content 流时， 每个 chunk 包含 "tool_calls": [] 空数组，导致每个字符被当作独立 block 处理 根因：streaming.rs:343 的 if let Some(tool_calls) 匹配空数组进入处理分支， 发送 content_block_stop 并重置 current_non_tool_block_type 为 None 修改：在 tool_calls 处理前添加 if !tool_calls.is_empty() 判断 示例输入： ```jsonl {"delta": {"content": null, "reasoning_content": "用户", "role": "assistant", "tool_calls": []}} {"delta": {"content": null, "reasoning_content": "用", "role": "assistant", "tool_calls": []}} {"delta": {"content": null, "reasoning_content": "中文", "role": "assistant", "tool_calls": []}} ... ``` 示例错误输出： ```text event: content_block_start → index 0, type thinking event: content_block_delta → thinking "用户" event: content_block_stop → index 0 event: content_block_start → index 1, type thinking event: content_block_delta → thinking "用" event: content_block_stop → index 1 event: content_block_start → index 2, type thinking event: content_block_delta → thinking "中文" event: content_block_stop → index 2 ... ``` Co-authored-by: WangQiang --- src-tauri/src/proxy/providers/streaming.rs | 272 +++++++++++---------- 1 file changed, 137 insertions(+), 135 deletions(-) diff --git a/src-tauri/src/proxy/providers/streaming.rs b/src-tauri/src/proxy/providers/streaming.rs index 3bfc8b305..c69e78496 100644 --- a/src-tauri/src/proxy/providers/streaming.rs +++ b/src-tauri/src/proxy/providers/streaming.rs @@ -341,151 +341,153 @@ pub fn create_anthropic_sse_stream( // 处理工具调用 if let Some(tool_calls) = &choice.delta.tool_calls { - if let Some(index) = current_non_tool_block_index.take() { - let event = json!({ - "type": "content_block_stop", - "index": index - }); - let sse_data = format!("event: content_block_stop\ndata: {}\n\n", - serde_json::to_string(&event).unwrap_or_default()); - yield Ok(Bytes::from(sse_data)); - } - current_non_tool_block_type = None; + if !tool_calls.is_empty() { + if let Some(index) = current_non_tool_block_index.take() { + let event = json!({ + "type": "content_block_stop", + "index": index + }); + let sse_data = format!("event: content_block_stop\ndata: {}\n\n", + serde_json::to_string(&event).unwrap_or_default()); + yield Ok(Bytes::from(sse_data)); + } + current_non_tool_block_type = None; - for tool_call in tool_calls { - let ( - anthropic_index, - id, - name, - should_start, - pending_after_start, - immediate_delta, - ) = { - let state = tool_blocks_by_index - .entry(tool_call.index) - .or_insert_with(|| { - let index = next_content_index; - next_content_index += 1; - ToolBlockState { - anthropic_index: index, - id: String::new(), - name: String::new(), - started: false, - pending_args: String::new(), - consecutive_whitespace: 0, - aborted: false, - } - }); - - // 如果此 tool call 已被中止（无限空白 bug），跳过后续处理 - if state.aborted { - continue; - } - - if let Some(id) = &tool_call.id { - state.id = id.clone(); - } - if let Some(function) = &tool_call.function { - if let Some(name) = &function.name { - state.name = name.clone(); - } - } - - let should_start = - !state.started - && !state.id.is_empty() - && !state.name.is_empty(); - if should_start { - state.started = true; - } - let pending_after_start = if should_start - && !state.pending_args.is_empty() - { - Some(std::mem::take(&mut state.pending_args)) - } else { - None - }; - let args_delta = tool_call - .function - .as_ref() - .and_then(|f| f.arguments.clone()); - let immediate_delta = if let Some(args) = args_delta { - // 无限空白 bug 检测：跟踪连续空白字符 - for ch in args.chars() { - if ch.is_whitespace() { - state.consecutive_whitespace += 1; - } else { - state.consecutive_whitespace = 0; - } - } - if state.consecutive_whitespace >= INFINITE_WHITESPACE_THRESHOLD { - log::warn!( - "[Copilot] 检测到无限空白 bug (tool: {}), 中止此 tool call 流", - state.name - ); - state.aborted = true; - None - } else if state.started { - Some(args) - } else { - state.pending_args.push_str(&args); - None - } - } else { - None - }; - ( - state.anthropic_index, - state.id.clone(), - state.name.clone(), + for tool_call in tool_calls { + let ( + anthropic_index, + id, + name, should_start, pending_after_start, immediate_delta, - ) - }; + ) = { + let state = tool_blocks_by_index + .entry(tool_call.index) + .or_insert_with(|| { + let index = next_content_index; + next_content_index += 1; + ToolBlockState { + anthropic_index: index, + id: String::new(), + name: String::new(), + started: false, + pending_args: String::new(), + consecutive_whitespace: 0, + aborted: false, + } + }); - if should_start { - let event = json!({ - "type": "content_block_start", - "index": anthropic_index, - "content_block": { - "type": "tool_use", - "id": id, - "name": name + // 如果此 tool call 已被中止（无限空白 bug），跳过后续处理 + if state.aborted { + continue; } - }); - let sse_data = format!("event: content_block_start\ndata: {}\n\n", - serde_json::to_string(&event).unwrap_or_default()); - yield Ok(Bytes::from(sse_data)); - open_tool_block_indices.insert(anthropic_index); - } - if let Some(args) = pending_after_start { - let event = json!({ - "type": "content_block_delta", - "index": anthropic_index, - "delta": { - "type": "input_json_delta", - "partial_json": args + if let Some(id) = &tool_call.id { + state.id = id.clone(); + } + if let Some(function) = &tool_call.function { + if let Some(name) = &function.name { + state.name = name.clone(); + } } - }); - let sse_data = format!("event: content_block_delta\ndata: {}\n\n", - serde_json::to_string(&event).unwrap_or_default()); - yield Ok(Bytes::from(sse_data)); - } - if let Some(args) = immediate_delta { - let event = json!({ - "type": "content_block_delta", - "index": anthropic_index, - "delta": { - "type": "input_json_delta", - "partial_json": args + let should_start = + !state.started + && !state.id.is_empty() + && !state.name.is_empty(); + if should_start { + state.started = true; } - }); - let sse_data = format!("event: content_block_delta\ndata: {}\n\n", - serde_json::to_string(&event).unwrap_or_default()); - yield Ok(Bytes::from(sse_data)); + let pending_after_start = if should_start + && !state.pending_args.is_empty() + { + Some(std::mem::take(&mut state.pending_args)) + } else { + None + }; + let args_delta = tool_call + .function + .as_ref() + .and_then(|f| f.arguments.clone()); + let immediate_delta = if let Some(args) = args_delta { + // 无限空白 bug 检测：跟踪连续空白字符 + for ch in args.chars() { + if ch.is_whitespace() { + state.consecutive_whitespace += 1; + } else { + state.consecutive_whitespace = 0; + } + } + if state.consecutive_whitespace >= INFINITE_WHITESPACE_THRESHOLD { + log::warn!( + "[Copilot] 检测到无限空白 bug (tool: {}), 中止此 tool call 流", + state.name + ); + state.aborted = true; + None + } else if state.started { + Some(args) + } else { + state.pending_args.push_str(&args); + None + } + } else { + None + }; + ( + state.anthropic_index, + state.id.clone(), + state.name.clone(), + should_start, + pending_after_start, + immediate_delta, + ) + }; + + if should_start { + let event = json!({ + "type": "content_block_start", + "index": anthropic_index, + "content_block": { + "type": "tool_use", + "id": id, + "name": name + } + }); + let sse_data = format!("event: content_block_start\ndata: {}\n\n", + serde_json::to_string(&event).unwrap_or_default()); + yield Ok(Bytes::from(sse_data)); + open_tool_block_indices.insert(anthropic_index); + } + + if let Some(args) = pending_after_start { + let event = json!({ + "type": "content_block_delta", + "index": anthropic_index, + "delta": { + "type": "input_json_delta", + "partial_json": args + } + }); + let sse_data = format!("event: content_block_delta\ndata: {}\n\n", + serde_json::to_string(&event).unwrap_or_default()); + yield Ok(Bytes::from(sse_data)); + } + + if let Some(args) = immediate_delta { + let event = json!({ + "type": "content_block_delta", + "index": anthropic_index, + "delta": { + "type": "input_json_delta", + "partial_json": args + } + }); + let sse_data = format!("event: content_block_delta\ndata: {}\n\n", + serde_json::to_string(&event).unwrap_or_default()); + yield Ok(Bytes::from(sse_data)); + } } } }