From 7c61e9a532e79dba2b98745bc346411abeae5185 Mon Sep 17 00:00:00 2001 From: Linghong Date: Sat, 25 Jul 2026 13:38:41 +0000 Subject: [PATCH] =?UTF-8?q?=E8=A1=A5=E9=BD=90=20Anthropic=20=E5=8E=9F?= =?UTF-8?q?=E7=94=9F=E5=B7=A5=E5=85=B7=E4=B8=8E=E4=B8=89=E5=8D=8F=E8=AE=AE?= =?UTF-8?q?=E6=B5=81=E5=BC=8F=E5=B7=A5=E5=85=B7=E8=B0=83=E7=94=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Anthropic 请求发送 tools 与对象形态 tool_choice 并解析 tool_use block，解除本地对 function tools 的拦截 Chat / Responses / Anthropic 三种协议的流式工具增量按槽位聚合，收尾校验参数为完整 JSON，截断流不返回半截参数 解除 App 侧 Anthropic 降级为文本 JSON 协议的两处守卫与对应提示词分支 新增依赖真实凭据的流式工具验收用例，默认 ignore Co-Authored-By: Claude Opus 5 --- .../src-tauri/src/agent/interaction.rs | 30 +- .../provider_request_builders.rs | 15 +- server-rs/crates/platform-llm/src/lib.rs | 911 ++++++++++++++++-- .../tests/live_stream_tool_calls.rs | 101 ++ 4 files changed, 927 insertions(+), 130 deletions(-) create mode 100644 server-rs/crates/platform-llm/tests/live_stream_tool_calls.rs diff --git a/apps/ai-game-creator-shell/src-tauri/src/agent/interaction.rs b/apps/ai-game-creator-shell/src-tauri/src/agent/interaction.rs index c47055f39..9677b2ce8 100644 --- a/apps/ai-game-creator-shell/src-tauri/src/agent/interaction.rs +++ b/apps/ai-game-creator-shell/src-tauri/src/agent/interaction.rs @@ -93,17 +93,13 @@ fn agent_interaction_function_tools() -> Vec { .collect() } -fn agent_interaction_system_prompt(agent_id: &str, native_tools: bool) -> String { +fn agent_interaction_system_prompt(agent_id: &str) -> String { let role_prompt = if agent_id == GAME_CREATOR_PROJECT_SUPERVISOR_AGENT_ID { game_creator_project_supervisor_chat_system_prompt() } else { game_creator_role_agent_chat_system_prompt() }; - let protocol = if native_tools { - "普通问答、身份说明、架构解释、方案讨论和必要澄清直接用自然语言回复。只有确实需要宿主持久能力时才调用一个 function tool，调用工具时不要同时输出回复文本。不要根据单个关键词决定是否执行，要理解整句的否定、假设、范围和上下文。" - } else { - "你必须只输出一个 JSON 对象，不要代码块或额外文字。允许的结构为：{\"action\":\"reply\",\"reply\":\"自然语言回复\"}、{\"action\":\"execute\"}、{\"action\":\"resume\"}、{\"action\":\"project_location\"}。不要根据单个关键词决定 action，要理解整句的否定、假设、范围和上下文。" - }; + let protocol = "普通问答、身份说明、架构解释、方案讨论和必要澄清直接用自然语言回复。只有确实需要宿主持久能力时才调用一个 function tool，调用工具时不要同时输出回复文本。不要根据单个关键词决定是否执行，要理解整句的否定、假设、范围和上下文。"; format!( "{role_prompt}\n\n你现在位于统一的 Agent interaction loop。{protocol} 高影响请求仍不明确时直接追问，不要擅自启动 Runtime。" ) @@ -114,7 +110,7 @@ fn build_agent_interaction_request_for_session( agent_id: &str, session_id: &str, prompt: &str, -) -> Result<(GameCreatorLlmConfig, String, LlmRunRequest, bool), String> { +) -> Result<(GameCreatorLlmConfig, String, LlmRunRequest), String> { let prompt = prompt.trim(); if prompt.is_empty() { return Err("交互内容不能为空".to_string()); @@ -125,7 +121,6 @@ fn build_agent_interaction_request_for_session( let (llm, config_path, context) = build_game_creator_role_agent_context_for_session(root, agent_id, Some(session_id))?; let api_kind = parse_game_creator_llm_api_kind(&llm.api_kind)?; - let native_tools = api_kind != LlmApiKind::Anthropic; let user_prompt = if context.trim().is_empty() { format!("用户这轮输入：\n{prompt}") } else { @@ -133,18 +128,15 @@ fn build_agent_interaction_request_for_session( "项目上下文如下。只把它当作背景，不要逐字复述。\n\n{context}\n\n用户这轮输入：\n{prompt}" ) }; - let mut request = LlmRunRequest::new(vec![ - LlmMessage::system(agent_interaction_system_prompt(agent_id, native_tools)), + let request = LlmRunRequest::new(vec![ + LlmMessage::system(agent_interaction_system_prompt(agent_id)), LlmMessage::user(user_prompt), ]) .with_api_kind(api_kind) - .with_max_output_tokens(AGENT_INTERACTION_MAX_OUTPUT_TOKENS); - if native_tools { - request = request - .with_function_tools(agent_interaction_function_tools()) - .with_tool_choice(platform_llm::LlmToolChoice::Auto); - } - Ok((llm, config_path, request, native_tools)) + .with_max_output_tokens(AGENT_INTERACTION_MAX_OUTPUT_TOKENS) + .with_function_tools(agent_interaction_function_tools()) + .with_tool_choice(platform_llm::LlmToolChoice::Auto); + Ok((llm, config_path, request)) } pub(crate) async fn decide_game_creator_agent_interaction_turn_for_session_at( @@ -157,10 +149,10 @@ pub(crate) async fn decide_game_creator_agent_interaction_turn_for_session_at where F: FnMut(&platform_llm::LlmStreamDelta), { - let (llm, config_path, request, native_tools) = + let (llm, config_path, request) = build_agent_interaction_request_for_session(root, agent_id, session_id, prompt)?; let client = build_game_creator_agent_runtime_llm_client(&llm, &config_path)?; - let response = if native_tools && llm.stream { + let response = if llm.stream { let fallback_request = request.clone(); match client.stream_run(request, |delta| on_delta(delta)).await { Ok(response) => response, diff --git a/apps/ai-game-creator-shell/src-tauri/src/agent/runtime_actions/provider_request_builders.rs b/apps/ai-game-creator-shell/src-tauri/src/agent/runtime_actions/provider_request_builders.rs index 6ff53dc37..2afc6b5f6 100644 --- a/apps/ai-game-creator-shell/src-tauri/src/agent/runtime_actions/provider_request_builders.rs +++ b/apps/ai-game-creator-shell/src-tauri/src/agent/runtime_actions/provider_request_builders.rs @@ -139,11 +139,7 @@ pub(in crate::agent) fn build_game_creator_agent_background_tool_plan_request( "command.start 使用 {\"program\":\"受信任 PATH 中的裸可执行名\"", ); let api_kind = parse_game_creator_llm_api_kind(&llm.api_kind)?; - let protocol_prompt = if api_kind == platform_llm::LlmApiKind::Anthropic { - "当前 Provider 不提供 function tools，请返回上述 schema 的单个完整 JSON object；不要解释、markdown 或代码围栏。" - } else { - "必须直接调用当前请求提供的原生函数：需要更新持久计划时调用 update_agent_plan，需要行动时调用对应动作工具，已有观察足够时调用 respond_to_user。只有步骤或状态真实变化时才单独调用 update_agent_plan；当前 in_progress 步骤已具备执行条件时必须在同一响应调用对应动作工具，不能只改计划解释。不要调用未广告的旧 submit_agent_tool_plan，也不要把计划或动作放在普通文本中。" - }; + let protocol_prompt = "必须直接调用当前请求提供的原生函数：需要更新持久计划时调用 update_agent_plan，需要行动时调用对应动作工具，已有观察足够时调用 respond_to_user。只有步骤或状态真实变化时才单独调用 update_agent_plan；当前 in_progress 步骤已具备执行条件时必须在同一响应调用对应动作工具，不能只改计划解释。不要调用未广告的旧 submit_agent_tool_plan，也不要把计划或动作放在普通文本中。"; let mut system_prompt = game_creator_agent_runtime_tool_plan_system_prompt_for_agent(agent_id); if autonomous_game_build { system_prompt.push_str( @@ -167,12 +163,9 @@ pub(in crate::agent) fn build_game_creator_agent_background_tool_plan_request( ]) .with_api_kind(api_kind) .with_max_output_tokens(AGENT_RUNTIME_TOOL_PLAN_MAX_OUTPUT_TOKENS) - .with_response_text_verbosity(platform_llm::LlmResponseTextVerbosity::Low); - if api_kind != platform_llm::LlmApiKind::Anthropic { - request = request - .with_function_tools(build_agent_runtime_native_function_tools(mcp_catalog)?) - .with_tool_choice(platform_llm::LlmToolChoice::Required); - } + .with_response_text_verbosity(platform_llm::LlmResponseTextVerbosity::Low) + .with_function_tools(build_agent_runtime_native_function_tools(mcp_catalog)?) + .with_tool_choice(platform_llm::LlmToolChoice::Required); request = apply_game_creator_llm_web_search( apply_game_creator_llm_reasoning_effort(request, &llm)?, &llm, diff --git a/server-rs/crates/platform-llm/src/lib.rs b/server-rs/crates/platform-llm/src/lib.rs index 62ea4aa8d..40cdeb750 100644 --- a/server-rs/crates/platform-llm/src/lib.rs +++ b/server-rs/crates/platform-llm/src/lib.rs @@ -120,6 +120,14 @@ impl LlmToolChoice { Self::Required => "required", } } + + // Anthropic 用 any 表达“必须调用某个工具”，与 OpenAI 的 required 同义。 + fn as_anthropic_type(self) -> &'static str { + match self { + Self::Auto => "auto", + Self::Required => "any", + } + } } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] @@ -380,6 +388,10 @@ struct AnthropicMessagesRequestBody { #[serde(skip_serializing_if = "Option::is_none")] system: Option, messages: Vec, + #[serde(skip_serializing_if = "Option::is_none")] + tools: Option>, + #[serde(skip_serializing_if = "Option::is_none")] + tool_choice: Option, } #[derive(Serialize)] @@ -388,6 +400,21 @@ struct AnthropicInputMessage { content: String, } +// Anthropic 工具与 OpenAI 的差异：schema 字段名为 input_schema，且没有 function 包装层与 strict。 +#[derive(Serialize)] +struct AnthropicTool { + name: String, + description: String, + input_schema: serde_json::Value, +} + +// Anthropic 的 tool_choice 必须是对象，发送裸字符串会被上游拒绝。 +#[derive(Serialize)] +struct AnthropicToolChoice { + #[serde(rename = "type")] + choice_type: &'static str, +} + #[derive(Serialize)] struct ResponsesReasoningOptions { effort: &'static str, @@ -454,16 +481,23 @@ struct ChatCompletionsMessage { tool_calls: Option>, } +// 流式分片只有首片带 id / name，后续片仅有 index 与 arguments 片段，因此字段全部可选。 #[derive(Deserialize)] struct ChatCompletionsToolCall { - id: String, - function: ChatCompletionsFunctionCall, + #[serde(default)] + id: Option, + #[serde(default)] + index: Option, + #[serde(default)] + function: Option, } #[derive(Deserialize)] struct ChatCompletionsFunctionCall { - name: String, - arguments: String, + #[serde(default)] + name: Option, + #[serde(default)] + arguments: Option, } #[derive(Deserialize)] @@ -546,6 +580,13 @@ struct AnthropicContentBlock { block_type: Option, #[serde(default)] text: Option, + // tool_use block 字段：id 与 name 标识调用，input 是已解析的 JSON object。 + #[serde(default)] + id: Option, + #[serde(default)] + name: Option, + #[serde(default)] + input: Option, } #[derive(Deserialize)] @@ -563,12 +604,117 @@ struct OpenAiCompatibleSseParser { terminated: bool, } -#[derive(Debug)] +#[derive(Debug, Default)] struct ParsedStreamEvent { delta_text: Option, finish_reason: Option, usage: Option, is_terminal: bool, + tool_fragments: Vec, +} + +// 三种协议的工具调用增量归一：slot 是协议各自的索引（Chat/Anthropic 的 index、 +// Responses 的 output_index），id 与 name 只在首个分片出现，参数按到达顺序拼接。 +#[derive(Debug, Default)] +struct ToolCallFragment { + slot: u64, + id: Option, + name: Option, + arguments_delta: Option, + // 上游给出完整参数时（Responses 的 .done）直接覆盖，避免依赖分片拼接结果。 + arguments_complete: Option, +} + +#[derive(Debug)] +struct PendingToolCall { + slot: u64, + id: Option, + name: Option, + arguments: String, +} + +// 流式累加状态：文本、终止原因、用量与按槽位聚合的工具调用。 +#[derive(Debug, Default)] +struct StreamAccumulation { + text: String, + finish_reason: Option, + usage: Option, + tool_calls: Vec, +} + +impl StreamAccumulation { + fn push_tool_fragment(&mut self, fragment: ToolCallFragment) { + if !self + .tool_calls + .iter() + .any(|pending| pending.slot == fragment.slot) + { + self.tool_calls.push(PendingToolCall { + slot: fragment.slot, + id: None, + name: None, + arguments: String::new(), + }); + } + let entry = self + .tool_calls + .iter_mut() + .find(|pending| pending.slot == fragment.slot) + .expect("slot was just ensured"); + + if let Some(id) = fragment.id { + entry.id = Some(id); + } + if let Some(name) = fragment.name { + entry.name = Some(name); + } + if let Some(delta) = fragment.arguments_delta { + entry.arguments.push_str(delta.as_str()); + } + // 上游给出的完整参数是权威值，直接覆盖分片拼接结果。 + if let Some(complete) = fragment.arguments_complete { + entry.arguments = complete; + } + } + + // 流结束后固化。参数必须是完整 JSON，否则说明流被截断，不能把半截参数交给业务层。 + fn finish_tool_calls(&self) -> Result, LlmError> { + self.tool_calls + .iter() + .map(|pending| { + let id = pending.id.clone().ok_or_else(|| { + LlmError::Deserialize(format!( + "LLM 流式工具调用缺少 id：slot={}", + pending.slot + )) + })?; + let name = pending.name.clone().ok_or_else(|| { + LlmError::Deserialize(format!( + "LLM 流式工具调用缺少函数名：slot={}", + pending.slot + )) + })?; + let arguments = pending.arguments.trim(); + if arguments.is_empty() { + return Ok(LlmToolCall { + id, + name, + arguments: "{}".to_string(), + }); + } + serde_json::from_str::(arguments).map_err(|error| { + LlmError::Deserialize(format!( + "LLM 流式工具调用参数不是完整 JSON：name={name}, error={error}" + )) + })?; + Ok(LlmToolCall { + id, + name, + arguments: arguments.to_string(), + }) + }) + .collect() + } } #[derive(Debug)] @@ -913,12 +1059,6 @@ impl LlmRunRequest { } if self.api_kind == LlmApiKind::Anthropic { - if !self.function_tools.is_empty() || self.tool_choice.is_some() { - return Err(LlmError::InvalidRequest( - "Anthropic api_kind 暂不支持 function tools".to_string(), - )); - } - if self.enable_web_search { return Err(LlmError::InvalidRequest( "Anthropic api_kind 暂不支持 web_search".to_string(), @@ -1125,9 +1265,7 @@ impl LlmClient { .map(str::to_string); let mut parser = OpenAiCompatibleSseParser::new(request.api_kind); - let mut accumulated_text = String::new(); - let mut finish_reason = None; - let mut usage = None; + let mut accumulation = StreamAccumulation::default(); let mut undecoded_chunk_bytes = Vec::new(); let emit_finish_only_delta = request.api_kind == LlmApiKind::OpenAiChat; let mut stream_terminated = false; @@ -1138,8 +1276,7 @@ impl LlmClient { Err(error) => { let llm_error = map_stream_read_error(error, 1); if retain_completed_stream_after_tail_error( - accumulated_text.as_str(), - &finish_reason, + &accumulation, "read_stream_failed", &llm_error, ) { @@ -1168,8 +1305,7 @@ impl LlmClient { Ok(decoded) => decoded, Err(error) => { if retain_completed_stream_after_tail_error( - accumulated_text.as_str(), - &finish_reason, + &accumulation, "decode_stream_failed", &error, ) { @@ -1193,9 +1329,7 @@ impl LlmClient { } stream_terminated = consume_stream_parser_result( parser.push_chunk(chunk_text.as_ref()), - &mut accumulated_text, - &mut finish_reason, - &mut usage, + &mut accumulation, emit_finish_only_delta, &mut on_delta, ) @@ -1222,8 +1356,7 @@ impl LlmClient { let llm_error = LlmError::Deserialize(format!("解析 LLM 流式 UTF-8 响应失败：{error}")); if retain_completed_stream_after_tail_error( - accumulated_text.as_str(), - &finish_reason, + &accumulation, "decode_stream_failed", &llm_error, ) { @@ -1245,9 +1378,7 @@ impl LlmClient { if !stream_terminated && !trailing_text.is_empty() { stream_terminated = consume_stream_parser_result( parser.push_chunk(trailing_text), - &mut accumulated_text, - &mut finish_reason, - &mut usage, + &mut accumulation, emit_finish_only_delta, &mut on_delta, ) @@ -1268,9 +1399,7 @@ impl LlmClient { if !stream_terminated { consume_stream_parser_result( parser.finish(), - &mut accumulated_text, - &mut finish_reason, - &mut usage, + &mut accumulation, emit_finish_only_delta, &mut on_delta, ) @@ -1287,8 +1416,39 @@ impl LlmClient { })?; } - let content = accumulated_text.trim().to_string(); - if content.is_empty() { + let tool_calls = accumulation.finish_tool_calls().map_err(|error| { + log_llm_raw_failure( + &self.config, + &request, + true, + 1, + "parse_stream_tool_calls_failed", + parser.raw_text().as_str(), + ); + error + })?; + + // 一致性断言：上游已表明本轮是工具调用，却一个都没累加出来，说明该网关的事件形状 + // 不在已支持范围内。此时必须显式失败让调用方回退非流式，不能静默丢掉调用。 + if tool_calls.is_empty() + && accumulation + .finish_reason + .as_deref() + .is_some_and(|reason| reason == "tool_use" || reason == "tool_calls") + { + log_llm_raw_failure( + &self.config, + &request, + true, + 1, + "stream_tool_calls_missing", + parser.raw_text().as_str(), + ); + return Err(LlmError::StreamUnavailable); + } + + let content = accumulation.text.trim().to_string(); + if content.is_empty() && tool_calls.is_empty() { log_llm_raw_failure( &self.config, &request, @@ -1304,10 +1464,10 @@ impl LlmClient { provider: self.config.provider(), model: resolved_model, text: content, - finish_reason, + finish_reason: accumulation.finish_reason, response_id, - usage, - tool_calls: Vec::new(), + usage: accumulation.usage, + tool_calls, }) } @@ -1561,9 +1721,7 @@ impl OpenAiCompatibleSseParser { fn consume_stream_parser_result( result: Result, SseEventDrainError>, - accumulated_text: &mut String, - finish_reason: &mut Option, - usage: &mut Option, + accumulation: &mut StreamAccumulation, emit_finish_only_delta: bool, on_delta: &mut F, ) -> Result @@ -1574,25 +1732,14 @@ where Ok(events) => (events, None), Err(error) => (error.parsed_events, Some(error.error)), }; - let stream_terminated = consume_stream_events( - events, - accumulated_text, - finish_reason, - usage, - emit_finish_only_delta, - on_delta, - ); + let stream_terminated = + consume_stream_events(events, accumulation, emit_finish_only_delta, on_delta); if stream_terminated { return Ok(true); } if let Some(error) = tail_error { - if retain_completed_stream_after_tail_error( - accumulated_text.as_str(), - finish_reason, - "parse_stream_failed", - &error, - ) { + if retain_completed_stream_after_tail_error(accumulation, "parse_stream_failed", &error) { return Ok(true); } return Err(error); @@ -1602,8 +1749,7 @@ where } fn retain_completed_stream_after_tail_error( - accumulated_text: &str, - finish_reason: &Option, + accumulation: &StreamAccumulation, stage: &str, error: &LlmError, ) -> bool { @@ -1614,8 +1760,12 @@ fn retain_completed_stream_after_tail_error( | LlmErrorKind::Transport | LlmErrorKind::Deserialize ); - let retain_response = - !accumulated_text.trim().is_empty() && finish_reason.is_some() && is_tolerable_tail_error; + // 工具调用尚未拼完整时不能保留：半截参数比直接失败更危险。 + let tool_calls_complete = accumulation.finish_tool_calls().is_ok(); + let retain_response = !accumulation.text.trim().is_empty() + && accumulation.finish_reason.is_some() + && tool_calls_complete + && is_tolerable_tail_error; if retain_response { warn!( @@ -1629,9 +1779,7 @@ fn retain_completed_stream_after_tail_error( fn consume_stream_events( events: Vec, - accumulated_text: &mut String, - finish_reason: &mut Option, - usage: &mut Option, + accumulation: &mut StreamAccumulation, emit_finish_only_delta: bool, on_delta: &mut F, ) -> bool @@ -1644,23 +1792,29 @@ where finish_reason: event_finish_reason, usage: event_usage, is_terminal, + tool_fragments, } = event; if let Some(event_usage) = event_usage { - *usage = Some(event_usage); + accumulation.usage = Some(event_usage); + } + + // 工具调用只累加，不进 on_delta：调用方的流式通道仍然只承载文本。 + for fragment in tool_fragments { + accumulation.push_tool_fragment(fragment); } let delta_text = delta_text.unwrap_or_default(); let has_delta = !delta_text.is_empty(); if has_delta { - accumulated_text.push_str(delta_text.as_str()); + accumulation.text.push_str(delta_text.as_str()); } if let Some(event_finish_reason) = event_finish_reason { - *finish_reason = Some(event_finish_reason.clone()); + accumulation.finish_reason = Some(event_finish_reason.clone()); if has_delta || emit_finish_only_delta { let update = LlmStreamDelta { - accumulated_text: accumulated_text.clone(), + accumulated_text: accumulation.text.clone(), delta_text, finish_reason: Some(event_finish_reason), }; @@ -1668,7 +1822,7 @@ where } } else if has_delta { let update = LlmStreamDelta { - accumulated_text: accumulated_text.clone(), + accumulated_text: accumulation.text.clone(), delta_text, finish_reason: None, }; @@ -1791,6 +1945,18 @@ fn build_anthropic_messages_request_body( }) .collect(); + let tools = (!request.function_tools.is_empty()).then(|| { + request + .function_tools + .iter() + .map(|function| AnthropicTool { + name: function.name.clone(), + description: function.description.clone(), + input_schema: function.parameters.clone(), + }) + .collect() + }); + AnthropicMessagesRequestBody { model: request.resolved_model(fallback_model).to_string(), max_tokens: request @@ -1799,6 +1965,12 @@ fn build_anthropic_messages_request_body( stream, system: (!system.is_empty()).then_some(system), messages, + tools, + tool_choice: request + .tool_choice + .map(|choice| AnthropicToolChoice { + choice_type: choice.as_anthropic_type(), + }), } } @@ -2141,12 +2313,14 @@ fn parse_anthropic_response( let parsed: AnthropicResponseEnvelope = serde_json::from_str(raw_text).map_err(|error| { LlmError::Deserialize(format!("解析 LLM Anthropic JSON 响应失败：{error}")) })?; + let tool_calls = extract_anthropic_tool_calls(&parsed); let content = extract_anthropic_text(&parsed) - .ok_or(LlmError::EmptyResponse)? + .unwrap_or_default() .trim() .to_string(); - if content.is_empty() { + // 纯 tool_use 响应没有 text block，此时不能按空响应处理。 + if content.is_empty() && tool_calls.is_empty() { return Err(LlmError::EmptyResponse); } @@ -2161,7 +2335,7 @@ fn parse_anthropic_response( completion_tokens: usage.output_tokens, total_tokens: usage.input_tokens.saturating_add(usage.output_tokens), }), - tool_calls: Vec::new(), + tool_calls, }) } @@ -2200,6 +2374,25 @@ fn extract_responses_tool_calls(parsed: &ResponsesResponseEnvelope) -> Vec Vec { + parsed + .content + .iter() + .filter(|block| block.block_type.as_deref() == Some("tool_use")) + .filter_map(|block| { + Some(LlmToolCall { + id: block.id.as_ref()?.clone(), + name: block.name.as_ref()?.clone(), + arguments: block + .input + .as_ref() + .map(serde_json::Value::to_string) + .unwrap_or_else(|| "{}".to_string()), + }) + }) + .collect() +} + fn extract_anthropic_text(parsed: &AnthropicResponseEnvelope) -> Option { let text = parsed .content @@ -2241,10 +2434,13 @@ fn extract_chat_tool_calls(choice: &ChatCompletionsChoice) -> Vec { }) .unwrap_or_default() .iter() - .map(|tool_call| LlmToolCall { - id: tool_call.id.clone(), - name: tool_call.function.name.clone(), - arguments: tool_call.function.arguments.clone(), + .filter_map(|tool_call| { + let function = tool_call.function.as_ref()?; + Some(LlmToolCall { + id: tool_call.id.as_ref()?.clone(), + name: function.name.as_ref()?.clone(), + arguments: function.arguments.clone().unwrap_or_default(), + }) }) .collect() } @@ -2316,10 +2512,8 @@ fn parse_sse_event_block( if data.trim() == "[DONE]" { return if api_kind == LlmApiKind::OpenAiChat { Ok(Some(ParsedStreamEvent { - delta_text: None, - finish_reason: None, - usage: None, is_terminal: true, + ..Default::default() })) } else { Ok(None) @@ -2352,10 +2546,8 @@ fn parse_sse_event_block( let Some(first_choice) = parsed.choices.first() else { return if let Some(usage) = parsed.usage { Ok(Some(ParsedStreamEvent { - delta_text: None, - finish_reason: None, usage: Some(usage), - is_terminal: false, + ..Default::default() })) } else { // OpenAI-compatible gateways may emit heartbeat or metadata-only @@ -2368,10 +2560,40 @@ fn parse_sse_event_block( delta_text: extract_message_text(first_choice), finish_reason: first_choice.finish_reason.clone(), usage: parsed.usage, - is_terminal: false, + tool_fragments: extract_chat_tool_fragments(first_choice), + ..Default::default() })) } +// Chat 分片：首片带 index + id + function.name，后续片只有 index + function.arguments。 +fn extract_chat_tool_fragments(choice: &ChatCompletionsChoice) -> Vec { + let Some(tool_calls) = choice + .delta + .as_ref() + .and_then(|delta| delta.tool_calls.as_deref()) + else { + return Vec::new(); + }; + + tool_calls + .iter() + .enumerate() + .map(|(position, tool_call)| ToolCallFragment { + slot: tool_call.index.unwrap_or(position as u64), + id: tool_call.id.clone(), + name: tool_call + .function + .as_ref() + .and_then(|function| function.name.clone()), + arguments_delta: tool_call + .function + .as_ref() + .and_then(|function| function.arguments.clone()), + arguments_complete: None, + }) + .collect() +} + fn parse_responses_sse_event(data: &str) -> Result, LlmError> { let parsed: serde_json::Value = serde_json::from_str(data).map_err(|error| { LlmError::Deserialize(format!("解析 LLM Responses SSE 事件失败：{error}")) @@ -2387,16 +2609,77 @@ fn parse_responses_sse_event(data: &str) -> Result, Ll .get("delta") .and_then(serde_json::Value::as_str) .map(str::to_string), - finish_reason: None, - usage: None, - is_terminal: false, + ..Default::default() })), + // completed 事件携带完整 output；有的网关只发它而不发增量事件，这里再取一遍， + // 槽位沿用 output 数组下标，与 output_index 语义一致，可安全覆盖增量拼接结果。 "response.completed" => Ok(Some(ParsedStreamEvent { - delta_text: None, finish_reason: Some("completed".to_string()), - usage: None, - is_terminal: false, + 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 Some(slot) = responses_output_slot(&parsed) else { + return Ok(None); + }; + 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() + })) + } + "response.function_call_arguments.delta" => { + let Some(slot) = responses_output_slot(&parsed) else { + return Ok(None); + }; + Ok(Some(ParsedStreamEvent { + tool_fragments: vec![ToolCallFragment { + slot, + arguments_delta: parsed + .get("delta") + .and_then(serde_json::Value::as_str) + .map(str::to_string), + ..Default::default() + }], + ..Default::default() + })) + } + "response.function_call_arguments.done" => { + let Some(slot) = responses_output_slot(&parsed) else { + return Ok(None); + }; + Ok(Some(ParsedStreamEvent { + tool_fragments: vec![ToolCallFragment { + slot, + arguments_complete: parsed + .get("arguments") + .and_then(serde_json::Value::as_str) + .map(str::to_string), + ..Default::default() + }], + ..Default::default() + })) + } "response.failed" | "error" => { let message = parsed .get("error") @@ -2414,6 +2697,52 @@ fn parse_responses_sse_event(data: &str) -> Result, Ll } } +fn extract_responses_completed_tool_fragments(parsed: &serde_json::Value) -> Vec { + let Some(items) = parsed + .get("response") + .and_then(|response| response.get("output")) + .and_then(serde_json::Value::as_array) + else { + return Vec::new(); + }; + + items + .iter() + .enumerate() + .filter(|(_, item)| { + item.get("type").and_then(serde_json::Value::as_str) == Some("function_call") + }) + .map(|(index, item)| ToolCallFragment { + slot: index as u64, + 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() + }) + .collect() +} + +fn responses_output_slot(parsed: &serde_json::Value) -> Option { + parsed + .get("output_index") + .and_then(serde_json::Value::as_u64) +} + +fn anthropic_block_slot(parsed: &serde_json::Value) -> Option { + parsed.get("index").and_then(serde_json::Value::as_u64) +} + fn parse_anthropic_sse_event(data: &str) -> Result, LlmError> { let parsed: serde_json::Value = serde_json::from_str(data).map_err(|error| { LlmError::Deserialize(format!("解析 LLM Anthropic SSE 事件失败：{error}")) @@ -2424,12 +2753,60 @@ fn parse_anthropic_sse_event(data: &str) -> Result, Ll .unwrap_or_default(); match event_type { + // tool_use block 的 id 与 name 只在 content_block_start 出现；此时 input 恒为空对象， + // 不能拿它初始化参数，否则会和后续 input_json_delta 拼出非法 JSON。 + "content_block_start" => { + let block = parsed.get("content_block"); + if block + .and_then(|block| block.get("type")) + .and_then(serde_json::Value::as_str) + != Some("tool_use") + { + return Ok(None); + } + let Some(slot) = anthropic_block_slot(&parsed) else { + return Ok(None); + }; + Ok(Some(ParsedStreamEvent { + tool_fragments: vec![ToolCallFragment { + slot, + id: block + .and_then(|block| block.get("id")) + .and_then(serde_json::Value::as_str) + .map(str::to_string), + name: block + .and_then(|block| block.get("name")) + .and_then(serde_json::Value::as_str) + .map(str::to_string), + ..Default::default() + }], + ..Default::default() + })) + } "content_block_delta" => { let delta = parsed.get("delta"); let delta_type = delta .and_then(|value| value.get("type")) .and_then(serde_json::Value::as_str) .unwrap_or_default(); + + if delta_type == "input_json_delta" { + let Some(slot) = anthropic_block_slot(&parsed) else { + return Ok(None); + }; + return Ok(Some(ParsedStreamEvent { + tool_fragments: vec![ToolCallFragment { + slot, + arguments_delta: delta + .and_then(|value| value.get("partial_json")) + .and_then(serde_json::Value::as_str) + .map(str::to_string), + ..Default::default() + }], + ..Default::default() + })); + } + if delta_type != "text_delta" { return Ok(None); } @@ -2439,20 +2816,16 @@ fn parse_anthropic_sse_event(data: &str) -> Result, Ll .and_then(|value| value.get("text")) .and_then(serde_json::Value::as_str) .map(str::to_string), - finish_reason: None, - usage: None, - is_terminal: false, + ..Default::default() })) } "message_delta" => Ok(Some(ParsedStreamEvent { - delta_text: None, finish_reason: parsed .get("delta") .and_then(|value| value.get("stop_reason")) .and_then(serde_json::Value::as_str) .map(str::to_string), - usage: None, - is_terminal: false, + ..Default::default() })), // message_stop 只是流终止信号；真正的 stop_reason 已由 message_delta 提供， // 这里不要伪造 finish_reason，否则会覆盖掉 end_turn 等真实值。 @@ -2639,23 +3012,120 @@ mod tests { } #[test] - fn run_request_rejects_function_tools_for_anthropic() { - let error = LlmRunRequest::single_turn("系统", "用户") + fn anthropic_request_body_maps_function_tools_to_input_schema() { + let config = LlmConfig::new( + LlmProvider::OpenAiCompatible, + "https://example.com/anthropic".to_string(), + "secret".to_string(), + "model-a".to_string(), + DEFAULT_REQUEST_TIMEOUT_MS, + DEFAULT_MAX_RETRIES, + DEFAULT_RETRY_BACKOFF_MS, + ) + .expect("config should be valid"); + let request = LlmRunRequest::single_turn("系统", "用户") .with_anthropic() - .with_function_tools(vec![LlmFunctionTool::new( - "submit_plan", - "提交计划", - serde_json::json!({ "type": "object" }), - )]) - .validate() - .expect_err("anthropic function tools should fail locally"); + .with_function_tools(vec![ + LlmFunctionTool::new( + "get_weather", + "查询天气", + serde_json::json!({ "type": "object", "properties": { "city": { "type": "string" } } }), + ) + .with_strict(true), + ]) + .with_tool_choice(LlmToolChoice::Required); + request.validate().expect("anthropic tools should validate"); + let body = build_request_body(&request, &config, false); + let json = serde_json::to_value(&body).expect("body should serialize"); + + assert_eq!(json["tools"][0]["name"], "get_weather"); + assert_eq!(json["tools"][0]["description"], "查询天气"); + assert_eq!(json["tools"][0]["input_schema"]["type"], "object"); + // Anthropic 没有 parameters / strict 字段，映射时必须丢弃。 + assert!(json["tools"][0].get("parameters").is_none()); + assert!(json["tools"][0].get("strict").is_none()); + // tool_choice 必须是对象；Required 对应 Anthropic 的 any。 + assert_eq!(json["tool_choice"], serde_json::json!({ "type": "any" })); + } + + #[test] + fn anthropic_request_body_omits_tool_fields_without_tools() { + let config = LlmConfig::new( + LlmProvider::OpenAiCompatible, + "https://example.com/anthropic".to_string(), + "secret".to_string(), + "model-a".to_string(), + DEFAULT_REQUEST_TIMEOUT_MS, + DEFAULT_MAX_RETRIES, + DEFAULT_RETRY_BACKOFF_MS, + ) + .expect("config should be valid"); + let request = LlmRunRequest::single_turn("系统", "用户").with_anthropic(); + + let json = serde_json::to_value(build_request_body(&request, &config, false)) + .expect("body should serialize"); + + assert!(json.get("tools").is_none()); + assert!(json.get("tool_choice").is_none()); + } + + #[test] + fn anthropic_response_parses_tool_use_blocks_without_text() { + let raw = r#"{ + "id": "msg_1", + "model": "model-a", + "stop_reason": "tool_use", + "content": [ + { "type": "tool_use", "id": "call_1", "name": "get_weather", "input": { "city": "杭州" } } + ] + }"#; + + let response = parse_anthropic_response(LlmProvider::OpenAiCompatible, "fallback", raw) + .expect("tool-only response should parse"); + + assert_eq!(response.text, ""); + assert_eq!(response.finish_reason.as_deref(), Some("tool_use")); assert_eq!( - error, - LlmError::InvalidRequest("Anthropic api_kind 暂不支持 function tools".to_string()) + response.tool_calls, + vec![LlmToolCall { + id: "call_1".to_string(), + name: "get_weather".to_string(), + arguments: r#"{"city":"杭州"}"#.to_string(), + }] ); } + #[test] + fn anthropic_response_keeps_text_alongside_tool_use() { + let raw = r#"{ + "id": "msg_2", + "model": "model-a", + "stop_reason": "tool_use", + "content": [ + { "type": "text", "text": "我来帮你查询。" }, + { "type": "tool_use", "id": "call_2", "name": "get_weather", "input": {} } + ] + }"#; + + let response = parse_anthropic_response(LlmProvider::OpenAiCompatible, "fallback", raw) + .expect("mixed response should parse"); + + assert_eq!(response.text, "我来帮你查询。"); + assert_eq!(response.tool_calls.len(), 1); + assert_eq!(response.tool_calls[0].arguments, "{}"); + } + + #[test] + fn anthropic_response_without_text_or_tool_calls_is_empty() { + let raw = r#"{ "id": "msg_3", "model": "model-a", "content": [] }"#; + + let error = parse_anthropic_response(LlmProvider::OpenAiCompatible, "fallback", raw) + .expect_err("empty content should fail"); + + assert_eq!(error, LlmError::EmptyResponse); + } + #[tokio::test] async fn run_sends_official_fallback_for_openai_compatible_clients() { let listener = TcpListener::bind("127.0.0.1:0").expect("listener should bind"); @@ -3872,6 +4342,247 @@ mod tests { ); } + // 以下三个流式工具用例的 SSE 原文取自真实端点：Anthropic 与 Chat/Responses 分别来自 + // MiniMax 的 anthropic 兼容层和 api.openai.com（gpt-4.1 / gpt-5.5）。 + fn weather_tool_request(api_kind: LlmApiKind) -> LlmRunRequest { + LlmRunRequest::single_turn("系统", "用户") + .with_api_kind(api_kind) + .with_function_tools(vec![LlmFunctionTool::new( + "get_weather", + "查询天气", + serde_json::json!({ "type": "object" }), + )]) + .with_tool_choice(LlmToolChoice::Auto) + } + + #[tokio::test] + async fn stream_run_accumulates_anthropic_tool_use_alongside_text() { + 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":"content_block_start","index":0,"content_block":{"type":"text","text":""}}"#, "\n\n", + r#"data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"我来"}}"#, "\n\n", + r#"data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"为您查询。"}}"#, "\n\n", + r#"data: {"type":"content_block_stop","index":0}"#, "\n\n", + r#"data: {"type":"content_block_start","index":1,"content_block":{"type":"tool_use","id":"call_019f98be1099","name":"get_weather","input":{}}}"#, "\n\n", + r#"data: {"type":"content_block_delta","index":1,"delta":{"type":"input_json_delta","partial_json":""}}"#, "\n\n", + r#"data: {"type":"content_block_delta","index":1,"delta":{"type":"input_json_delta","partial_json":"{"}}"#, "\n\n", + r#"data: {"type":"content_block_delta","index":1,"delta":{"type":"input_json_delta","partial_json":"\"city\":\"杭州\""}}"#, "\n\n", + r#"data: {"type":"content_block_delta","index":1,"delta":{"type":"input_json_delta","partial_json":"}"}}"#, "\n\n", + r#"data: {"type":"content_block_stop","index":1}"#, "\n\n", + r#"data: {"type":"message_delta","delta":{"stop_reason":"tool_use"}}"#, "\n\n", + r#"data: {"type":"message_stop"}"#, "\n\n" + ) + .to_string(), + extra_headers: Vec::new(), + }]); + + let client = build_test_client(server_url, 0); + let mut updates = Vec::new(); + let response = client + .stream_run(weather_tool_request(LlmApiKind::Anthropic), |delta| { + updates.push(delta.delta_text.clone()); + }) + .await + .expect("anthropic tool stream should succeed"); + + // 工具增量不进 on_delta，回调里只应看到文本。 + assert_eq!(updates, vec!["我来".to_string(), "为您查询。".to_string()]); + assert_eq!(response.text, "我来为您查询。"); + assert_eq!(response.finish_reason.as_deref(), Some("tool_use")); + assert_eq!( + response.tool_calls, + vec![LlmToolCall { + id: "call_019f98be1099".to_string(), + name: "get_weather".to_string(), + arguments: r#"{"city":"杭州"}"#.to_string(), + }] + ); + } + + #[tokio::test] + async fn stream_run_accumulates_chat_tool_call_fragments() { + let server_url = spawn_mock_server(vec![MockResponse { + status_line: "200 OK", + content_type: "text/event-stream; charset=utf-8", + body: concat!( + r#"data: {"choices":[{"index":0,"delta":{"role":"assistant","content":null,"tool_calls":[{"index":0,"id":"call_7gOveph","type":"function","function":{"name":"get_weather","arguments":""}}]},"finish_reason":null}]}"#, "\n\n", + r#"data: {"choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"function":{"arguments":"{\""}}]},"finish_reason":null}]}"#, "\n\n", + r#"data: {"choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"function":{"arguments":"city"}}]},"finish_reason":null}]}"#, "\n\n", + r#"data: {"choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"function":{"arguments":"\":\""}}]},"finish_reason":null}]}"#, "\n\n", + r#"data: {"choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"function":{"arguments":"杭州"}}]},"finish_reason":null}]}"#, "\n\n", + r#"data: {"choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"function":{"arguments":"\"}"}}]},"finish_reason":null}]}"#, "\n\n", + r#"data: {"choices":[{"index":0,"delta":{},"finish_reason":"tool_calls"}]}"#, "\n\n", + "data: [DONE]\n\n" + ) + .to_string(), + extra_headers: Vec::new(), + }]); + + let client = build_test_client(server_url, 0); + let response = client + .stream_run(weather_tool_request(LlmApiKind::OpenAiChat), |_| {}) + .await + .expect("chat tool stream should succeed"); + + // 纯工具调用没有文本，放宽后的空响应判定必须放行。 + assert_eq!(response.text, ""); + assert_eq!(response.finish_reason.as_deref(), Some("tool_calls")); + assert_eq!( + response.tool_calls, + vec![LlmToolCall { + id: "call_7gOveph".to_string(), + name: "get_weather".to_string(), + arguments: r#"{"city":"杭州"}"#.to_string(), + }] + ); + } + + #[tokio::test] + async fn stream_run_accumulates_responses_function_call() { + 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":{"id":"fc_0","type":"function_call","status":"in_progress","arguments":"","call_id":"call_EkOU4","name":"get_weather"},"output_index":0,"sequence_number":2}"#, "\n\n", + r#"data: {"type":"response.function_call_arguments.delta","delta":"{\"","item_id":"fc_0","output_index":0,"sequence_number":3}"#, "\n\n", + r#"data: {"type":"response.function_call_arguments.delta","delta":"city\":\"杭州","item_id":"fc_0","output_index":0,"sequence_number":4}"#, "\n\n", + r#"data: {"type":"response.function_call_arguments.delta","delta":"\"}","item_id":"fc_0","output_index":0,"sequence_number":5}"#, "\n\n", + r#"data: {"type":"response.function_call_arguments.done","item_id":"fc_0","output_index":0,"arguments":"{\"city\":\"杭州\"}","sequence_number":6}"#, "\n\n", + r#"data: {"type":"response.completed"}"#, "\n\n" + ) + .to_string(), + extra_headers: Vec::new(), + }]); + + let client = build_test_client(server_url, 0); + let response = client + .stream_run(weather_tool_request(LlmApiKind::OpenAiResponses), |_| {}) + .await + .expect("responses tool stream should succeed"); + + assert_eq!(response.finish_reason.as_deref(), Some("completed")); + assert_eq!( + response.tool_calls, + vec![LlmToolCall { + // call_id 优先于 item id，与非流式解析保持一致。 + id: "call_EkOU4".to_string(), + name: "get_weather".to_string(), + arguments: r#"{"city":"杭州"}"#.to_string(), + }] + ); + } + + #[tokio::test] + async fn stream_run_recovers_responses_tool_calls_from_completed_event_only() { + // 只发 completed、不发增量事件的网关也必须能解出工具调用。 + 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.completed","response":{"output":[{"id":"msg_0","type":"message","content":[{"type":"output_text","text":"我来查询。"}]},{"id":"fc_0","type":"function_call","call_id":"call_only","name":"get_weather","arguments":"{\"city\":\"杭州\"}"}]}}"#, "\n\n" + ) + .to_string(), + extra_headers: Vec::new(), + }]); + + let client = build_test_client(server_url, 0); + let response = client + .stream_run(weather_tool_request(LlmApiKind::OpenAiResponses), |_| {}) + .await + .expect("completed-only stream should succeed"); + + assert_eq!( + response.tool_calls, + vec![LlmToolCall { + id: "call_only".to_string(), + name: "get_weather".to_string(), + arguments: r#"{"city":"杭州"}"#.to_string(), + }] + ); + } + + #[tokio::test] + async fn stream_run_accumulates_parallel_anthropic_tool_calls() { + 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":"content_block_start","index":1,"content_block":{"type":"tool_use","id":"call_a","name":"get_weather","input":{}}}"#, "\n\n", + r#"data: {"type":"content_block_delta","index":1,"delta":{"type":"input_json_delta","partial_json":"{\"city\":\"杭州\"}"}}"#, "\n\n", + r#"data: {"type":"content_block_start","index":2,"content_block":{"type":"tool_use","id":"call_b","name":"get_air_quality","input":{}}}"#, "\n\n", + r#"data: {"type":"content_block_delta","index":2,"delta":{"type":"input_json_delta","partial_json":"{\"city\":\"杭州\"}"}}"#, "\n\n", + r#"data: {"type":"message_delta","delta":{"stop_reason":"tool_use"}}"#, "\n\n" + ) + .to_string(), + extra_headers: Vec::new(), + }]); + + let client = build_test_client(server_url, 0); + let response = client + .stream_run(weather_tool_request(LlmApiKind::Anthropic), |_| {}) + .await + .expect("parallel tool stream should succeed"); + + let names = response + .tool_calls + .iter() + .map(|call| call.name.as_str()) + .collect::>(); + assert_eq!(names, vec!["get_weather", "get_air_quality"]); + assert_eq!(response.tool_calls[1].id, "call_b"); + } + + #[tokio::test] + async fn stream_run_rejects_truncated_tool_arguments() { + // 参数只拼到一半就断流，不能把半截 JSON 交给业务层。 + 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":"content_block_start","index":1,"content_block":{"type":"tool_use","id":"call_1","name":"get_weather","input":{}}}"#, "\n\n", + r#"data: {"type":"content_block_delta","index":1,"delta":{"type":"input_json_delta","partial_json":"{\"city\":"}}"#, "\n\n", + r#"data: {"type":"message_delta","delta":{"stop_reason":"tool_use"}}"#, "\n\n" + ) + .to_string(), + extra_headers: Vec::new(), + }]); + + let client = build_test_client(server_url, 0); + let error = client + .stream_run(weather_tool_request(LlmApiKind::Anthropic), |_| {}) + .await + .expect_err("truncated arguments should fail"); + + assert!(matches!(error, LlmError::Deserialize(_))); + } + + #[tokio::test] + async fn stream_run_falls_back_when_tool_use_yields_no_fragments() { + // 上游说了本轮是工具调用，但事件形状不在已支持范围内，一个分片都没解出来。 + // 这时必须显式失败让调用方回退非流式，不能把解说文本当成最终回复返回。 + 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":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"我来帮你查询。"}}"#, "\n\n", + r#"data: {"type":"unknown_vendor_tool_event","index":9,"payload":{"name":"get_weather"}}"#, "\n\n", + r#"data: {"type":"message_delta","delta":{"stop_reason":"tool_use"}}"#, "\n\n" + ) + .to_string(), + extra_headers: Vec::new(), + }]); + + let client = build_test_client(server_url, 0); + let error = client + .stream_run(weather_tool_request(LlmApiKind::Anthropic), |_| {}) + .await + .expect_err("unparsed tool call should fall back"); + + assert_eq!(error, LlmError::StreamUnavailable); + } + #[test] fn multimodal_raw_failure_log_omits_request_and_image_data() { let config = LlmConfig::new( diff --git a/server-rs/crates/platform-llm/tests/live_stream_tool_calls.rs b/server-rs/crates/platform-llm/tests/live_stream_tool_calls.rs new file mode 100644 index 000000000..14f786cf3 --- /dev/null +++ b/server-rs/crates/platform-llm/tests/live_stream_tool_calls.rs @@ -0,0 +1,101 @@ +//! 真实端点的流式工具调用验收。默认 `#[ignore]`，只在显式指定环境变量时运行： +//! +//! ```powershell +//! $env:PLATFORM_LLM_LIVE_BASE_URL = 'https://api.minimaxi.com/anthropic' +//! $env:PLATFORM_LLM_LIVE_API_KEY = '...' +//! $env:PLATFORM_LLM_LIVE_MODEL = 'MiniMax-M3' +//! $env:PLATFORM_LLM_LIVE_API_KIND = 'anthropic' # 或 openai_chat / openai_responses +//! cargo test -p platform-llm --test live_stream_tool_calls -- --ignored --nocapture +//! ``` +//! +//! 单测里的 SSE 是转录的真实报文，这个用例负责证明转录没有偏差。 + +use platform_llm::{ + LlmApiKind, LlmClient, LlmConfig, LlmFunctionTool, LlmMessage, LlmProvider, LlmRunRequest, + LlmToolChoice, +}; + +fn env_var(name: &str) -> Option { + std::env::var(name).ok().filter(|value| !value.trim().is_empty()) +} + +fn parse_api_kind(value: &str) -> LlmApiKind { + match value.trim().to_ascii_lowercase().replace('-', "_").as_str() { + "anthropic" => LlmApiKind::Anthropic, + "openai_chat" => LlmApiKind::OpenAiChat, + _ => LlmApiKind::OpenAiResponses, + } +} + +#[tokio::test] +#[ignore = "需要真实 Provider 凭据，用 --ignored 显式运行"] +async fn live_stream_run_returns_native_tool_calls() { + let (Some(base_url), Some(api_key), Some(model)) = ( + env_var("PLATFORM_LLM_LIVE_BASE_URL"), + env_var("PLATFORM_LLM_LIVE_API_KEY"), + env_var("PLATFORM_LLM_LIVE_MODEL"), + ) else { + panic!("缺少 PLATFORM_LLM_LIVE_BASE_URL / _API_KEY / _MODEL"); + }; + let api_kind = parse_api_kind(&env_var("PLATFORM_LLM_LIVE_API_KIND").unwrap_or_default()); + + let config = LlmConfig::new( + LlmProvider::OpenAiCompatible, + base_url, + api_key, + model, + 120_000, + 0, + 1_000, + ) + .expect("live config should be valid"); + let client = LlmClient::new(config).expect("live client should be created"); + + let request = LlmRunRequest::new(vec![ + LlmMessage::system("你可以使用工具。需要外部数据时必须调用工具，不要凭空回答。"), + LlmMessage::user("杭州现在天气怎么样？"), + ]) + .with_api_kind(api_kind) + .with_max_output_tokens(512) + .with_function_tools(vec![LlmFunctionTool::new( + "get_weather", + "查询指定城市的当前天气。", + serde_json::json!({ + "type": "object", + "properties": { "city": { "type": "string" } }, + "required": ["city"] + }), + )]) + .with_tool_choice(LlmToolChoice::Required); + + let mut streamed_chars = 0usize; + let response = client + .stream_run(request, |delta| { + streamed_chars += delta.delta_text.chars().count(); + }) + .await + .expect("live stream_run should succeed"); + + println!( + "api_kind={api_kind:?} finish_reason={:?} streamed_chars={streamed_chars} text={:?}", + response.finish_reason, response.text + ); + for call in &response.tool_calls { + println!("tool_call id={} name={} args={}", call.id, call.name, call.arguments); + } + + assert!( + !response.tool_calls.is_empty(), + "流式必须解析出工具调用，实际 finish_reason={:?}", + response.finish_reason + ); + let call = &response.tool_calls[0]; + assert_eq!(call.name, "get_weather"); + assert!(!call.id.trim().is_empty(), "工具调用必须带 id"); + let arguments: serde_json::Value = + serde_json::from_str(&call.arguments).expect("参数必须是完整 JSON"); + assert!( + arguments.get("city").is_some(), + "参数应包含 city，实际为 {arguments}" + ); +}