//! 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 eventsource_stream::Eventsource; use futures::StreamExt; use reqwest::Client; use serde::{Deserialize, Serialize}; use tracing::{debug, error, warn}; use crate::provider::{ CompletionRequest, CompletionResponse, LlmProvider, StreamChunk, StreamResult, TokenUsage, ToolCall, ToolCallDelta, }; use crate::retry::{ retry_with_backoff, AttemptOutcome, is_reqwest_error_retryable, is_status_retryable, }; // ============================================================ // OpenAI API 请求/响应结构体 // ============================================================ /// OpenAI 兼容请求体 #[derive(Debug, Clone, Serialize)] struct OpenAiRequest { model: String, messages: Vec, #[serde(skip_serializing_if = "Option::is_none")] temperature: Option, #[serde(skip_serializing_if = "Option::is_none")] max_tokens: Option, stream: bool, #[serde(skip_serializing_if = "Option::is_none")] tools: Option>, #[serde(skip_serializing_if = "Option::is_none")] tool_choice: Option, /// 流式时请求末 chunk 携带 usage(OpenAI 官方 + DeepSeek/GLM 兼容) #[serde(skip_serializing_if = "Option::is_none")] stream_options: Option, } /// OpenAI 消息格式 #[derive(Debug, Clone, Serialize, Deserialize)] struct OpenAiMessage { role: String, content: String, #[serde(skip_serializing_if = "Option::is_none")] tool_call_id: Option, #[serde(skip_serializing_if = "Option::is_none")] tool_calls: Option>, } /// OpenAI 同步响应 #[derive(Debug, Deserialize)] struct OpenAiResponse { choices: Vec, model: String, usage: Option, } #[derive(Debug, Deserialize)] struct OpenAiChoice { message: OpenAiMessageResp, finish_reason: Option, } #[derive(Debug, Deserialize)] struct OpenAiMessageResp { content: Option, tool_calls: Option>, } #[derive(Debug, Deserialize)] struct OpenAiToolCallResp { id: String, #[serde(rename = "type")] call_type: String, function: OpenAiFunctionResp, } #[derive(Debug, Deserialize)] struct OpenAiFunctionResp { name: String, arguments: String, } #[derive(Debug, Deserialize)] struct OpenAiUsage { prompt_tokens: u32, completion_tokens: u32, total_tokens: u32, } /// SSE 流式响应 chunk #[derive(Debug, Deserialize)] struct OpenAiStreamChunk { choices: Vec, /// 末 chunk(choices 为空)携带的累计 usage #[serde(default)] usage: Option, } #[derive(Debug, Deserialize)] struct OpenAiStreamChoice { delta: OpenAiStreamDelta, finish_reason: Option, } #[derive(Debug, Deserialize)] struct OpenAiStreamDelta { content: Option, tool_calls: Option>, } #[derive(Debug, Deserialize)] struct OpenAiStreamToolCall { index: u32, id: Option, function: Option, } #[derive(Debug, Deserialize)] struct OpenAiStreamFunction { name: Option, arguments: Option, } // ============================================================ // SSE 解析纯函数(与 HTTP 解耦,便于单测) // ============================================================ /// 将一条 OpenAI 兼容 SSE 事件 data 解析为 StreamChunk,并按需更新 usage 累加器。 /// /// - `[DONE]` → 返回 `finished=true` 的终态 chunk,`usage` 取自累加器(`take()`)。 /// - 普通文本/工具增量 chunk → 返回对应 `StreamChunk`,usage 字段恒为 None(usage 仅在终态带出)。 /// - usage(`stream_options.include_usage` 时末段或 usage-only chunk 携带)→ 覆盖累加器(覆盖语义保对)。 /// - 解析失败 → 返回空 chunk(与原内联实现一致)。 /// /// 等价性:delta / tool_calls / finished / 解析失败等分支与原 stream() 闭包逐字一致; /// usage 透传([DONE] 终态 take() 带出、usage chunk 覆盖累加器)为本次新增能力, /// 对应 StreamChunk 新增的 usage 字段 + 请求体新增 stream_options.include_usage。 pub(crate) fn apply_openai_sse(data: &str, usage_accum: &mut Option) -> StreamChunk { // OpenAI 发送 "data: [DONE]" 表示流结束,带出累积 usage if data == "[DONE]" { return StreamChunk { delta: String::new(), finished: true, tool_calls: None, usage: usage_accum.take(), error: None, }; } match serde_json::from_str::(data) { Ok(chunk) => { // 提取 usage(带 include_usage 时末段 chunk 携带,覆盖累积) if let Some(u) = chunk.usage { *usage_accum = Some(TokenUsage { prompt_tokens: u.prompt_tokens, completion_tokens: u.completion_tokens, total_tokens: u.total_tokens, }); } if let Some(choice) = chunk.choices.into_iter().next() { let delta_text = choice.delta.content.unwrap_or_default(); // "length" = max_tokens 截断,属正常终止(非断连),纳入 finished let finished = choice.finish_reason.as_deref() == Some("stop") || choice.finish_reason.as_deref() == Some("tool_calls") || choice.finish_reason.as_deref() == Some("length"); let tool_calls = choice.delta.tool_calls.map(|tcs| { tcs.into_iter() .map(|tc| ToolCallDelta { index: tc.index, id: tc.id, function_name: tc.function.as_ref().and_then(|f| f.name.clone()), function_arguments: tc.function.and_then(|f| f.arguments), }) .collect() }); StreamChunk { delta: delta_text, finished, tool_calls, usage: None, error: None, } } else { // choices 为空 = usage-only chunk,不输出文本(usage 已累积) StreamChunk { delta: String::new(), finished: false, tool_calls: None, usage: None, error: None, } } } Err(e) => { debug!("SSE 数据解析失败: {} — data: {}", e, data); StreamChunk { delta: String::new(), finished: false, tool_calls: None, usage: None, error: None, } } } } // ============================================================ // 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 { // connect_timeout 防连接阶段无限 hang(网络静默断);不设总 timeout—— // reqwest 的 .timeout() 会限制整个响应 body 时长,流式长生成任务会被误砍。 // 读取阶段的中途静默由上层 stream_llm 的 idle timeout 兜底。 let client = Client::builder() .connect_timeout(std::time::Duration::from_secs(30)) .build() .unwrap_or_else(|e| { warn!("reqwest builder 失败,回退默认 client: {}", e); Client::new() }); 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 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", }; 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: m.content, tool_call_id: m.tool_call_id, tool_calls, } }) .collect(); 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, // 流式请求末 chunk 带 usage(同步调用 complete 不需要) stream_options: if req.stream { Some(serde_json::json!({ "include_usage": true })) } else { None }, } } /// 解析同步响应中的工具调用 fn parse_tool_calls(calls: Vec) -> Vec { calls .into_iter() .map(|c| ToolCall::new(c.id, c.function.name, c.function.arguments)) .collect() } } #[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 同步调用"); // 指数退避重试(B-260616-07): 包裹 send + 状态码判定。 // 单请求 60s timeout 保持不变(FR-R4),重试是额外层: 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, }) } }) .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 流式调用"); let resp = self .client .post(self.chat_url()) .header("Authorization", format!("Bearer {}", self.api_key)) .header("Content-Type", "application/json") .json(&openai_req) .send() .await?; 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); } // 累积流式 usage:开 include_usage 后,末段正常 chunk(finish_reason)及额外 usage-only chunk(choices=[])都带 usage。 // usage 解析/累积逻辑抽到 apply_openai_sse 纯函数,便于单测;此处闭包只负责传 data 与传递 last_usage。 let mut last_usage: Option = None; let stream = resp .bytes_stream() .eventsource() .map(move |event| match event { Ok(event) => Ok(apply_openai_sse(&event.data, &mut last_usage)), Err(e) => { // 保留 #[source] 因果链: anyhow!("...{}", e) 仅把 e 的 Display 塞进 message, // 丢掉 source(无法 downcast/遍历)。改用 Error::from(e).context(...): // Display 不变(仍为 "SSE 流错误: {e}"), 且 e 作为 .source() 可追溯。 // 顺序: 先 format(e) 构造 context 文案, 再 Error::from(e) move e 进 source。 let ctx = format!("SSE 流错误: {}", e); error!("{}", ctx); Err(anyhow::Error::from(e).context(ctx)) } }); 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); } }