//! OpenAI 兼容 Provider — 通过 /v1/chat/completions 端点实现 //! //! 覆盖: OpenAI / GLM (open.bigmodel.cn) / DeepSeek / Claude OpenAI 兼容模式 //! 支持: 同步调用 + SSE 流式 + Function Calling / Tool Use use std::time::Duration; use async_trait::async_trait; use futures::StreamExt; use reqwest::Client; use tracing::{debug, error, warn}; use crate::provider::{ tool_call_id_or_fallback, CompletionRequest, CompletionResponse, LlmProvider, 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, }; // 协议数据结构(OpenAiRequest / OpenAiResponse / OpenAiStreamChunk …)与 SSE 解析纯函数 // apply_openai_sse 抽离到 openai_helpers.rs,结构对齐 anthropic_helpers.rs。 // Rust impl 块不可跨文件,Provider struct + impl 仍留本文件。 use crate::openai_helpers::{ apply_openai_sse, OpenAiMessage, OpenAiRequest, OpenAiResponse, OpenAiToolCallResp, }; // ============================================================ // OpenAI Compat Provider // ============================================================ /// OpenAI 兼容 LLM Provider pub struct OpenAICompatProvider { client: Client, api_key: String, base_url: String, default_model: String, } impl OpenAICompatProvider { /// 创建 Provider /// /// - `base_url`: 如 "https://api.openai.com", "https://open.bigmodel.cn/api/paas", "https://api.deepseek.com" /// - `api_key`: API 密钥 /// - `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(与 Anthropic 共用 DRY)。 // connect_timeout/回退策略集中此处,未来改一处即可(见 lib.rs::build_provider_client)。 let client = crate::build_provider_client(); Self { client, api_key: api_key.into(), base_url: base_url.into(), default_model: default_model.into(), } } /// 构建完整 API URL /// /// 智能拼接,兼容三种 base_url 约定: /// - 已含完整端点(…/chat/completions)→ 直接用 /// - 已含版本段(…/v1 …/v4 等,如 GLM 的 /api/paas/v4)→ 补 /chat/completions /// - 仅域名无版本(如 api.openai.com / api.deepseek.com)→ 补 /v1/chat/completions(OpenAI 约定) fn chat_url(&self) -> String { let base = self.base_url.trim_end_matches('/'); if base.ends_with("/chat/completions") { return base.to_string(); } if Self::ends_with_version(base) { return format!("{}/chat/completions", base); } format!("{}/v1/chat/completions", base) } /// base_url 是否以 `/v<数字>` 结尾(如 /v1 /v4) fn ends_with_version(base: &str) -> bool { match base.rsplit_once('/') { Some((_, last)) if last.starts_with('v') && last.len() > 1 => { last[1..].bytes().all(|b| b.is_ascii_digit()) } _ => false, } } /// 构建 embeddings API URL(与 chat_url 同套智能拼接规则) fn embed_url(&self) -> String { let base = self.base_url.trim_end_matches('/'); if base.ends_with("/embeddings") { return base.to_string(); } if Self::ends_with_version(base) { return format!("{}/embeddings", base); } format!("{}/v1/embeddings", base) } /// 将通用请求转换为 OpenAI 格式 fn convert_request(&self, req: CompletionRequest) -> OpenAiRequest { let model = if req.model.is_empty() { self.default_model.clone() } else { req.model }; let mut messages: Vec = req .messages .into_iter() .map(|m| { let role = match m.role { crate::provider::MessageRole::System => "system", crate::provider::MessageRole::User => "user", crate::provider::MessageRole::Assistant => "assistant", crate::provider::MessageRole::Tool => "tool", }; // 多模态 content(须在 move m.tool_calls 之前算,借用 m)。 // 含图消息走 content 数组(text/image_url);纯文本走字符串简写 // (保持与现有纯文本端点零回归)。image_url 支持 data URI(base64)与 http(s) URL。 let content = if m.has_image() { let parts: 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: _ } => { let final_url = match (base64, url, media_type) { (Some(b), _, Some(mt)) => { format!("data:{};base64,{}", mt, b) } (None, Some(u), _) => u, // 完整性兜底:当前 image_base64 构造器强制 media_type:Some, // image_url 构造器提供 url:Some,二者分别命中上两个分支; // 此分支仅在 parts 来源被外部直接构造且字段均缺时才可达 //(如 url:None+base64:None 或 base64:Some+media_type:None)。 // 退化为空串(OpenAI 对空 image_url.url 会 400), // 由调用方保证 parts 合法性,provider 层不做静默伪造。 _ => String::new(), }; serde_json::json!({ "type": "image_url", "image_url": { "url": final_url }, }) } }) .collect(); serde_json::Value::Array(parts) } else { serde_json::Value::String(m.content.clone()) }; let tool_calls = m.tool_calls.map(|calls| { calls .into_iter() .map(|tc| { serde_json::json!({ "id": tc.id, "type": tc.call_type, "function": { "name": tc.function.name, "arguments": tc.function.arguments, } }) }) .collect() }); OpenAiMessage { role: role.to_string(), content, tool_call_id: m.tool_call_id, tool_calls, reasoning_content: m.reasoning_content, } }) .collect(); // 保证首条 user/system(OpenAI 协议要求首条非 assistant/tool)。 // 对齐 AnthropicCompatProvider::ensure_leading_user:上游绕过 sanitize 的调用方 // (标题生成/知识注入/工作流 AI 节点等直构造 CompletionRequest 的路径)可能传入首条 // assistant 的序列(会话恢复/续发/片段截取),补 user 占位保留上下文,首条合法。 Self::ensure_leading_user(&mut messages); // 治 DeepSeek/OpenAI 400(三元组完整性 P0)。OpenAI 协议铁律: // (a) assistant 的每个 tool_call.id 必须有后续 tool(role=tool, tool_call_id 匹配)响应, // 否则 "insufficient tool messages" 400(assistant 调了工具但无结果)。 // (b) 反之,每条 tool 消息必须紧跟一个含 tool_calls(同 tool_call_id)的 assistant, // 否则 "Messages with role tool must be a response to a preceding message // with tool_calls" 400(tool 无配对头)。 // // 旧逻辑只检查「下一条 role 是否为 tool」(粗粒度),漏两类 orphan: // 1) 部分 tool_call 无响应:assistant(tc=[a,b]) → tool(a)(b 丢失)→ 旧逻辑因下一条是 // tool 不剥 → 发出未闭合的 b → 400。修法:按 tool_call_id 精确配对,剥未闭合 id。 // 2) orphan tool_result(tool 无前置 assistant tool_calls 配对):DB/直构造路径绕过 // ContextManager::sanitize_messages(标题/知识注入/工作流节点),tool 残留无头 → // 旧逻辑不处理 → 400。修法:剥 assistant tool_calls 时同步丢弃同 id 的 orphan // tool(一致性:不留无头 result),并对独立 orphan tool(全程无配对头)直接丢弃。 // // 正常三元组形如:assistant(tc=[a]) → tool(a) → assistant(tc=[b]) → tool(b),各 id 闭合, // 本守卫零介入。仅异常截断/恢复/直构造路径触发(防 400 兜底)。 // view-only:仅改发送视图(本函数消费 req.messages 所有权),持久化由调用方/上层 sanitize 全量保留。 sanitize_openai_triplets(&mut messages); let tools = req.tools.map(|defs| { defs.into_iter() .map(|d| serde_json::to_value(d).unwrap_or_default()) .collect() }); OpenAiRequest { model, messages, temperature: req.temperature, max_tokens: req.max_tokens, stream: req.stream, tools, tool_choice: req.tool_choice, reasoning_content: req.reasoning_content, // 流式请求末 chunk 带 usage(同步调用 complete 不需要) stream_options: if req.stream { Some(serde_json::json!({ "include_usage": true })) } else { None }, } } /// 生成 messages 诊断摘要(每条 role + content 形态 + tool 标记),不含敏感数据。 /// 流中途 error 时附摘要定位哪条非法(对齐 `AnthropicCompatProvider::summarize_messages`)。 fn summarize_openai_messages(messages: &[OpenAiMessage]) -> String { let lines: Vec = messages .iter() .enumerate() .map(|(i, m)| { let role = m.role.as_str(); let desc = match &m.content { serde_json::Value::String(s) => format!("text({}B)", s.len()), 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) ), "image_url" => "image".to_string(), _ => ty.to_string(), } }) .collect(); format!("[{}]", parts.join(",")) } _ => "?".to_string(), }; let tool_mark = match (&m.tool_calls, &m.tool_call_id) { (Some(tcs), _) => format!(" tool_calls={}", tcs.len()), (None, Some(tid)) => format!(" tool_result[tid={}]", tid), (None, None) => String::new(), }; format!("#{}:{} {}{}", i, role, desc, tool_mark) }) .collect(); format!("{} msgs: {}", lines.len(), lines.join(" | ")) } /// 保证 messages 首条为 user/system(OpenAI 协议要求首条非 assistant/tool)。 /// /// 对齐 `AnthropicCompatProvider::ensure_leading_user`。上游绕过 `ContextManager::sanitize_messages` /// 的调用方(标题生成/知识注入/工作流 AI 节点等直构造 CompletionRequest 的路径)可能传入首条 /// assistant 的序列——会话恢复、续发或历史片段截取时,真正的首条 user 已被裁剪/压缩掉。 /// /// **用"补"而非"砍"**:开头插一条 user 占位,保留全部上下文(砍会丢工具调用历史,多轮砍到空)。 /// 占位 user 紧贴原首条,不破坏 user/assistant 交替;仅异常路径触发(正常首条本就是 user)。 fn ensure_leading_user(messages: &mut Vec) { let first_role = messages.first().map(|m| m.role.as_str()).unwrap_or(""); if first_role == "user" || first_role == "system" { return; } warn!( first_role, msg_count = messages.len(), "ensure_leading_user: 首条非 user/system,补 user 占位(保留上下文,防 OpenAI 首条 assistant/tool 非法)" ); messages.insert( 0, OpenAiMessage { role: "user".into(), content: serde_json::Value::String("(continued from previous context)".into()), tool_call_id: None, tool_calls: None, reasoning_content: None, }, ); } /// 解析同步响应中的工具调用。 /// /// 兜底(CR-空 id):id 空时按数组 index 生成 `gen_tool_{index}` fallback。 /// SenseNova 等兼容缺陷 provider 发空 id,多 tool_call 同 id 致结果路由全落首个。 /// 详见 `tool_call_id_or_fallback`。正常 provider id 非空原样透传。 fn parse_tool_calls(calls: Vec) -> Vec { calls .into_iter() .enumerate() .map(|(i, c)| { let id = tool_call_id_or_fallback(&c.id, i, "gen_tool"); ToolCall::new(id, c.function.name, c.function.arguments) }) .collect() } } /// 从 OpenAiMessage 的 tool_calls 数组里取每个 call 的 id(tool_calls 形如 /// [{id, type, function:{name, arguments}}, ...])。非数组 / 缺 id 的条目跳过。 fn extract_tool_call_ids(msg: &OpenAiMessage) -> Vec { let Some(arr) = msg.tool_calls.as_ref() else { return Vec::new(); }; arr.iter() .filter_map(|tc| tc.get("id").and_then(|v| v.as_str()).map(|s| s.to_string())) .collect() } /// 三元组一致性自愈(view-only,发送视图):保证 OpenAI 协议 tool_call/tool_result /// 双向闭合,防 DeepSeek/OpenAI 400。详见 [`OpenAICompatProvider::convert_request`] 调用处注释。 /// /// 两轮扫描: /// 1) 收集 resolved_ids = 所有 tool 消息的 tool_call_id(这些 id 有 result 响应)。 /// 2) assistant(tool_calls):剥未在 resolved_ids 内的 call.id;剥空则 tool_calls=None。 /// (头被剥后,其 tool_call.id 不再进 head_ids,故 step3 会同步丢弃对应 orphan tool。) /// 3) tool:tool_call_id 不在任何保留 assistant 头(任意 assistant 仍含此 id)→ orphan /// tool_result,丢弃。这覆盖「头被剥后残留的 tool」与「全程无配对头的 tool」两类。 /// /// 一致性:剥 assistant tool_call → 该 id 不进 head_ids → 对应 tool 在 step3 被丢; /// 反之剥 orphan tool 不动 assistant(若 assistant 的所有 id 都被剥则 tool_calls=None)。 /// 正常三元组(各 id 闭合)零介入。 /// /// 设计取舍:OpenAI 协议 assistant(tool_calls) 需有效函数结构,补头(像 Anthropic /// TOOL_MISSING_PREFIX)风险高于丢弃——故选「丢弃 orphan」而非「补头」。Anthropic /// 路径由 drop_reverse_orphans 补头自愈(保留 LLM 可见的工具结果);OpenAI 路径走丢弃, /// 二者各自适配协议特性(Anthropic 严格交替 + 补头可行;OpenAI tool 必须紧跟 tool_calls)。 fn sanitize_openai_triplets(messages: &mut Vec) { use std::collections::HashSet; // step 1:resolved_ids = 所有 tool 消息提供的 tool_call_id(有 result 响应的 id)。 let resolved_ids: HashSet = messages .iter() .filter(|m| m.role == "tool") .filter_map(|m| m.tool_call_id.clone()) .collect(); let mut stripped_heads = 0u32; let mut total_stripped = 0u32; // step 2:assistant 剥未闭合 tool_call(无对应 tool result 响应)。 for m in messages.iter_mut() { if m.role != "assistant" { continue; } let Some(calls) = m.tool_calls.as_ref() else { continue; }; if calls.is_empty() { continue; } let kept: Vec = calls .iter() .filter(|tc| { tc.get("id") .and_then(|v| v.as_str()) .is_some_and(|id| resolved_ids.contains(id)) }) .cloned() .collect(); let stripped_count = calls.len() - kept.len(); if stripped_count == 0 { continue; } m.tool_calls = if kept.is_empty() { None } else { Some(kept) }; stripped_heads += 1; total_stripped += stripped_count as u32; tracing::warn!( stripped_count, "[openai] assistant 含未闭合 tool_calls(无对应 tool result),已剥离 {} 个(防 insufficient tool messages 400)", stripped_count, ); } // step 3:head_ids = step2 后仍保留在任意 assistant 头的 id(有头配对的 tool 才保留)。 let head_ids: HashSet = messages .iter() .filter(|m| m.role == "assistant") .flat_map(extract_tool_call_ids) .collect(); let original_len = messages.len(); let mut dropped_orphan_tools = 0u32; messages.retain(|m| { if m.role != "tool" { return true; } let id = match m.tool_call_id.as_deref() { None => { // 无 tool_call_id 的 tool 消息(异常数据):无法配对,丢弃(发出去必 400)。 dropped_orphan_tools += 1; tracing::warn!( "[openai] tool 消息缺少 tool_call_id,已丢弃(无 id 无法配对 assistant tool_calls,防 400)" ); return false; } Some(id) => id, }; if head_ids.contains(id) { // 有配对头 → 保留(正常三元组)。 return true; } // 无配对头(id 不在任何保留 assistant 头内)→ orphan tool_result,丢弃。 // 含两类:(a) assistant 头被 step2 剥后残留的 tool;(b) 全程无配对头的直构造/DB 残留。 dropped_orphan_tools += 1; tracing::warn!( tool_call_id = %id, "[openai] orphan tool result(无配对 assistant tool_calls),已丢弃(防 'tool must be response to preceding tool_calls' 400)", ); false }); if stripped_heads > 0 || dropped_orphan_tools > 0 { tracing::warn!( stripped_heads, total_stripped, dropped_orphan_tools, before = original_len, after = messages.len(), "[openai] tool_call 三元组自愈(view-only, 持久化不受影响)" ); } } #[async_trait] impl LlmProvider for OpenAICompatProvider { /// 文本嵌入: POST /v1/embeddings(OpenAI 兼容,智谱/阿里百炼/OpenAI 通用) async fn embed(&self, model: &str, texts: Vec) -> anyhow::Result>> { #[derive(serde::Deserialize)] struct EmbedData { embedding: Vec, index: usize } #[derive(serde::Deserialize)] struct EmbedResponse { data: Vec } let resp = self .client .post(self.embed_url()) .header("Authorization", format!("Bearer {}", self.api_key)) .header("Content-Type", "application/json") .json(&serde_json::json!({ "model": model, "input": texts })) .send() .await?; if !resp.status().is_success() { let status = resp.status(); let body = resp.text().await.unwrap_or_default(); anyhow::bail!("Embedding API 错误 {}: {}", status, body); } let mut body: EmbedResponse = resp.json().await?; // 按 index 排序保证与输入顺序一致(API 不保证返回顺序) body.data.sort_by_key(|d| d.index); Ok(body.data.into_iter().map(|d| d.embedding).collect()) } async fn complete(&self, request: CompletionRequest) -> anyhow::Result { let mut req = request; req.stream = false; let openai_req = self.convert_request(req); debug!(model = %openai_req.model, "OpenAI 同步调用"); // 指数退避重试: 包裹 send + 状态码判定。 // 单请求 60s timeout 保持不变,重试是额外层: 3 次 × 60s 最坏 180s, // 由 retry_with_backoff 内部 30s 总预算主动止损。 let label = format!("OpenAI[{}]", openai_req.model); retry_with_backoff(&label, move |_| { let client = self.client.clone(); let url = self.chat_url(); let api_key = self.api_key.clone(); let openai_req = openai_req.clone(); async move { // send let resp = client .post(url) .header("Authorization", format!("Bearer {}", api_key)) .header("Content-Type", "application/json") .timeout(Duration::from_secs(60)) .json(&openai_req) .send() .await; let resp = match resp { Ok(r) => r, Err(e) => { if is_reqwest_error_retryable(&e) { return AttemptOutcome::Retryable(format!("请求失败(可重试): {}", e)); } return AttemptOutcome::Fatal(format!("请求失败(不可重试): {}", e)); } }; // 状态码判定 if !resp.status().is_success() { let status = resp.status().as_u16(); let body = resp.text().await.unwrap_or_default(); let msg = format!("LLM API 错误 {}: {}", status, body); if is_status_retryable(status) { warn!(%status, "OpenAI 同步调用可重试状态码"); return AttemptOutcome::Retryable(msg); } error!(%status, %body, "LLM API 调用失败(不可重试)"); return AttemptOutcome::Fatal(msg); } // body 解析: 解析错属 Fatal(响应已成功送达,重试也会因同样格式失败) let body: OpenAiResponse = match resp.json().await { Ok(b) => b, Err(e) => return AttemptOutcome::Fatal(format!("响应解析失败: {}", e)), }; let choice = match body.choices.into_iter().next() { Some(c) => c, None => return AttemptOutcome::Fatal("LLM 响应无 choices".to_string()), }; let text = choice.message.content.unwrap_or_default(); let tool_calls = choice.message.tool_calls.map(Self::parse_tool_calls); let usage = body.usage.map(|u| TokenUsage { prompt_tokens: u.prompt_tokens, completion_tokens: u.completion_tokens, total_tokens: u.total_tokens, }).unwrap_or(TokenUsage { prompt_tokens: 0, completion_tokens: 0, total_tokens: 0, }); AttemptOutcome::Ok(CompletionResponse { text, model: body.model, usage, tool_calls, reasoning_content: choice.message.reasoning_content, }) } }) .await } async fn stream(&self, request: CompletionRequest) -> anyhow::Result { let mut req = request; req.stream = true; let openai_req = self.convert_request(req); debug!(model = %openai_req.model, "OpenAI 流式调用"); // send 阶段需 timeout 防 hang(同 Anthropic 路径)。 // 不能用 reqwest .timeout()(会砍流式 body),改用 tokio::time::timeout 包裹 send。 let send_future = self .client .post(self.chat_url()) .header("Authorization", format!("Bearer {}", self.api_key)) .header("Content-Type", "application/json") .json(&openai_req) .send(); let resp = match tokio::time::timeout(Duration::from_secs(60), send_future).await { Ok(Ok(r)) => r, Ok(Err(e)) => { tracing::error!(error = %e, is_timeout = e.is_timeout(), "OpenAI 流式 send 失败"); return Err(e.into()); } Err(_elapsed) => { tracing::error!(url = %self.chat_url(), "OpenAI 流式 send 超时(60s 未返回响应头)"); anyhow::bail!("流式请求超时(60秒未收到 HTTP 响应,可能服务不可达或被防火墙拦截)"); } }; if !resp.status().is_success() { let status = resp.status(); let body = resp.text().await.unwrap_or_default(); error!(%status, %body, "LLM 流式 API 调用失败"); anyhow::bail!("LLM 流式 API 错误 {}: {}", status, body); } // 原生 SSE 解析器替代 eventsource-stream 库。 // eventsource-stream 在 Windows 上对 Deepseek 等响应报 "error decoding response body" // (严格 UTF-8 + SSE 协议校验,跨 chunk 字符/不完整事件均报错且不可恢复)。 // 原生解析器:bytes 累积 + from_utf8_lossy 宽松处理 + \n\n 分隔,容错不中断流。 let mut last_usage: Option = None; // MidStream error(中转站按 OpenAI 协议在流中途发 error 帧)时附 messages 摘要定位哪条非法 // (对齐 anthropic_compat 672)。 let messages_summary = Self::summarize_openai_messages(&openai_req.messages); let sse = crate::sse_parser::SseStream::new(resp.bytes_stream()); let stream = sse.flat_map(move |result: Result, String>| { let mut chunks: Vec> = Vec::new(); match result { Ok(events) => { for data in events { let mut chunk = apply_openai_sse(&data, &mut last_usage); if let Some(err) = chunk.error.as_mut() { *err = format!("{} | messages 摘要: {}", err, messages_summary); } chunks.push(Ok(chunk)); } } Err(e) => { let ctx = format!("SSE 流错误: {}", e); error!("{}", ctx); chunks.push(Err(anyhow::anyhow!("{}", ctx))); } } futures::stream::iter(chunks) }); Ok(Box::pin(stream)) } fn name(&self) -> &str { &self.default_model } fn endpoint(&self) -> String { self.chat_url() } } // ============================================================ // 单测(不发真实 HTTP,喂构造的 SSE data 字符串序列) // ============================================================ #[cfg(test)] mod tests { use super::*; /// 辅助:构造普通文本 delta chunk 的 SSE data fn text_chunk(content: &str, finish_reason: Option<&str>) -> String { let fr = match finish_reason { Some(r) => format!(", \"finish_reason\": \"{}\"", r), None => String::from(", \"finish_reason\": null"), }; format!( r#"{{"choices":[{{"delta":{{"content":"{}"}}{}}}]}}"#, content, fr ) } /// 辅助:构造带 usage 的 chunk(choices 为空 → usage-only 末 chunk,对应 include_usage) fn usage_only_chunk(prompt: u32, completion: u32) -> String { format!( r#"{{"choices":[],"usage":{{"prompt_tokens":{},"completion_tokens":{},"total_tokens":{}}}}}"#, prompt, completion, prompt + completion ) } /// 辅助:构造既有 content 又带 usage 的末段 chunk(部分兼容端点会把 usage 挂到正常末 chunk 上) fn text_chunk_with_usage(content: &str, finish_reason: &str, prompt: u32, completion: u32) -> String { format!( r#"{{"choices":[{{"delta":{{"content":"{}"}},"finish_reason":"{}"}}],"usage":{{"prompt_tokens":{},"completion_tokens":{},"total_tokens":{}}}}}"#, content, finish_reason, prompt, completion, prompt + completion ) } /// 多 chunk 文本流后,末 chunk 携带 usage(include_usage 覆盖语义) #[test] fn openai_sse_multi_chunk_with_terminal_usage() { let mut acc: Option = None; // 1) 首个文本增量,无 usage let c = apply_openai_sse(&text_chunk("Hello", None), &mut acc); assert_eq!(c.delta, "Hello"); assert!(!c.finished); assert!(c.usage.is_none()); assert!(acc.is_none(), "无 usage 的 chunk 不应改累加器"); // 2) 第二个文本增量 let c = apply_openai_sse(&text_chunk(" world", None), &mut acc); assert_eq!(c.delta, " world"); assert!(!c.finished); assert!(acc.is_none()); // 3) 末段正常 chunk 带 finish_reason=stop(仍是文本 delta,不带 usage) let c = apply_openai_sse(&text_chunk("", Some("stop")), &mut acc); assert!(c.finished); assert_eq!(c.delta, ""); assert!(acc.is_none(), "此 chunk 无 usage 字段,累加器仍为 None"); // 4) usage-only chunk(choices=[])携带累计 usage → 覆盖累加器 let c = apply_openai_sse(&usage_only_chunk(12, 34), &mut acc); assert!(!c.finished); assert!(c.usage.is_none(), "非 [DONE] chunk 不带出 usage"); let acc = acc.expect("累加器应已被 usage-only chunk 覆盖写入"); assert_eq!(acc.prompt_tokens, 12); assert_eq!(acc.completion_tokens, 34); assert_eq!(acc.total_tokens, 46); } /// usage 挂在正常末段 chunk(含 finish_reason)上,而非独立 usage-only chunk #[test] fn openai_sse_usage_on_terminal_text_chunk() { let mut acc: Option = None; let c = apply_openai_sse(&text_chunk_with_usage("", "stop", 100, 200), &mut acc); assert!(c.finished); assert!(c.usage.is_none(), "非 [DONE] 不带出 usage,仅覆盖累加器"); let acc = acc.expect("末段 chunk 的 usage 应已覆盖累加器"); assert_eq!(acc.prompt_tokens, 100); assert_eq!(acc.completion_tokens, 200); assert_eq!(acc.total_tokens, 300); } /// [DONE] 时 take() 带出累积 usage,且取走后累加器清空 #[test] fn openai_sse_done_takes_accumulated_usage() { let mut acc: Option = None; apply_openai_sse(&text_chunk("x", None), &mut acc); apply_openai_sse(&usage_only_chunk(5, 7), &mut acc); let c = apply_openai_sse("[DONE]", &mut acc); assert!(c.finished); let u = c.usage.expect("[DONE] 应带出累积 usage"); assert_eq!(u.prompt_tokens, 5); assert_eq!(u.completion_tokens, 7); assert_eq!(u.total_tokens, 12); assert!(acc.is_none(), "take() 后累加器应清空"); } /// 无 usage 的流:[DONE] 时 usage 字段为 None #[test] fn openai_sse_done_without_usage() { let mut acc: Option = None; apply_openai_sse(&text_chunk("hi", None), &mut acc); let c = apply_openai_sse("[DONE]", &mut acc); assert!(c.finished); assert!(c.usage.is_none(), "全程无 usage 时 [DONE] usage 应为 None"); assert!(acc.is_none()); } /// 后续 usage chunk 覆盖先前 usage(多轮 / 重发场景) #[test] fn openai_sse_later_usage_overrides_earlier() { let mut acc: Option = None; apply_openai_sse(&usage_only_chunk(1, 1), &mut acc); apply_openai_sse(&usage_only_chunk(50, 60), &mut acc); let c = apply_openai_sse("[DONE]", &mut acc); let u = c.usage.unwrap(); assert_eq!(u.prompt_tokens, 50, "末 usage 应覆盖前值"); assert_eq!(u.completion_tokens, 60); assert_eq!(u.total_tokens, 110); } /// finish_reason=length(max_tokens 截断)按正常终止处理 #[test] fn openai_sse_length_finish_reason_treated_as_finished() { let mut acc: Option = None; let c = apply_openai_sse(&text_chunk("...", Some("length")), &mut acc); assert!(c.finished, "length 应视为正常终止"); assert!(acc.is_none()); } /// 非法 JSON data → 返回空 chunk,不 panic、不改累加器 #[test] fn openai_sse_malformed_json_yields_empty_chunk() { let mut acc: Option = None; let c = apply_openai_sse("not a json", &mut acc); assert_eq!(c.delta, ""); assert!(!c.finished); assert!(c.usage.is_none()); assert!(acc.is_none()); } /// tool_calls 增量解析 #[test] fn openai_sse_tool_call_delta() { let mut acc: Option = None; let data = r#"{"choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_1","function":{"name":"get_weather","arguments":"{\"q\":"}}]}}]}"#; let c = apply_openai_sse(data, &mut acc); assert!(acc.is_none()); let tcs = c.tool_calls.expect("应有 tool_calls 增量"); assert_eq!(tcs.len(), 1); assert_eq!(tcs[0].index, 0); assert_eq!(tcs[0].id.as_deref(), Some("call_1")); assert_eq!(tcs[0].function_name.as_deref(), Some("get_weather")); assert_eq!(tcs[0].function_arguments.as_deref(), Some("{\"q\":")); assert!(!c.finished); } /// 流中途 error 事件 → error 为 Some(msg),finished=false(避免残缺被当正常完成入库),不污染 usage 累加 #[test] fn openai_sse_midstream_error_event() { let mut acc: Option = None; // 先累积一段 usage,验证 error 分支不污染累加器 apply_openai_sse(&usage_only_chunk(10, 20), &mut acc); let data = r#"{"choices":[],"error":{"message":"context length exceeded","type":"invalid_request_error"}}"#; let c = apply_openai_sse(data, &mut acc); assert!(!c.finished, "error 帧不应走 finished 完成路径"); assert_eq!(c.delta, ""); assert!(c.tool_calls.is_none()); assert!(c.usage.is_none(), "error 帧不应带出 usage"); let err = c.error.expect("error 帧应映射为 Some(msg)"); assert_eq!(err, "context length exceeded"); // 累加器保持原值(未被覆盖/清空) let acc = acc.expect("累加器应保留先前 usage 不受 error 影响"); assert_eq!(acc.prompt_tokens, 10); assert_eq!(acc.completion_tokens, 20); } /// error 无 message 字段 → 兜底 "stream error" 字符串 #[test] fn openai_sse_midstream_error_without_message_falls_back() { let mut acc: Option = None; // error 形态异常(只有 type,无 message) let data = r#"{"choices":[],"error":{"type":"server_error"}}"#; let c = apply_openai_sse(data, &mut acc); assert!(!c.finished); assert_eq!(c.error.as_deref(), Some("stream error"), "无 message 字段应兜底"); } // ---------- 多模态 convert_request ---------- /// 含图消息 → content 数组(text + image_url data URI);纯文本 → 字符串简写 #[test] fn openai_convert_multimodal_content() { let provider = OpenAICompatProvider::new("https://api.openai.com", "k", "gpt-4o"); let req = CompletionRequest { model: "gpt-4o".into(), messages: vec![ChatMessage::user_parts( "看图", vec![ crate::provider::ContentPart::image_base64("image/png", "iVBOR"), crate::provider::ContentPart::image_url("https://x/a.png"), ], )], temperature: None, max_tokens: None, stream: false, tools: None, tool_choice: None, reasoning_content: None, }; let out = provider.convert_request(req); let msg = &out.messages[0]; // content 是数组:[text "看图", image_url(data URI), image_url(http url)] let arr = msg.content.as_array().expect("含图 → content 数组"); assert_eq!(arr.len(), 3); assert_eq!(arr[0]["type"], "text"); assert_eq!(arr[0]["text"], "看图"); assert_eq!(arr[1]["type"], "image_url"); assert_eq!( arr[1]["image_url"]["url"], "data:image/png;base64,iVBOR" ); assert_eq!(arr[2]["image_url"]["url"], "https://x/a.png"); } /// 纯文本消息 → content 仍是字符串简写(无图不数组化,对齐纯文本端点兼容) #[test] fn openai_convert_text_only_remains_string() { let provider = OpenAICompatProvider::new("https://api.openai.com", "k", "gpt-4o"); let req = CompletionRequest { model: "gpt-4o".into(), messages: vec![ChatMessage::user("hello")], temperature: None, max_tokens: None, stream: false, tools: None, tool_choice: None, reasoning_content: None, }; let out = provider.convert_request(req); let msg = &out.messages[0]; assert_eq!(msg.content, serde_json::Value::String("hello".into())); } // ---------- ensure_leading_user(首条非 user/system → 补 user 占位,OpenAI 对称 Anthropic)---------- /// 首条 assistant → 补 user 占位(对齐 Anthropic)。上游绕过 sanitize 的 /// 调用方(title/knowledge_inject/工作流节点)可能传入首条 assistant 序列,补占位保留上下文。 #[test] fn openai_ensure_leading_user_first_assistant_gets_placeholder() { let provider = OpenAICompatProvider::new("https://api.openai.com", "k", "gpt-4o"); let req = CompletionRequest { model: "gpt-4o".into(), messages: vec![ ChatMessage::assistant("我来帮你"), ChatMessage::user("继续"), ], temperature: None, max_tokens: None, stream: false, tools: None, tool_choice: None, reasoning_content: None, }; let out = provider.convert_request(req); assert_eq!(out.messages.len(), 3, "占位 + 原 2 条"); assert_eq!(out.messages[0].role.as_str(), "user", "首条应为 user(补占位)"); assert_eq!(out.messages[1].role.as_str(), "assistant"); assert_eq!(out.messages[2].role.as_str(), "user"); } /// 正常序列(user 开头)不补占位——零回归。 #[test] fn openai_ensure_leading_user_normal_unchanged() { let provider = OpenAICompatProvider::new("https://api.openai.com", "k", "gpt-4o"); let req = CompletionRequest { model: "gpt-4o".into(), messages: vec![ ChatMessage::user("hello"), ChatMessage::assistant("hi"), ], temperature: None, max_tokens: None, stream: false, tools: None, tool_choice: None, reasoning_content: None, }; let out = provider.convert_request(req); assert_eq!(out.messages.len(), 2, "正常序列不补占位"); assert_eq!(out.messages[0].role.as_str(), "user"); } // ---------- 三元组一致性自愈(P0:治 DeepSeek/OpenAI 400) ---------- /// 辅助:取 assistant 消息的 tool_call id 列表(发出去的形态)。 fn openai_tool_call_ids(m: &OpenAiMessage) -> Vec { m.tool_calls .as_ref() .map(|arr| { arr.iter() .filter_map(|tc| tc.get("id").and_then(|v| v.as_str()).map(String::from)) .collect() }) .unwrap_or_default() } /// 正常三元组(各 id 闭合)零介入:assistant(tc=[a]) → tool(a) → assistant(tc=[b]) → tool(b)。 /// 约束铁律:不破正常三元组。 #[test] fn openai_sanitize_keeps_closed_triplets() { let provider = OpenAICompatProvider::new("https://api.deepseek.com", "k", "deepseek-chat"); let req = CompletionRequest { model: "deepseek-chat".into(), messages: vec![ ChatMessage::user("查天气"), ChatMessage::assistant_with_tools( "调用中", vec![ToolCall::new("call_a", "get_weather", "{}")], ), ChatMessage::tool_result("call_a", "晴"), ChatMessage::assistant_with_tools( "再查", vec![ToolCall::new("call_b", "get_weather", "{}")], ), ChatMessage::tool_result("call_b", "雨"), ], temperature: None, max_tokens: None, stream: false, tools: None, tool_choice: None, reasoning_content: None, }; let out = provider.convert_request(req); // 5 条全保留(正常三元组不剥不丢)。 assert_eq!(out.messages.len(), 5, "正常三元组零介入,不应剥/丢任何消息"); // 两个 assistant 头的 tool_calls 完整保留。 let heads: Vec<&OpenAiMessage> = out .messages .iter() .filter(|m| m.role == "assistant") .collect(); assert_eq!(openai_tool_call_ids(heads[0]), vec!["call_a".to_string()]); assert_eq!(openai_tool_call_ids(heads[1]), vec!["call_b".to_string()]); } /// 末尾 assistant tool_calls 无 result(残末尾)→ 剥离 tool_calls(保留 assistant 文本)。 /// 防 "insufficient tool messages" 400。 #[test] fn openai_sanitize_strips_tail_unresolved_tool_calls() { let provider = OpenAICompatProvider::new("https://api.deepseek.com", "k", "deepseek-chat"); let req = CompletionRequest { model: "deepseek-chat".into(), messages: vec![ ChatMessage::user("查天气"), ChatMessage::assistant_with_tools( "调工具但 result 还没回来", vec![ToolCall::new("call_x", "get_weather", "{}")], ), ], temperature: None, max_tokens: None, stream: false, tools: None, tool_choice: None, reasoning_content: None, }; let out = provider.convert_request(req); // assistant 保留(content 不丢),但 tool_calls 被剥。 let asst = out .messages .iter() .find(|m| m.role == "assistant") .expect("assistant 应保留"); assert!( asst.tool_calls.is_none(), "未闭合 tool_calls 应被剥离" ); } /// orphan tool_result(无配对 assistant tool_calls 头)→ 丢弃。 /// 直构造/DB 残留路径绕过 ContextManager::sanitize_messages 时由本守卫兜底。 /// 防 "Messages with role tool must be a response to a preceding message with tool_calls" 400。 #[test] fn openai_sanitize_drops_orphan_tool_result_no_head() { let provider = OpenAICompatProvider::new("https://api.deepseek.com", "k", "deepseek-chat"); let req = CompletionRequest { model: "deepseek-chat".into(), messages: vec![ ChatMessage::user("问"), // 无头的 orphan tool_result(头被裁剪/丢失)。 ChatMessage::tool_result("orphan_id", "结果"), ChatMessage::assistant("回复"), ], temperature: None, max_tokens: None, stream: false, tools: None, tool_choice: None, reasoning_content: None, }; let out = provider.convert_request(req); // orphan tool 被丢弃,剩 user + assistant。 let tools: Vec<&OpenAiMessage> = out .messages .iter() .filter(|m| m.role == "tool") .collect(); assert!(tools.is_empty(), "无配对头的 orphan tool_result 应丢弃, 实际 {:?}", tools); assert_eq!(out.messages.len(), 2, "应剩 user + assistant"); } /// assistant tool_calls 剥离后,对应 orphan tool_result 同步丢弃(一致性)。 /// 场景:assistant(tc=[a,b]) → tool(a)(b 的 result 丢失)。旧逻辑因下一条是 tool /// 不剥 → 发出未闭合 b → 400。新逻辑按 id 精确配对:剥 b(保留 a),tool(a) 保留。 #[test] fn openai_sanitize_partial_triplet_strips_unresolved_id() { let provider = OpenAICompatProvider::new("https://api.deepseek.com", "k", "deepseek-chat"); let req = CompletionRequest { model: "deepseek-chat".into(), messages: vec![ ChatMessage::user("问"), ChatMessage::assistant_with_tools( "调两工具", vec![ ToolCall::new("call_a", "tool_a", "{}"), ToolCall::new("call_b", "tool_b", "{}"), ], ), // 只回了 call_a,call_b 的 result 丢失。 ChatMessage::tool_result("call_a", "a 结果"), ], temperature: None, max_tokens: None, stream: false, tools: None, tool_choice: None, reasoning_content: None, }; let out = provider.convert_request(req); let asst = out .messages .iter() .find(|m| m.role == "assistant") .expect("assistant 应保留"); // 只保留 call_a(已闭合),剥 call_b(未闭合)。 assert_eq!( openai_tool_call_ids(asst), vec!["call_a".to_string()], "部分闭合头应只留已闭合 call_a, 剥未闭合 call_b" ); // tool(call_a) 保留(有配对头)。 let tools: Vec<&OpenAiMessage> = out .messages .iter() .filter(|m| m.role == "tool") .collect(); assert_eq!(tools.len(), 1, "call_a 的 tool_result 应保留"); } /// 全未闭合三元组:assistant(tc=[a]) 但全程无 tool(a) → 剥 tool_calls, /// 且不残留任何 orphan tool(本就无 tool 消息)。 #[test] fn openai_sanitize_fully_unresolved_strips_all() { let provider = OpenAICompatProvider::new("https://api.deepseek.com", "k", "deepseek-chat"); let req = CompletionRequest { model: "deepseek-chat".into(), messages: vec![ ChatMessage::user("问"), ChatMessage::assistant_with_tools( "调工具无结果", vec![ ToolCall::new("call_y", "tool_y", "{}"), ToolCall::new("call_z", "tool_z", "{}"), ], ), ChatMessage::assistant("纯文本续"), ], temperature: None, max_tokens: None, stream: false, tools: None, tool_choice: None, reasoning_content: None, }; let out = provider.convert_request(req); let heads: Vec<&OpenAiMessage> = out .messages .iter() .filter(|m| m.role == "assistant") .collect(); // 第一个 assistant(原含 tool_calls)应被剥空;第二个纯文本不变。 assert!( heads[0].tool_calls.is_none(), "全未闭合 tool_calls 应全部剥离" ); assert!(heads[1].tool_calls.is_none(), "纯文本 assistant 无 tool_calls"); } /// 无 tool_call_id 的 tool 消息(异常数据)→ 丢弃(发出去必 400)。 #[test] fn openai_sanitize_drops_tool_without_call_id() { let provider = OpenAICompatProvider::new("https://api.deepseek.com", "k", "deepseek-chat"); let mut bad_tool = ChatMessage::tool_result("temp", "结果"); bad_tool.tool_call_id = None; // 异常:无 id let req = CompletionRequest { model: "deepseek-chat".into(), messages: vec![ChatMessage::user("问"), bad_tool], temperature: None, max_tokens: None, stream: false, tools: None, tool_choice: None, reasoning_content: None, }; let out = provider.convert_request(req); let tools: Vec<&OpenAiMessage> = out .messages .iter() .filter(|m| m.role == "tool") .collect(); assert!( tools.is_empty(), "无 tool_call_id 的 tool 消息应丢弃, 实际 {:?}", tools ); } /// CR-空 id:parse_tool_calls 对空 id 按 index 生成 gen_tool_{i} fallback,非空原样。 /// 根因:SenseNova 等兼容缺陷 provider 发空 tool_call.id,多 tool_call 同 id(空串) /// 致 audit/mod.rs:203 seen_ids 去重只留首个 → 所有工具结果路由到首个。 #[test] fn openai_parse_tool_calls_empty_id_fallback_unique() { let calls = vec![ OpenAiToolCallResp { id: String::new(), call_type: "function".into(), function: OpenAiFunctionResp { name: "list_dir".into(), arguments: r#"{"path":"docs"}"#.into() }, }, OpenAiToolCallResp { id: String::new(), call_type: "function".into(), function: OpenAiFunctionResp { name: "list_dir".into(), arguments: r#"{"path":"crates"}"#.into() }, }, OpenAiToolCallResp { id: "call_abc123".into(), call_type: "function".into(), function: OpenAiFunctionResp { name: "read_file".into(), arguments: r#"{"path":"根"}"#.into() }, }, ]; let parsed = OpenAICompatProvider::parse_tool_calls(calls); assert_eq!(parsed.len(), 3); // 空 id → fallback(按 index),保证唯一 assert_eq!(parsed[0].id, "gen_tool_0"); assert_eq!(parsed[1].id, "gen_tool_1"); // 非空 id 原样透传 assert_eq!(parsed[2].id, "call_abc123"); // name/args 透传无损 assert_eq!(parsed[0].function.name, "list_dir"); assert_eq!(parsed[1].function.arguments, r#"{"path":"crates"}"#); // 关键:所有 id 互异(去重后不丢工具) let mut ids: Vec<&str> = parsed.iter().map(|c| c.id.as_str()).collect(); ids.sort(); let unique: Vec<&str> = { let mut u = ids.clone(); u.dedup(); u }; assert_eq!(ids.len(), unique.len(), "id 应全部唯一,实际 {:?}", ids); } /// CR-空 id 流式:SSE chunk 携带 `"id":""`(SenseNova 兼容缺陷)→ ToolCallDelta.id /// 转为 `gen_stream_{index}` fallback(非 None),保证下游 accumulate_tool_calls 写入 /// draft.id 非空。chunk 完全无 id 字段(None)保持 None(OpenAI 协议:仅首 chunk 有 id, /// 后续 chunk 无 id 不应覆盖首 chunk 权威 id),由 agentic 转换点兜底。 #[test] fn openai_stream_chunk_empty_id_fallback() { let mut acc: Option = None; // chunk 1: tool_call index=0, id="" → fallback gen_stream_0 let data1 = r#"{"choices":[{"delta":{"tool_calls":[{"index":0,"id":"","type":"function","function":{"name":"list_dir","arguments":"{\"path\":\"docs\"}"}}]}}]}"#; let c1 = apply_openai_sse(data1, &mut acc); let tc1 = c1.tool_calls.as_ref().expect("应有 tool_calls").first().unwrap(); assert_eq!(tc1.index, 0); assert_eq!(tc1.id.as_deref(), Some("gen_stream_0"), "空 id 应转 fallback"); // chunk 2: tool_call index=1, id="" → fallback gen_stream_1(与 index=0 不同,唯一) let data2 = r#"{"choices":[{"delta":{"tool_calls":[{"index":1,"id":"","type":"function","function":{"name":"read_file","arguments":""}}]}}]}"#; let c2 = apply_openai_sse(data2, &mut acc); let tc2 = c2.tool_calls.as_ref().expect("应有 tool_calls").first().unwrap(); assert_eq!(tc2.id.as_deref(), Some("gen_stream_1"), "不同 index fallback 应不同"); // chunk 3: tool_call index=0, 无 id 字段(None)→ 保持 None(不覆盖首 chunk) let data3 = r#"{"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"更多参数"}}]}}]}"#; let c3 = apply_openai_sse(data3, &mut acc); let tc3 = c3.tool_calls.as_ref().expect("应有 tool_calls").first().unwrap(); assert!(tc3.id.is_none(), "无 id 字段 chunk 应保持 None,不覆盖首 chunk 权威 id"); // chunk 4: tool_call 非空 id → 原样透传 let data4 = r#"{"choices":[{"delta":{"tool_calls":[{"index":2,"id":"call_xyz","type":"function","function":{"name":"write"}}]}}]}"#; let c4 = apply_openai_sse(data4, &mut acc); let tc4 = c4.tool_calls.as_ref().expect("应有 tool_calls").first().unwrap(); assert_eq!(tc4.id.as_deref(), Some("call_xyz"), "非空 id 原样透传"); } }