From a3835e2acb578559aa87085098c8179e8d79c215 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=BB=9D=E5=B0=98?= <237809796@qq.com> Date: Fri, 19 Jun 2026 04:36:42 +0800 Subject: [PATCH] =?UTF-8?q?=E9=87=8D=E6=9E=84:=20=E6=8B=86openai=5Fcompat?= =?UTF-8?q?=E5=8D=8F=E8=AE=AE=E9=80=82=E9=85=8D(strategy=E6=A0=B8=E5=BF=83?= =?UTF-8?q?=E5=BA=93=C2=B7=E5=AF=B9=E9=BD=90anthropic)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新建 df-ai/openai_helpers.rs(231行): apply_openai_sse纯函数 + 13 struct(OpenAi*字段pub(crate)) - openai_compat.rs 809→597: 删抽离+use引入, Provider impl保留(impl块约束: new/convert_request/parse_tool_calls/LlmProvider trait) - lib.rs: pub mod openai_helpers 主代兜底: cargo check --workspace 0 + test df-ai 119 strategy: 核心库自底向上, impl块约束(Provider方法保留只抽纯函数/类型), 对齐anthropic_helpers模式 git add指定(df-ai/*) 协议适配双拆完整: anthropic_compat + openai_compat 均helpers抽离 --- crates/df-ai/src/lib.rs | 1 + crates/df-ai/src/openai_compat.rs | 226 +--------------------------- crates/df-ai/src/openai_helpers.rs | 231 +++++++++++++++++++++++++++++ 3 files changed, 239 insertions(+), 219 deletions(-) create mode 100644 crates/df-ai/src/openai_helpers.rs diff --git a/crates/df-ai/src/lib.rs b/crates/df-ai/src/lib.rs index 4dd3795..1e425c3 100644 --- a/crates/df-ai/src/lib.rs +++ b/crates/df-ai/src/lib.rs @@ -9,6 +9,7 @@ pub mod coordinator; pub mod model_fetch; pub mod model_probe; pub mod openai_compat; +pub mod openai_helpers; pub mod provider; pub mod router; // CR-30-1: 流前重试退避对外复用。complete() 的 retry_with_backoff 仍 crate 内用, diff --git a/crates/df-ai/src/openai_compat.rs b/crates/df-ai/src/openai_compat.rs index 213eb10..9d684d4 100644 --- a/crates/df-ai/src/openai_compat.rs +++ b/crates/df-ai/src/openai_compat.rs @@ -9,12 +9,10 @@ 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, + CompletionRequest, CompletionResponse, LlmProvider, StreamResult, TokenUsage, ToolCall, }; // ChatMessage 仅单测构造 CompletionRequest 用，避免非 test 构建的 unused import 警告。 #[cfg(test)] @@ -23,222 +21,12 @@ 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, - /// DeepSeek thinking 模式 - #[serde(skip_serializing_if = "Option::is_none")] - pub reasoning_content: Option, -} - -/// OpenAI 消息格式 -/// -/// `content` 为 `serde_json::Value`：纯文本消息走字符串简写（与老端点兼容）， -/// 含图消息走 `[{type:"text",text},{type:"image_url",image_url:{url}}]` 数组。 -#[derive(Debug, Clone, Serialize, Deserialize)] -struct OpenAiMessage { - role: String, - content: serde_json::Value, - #[serde(skip_serializing_if = "Option::is_none")] - tool_call_id: Option, - #[serde(skip_serializing_if = "Option::is_none")] - tool_calls: Option>, - /// DeepSeek thinking 模式推理内容 - #[serde(skip_serializing_if = "Option::is_none")] - pub reasoning_content: Option, -} - -/// OpenAI 同步响应 -#[derive(Debug, Deserialize)] -struct OpenAiResponse { - choices: Vec, - model: String, - usage: Option, -} - -#[derive(Debug, Deserialize)] -struct OpenAiChoice { - message: OpenAiMessageResp, - // SW-260618-24: OpenAI 响应反序列化字段,保留以备调试/未来消费(如日志记录调用终止原因),标注意图消除 dead_code warning - #[allow(dead_code)] - finish_reason: Option, -} - -#[derive(Debug, Deserialize)] -struct OpenAiMessageResp { - content: Option, - tool_calls: Option>, - /// DeepSeek thinking 模式响应中的推理内容 - #[serde(default, skip_serializing_if = "Option::is_none")] - pub reasoning_content: Option, -} - -#[derive(Debug, Deserialize)] -struct OpenAiToolCallResp { - id: String, - // SW-260618-24: 反序列化 #[serde(rename="type")] 字段,保留以对齐 OpenAI 响应结构,标注意图消除 dead_code warning - #[allow(dead_code)] - #[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>, - /// DeepSeek thinking 模式流式推理内容 - #[serde(default, skip_serializing_if = "Option::is_none")] - pub reasoning_content: 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, - reasoning_content: 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, - reasoning_content: choice.delta.reasoning_content, - } - } else { - // choices 为空 = usage-only chunk，不输出文本（usage 已累积） - StreamChunk { - delta: String::new(), - finished: false, - tool_calls: None, - usage: None, - error: None, - reasoning_content: None, - } - } - } - Err(e) => { - debug!("SSE 数据解析失败: {} — data: {}", e, data); - StreamChunk { - delta: String::new(), - finished: false, - tool_calls: None, - usage: None, - error: None, - reasoning_content: None, - } - } - } -} +// 协议数据结构（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 diff --git a/crates/df-ai/src/openai_helpers.rs b/crates/df-ai/src/openai_helpers.rs new file mode 100644 index 0000000..a092449 --- /dev/null +++ b/crates/df-ai/src/openai_helpers.rs @@ -0,0 +1,231 @@ +//! OpenAI 兼容协议适配 — 协议数据结构与 SSE 事件解析（纯函数）。 +//! +//! 本模块从 `openai_compat.rs` 抽离，承载与 HTTP 无关的纯协议逻辑： +//! - 请求/响应结构体（`OpenAiRequest` / `OpenAiResponse` / `OpenAiStreamChunk` 等） +//! - SSE 事件 data → `StreamChunk` 转换纯函数（`apply_openai_sse`） +//! +//! Provider struct + impl（含 HTTP 调用 / `convert_request` / `chat_url`）仍留在 +//! `openai_compat.rs`，Rust impl 块不可跨文件，故仅搬迁 impl 块外部的类型/纯函数。 +//! 零行为变更（纯搬迁）。结构对齐 `anthropic_helpers.rs`。 + +use serde::{Deserialize, Serialize}; +use tracing::debug; + +use crate::provider::{StreamChunk, TokenUsage, ToolCallDelta}; + +// ============================================================ +// OpenAI API 请求/响应结构体 +// ============================================================ + +/// OpenAI 兼容请求体 +#[derive(Debug, Clone, Serialize)] +pub(crate) struct OpenAiRequest { + pub model: String, + pub messages: Vec, + #[serde(skip_serializing_if = "Option::is_none")] + pub temperature: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub max_tokens: Option, + pub stream: bool, + #[serde(skip_serializing_if = "Option::is_none")] + pub tools: Option>, + #[serde(skip_serializing_if = "Option::is_none")] + pub tool_choice: Option, + /// 流式时请求末 chunk 携带 usage（OpenAI 官方 + DeepSeek/GLM 兼容） + #[serde(skip_serializing_if = "Option::is_none")] + pub stream_options: Option, + /// DeepSeek thinking 模式 + #[serde(skip_serializing_if = "Option::is_none")] + pub reasoning_content: Option, +} + +/// OpenAI 消息格式 +/// +/// `content` 为 `serde_json::Value`：纯文本消息走字符串简写（与老端点兼容）， +/// 含图消息走 `[{type:"text",text},{type:"image_url",image_url:{url}}]` 数组。 +#[derive(Debug, Clone, Serialize, Deserialize)] +pub(crate) struct OpenAiMessage { + pub role: String, + pub content: serde_json::Value, + #[serde(skip_serializing_if = "Option::is_none")] + pub tool_call_id: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub tool_calls: Option>, + /// DeepSeek thinking 模式推理内容 + #[serde(skip_serializing_if = "Option::is_none")] + pub reasoning_content: Option, +} + +/// OpenAI 同步响应 +#[derive(Debug, Deserialize)] +pub(crate) struct OpenAiResponse { + pub choices: Vec, + pub model: String, + pub usage: Option, +} + +#[derive(Debug, Deserialize)] +pub(crate) struct OpenAiChoice { + pub message: OpenAiMessageResp, + // SW-260618-24: OpenAI 响应反序列化字段,保留以备调试/未来消费(如日志记录调用终止原因),标注意图消除 dead_code warning + #[allow(dead_code)] + pub finish_reason: Option, +} + +#[derive(Debug, Deserialize)] +pub(crate) struct OpenAiMessageResp { + pub content: Option, + pub tool_calls: Option>, + /// DeepSeek thinking 模式响应中的推理内容 + #[serde(default, skip_serializing_if = "Option::is_none")] + pub reasoning_content: Option, +} + +#[derive(Debug, Deserialize)] +pub(crate) struct OpenAiToolCallResp { + pub id: String, + // SW-260618-24: 反序列化 #[serde(rename="type")] 字段,保留以对齐 OpenAI 响应结构,标注意图消除 dead_code warning + #[allow(dead_code)] + #[serde(rename = "type")] + pub call_type: String, + pub function: OpenAiFunctionResp, +} + +#[derive(Debug, Deserialize)] +pub(crate) struct OpenAiFunctionResp { + pub name: String, + pub arguments: String, +} + +#[derive(Debug, Deserialize)] +pub(crate) struct OpenAiUsage { + pub prompt_tokens: u32, + pub completion_tokens: u32, + pub total_tokens: u32, +} + +/// SSE 流式响应 chunk +#[derive(Debug, Deserialize)] +pub(crate) struct OpenAiStreamChunk { + pub choices: Vec, + /// 末 chunk（choices 为空）携带的累计 usage + #[serde(default)] + pub usage: Option, +} + +#[derive(Debug, Deserialize)] +pub(crate) struct OpenAiStreamChoice { + pub delta: OpenAiStreamDelta, + pub finish_reason: Option, +} + +#[derive(Debug, Deserialize)] +pub(crate) struct OpenAiStreamDelta { + pub content: Option, + pub tool_calls: Option>, + /// DeepSeek thinking 模式流式推理内容 + #[serde(default, skip_serializing_if = "Option::is_none")] + pub reasoning_content: Option, +} + +#[derive(Debug, Deserialize)] +pub(crate) struct OpenAiStreamToolCall { + pub index: u32, + pub id: Option, + pub function: Option, +} + +#[derive(Debug, Deserialize)] +pub(crate) struct OpenAiStreamFunction { + pub name: Option, + pub 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, + reasoning_content: 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, + reasoning_content: choice.delta.reasoning_content, + } + } else { + // choices 为空 = usage-only chunk，不输出文本（usage 已累积） + StreamChunk { + delta: String::new(), + finished: false, + tool_calls: None, + usage: None, + error: None, + reasoning_content: None, + } + } + } + Err(e) => { + debug!("SSE 数据解析失败: {} — data: {}", e, data); + StreamChunk { + delta: String::new(), + finished: false, + tool_calls: None, + usage: None, + error: None, + reasoning_content: None, + } + } + } +}