重构: 拆openai_compat协议适配(strategy核心库·对齐anthropic)
- 新建 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抽离
This commit is contained in:
@@ -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 内用,
|
||||
|
||||
@@ -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<OpenAiMessage>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
temperature: Option<f32>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
max_tokens: Option<u32>,
|
||||
stream: bool,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
tools: Option<Vec<serde_json::Value>>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
tool_choice: Option<serde_json::Value>,
|
||||
/// 流式时请求末 chunk 携带 usage（OpenAI 官方 + DeepSeek/GLM 兼容）
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
stream_options: Option<serde_json::Value>,
|
||||
/// DeepSeek thinking 模式
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub reasoning_content: Option<String>,
|
||||
}
|
||||
|
||||
/// 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<String>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
tool_calls: Option<Vec<serde_json::Value>>,
|
||||
/// DeepSeek thinking 模式推理内容
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub reasoning_content: Option<String>,
|
||||
}
|
||||
|
||||
/// OpenAI 同步响应
|
||||
#[derive(Debug, Deserialize)]
|
||||
struct OpenAiResponse {
|
||||
choices: Vec<OpenAiChoice>,
|
||||
model: String,
|
||||
usage: Option<OpenAiUsage>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
struct OpenAiChoice {
|
||||
message: OpenAiMessageResp,
|
||||
// SW-260618-24: OpenAI 响应反序列化字段,保留以备调试/未来消费(如日志记录调用终止原因),标注意图消除 dead_code warning
|
||||
#[allow(dead_code)]
|
||||
finish_reason: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
struct OpenAiMessageResp {
|
||||
content: Option<String>,
|
||||
tool_calls: Option<Vec<OpenAiToolCallResp>>,
|
||||
/// DeepSeek thinking 模式响应中的推理内容
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub reasoning_content: Option<String>,
|
||||
}
|
||||
|
||||
#[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<OpenAiStreamChoice>,
|
||||
/// 末 chunk（choices 为空）携带的累计 usage
|
||||
#[serde(default)]
|
||||
usage: Option<OpenAiUsage>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
struct OpenAiStreamChoice {
|
||||
delta: OpenAiStreamDelta,
|
||||
finish_reason: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
struct OpenAiStreamDelta {
|
||||
content: Option<String>,
|
||||
tool_calls: Option<Vec<OpenAiStreamToolCall>>,
|
||||
/// DeepSeek thinking 模式流式推理内容
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub reasoning_content: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
struct OpenAiStreamToolCall {
|
||||
index: u32,
|
||||
id: Option<String>,
|
||||
function: Option<OpenAiStreamFunction>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
struct OpenAiStreamFunction {
|
||||
name: Option<String>,
|
||||
arguments: Option<String>,
|
||||
}
|
||||
|
||||
// ============================================================
|
||||
// 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<TokenUsage>) -> 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,
|
||||
// 协议数据结构（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,
|
||||
};
|
||||
}
|
||||
|
||||
match serde_json::from_str::<OpenAiStreamChunk>(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,
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ============================================================
|
||||
// OpenAI Compat Provider
|
||||
|
||||
231
crates/df-ai/src/openai_helpers.rs
Normal file
231
crates/df-ai/src/openai_helpers.rs
Normal file
@@ -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<OpenAiMessage>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub temperature: Option<f32>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub max_tokens: Option<u32>,
|
||||
pub stream: bool,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub tools: Option<Vec<serde_json::Value>>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub tool_choice: Option<serde_json::Value>,
|
||||
/// 流式时请求末 chunk 携带 usage（OpenAI 官方 + DeepSeek/GLM 兼容）
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub stream_options: Option<serde_json::Value>,
|
||||
/// DeepSeek thinking 模式
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub reasoning_content: Option<String>,
|
||||
}
|
||||
|
||||
/// 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<String>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub tool_calls: Option<Vec<serde_json::Value>>,
|
||||
/// DeepSeek thinking 模式推理内容
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub reasoning_content: Option<String>,
|
||||
}
|
||||
|
||||
/// OpenAI 同步响应
|
||||
#[derive(Debug, Deserialize)]
|
||||
pub(crate) struct OpenAiResponse {
|
||||
pub choices: Vec<OpenAiChoice>,
|
||||
pub model: String,
|
||||
pub usage: Option<OpenAiUsage>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
pub(crate) struct OpenAiChoice {
|
||||
pub message: OpenAiMessageResp,
|
||||
// SW-260618-24: OpenAI 响应反序列化字段,保留以备调试/未来消费(如日志记录调用终止原因),标注意图消除 dead_code warning
|
||||
#[allow(dead_code)]
|
||||
pub finish_reason: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
pub(crate) struct OpenAiMessageResp {
|
||||
pub content: Option<String>,
|
||||
pub tool_calls: Option<Vec<OpenAiToolCallResp>>,
|
||||
/// DeepSeek thinking 模式响应中的推理内容
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub reasoning_content: Option<String>,
|
||||
}
|
||||
|
||||
#[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<OpenAiStreamChoice>,
|
||||
/// 末 chunk（choices 为空）携带的累计 usage
|
||||
#[serde(default)]
|
||||
pub usage: Option<OpenAiUsage>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
pub(crate) struct OpenAiStreamChoice {
|
||||
pub delta: OpenAiStreamDelta,
|
||||
pub finish_reason: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
pub(crate) struct OpenAiStreamDelta {
|
||||
pub content: Option<String>,
|
||||
pub tool_calls: Option<Vec<OpenAiStreamToolCall>>,
|
||||
/// DeepSeek thinking 模式流式推理内容
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub reasoning_content: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
pub(crate) struct OpenAiStreamToolCall {
|
||||
pub index: u32,
|
||||
pub id: Option<String>,
|
||||
pub function: Option<OpenAiStreamFunction>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
pub(crate) struct OpenAiStreamFunction {
|
||||
pub name: Option<String>,
|
||||
pub arguments: Option<String>,
|
||||
}
|
||||
|
||||
// ============================================================
|
||||
// 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<TokenUsage>) -> 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::<OpenAiStreamChunk>(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,
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user