//! Anthropic 兼容 Provider — 通过 /v1/messages 端点实现 //! //! 覆盖: Claude 官方 / GLM 订阅端点 (open.bigmodel.cn/api/anthropic) / 任意 Messages API 网关 //! 支持: 同步调用 + SSE 流式 + Tool Use //! //! 与 OpenAI 协议的关键差异由本模块内部完成转换,对外仍暴露统一的 LlmProvider trait, //! 上层 (Agentic Loop / AiNode) 无需感知协议。 //! //! 协议数据结构(请求/响应 struct)与 SSE 事件解析纯函数已抽至 `anthropic_helpers`, //! 本模块仅保留 Provider struct + impl(HTTP 调用),Rust impl 块不可跨文件故作此切分。 use async_trait::async_trait; use eventsource_stream::Eventsource; use futures::StreamExt; use reqwest::Client; use std::time::Duration; use tracing::{debug, error, warn}; use crate::provider::{ CompletionRequest, CompletionResponse, LlmProvider, MessageRole, StreamResult, TokenUsage, ToolCall, }; // ChatMessage 仅单测构造 CompletionRequest 用,避免非 test 构建的 unused import 警告。 #[cfg(test)] use crate::provider::ChatMessage; use crate::retry::{ retry_with_backoff, AttemptOutcome, is_reqwest_error_retryable, is_status_retryable, }; // 协议数据结构 + SSE 纯解析(apply_anthropic_event / AnthropicRequest 等)抽至 anthropic_helpers, // 此处 use 引入以保持本模块内引用路径不变(零行为变更搬迁)。 use crate::anthropic_helpers::{ apply_anthropic_event, AnthropicRequest, AnthropicResponse, AnthropicToolDef, ANTHROPIC_VERSION, DEFAULT_MAX_TOKENS, }; // ============================================================ // Provider 实现 // ============================================================ /// Anthropic 兼容 LLM Provider pub struct AnthropicCompatProvider { client: Client, api_key: String, base_url: String, default_model: String, } impl AnthropicCompatProvider { /// 创建 Provider /// /// - `base_url`: 如 `https://api.anthropic.com`、`https://open.bigmodel.cn/api/anthropic` /// - `api_key`: API 密钥(Anthropic 用 x-api-key 头,非 Bearer) /// - `default_model`: 默认模型名称 pub fn new(base_url: impl Into, api_key: impl Into, default_model: impl Into) -> Self { // SW-260618-10: reqwest Client 构建抽 crate::build_provider_client(与 OpenAI 共用 DRY)。 let client = crate::build_provider_client(); Self { client, api_key: api_key.into(), base_url: base_url.into(), default_model: default_model.into(), } } /// 构建 messages 端点 URL /// /// - 已含 `/v1/messages` → 直接用 /// - 以 `/v1` 结尾 → 补 `/messages` /// - 否则(如 `.../api/anthropic`、`api.anthropic.com`)→ 补 `/v1/messages` fn messages_url(&self) -> String { let base = self.base_url.trim_end_matches('/'); if base.ends_with("/v1/messages") { return base.to_string(); } if base.ends_with("/v1") { return format!("{}/messages", base); } format!("{}/v1/messages", base) } /// 将统一 CompletionRequest 转换为 Anthropic 请求体 /// /// 转换要点: /// - system 消息从 messages 抽离到顶层 system 字段 /// - assistant 带 tool_calls → content 数组含 text + tool_use 块 /// - tool_result(role=Tool)→ user 消息含 tool_result 块;连续多个合并为一条 user /// - tool_definitions 的 parameters → input_schema fn convert_request(&self, req: CompletionRequest) -> AnthropicRequest { let model = if req.model.is_empty() { self.default_model.clone() } else { req.model }; // 抽离 system 消息 let system: Option = { let sys: Vec = req .messages .iter() .filter(|m| matches!(m.role, MessageRole::System)) .map(|m| m.content.clone()) .collect(); if sys.is_empty() { None } else { Some(sys.join("\n\n")) } }; // 构建非 system 消息(保留顺序,合并连续 tool_result) let mut messages: Vec = Vec::new(); let mut pending_tool_results: Vec = Vec::new(); for m in req.messages.iter() { match m.role { MessageRole::System => continue, MessageRole::Tool => { // 累积 tool_result 块,遇到非 Tool 消息时 flush。 // tool_use_id 为 None 时绝不能发 null(GLM anthropic 端点会 500 // 报 'ClaudeContentBlockToolResult' object has no attribute 'id', // 致会话卡死持续 500),跳过该块并告警。 match &m.tool_call_id { Some(tid) if !tid.is_empty() => { pending_tool_results.push(serde_json::json!({ "type": "tool_result", "tool_use_id": tid, "content": m.content, })); } _ => { warn!( content_preview = %m.content.chars().take(80).collect::(), "Anthropic tool_result 缺少 tool_use_id,已跳过该块(发 null 会触发 GLM 500)" ); } } } MessageRole::User => { Self::flush_tool_results(&mut messages, &mut pending_tool_results); // F-260614-05 Phase 2a: 多模态 user 消息 → content blocks 数组(text/image)。 // 含图时把 content + parts 拍平成 blocks:Text 片 → {type:text}, // Image 片 → {type:image, source:{type:base64, media_type, data}}。 // Anthropic 协议要求 image 必须内嵌 base64(不接受 URL 直传)。 // 现状:前端(Phase2b)只产 base64 模式图片片,url 模式当前不可达。 // 未来若加 url 图片输入,必须在 commands 层补 url→base64 预拉 //(provider 不发额外 HTTP),否则下方兜底会发空 data 致 Anthropic 400。 // 纯文本消息(无图)保持原字符串简写,与现有端点零回归。 if m.has_image() { let blocks: Vec = m .flattened_parts() .into_iter() .map(|p| match p { crate::provider::ContentPart::Text { text } => serde_json::json!({ "type": "text", "text": text, }), crate::provider::ContentPart::Image { url, base64, media_type, alt: _ } => { // p 已被 match 取得所有权,直接 move media_type/base64 避免大 base64 clone。 let mt = media_type.unwrap_or_else(|| "image/png".into()); let data = base64.unwrap_or_else(|| { // 完整性兜底:当前 url 模式不可达(前端 Phase2b 只产 base64 图片片)。 // 若未来接入 url 图片输入而 commands 层未补 url→base64 预拉, // 此处会发空 data 致 Anthropic 400,warn 留痕但不阻塞(避免静默吞数据)。 if url.is_some() { warn!( url = ?url, "Anthropic user 消息含 Image(url) 但 base64 缺失,将发空 data(commands 层未补 url→base64 预拉)" ); } String::new() }); serde_json::json!({ "type": "image", "source": { "type": "base64", "media_type": mt, "data": data, } }) } }) .collect(); messages.push(serde_json::json!({ "role": "user", "content": blocks })); } else { messages.push(serde_json::json!({ "role": "user", "content": m.content })); } } MessageRole::Assistant => { Self::flush_tool_results(&mut messages, &mut pending_tool_results); let mut content: Vec = Vec::new(); if !m.content.is_empty() { content.push(serde_json::json!({ "type": "text", "text": m.content })); } if let Some(calls) = &m.tool_calls { for tc in calls { // B-260618-25: arguments 非法 JSON(流式中断残留 / ToolCall::new 默认空串) // → 空 object 兜底。Anthropic/GLM 要求 tool_use.input 必为 object, // null 直触发 1214「messages 参数非法」。 let input: serde_json::Value = serde_json::from_str(&tc.function.arguments) .unwrap_or_else(|_| serde_json::json!({})); content.push(serde_json::json!({ "type": "tool_use", "id": tc.id, "name": tc.function.name, "input": input, })); } } if content.is_empty() { content.push(serde_json::json!({ "type": "text", "text": "" })); } messages.push(serde_json::json!({ "role": "assistant", "content": content })); } } } Self::flush_tool_results(&mut messages, &mut pending_tool_results); // B-260619-03: 合并相邻 user 块。Anthropic 协议要求 user/assistant 严格交替,连续 user // 触发 GLM 1214。场景:drainQueue 续发(前一轮以 tool_result 结尾 + 新 user)→ flush 把 // tool_result 转成 user 后紧跟 push 新 user → 连续两 user。合并成一条 user 含 // [tool_result..., text] blocks(Anthropic 允许一条 user 多 blocks),打破恶性循环。 Self::merge_consecutive_users(&mut messages); let tools = req.tools.map(|defs| { defs.into_iter() .map(|d| AnthropicToolDef { name: d.function.name, description: Some(d.function.description).filter(|s| !s.is_empty()), // B-260618-25: input_schema 非 object(未来误用)→ 兜底 {"type":"object"}, // 防 Anthropic 拒非法 tool schema(当前全走 object_schema 恒 object,纯防御)。 input_schema: if d.function.parameters.is_object() { d.function.parameters } else { serde_json::json!({"type": "object"}) }, }) .collect() }); AnthropicRequest { model, messages, max_tokens: req.max_tokens.unwrap_or(DEFAULT_MAX_TOKENS), system, temperature: req.temperature, stream: req.stream, tools, tool_choice: req.tool_choice, } } /// 将累积的 tool_result 块作为一条 user 消息 flush 进消息列表 fn flush_tool_results( messages: &mut Vec, pending: &mut Vec, ) { if pending.is_empty() { return; } let blocks: Vec = pending.drain(..).collect(); messages.push(serde_json::json!({ "role": "user", "content": blocks })); } /// B-260619-03: 合并相邻 user 消息为一条(content 拼成 blocks 数组)。 /// Anthropic 协议要求 user/assistant 严格交替,连续 user 触发 1214。 /// 触发场景:flush_tool_results 把 tool_result 转 user 后紧跟新 user(drainQueue 续发, /// 前一轮以 tool_result 结尾)。合并成一条 user 含 [tool_result..., text] blocks,合法。 /// content 形态:string(简写)或 array(blocks),统一规范化为 array 后拼接。 fn merge_consecutive_users(messages: &mut Vec) { let is_user = |m: &serde_json::Value| m.get("role").and_then(|r| r.as_str()) == Some("user"); let mut i = 0; while i + 1 < messages.len() { if is_user(&messages[i]) && is_user(&messages[i + 1]) { let next_content = messages[i + 1].get("content").cloned().unwrap_or(serde_json::Value::Null); let cur = &mut messages[i]; // 当前 user content 规范化为 array let mut blocks: Vec = match cur.get("content").cloned() { Some(serde_json::Value::String(s)) => vec![serde_json::json!({ "type": "text", "text": s })], Some(serde_json::Value::Array(a)) => a, _ => vec![], }; // 并入下一个 user content match next_content { serde_json::Value::String(s) => blocks.push(serde_json::json!({ "type": "text", "text": s })), serde_json::Value::Array(a) => blocks.extend(a), _ => {} } cur["content"] = serde_json::Value::Array(blocks); messages.remove(i + 1); // 不增 i:继续合并 i 与新 i+1(可能多个连续 user) } else { i += 1; } } } /// 生成 messages 诊断摘要(每条 role + content 形态 + tool 标记),不含敏感数据。 /// B-260618-27: 1214 类错误时随 bail 文案直达前端 raw,定位哪条/字段非法。 fn summarize_messages(messages: &[serde_json::Value]) -> String { let lines: Vec = messages .iter() .enumerate() .map(|(i, m)| { let role = m.get("role").and_then(|r| r.as_str()).unwrap_or("?"); let desc = match m.get("content") { Some(serde_json::Value::String(s)) => format!("text({}B)", s.len()), Some(serde_json::Value::Array(blocks)) => { let parts: Vec = blocks .iter() .map(|b| { let ty = b.get("type").and_then(|t| t.as_str()).unwrap_or("?"); match ty { "text" => format!( "text({}B)", b.get("text").and_then(|t| t.as_str()).map(|s| s.len()).unwrap_or(0) ), "tool_use" => format!( "tool_use[id={},input_obj={}]", b.get("id").and_then(|t| t.as_str()).unwrap_or(""), b.get("input").map(|v| v.is_object()).unwrap_or(false) ), "tool_result" => format!( "tool_result[tid={}]", b.get("tool_use_id").and_then(|t| t.as_str()).unwrap_or("") ), "image" => "image".to_string(), _ => ty.to_string(), } }) .collect(); format!("[{}]", parts.join(",")) } _ => "?".to_string(), }; format!("#{}:{} {}", i, role, desc) }) .collect(); format!("{} msgs: {}", lines.len(), lines.join(" | ")) } /// B-260618-27: 协议预检——扫 messages 发现确定非法形态,命中返回原因(仅诊断不修复)。 /// 覆盖:首条非 user / 连续同 role / tool_use input 非 object / 空 content / orphan tool_result /// (tool_use_id 无前置 tool_use,常见于裁剪/过滤后 assistant 被删但 tool_result 留)。 fn precheck_messages(messages: &[serde_json::Value]) -> Result<(), String> { if messages.is_empty() { return Err("messages 为空".into()); } let first_role = messages[0].get("role").and_then(|r| r.as_str()).unwrap_or(""); if first_role != "user" { return Err(format!("首条 role={} 非法(须 user)", first_role)); } for w in messages.windows(2) { let r0 = w[0].get("role").and_then(|r| r.as_str()).unwrap_or(""); let r1 = w[1].get("role").and_then(|r| r.as_str()).unwrap_or(""); if r0 == r1 && (r0 == "user" || r0 == "assistant") { return Err(format!("连续同 role={}", r0)); } } let mut tool_use_ids: Vec<&str> = Vec::new(); for (i, m) in messages.iter().enumerate() { let role = m.get("role").and_then(|r| r.as_str()).unwrap_or(""); match m.get("content") { Some(serde_json::Value::String(s)) if s.is_empty() && role == "user" => { return Err(format!("#{} user content 空", i)); } Some(serde_json::Value::Array(blocks)) => { if blocks.is_empty() && role == "user" { return Err(format!("#{} user content 空数组", i)); } for b in blocks { match b.get("type").and_then(|t| t.as_str()).unwrap_or("") { "tool_use" => { let id = b.get("id").and_then(|t| t.as_str()).unwrap_or(""); tool_use_ids.push(id); if !b.get("input").map(|v| v.is_object()).unwrap_or(false) { return Err(format!("#{} tool_use input 非 object", i)); } } "tool_result" => { let tid = b.get("tool_use_id").and_then(|t| t.as_str()).unwrap_or(""); if !tid.is_empty() && !tool_use_ids.contains(&tid) { return Err(format!( "#{} orphan tool_result(tid={} 无前置 tool_use)", i, tid )); } } _ => {} } } } _ => {} } } Ok(()) } /// 统一鉴权头:x-api-key + anthropic-version fn auth_headers(&self, rb: reqwest::RequestBuilder) -> reqwest::RequestBuilder { rb.header("x-api-key", &self.api_key) .header("anthropic-version", ANTHROPIC_VERSION) .header("Content-Type", "application/json") } } #[async_trait] impl LlmProvider for AnthropicCompatProvider { async fn complete(&self, request: CompletionRequest) -> anyhow::Result { let mut req = request; req.stream = false; let body = self.convert_request(req); // B-260618-27: 协议预检——命中非法 bail 含 messages 摘要,把 GLM 模糊 1214 转明确诊断 if let Err(reason) = Self::precheck_messages(&body.messages) { let summary = Self::summarize_messages(&body.messages); warn!(%reason, %summary, "Anthropic messages 协议预检失败"); anyhow::bail!("messages 预检失败: {} | 摘要: {}", reason, summary); } debug!(model = %body.model, "Anthropic 同步调用"); // 指数退避重试(B-260616-07): 包裹 send + 状态码判定。 // 同时补 FR-R4 遗漏: Anthropic 同步路径此前无单请求 timeout(建连后挂起会无限 hang), // 此处加 60s timeout,与 OpenAI 路径对齐。 let label = format!("Anthropic[{}]", body.model); retry_with_backoff(&label, move |_| { let client = self.client.clone(); let url = self.messages_url(); let api_key = self.api_key.clone(); let body = body.clone(); async move { // send: 补 60s 单请求 timeout(此前缺失,FR-R4 遗漏) let rb = client .post(url) .header("x-api-key", &api_key) .header("anthropic-version", ANTHROPIC_VERSION) .header("Content-Type", "application/json") .timeout(Duration::from_secs(60)) .version(reqwest::Version::HTTP_11) .json(&body); let resp = match rb.send().await { Ok(r) => r, Err(e) => { // B-260618-26: 记 reqwest 错误源因链(is_*/source)。原仅 Display // "error sending request for url" 无法定位 reset/TLS/超时/body 真因。 tracing::error!( is_timeout = e.is_timeout(), is_connect = e.is_connect(), is_body = e.is_body(), is_request = e.is_request(), url = ?e.url(), source = ?std::error::Error::source(&e), "Anthropic 同步 send 失败" ); if is_reqwest_error_retryable(&e) { return AttemptOutcome::Retryable(format!("请求失败(可重试): {}", e)); } return AttemptOutcome::Fatal(format!( "请求失败(不可重试): {} [timeout={} connect={} body={} src={:?}]", e, e.is_timeout(), e.is_connect(), e.is_body(), std::error::Error::source(&e) )); } }; if !resp.status().is_success() { let status = resp.status().as_u16(); let text = resp.text().await.unwrap_or_default(); let msg = format!("Anthropic API 错误 {}: {}", status, text); if is_status_retryable(status) { warn!(%status, "Anthropic 同步调用可重试状态码"); return AttemptOutcome::Retryable(msg); } error!(%status, %text, "Anthropic API 调用失败(不可重试)"); return AttemptOutcome::Fatal(msg); } let resp: AnthropicResponse = match resp.json().await { Ok(r) => r, Err(e) => return AttemptOutcome::Fatal(format!("响应解析失败: {}", e)), }; // content 块中拼接 text,收集 tool_use let mut text = String::new(); let mut tool_calls: Vec = Vec::new(); for block in resp.content { match block.block_type.as_str() { "text" => { if let Some(t) = block.text { text.push_str(&t); } } "tool_use" => { let id = match block.id { Some(id) if !id.is_empty() => id, _ => { warn!( name = ?block.name, "Anthropic tool_use 块缺少 id,已跳过(空 id 会回传空 tool_use_id 触发 500)" ); continue; } }; let name = block.name.unwrap_or_default(); let args = block .input .map(|v| serde_json::to_string(&v).unwrap_or_default()) .unwrap_or_default(); tool_calls.push(ToolCall::new(id, name, args)); } other => warn!(block_type = other, "Anthropic 响应含未知 content 块类型,已忽略"), } } let usage = TokenUsage { prompt_tokens: resp.usage.input_tokens, completion_tokens: resp.usage.output_tokens, total_tokens: resp.usage.input_tokens + resp.usage.output_tokens, }; AttemptOutcome::Ok(CompletionResponse { text, model: resp.model, usage, tool_calls: if tool_calls.is_empty() { None } else { Some(tool_calls) }, reasoning_content: None, }) } }) .await } async fn stream(&self, request: CompletionRequest) -> anyhow::Result { let mut req = request; req.stream = true; let body = self.convert_request(req); // B-260618-27: 协议预检——命中非法 bail 含 messages 摘要,把 GLM 模糊 1214 转明确诊断 if let Err(reason) = Self::precheck_messages(&body.messages) { let summary = Self::summarize_messages(&body.messages); warn!(%reason, %summary, "Anthropic messages 协议预检失败"); anyhow::bail!("messages 预检失败: {} | 摘要: {}", reason, summary); } debug!(model = %body.model, "Anthropic 流式调用"); let resp = match self .auth_headers(self.client.post(self.messages_url())) .json(&body) .version(reqwest::Version::HTTP_11) .send() .await { Ok(r) => r, Err(e) => { // B-260618-26: 记 reqwest 错误源因链。原 ? 转 anyhow 仅 Display // "error sending request for url" 无法定位 reset/TLS/超时/body 真因。 tracing::error!( is_timeout = e.is_timeout(), is_connect = e.is_connect(), is_body = e.is_body(), is_request = e.is_request(), url = %self.messages_url(), source = ?std::error::Error::source(&e), "Anthropic 流式 send 失败" ); anyhow::bail!( "send 失败: {} [timeout={} connect={} body={} request={} src={:?}]", e, e.is_timeout(), e.is_connect(), e.is_body(), e.is_request(), std::error::Error::source(&e) ); } }; if !resp.status().is_success() { let status = resp.status(); let text = resp.text().await.unwrap_or_default(); error!(%status, %text, "Anthropic 流式 API 调用失败"); anyhow::bail!("Anthropic 流式 API 错误 {}: {}", status, text); } // 流式解析:eventsource 逐事件处理,按 type 字段分发转 StreamChunk。 // 事件解析/usage 累积逻辑抽到 apply_anthropic_event 纯函数,便于单测;此处闭包只负责传 data。 // usage 累积:message_start 给 input_tokens,message_delta 给累计 output_tokens(非增量),message_stop 带出。 let mut usage_accum: Option = None; // B-260618-28: MidStream error(如 GLM 1214 messages 非法)时附 messages 摘要定位哪条非法。 // precheck(Init 路径,发送前)漏的 case,靠此在 SSE error 事件暴露实际 messages 结构到前端 raw。 let messages_summary = Self::summarize_messages(&body.messages); let stream = resp .bytes_stream() .eventsource() .map(move |event| match event { Ok(ev) => { let mut chunk = apply_anthropic_event(&ev.data, &mut usage_accum); // GLM 中途 error(如 1214)→ chunk.error 附 messages 摘要,经 stream_recv MidStream // 路径 emit AiError raw,前端直接看到实际 messages 结构定位非法字段。 if let Some(err) = chunk.error.as_mut() { *err = format!("{} | messages 摘要: {}", err, messages_summary); } Ok(chunk) } Err(e) => { // 保留 #[source] 因果链: anyhow!("...{}", e) 仅把 e 的 Display 塞进 message, // 丢掉 source(无法 downcast/遍历)。改用 Error::from(e).context(...): // Display 不变(仍为 "Anthropic SSE 错误: {e}"), 且 e 作为 .source() 可追溯。 // 顺序: 先 format(e) 构造 context 文案, 再 Error::from(e) move e 进 source。 let ctx = format!("Anthropic SSE 错误: {}", e); error!(error = %e, "Anthropic SSE 事件流错误"); Err(anyhow::Error::from(e).context(ctx)) } }); Ok(Box::pin(stream)) } fn name(&self) -> &str { "anthropic-compat" } fn endpoint(&self) -> String { self.messages_url() } } // ============================================================ // 单测(不发真实 HTTP,喂构造的 SSE data 字符串序列) // ============================================================ #[cfg(test)] mod tests { use super::*; // apply_anthropic_event / TokenUsage 经 super::*(含 anthropic_helpers::apply_anthropic_event 的 use)可见。 /// 辅助:构造 message_start 事件 fn message_start(input_tokens: u32) -> String { format!( r#"{{"type":"message_start","message":{{"usage":{{"input_tokens":{},"output_tokens":0}}}}}}"#, input_tokens ) } /// 辅助:构造 message_delta 事件(output_tokens 为累计值) fn message_delta(output_tokens: u32) -> String { format!( r#"{{"type":"message_delta","delta":{{"stop_reason":"end_turn"}},"usage":{{"output_tokens":{}}}}}"#, output_tokens ) } /// 辅助:构造文本增量 content_block_delta fn text_delta(text: &str) -> String { format!( r#"{{"type":"content_block_delta","index":0,"delta":{{"type":"text_delta","text":"{}"}}}}"#, text ) } /// 辅助:构造 message_stop 事件 fn message_stop() -> &'static str { r#"{"type":"message_stop"}"# } /// 完整流:message_start 初始化 input + 多次 message_delta 累计覆盖 output + message_stop 带出 #[test] fn anthropic_full_stream_accumulates_usage() { let mut acc: Option = None; // 1) message_start:input=42,output=0 let c = apply_anthropic_event(&message_start(42), &mut acc); assert!(!c.finished); assert!(c.usage.is_none()); let a = acc.as_ref().expect("message_start 应初始化累加器"); assert_eq!(a.prompt_tokens, 42); assert_eq!(a.completion_tokens, 0); assert_eq!(a.total_tokens, 42); // 2) 文本增量不影响 usage let c = apply_anthropic_event(&text_delta("Hello"), &mut acc); assert_eq!(c.delta, "Hello"); assert!(!c.finished); assert_eq!(acc.as_ref().unwrap().completion_tokens, 0, "文本增量不应改 output"); // 3) message_delta:output_tokens=10(累计值,覆盖) let c = apply_anthropic_event(&message_delta(10), &mut acc); assert!(!c.finished); let a = acc.as_ref().unwrap(); assert_eq!(a.prompt_tokens, 42, "input 保持"); assert_eq!(a.completion_tokens, 10, "output 被覆盖为累计值"); assert_eq!(a.total_tokens, 52, "total 重算 = input+output"); // 4) 再次 message_delta:output_tokens=30(更大累计值,再覆盖) let _ = apply_anthropic_event(&message_delta(30), &mut acc); let a = acc.as_ref().unwrap(); assert_eq!(a.completion_tokens, 30, "后续累计值覆盖前值"); assert_eq!(a.total_tokens, 72); // 5) message_stop:带出累积 usage,finished=true,累加器清空 let c = apply_anthropic_event(message_stop(), &mut acc); assert!(c.finished); let u = c.usage.expect("message_stop 应带出累积 usage"); assert_eq!(u.prompt_tokens, 42); assert_eq!(u.completion_tokens, 30); assert_eq!(u.total_tokens, 72); assert!(acc.is_none(), "take() 后累加器应清空"); } /// message_delta 在没有 message_start 时也能补全累加器(get_or_insert 兜底) #[test] fn anthropic_message_delta_without_start_uses_default_input() { let mut acc: Option = None; let _ = apply_anthropic_event(&message_delta(15), &mut acc); let a = acc.as_ref().unwrap(); assert_eq!(a.prompt_tokens, 0, "无 message_start 时 input 兜底为 0"); assert_eq!(a.completion_tokens, 15); assert_eq!(a.total_tokens, 15); } /// message_delta 的 output_tokens 必须是累计覆盖(非累加):连续两个 delta 5 和 8,结果应是 8 不是 13 #[test] fn anthropic_message_delta_output_is_cumulative_not_incremental() { let mut acc: Option = None; apply_anthropic_event(&message_start(100), &mut acc); apply_anthropic_event(&message_delta(5), &mut acc); apply_anthropic_event(&message_delta(8), &mut acc); let c = apply_anthropic_event(message_stop(), &mut acc); let u = c.usage.unwrap(); assert_eq!(u.completion_tokens, 8, "output_tokens 是累计值,覆盖而非累加"); assert_eq!(u.total_tokens, 108); } /// 无 usage 字段的流:message_stop 时 usage 为 None #[test] fn anthropic_message_stop_without_any_usage() { let mut acc: Option = None; let _ = apply_anthropic_event(&text_delta("hi"), &mut acc); assert!(acc.is_none(), "文本增量不初始化累加器"); let c = apply_anthropic_event(message_stop(), &mut acc); assert!(c.finished); assert!(c.usage.is_none(), "无 usage 时 message_stop usage 为 None"); } /// content_block_start (tool_use) 带 id+name #[test] fn anthropic_content_block_start_tool_use() { let mut acc: Option = None; let data = r#"{"type":"content_block_start","index":1,"content_block":{"type":"tool_use","id":"tool_1","name":"get_weather"}}"#; let c = apply_anthropic_event(data, &mut acc); assert!(acc.is_none(), "content_block_start 不动 usage"); let tcs = c.tool_calls.expect("应有 tool_calls"); assert_eq!(tcs.len(), 1); assert_eq!(tcs[0].index, 1); assert_eq!(tcs[0].id.as_deref(), Some("tool_1")); assert_eq!(tcs[0].function_name.as_deref(), Some("get_weather")); assert!(tcs[0].function_arguments.is_none()); assert!(!c.finished); } /// content_block_delta (input_json_delta) → 工具入参增量 #[test] fn anthropic_content_block_delta_input_json() { let mut acc: Option = None; let data = r#"{"type":"content_block_delta","index":2,"delta":{"type":"input_json_delta","partial_json":"{\"q\":"}}"#; let c = apply_anthropic_event(data, &mut acc); let tcs = c.tool_calls.expect("应有 tool_calls 增量"); assert_eq!(tcs[0].index, 2); assert_eq!(tcs[0].function_arguments.as_deref(), Some("{\"q\":")); assert!(tcs[0].id.is_none()); assert_eq!(c.delta, ""); assert!(!c.finished); } /// error 事件 → error=Some + finished=false(R-P1-1:避免残缺响应被当正常完成入库, /// 由 stream_llm 识别 error 非空发 AiError + 丢弃残缺,与 OpenAI 路径 Err 一致) #[test] fn anthropic_error_event_yields_error_not_finished() { let mut acc: Option = None; apply_anthropic_event(&message_start(10), &mut acc); let c = apply_anthropic_event(r#"{"type":"error","error":{"message":"overloaded"}}"#, &mut acc); assert!(!c.finished, "error 不走 finished 完成路径,否则残缺响应会被当正常完成"); assert_eq!(c.error.as_deref(), Some("overloaded"), "error 事件应携带错误消息"); assert!(c.usage.is_none(), "error 不带出 usage"); assert!(c.delta.is_empty(), "error 不带文本增量"); assert!(acc.is_some(), "error 不应清空已累积的 usage(与原实现一致)"); } /// error 事件无 error.message 字段时兜底为 "stream error" #[test] fn anthropic_error_event_missing_message_falls_back() { let mut acc: Option = None; let c = apply_anthropic_event(r#"{"type":"error"}"#, &mut acc); assert_eq!(c.error.as_deref(), Some("stream error")); assert!(!c.finished); } /// ping / content_block_stop 等事件 → 空且不 finished #[test] fn anthropic_ping_and_block_stop_yield_empty_chunk() { let mut acc: Option = None; let c = apply_anthropic_event(r#"{"type":"ping"}"#, &mut acc); assert!(!c.finished); assert_eq!(c.delta, ""); assert!(acc.is_none()); let c = apply_anthropic_event(r#"{"type":"content_block_stop","index":0}"#, &mut acc); assert!(!c.finished); assert_eq!(c.delta, ""); } /// 非法 JSON → 空 chunk,不 panic #[test] fn anthropic_malformed_json_yields_empty_chunk() { let mut acc: Option = None; let c = apply_anthropic_event("not json", &mut acc); assert!(!c.finished); assert_eq!(c.delta, ""); assert!(acc.is_none()); } // ---------- F-260614-05 Phase 2a 多模态 convert_request ---------- /// 含图 user 消息 → content blocks(text + image source.base64) #[test] fn anthropic_convert_multimodal_user_blocks() { let provider = AnthropicCompatProvider::new("https://api.anthropic.com", "k", "claude-3-5-sonnet"); let req = CompletionRequest { model: "claude-3-5-sonnet".into(), messages: vec![ChatMessage::user_parts( "看图", vec![crate::provider::ContentPart::image_base64("image/png", "iVBOR")], )], temperature: None, max_tokens: None, stream: false, tools: None, tool_choice: None, reasoning_content: None, }; let body = provider.convert_request(req); // user 消息 content 应为数组形态 let user_msg = body .messages .iter() .find(|m| m.get("role").and_then(|r| r.as_str()) == Some("user")) .expect("应有 user 消息"); let content = user_msg.get("content").and_then(|c| c.as_array()).expect("user content 应为数组"); // 顺序:content "看图" → text 块,image 片 → image 块 assert_eq!(content.len(), 2); assert_eq!(content[0]["type"], "text"); assert_eq!(content[0]["text"], "看图"); assert_eq!(content[1]["type"], "image"); assert_eq!(content[1]["source"]["type"], "base64"); assert_eq!(content[1]["source"]["media_type"], "image/png"); assert_eq!(content[1]["source"]["data"], "iVBOR"); } /// 纯文本 user 消息 → content 仍是字符串简写(无图不数组化) #[test] fn anthropic_convert_text_only_user_remains_string() { let provider = AnthropicCompatProvider::new("https://api.anthropic.com", "k", "claude-3-5-sonnet"); let req = CompletionRequest { model: "claude-3-5-sonnet".into(), messages: vec![ChatMessage::user("hello")], temperature: None, max_tokens: None, stream: false, tools: None, tool_choice: None, reasoning_content: None, }; let body = provider.convert_request(req); let user_msg = body .messages .iter() .find(|m| m.get("role").and_then(|r| r.as_str()) == Some("user")) .expect("应有 user 消息"); // 纯文本 → 字符串简写(非数组) assert_eq!(user_msg.get("content").and_then(|c| c.as_str()), Some("hello")); } }