xt/agcore
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(llm): 完成 Phase 5 热身准备（Ollama / non_exhaustive / ProviderConfig）

Browse Source 
Phase 5 三个 Step 全部落地：

Step 5.2 — Ollama Provider
- 新增 OllamaProvider newtype 包装（默认 localhost:11434/v1，零 API key）
- ProviderType 新增 Ollama 变体与 FromStr 解析

Step 5.3 — #[non_exhaustive] 前置标记
- ProviderType / StopReason / FinishReason / EvictionPolicy 加 #[non_exhaustive]
- 编译期兼容护栏，避免下游 silent break

Step 5.1 — ProviderConfig 扩展
- 加 timeout_secs / max_retries 字段、Default、from_env(prefix)
- create_provider 各分支通过 pub(crate) from_parts 一次性构造并注入 timeout
  （同时避开 Anthropic 的 default_headers 与双重 client 构造）
- map_reqwest_error 改为方法读取 self.timeout_secs（移除硬编码 120s）
- AnthropicProvider::with_timeout 同值短路，with_client 标 #[deprecated]
- DeepSeek / Qwen 加公开 with_client，new_with_client 走代理
- 7 个新测试：5 个 from_env 单元测试 + 3 个 timeout 传导 wiremock
  （OpenAI Chat / DeepSeek / Anthropic）
- Cargo.toml 加 temp-env dev-dep
This commit is contained in:
徐涛
2026-07-05 08:12:40 +08:00
parent 9e476e79bb
commit 98dfe6c1ed
10 changed files with 807 additions and 199 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/llm/provider/anthropic.rs
+132 -55
View File
@@ -12,10 +12,10 @@ use async_trait::async_trait;
use bytes::Bytes;
use futures_core::Stream;
use futures_util::StreamExt;
use reqwest::header::{HeaderMap, HeaderValue};
use reqwest::Client;
use reqwest::header::{HeaderMap, HeaderValue};
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
use serde_json::{Value, json};
use tracing::{debug, error, info, warn};
use super::{LlmProvider, ProviderCapabilities, ProviderFeatures};
@@ -39,16 +39,20 @@ pub struct AnthropicProvider {
#[allow(dead_code)]
api_key: String,
model: String,
/// HTTP 请求超时秒数。由 `ProviderConfig::timeout_secs` 传入，
/// 在 `LlmError::Timeout { duration }` 中回显。`reqwest::Client` 不暴露 timeout getter，
/// 因此单独存储以便错误消息与配置保持一致。
timeout_secs: u64,
}
impl AnthropicProvider {
pub fn new(base_url: String, api_key: String, model: String) -> Self {
let key_header = HeaderValue::from_str(&api_key)
.expect("Anthropic API key 包含无效的 HTTP 头部字符");
pub fn new(base_url: String, api_key: String, model: String, timeout_secs: u64) -> Self {
let key_header =
HeaderValue::from_str(&api_key).expect("Anthropic API key 包含无效的 HTTP 头部字符");
let version_header = HeaderValue::from_static("2023-06-01");
let http_client = Client::builder()
.timeout(Duration::from_secs(120))
.timeout(Duration::from_secs(timeout_secs))
.default_headers({
let mut headers = HeaderMap::new();
headers.insert("x-api-key", key_header);
@@ -67,14 +71,80 @@ impl AnthropicProvider {
},
api_key,
model,
timeout_secs,
}
}
/// ⚠️ 替换 HTTP Client，**丢弃** `new()` 中设置的默认 headers（`x-api-key` / `anthropic-version`）。
///
/// 调用此方法后，所有 Anthropic API 请求将以**无认证头**发送出去，预期会 401/403 失败。
/// 推荐改用 [`Self::with_timeout`]，它会重建 client 并保留默认 headers。
///
/// 此方法仍保留以兼容调用方自定义 client 但不需要默认 headers 的极端场景。
#[deprecated(
since = "0.2.0",
note = "此方法会丢弃默认 headers（x-api-key / anthropic-version），改为使用 `with_timeout` 或带 headers 的 `Client::builder()`"
)]
pub fn with_client(mut self, client: Client) -> Self {
self.http_client = client;
self
}
/// 替换 HTTP Client 的超时配置（重建底层 client，保留默认 headers）。
///
/// ⚠️ 副作用：此方法**完全重建** `http_client`，调用后通过 `with_client` 注入的 Client
/// 将被替换。headers 构造逻辑与 `new()` 中的保持一致（`x-api-key` / `anthropic-version`）。
///
/// ponytail: 同值调用短路。当 `secs == self.timeout_secs` 时跳过 client 重建，
/// 避免 `create_provider` 路径 `new(timeout).with_timeout(timeout)` 的双重构造。
pub fn with_timeout(mut self, secs: u64) -> Result<Self, LlmError> {
if secs == self.timeout_secs {
return Ok(self);
}
// ponytail: 重建 http_client 时保留已有默认 headers（x-api-key / anthropic-version）。
// 如后续 AnthropicProvider 的 headers 变为动态，此方法需同步更新。
let key_header = HeaderValue::from_str(&self.api_key)
.map_err(|_| LlmError::Other("Anthropic API key 包含无效的 HTTP 头部字符".into()))?;
let version_header = HeaderValue::from_static("2023-06-01");
self.http_client = Client::builder()
.timeout(Duration::from_secs(secs))
.default_headers({
let mut headers = HeaderMap::new();
headers.insert("x-api-key", key_header);
headers.insert("anthropic-version", version_header);
headers
})
.build()
.map_err(|e| LlmError::Other(format!("创建 Anthropic HTTP 客户端失败: {e}")))?;
self.timeout_secs = secs;
Ok(self)
}
/// 一次性构造 —— `create_provider` 路径专用，避免 `new(...)` + `with_timeout(...)` 的双重 client 构造。
///
/// 调用方负责预先构造好符合 Anthropic 协议要求的 `http_client`（带正确的 `x-api-key` /
/// `anthropic-version` 默认 headers + 指定 timeout）。
pub(crate) fn from_parts(
base_url: String,
api_key: String,
model: String,
http_client: Client,
timeout_secs: u64,
) -> Self {
Self {
http_client,
base_url: if base_url.is_empty() {
"https://api.anthropic.com".to_string()
} else {
base_url
},
api_key,
model,
timeout_secs,
}
}
fn resolve_max_tokens(&self, request: &MessageRequest) -> u32 {
request.max_tokens.unwrap_or(DEFAULT_MAX_TOKENS)
}
@@ -104,12 +174,14 @@ impl AnthropicProvider {
Message::User { content } => {
api_messages.push(AnthropicMessage::user(content));
}
Message::UserImage { data, mime_type, detail } => {
Message::UserImage {
data,
mime_type,
detail,
} => {
// Anthropic image format: {type: "image", source: {type: "base64", media_type, data}}
let source = if data.starts_with("http://") || data.starts_with("https://") {
AnthropicImageSource::Url {
url: data.clone(),
}
AnthropicImageSource::Url { url: data.clone() }
} else {
AnthropicImageSource::Base64 {
media_type: mime_type.clone(),
@@ -194,7 +266,7 @@ impl AnthropicProvider {
.json(&body)
.send()
.await
.map_err(Self::map_reqwest_error)?;
.map_err(|e| self.map_reqwest_error(e))?;
let status = response.status();
if !status.is_success() {
@@ -215,8 +287,7 @@ impl AnthropicProvider {
async fn chat_stream_inner(
&self,
request: MessageRequest,
) -> Result<Pin<Box<dyn Stream<Item = Result<StreamEvent, LlmError>> + Send>>, LlmError>
{
) -> Result<Pin<Box<dyn Stream<Item = Result<StreamEvent, LlmError>> + Send>>, LlmError> {
let mut body = self.build_request_body(request)?;
body.stream = Some(true);
@@ -230,16 +301,16 @@ impl AnthropicProvider {
.json(&body)
.send()
.await
.map_err(Self::map_reqwest_error)?;
.map_err(|e| self.map_reqwest_error(e))?;
let status = response.status();
if !status.is_success() {
return Err(Self::handle_error_response(response).await);
}
let byte_stream = response.bytes_stream().map(|r| {
r.map_err(|e| LlmError::Other(format!("流式读取失败: {e}")))
});
let byte_stream = response
.bytes_stream()
.map(|r| r.map_err(|e| LlmError::Other(format!("流式读取失败: {e}"))));
let byte_stream: Pin<Box<dyn Stream<Item = Result<Bytes, LlmError>> + Send>> =
Box::pin(byte_stream);
@@ -247,10 +318,10 @@ impl AnthropicProvider {
Ok(Box::pin(AnthropicSseStream::new(byte_stream)))
}
fn map_reqwest_error(e: reqwest::Error) -> LlmError {
fn map_reqwest_error(&self, e: reqwest::Error) -> LlmError {
if e.is_timeout() {
LlmError::Timeout {
duration: Duration::from_secs(120),
duration: Duration::from_secs(self.timeout_secs),
}
} else if e.is_connect() {
LlmError::Other(format!("连接失败: {e}"))
@@ -291,13 +362,12 @@ impl AnthropicProvider {
blocks.push(ContentBlock::Text { text });
}
AnthropicContentBlockResp::ToolUse { id, name, input } => {
blocks.push(ContentBlock::ToolUse {
id,
name,
input,
});
blocks.push(ContentBlock::ToolUse { id, name, input });
}
AnthropicContentBlockResp::Thinking { thinking, signature } => {
AnthropicContentBlockResp::Thinking {
thinking,
signature,
} => {
blocks.push(ContentBlock::Thinking {
text: thinking,
signature,
@@ -336,8 +406,7 @@ impl LlmProvider for AnthropicProvider {
async fn chat_stream(
&self,
request: MessageRequest,
) -> Result<Pin<Box<dyn Stream<Item = Result<StreamEvent, LlmError>> + Send>>, LlmError>
{
) -> Result<Pin<Box<dyn Stream<Item = Result<StreamEvent, LlmError>> + Send>>, LlmError> {
self.chat_stream_inner(request).await
}
@@ -418,7 +487,9 @@ impl AnthropicMessage {
#[derive(Debug, Serialize)]
#[serde(tag = "type", rename_all = "snake_case")]
enum AnthropicContentPart {
Text { text: String },
Text {
text: String,
},
Image {
source: AnthropicImageSource,
},
@@ -437,13 +508,8 @@ enum AnthropicContentPart {
#[derive(Debug, Serialize)]
#[serde(tag = "type", rename_all = "snake_case")]
enum AnthropicImageSource {
Base64 {
media_type: String,
data: String,
},
Url {
url: String,
},
Base64 { media_type: String, data: String },
Url { url: String },
}
fn content_to_parts(blocks: &[ContentBlock]) -> Vec<AnthropicContentPart> {
@@ -523,9 +589,18 @@ struct AnthropicUsage {
#[derive(Debug, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
enum AnthropicContentBlockResp {
Text { text: String },
ToolUse { id: String, name: String, input: Value },
Thinking { thinking: String, signature: Option<String> },
Text {
text: String,
},
ToolUse {
id: String,
name: String,
input: Value,
},
Thinking {
thinking: String,
signature: Option<String>,
},
}
// =============================================================================
@@ -662,18 +737,13 @@ impl AnthropicSseStream {
// 先把所有字段提前，避免 match 中 part-move
let block_type = match &content_block {
AnthropicContentBlockStart::Text { .. } => ContentBlockType::Text,
AnthropicContentBlockStart::ToolUse { id, name } => {
ContentBlockType::ToolUse {
id: id.clone(),
name: name.clone(),
}
}
AnthropicContentBlockStart::ToolUse { id, name } => ContentBlockType::ToolUse {
id: id.clone(),
name: name.clone(),
},
AnthropicContentBlockStart::Thinking { .. } => ContentBlockType::Thinking,
};
events.push(StreamEvent::ContentBlockStart {
index,
block_type,
});
events.push(StreamEvent::ContentBlockStart { index, block_type });
let builder = match content_block {
AnthropicContentBlockStart::Text { text } => {
crate::llm::types::response_v2::ContentBlockBuilder::Text(text)
@@ -742,7 +812,9 @@ impl AnthropicSseStream {
completion_tokens_details: None,
prompt_tokens_details: None,
};
events.push(StreamEvent::CostUpdate { usage: partial_usage });
events.push(StreamEvent::CostUpdate {
usage: partial_usage,
});
}
}
AnthropicSseEvent::MessageStop => {
@@ -753,7 +825,9 @@ impl AnthropicSseStream {
self.saw_terminal = true;
match self.partial.clone().finalize() {
Ok(full) => {
events.push(StreamEvent::MessageComplete { full_response: full });
events.push(StreamEvent::MessageComplete {
full_response: full,
});
}
Err(e) => {
events.push(StreamEvent::Error {
@@ -780,10 +854,7 @@ fn _unused_marker() {}
impl Stream for AnthropicSseStream {
type Item = Result<StreamEvent, LlmError>;
fn poll_next(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Option<Self::Item>> {
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
loop {
if let Some(data) = self.next_event_line() {
let mut events = self.handle_event_json(&data);
@@ -835,7 +906,12 @@ mod tests {
fn make_provider(base_url: String) -> AnthropicProvider {
// 跳过默认 header 注入：测试用自定义 base_url 直接 mock
AnthropicProvider::new(base_url, "sk-ant-test".into(), "claude-sonnet-4-20250514".into())
AnthropicProvider::new(
base_url,
"sk-ant-test".into(),
"claude-sonnet-4-20250514".into(),
30,
)
}
#[tokio::test]
@@ -979,6 +1055,7 @@ event: message_stop\ndata: {\"type\":\"message_stop\"}\n\n";
"http://x".into(),
"k".into(),
"claude-sonnet-4-20250514".into(),
30,
)
.capabilities();
assert_eq!(caps.provider_name, "anthropic");
src/llm/provider/ollama.rs
+72
View File
@@ -0,0 +1,72 @@
//! Ollama Provider —— OpenAI-compatible 协议的 newtype 包装，零 API key。
//!
//! 默认 base_url = `http://localhost:11434/v1`，空 api_key 也可工作。
//! 实现方式同 `DeepSeekProvider` / `QwenProvider`，共享 `GenericOpenaiProvider`
//! 的 HTTP/SSE/转换逻辑，仅配置不同。
use std::pin::Pin;
use async_trait::async_trait;
use futures_core::Stream;
use reqwest::Client;
use super::openai::GenericOpenaiProvider;
use super::{LlmProvider, ProviderCapabilities};
use crate::llm::error::LlmError;
use crate::llm::types::request_v2::MessageRequest;
use crate::llm::types::response_v2::{MessageResponse, StreamEvent};
/// Ollama 本地 Provider —— OpenAI-compatible 协议的 newtype 包装。
///
/// Ollama 在 `localhost:11434` 暴露与 OpenAI 兼容的 `/v1/chat/completions`
/// 接口，因此完全复用 `GenericOpenaiProvider` 的实现。允许空 `api_key`。
pub struct OllamaProvider(pub GenericOpenaiProvider);
impl OllamaProvider {
/// 构造 Ollama Provider。
///
/// - `base_url` 为空时使用默认 `http://localhost:11434/v1`
/// - `api_key` 可为空字符串（Ollama 不校验）
pub fn new(base_url: String, api_key: String, model: String, timeout_secs: u64) -> Self {
let url = if base_url.is_empty() {
"http://localhost:11434/v1".to_string()
} else {
base_url
};
Self(GenericOpenaiProvider::new_with_name(
url,
api_key,
model,
"ollama",
timeout_secs,
))
}
/// 替换默认 HTTP Client（用于 timeout 注入等场景）。
///
/// 与 `OpenaiChatProvider::with_client`、`DeepSeekProvider::with_client`、
/// `QwenProvider::with_client` 签名一致。
pub fn with_client(self, client: Client) -> Self {
Self(self.0.with_client(client))
}
}
#[async_trait]
impl LlmProvider for OllamaProvider {
async fn chat(&self, request: MessageRequest) -> Result<MessageResponse, LlmError> {
self.0.chat(request).await
}
async fn chat_stream(
&self,
request: MessageRequest,
) -> Result<Pin<Box<dyn Stream<Item = Result<StreamEvent, LlmError>> + Send>>, LlmError> {
self.0.chat_stream(request).await
}
fn capabilities(&self) -> ProviderCapabilities {
let mut caps = self.0.capabilities();
caps.provider_name = "ollama";
caps
}
}
src/llm/provider/openai.rs
+92 -55
View File
@@ -52,31 +52,63 @@ pub struct GenericOpenaiProvider {
model: String,
provider_name: &'static str,
extra_headers: Vec<(String, String)>,
/// HTTP 请求超时秒数。由 `ProviderConfig::timeout_secs` 传入，
/// 在 `LlmError::Timeout { duration }` 中回显。`reqwest::Client` 不暴露 timeout getter，
/// 因此单独存储以便错误消息与配置保持一致。
timeout_secs: u64,
}
impl GenericOpenaiProvider {
/// 基础构造器。
pub fn new_with_name(
/// 一次性构造 —— `create_provider` 路径专用，避免 `new_with_name` + `with_client` 的双重 client 构造。
///
/// 调用方负责预先构造好带正确 timeout 的 `http_client`。`extra_headers` 与 `timeout_secs`
/// 一并设置字段，避免后续修改。
pub(crate) fn from_parts(
base_url: String,
api_key: String,
model: String,
provider_name: &'static str,
http_client: Client,
extra_headers: Vec<(String, String)>,
timeout_secs: u64,
) -> Self {
let http_client = Client::builder()
.timeout(Duration::from_secs(120))
.build()
.expect("创建 HTTP 客户端失败");
Self {
http_client,
base_url,
api_key,
model,
provider_name,
extra_headers: Vec::new(),
extra_headers,
timeout_secs,
}
}
/// 基础构造器。
///
/// `timeout_secs` 应用于 `reqwest::Client` 的请求超时配置。
/// 应由 `ProviderConfig::timeout_secs` 传入（调用方如不知道，可传 30）。
pub fn new_with_name(
base_url: String,
api_key: String,
model: String,
provider_name: &'static str,
timeout_secs: u64,
) -> Self {
let http_client = Client::builder()
.timeout(Duration::from_secs(timeout_secs))
.build()
.expect("创建 HTTP 客户端失败");
Self::from_parts(
base_url,
api_key,
model,
provider_name,
http_client,
Vec::new(),
timeout_secs,
)
}
/// 带额外请求头的构造器（如 Qwen 需要 SSE 启用头）。
pub fn new_with_name_and_headers(
base_url: String,
@@ -84,10 +116,21 @@ impl GenericOpenaiProvider {
model: String,
provider_name: &'static str,
extra_headers: Vec<(String, String)>,
timeout_secs: u64,
) -> Self {
let mut base = Self::new_with_name(base_url, api_key, model, provider_name);
base.extra_headers = extra_headers;
base
let http_client = Client::builder()
.timeout(Duration::from_secs(timeout_secs))
.build()
.expect("创建 HTTP 客户端失败");
Self::from_parts(
base_url,
api_key,
model,
provider_name,
http_client,
extra_headers,
timeout_secs,
)
}
pub fn with_client(mut self, client: Client) -> Self {
@@ -119,10 +162,10 @@ impl GenericOpenaiProvider {
Ok(builder.json(body))
}
fn map_reqwest_error(e: reqwest::Error) -> LlmError {
fn map_reqwest_error(&self, e: reqwest::Error) -> LlmError {
if e.is_timeout() {
LlmError::Timeout {
duration: Duration::from_secs(120),
duration: Duration::from_secs(self.timeout_secs),
}
} else if e.is_connect() {
LlmError::Other(format!("连接失败: {}", e))
@@ -135,9 +178,7 @@ impl GenericOpenaiProvider {
///
/// ponytail: Qwen 等部分 OpenAI-compatible 提供方可能返回非标准 error body
/// （无法解析为 JSON），此处直接用 status code + 原始 body 兜底。
async fn handle_error_response(
response: reqwest::Response,
) -> LlmError {
async fn handle_error_response(response: reqwest::Response) -> LlmError {
let status = response.status().as_u16();
let body = response.text().await.unwrap_or_default();
@@ -148,10 +189,7 @@ impl GenericOpenaiProvider {
// 与 OpenAI 完全一致；DeepSeek/Qwen 通常遵循。
LlmError::RateLimit { retry_after: None }
}
_ if status >= 500 => LlmError::Request {
status,
body,
},
_ if status >= 500 => LlmError::Request { status, body },
_ if status == 400 && body.contains("context_length_exceeded") => {
LlmError::ContextLength {
actual: 0,
@@ -200,7 +238,9 @@ impl GenericOpenaiProvider {
stop_sequences[0].clone(),
))
} else {
Some(crate::llm::types::shared::StopSequence::Multiple(stop_sequences))
Some(crate::llm::types::shared::StopSequence::Multiple(
stop_sequences,
))
};
let frequency_penalty = request.get_extra_opt("frequency_penalty");
@@ -245,9 +285,7 @@ impl GenericOpenaiProvider {
let stop_reason = match choice.finish_reason {
Some(FinishReason::Stop) => StopReason::Stop,
Some(FinishReason::Length) => StopReason::Length,
Some(FinishReason::ToolCalls) | Some(FinishReason::FunctionCall) => {
StopReason::ToolUse
}
Some(FinishReason::ToolCalls) | Some(FinishReason::FunctionCall) => StopReason::ToolUse,
Some(FinishReason::ContentFilter) => StopReason::ContentFilter,
Some(FinishReason::Other) | None => StopReason::Stop,
};
@@ -263,7 +301,10 @@ impl GenericOpenaiProvider {
}
/// 非流式 `chat()` 入口。
pub async fn chat_blocking(&self, request: MessageRequest) -> Result<MessageResponse, LlmError> {
pub async fn chat_blocking(
&self,
request: MessageRequest,
) -> Result<MessageResponse, LlmError> {
let req = self.convert_request(request)?;
let url = format!("{}/chat/completions", self.base_url.trim_end_matches('/'));
@@ -280,7 +321,7 @@ impl GenericOpenaiProvider {
.await
.map_err(|e| {
error!(error = %e, "请求失败");
Self::map_reqwest_error(e)
self.map_reqwest_error(e)
})?;
let status = response.status();
@@ -303,8 +344,7 @@ impl GenericOpenaiProvider {
pub async fn chat_stream_inner(
&self,
request: MessageRequest,
) -> Result<Pin<Box<dyn Stream<Item = Result<StreamEvent, LlmError>> + Send>>, LlmError>
{
) -> Result<Pin<Box<dyn Stream<Item = Result<StreamEvent, LlmError>> + Send>>, LlmError> {
let mut req = self.convert_request(request)?;
req.stream = Some(true);
req.stream_options = Some(StreamOptions {
@@ -322,7 +362,7 @@ impl GenericOpenaiProvider {
.await
.map_err(|e| {
error!(error = %e, "流式请求失败");
Self::map_reqwest_error(e)
self.map_reqwest_error(e)
})?;
let status = response.status();
@@ -331,9 +371,9 @@ impl GenericOpenaiProvider {
}
let byte_stream: std::pin::Pin<Box<dyn Stream<Item = Result<Bytes, LlmError>> + Send>> = {
let s = response.bytes_stream().map(|r| {
r.map_err(|e| LlmError::Other(format!("流式读取失败: {}", e)))
});
let s = response
.bytes_stream()
.map(|r| r.map_err(|e| LlmError::Other(format!("流式读取失败: {}", e))));
Box::pin(s)
};
@@ -368,8 +408,7 @@ impl LlmProvider for GenericOpenaiProvider {
async fn chat_stream(
&self,
request: MessageRequest,
) -> Result<Pin<Box<dyn Stream<Item = Result<StreamEvent, LlmError>> + Send>>, LlmError>
{
) -> Result<Pin<Box<dyn Stream<Item = Result<StreamEvent, LlmError>> + Send>>, LlmError> {
self.chat_stream_inner(request).await
}
@@ -389,12 +428,13 @@ impl LlmProvider for GenericOpenaiProvider {
pub struct OpenaiChatProvider(pub GenericOpenaiProvider);
impl OpenaiChatProvider {
pub fn new(base_url: String, api_key: String, model: String) -> Self {
pub fn new(base_url: String, api_key: String, model: String, timeout_secs: u64) -> Self {
Self(GenericOpenaiProvider::new_with_name(
base_url,
api_key,
model,
"openai",
timeout_secs,
))
}
@@ -412,8 +452,7 @@ impl LlmProvider for OpenaiChatProvider {
async fn chat_stream(
&self,
request: MessageRequest,
) -> Result<Pin<Box<dyn Stream<Item = Result<StreamEvent, LlmError>> + Send>>, LlmError>
{
) -> Result<Pin<Box<dyn Stream<Item = Result<StreamEvent, LlmError>> + Send>>, LlmError> {
self.0.chat_stream(request).await
}
@@ -436,7 +475,10 @@ enum BlockState {
/// 正在累积 text block。
InText { block_index: u32 },
/// 正在累积 tool_call block。
InTool { block_index: u32, tool_call_index: u32 },
InTool {
block_index: u32,
tool_call_index: u32,
},
/// 正在累积 refusal block。
InRefusal { block_index: u32 },
}
@@ -456,9 +498,7 @@ pub struct ChunkToEventStream {
}
impl ChunkToEventStream {
fn new(
chunks: Pin<Box<dyn Stream<Item = Result<Bytes, LlmError>> + Send>>,
) -> Self {
fn new(chunks: Pin<Box<dyn Stream<Item = Result<Bytes, LlmError>> + Send>>) -> Self {
Self {
chunks,
buffer: String::new(),
@@ -500,12 +540,7 @@ impl ChunkToEventStream {
let mut events = Vec::new();
// 元信息：MessageStart（仅在第一次见到 role=assistant 时）。
if self.partial.id.is_none()
&& chunk
.choices
.iter()
.any(|c| c.delta.role.is_some())
{
if self.partial.id.is_none() && chunk.choices.iter().any(|c| c.delta.role.is_some()) {
events.push(StreamEvent::MessageStart {
id: chunk.id.clone(),
model: chunk.model.clone(),
@@ -650,9 +685,7 @@ impl ChunkToEventStream {
self.partial.stop_reason = Some(match fr {
FinishReason::Stop => StopReason::Stop,
FinishReason::Length => StopReason::Length,
FinishReason::ToolCalls | FinishReason::FunctionCall => {
StopReason::ToolUse
}
FinishReason::ToolCalls | FinishReason::FunctionCall => StopReason::ToolUse,
FinishReason::ContentFilter => StopReason::ContentFilter,
FinishReason::Other => StopReason::Other,
});
@@ -692,7 +725,9 @@ impl ChunkToEventStream {
match self.partial.clone().finalize() {
Ok(full) => {
events.push(StreamEvent::MessageComplete { full_response: full });
events.push(StreamEvent::MessageComplete {
full_response: full,
});
}
Err(e) => {
events.push(StreamEvent::Error {
@@ -707,10 +742,7 @@ impl ChunkToEventStream {
impl Stream for ChunkToEventStream {
type Item = Result<StreamEvent, LlmError>;
fn poll_next(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Option<Self::Item>> {
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
loop {
// 先尝试从 buffer 取一行处理
if let Some(line) = self.next_line() {
@@ -814,6 +846,7 @@ mod tests {
"sk-test".into(),
"gpt-4o".into(),
"openai",
30,
);
let response = provider
.chat_blocking(MessageRequest {
@@ -842,6 +875,7 @@ mod tests {
"sk-test".into(),
"gpt-4o".into(),
"openai",
30,
);
let err = provider
.chat_blocking(MessageRequest {
@@ -868,6 +902,7 @@ mod tests {
"sk-test".into(),
"gpt-4o".into(),
"openai",
30,
);
let err = provider
.chat_blocking(MessageRequest {
@@ -906,6 +941,7 @@ data: [DONE]\n\n";
"sk-test".into(),
"gpt-4o".into(),
"openai",
30,
);
let mut stream = provider
.chat_stream_inner(MessageRequest {
@@ -974,6 +1010,7 @@ data: [DONE]\n\n";
"k".into(),
"gpt-4o".into(),
"openai",
30,
);
let ir = provider.convert_response(resp).unwrap();
assert_eq!(ir.stop_reason, StopReason::ToolUse);
src/llm/provider/openai_compat.rs
+32 -39
View File
@@ -15,12 +15,12 @@ use std::pin::Pin;
use async_trait::async_trait;
use futures_core::Stream;
use super::openai::GenericOpenaiProvider;
use super::ProviderCapabilities;
use super::openai::GenericOpenaiProvider;
use crate::llm::error::LlmError;
use crate::llm::provider::LlmProvider;
use crate::llm::types::request_v2::MessageRequest;
use crate::llm::types::response_v2::{MessageResponse, StreamEvent};
use crate::llm::provider::LlmProvider;
// =============================================================================
// DeepSeek
@@ -29,7 +29,7 @@ use crate::llm::provider::LlmProvider;
pub struct DeepSeekProvider(pub GenericOpenaiProvider);
impl DeepSeekProvider {
pub fn new(base_url: String, api_key: String, model: String) -> Self {
pub fn new(base_url: String, api_key: String, model: String, timeout_secs: u64) -> Self {
let url = if base_url.is_empty() {
"https://api.deepseek.com".to_string()
} else {
@@ -40,26 +40,27 @@ impl DeepSeekProvider {
api_key,
model,
"deepseek",
timeout_secs,
))
}
}
impl DeepSeekProvider {
/// 替换默认 HTTP Client（用于 timeout 注入等场景）。
pub fn with_client(self, client: reqwest::Client) -> Self {
Self(self.0.with_client(client))
}
/// 测试中（带 mock_client）使用的构造器。
///
/// ponytail: 此处 `30` 是 `timeout_secs` 字段的占位值，仅用于 `map_reqwest_error`
/// 错误消息中的回显。实际请求超时由传入的 `client` 控制（通常测试用的 mock client
/// 无超时），不影响行为。
pub fn new_with_client(
base_url: String,
api_key: String,
model: String,
client: reqwest::Client,
) -> Self {
let url = if base_url.is_empty() {
"https://api.deepseek.com".to_string()
} else {
base_url
};
let mut inner = GenericOpenaiProvider::new_with_name(url, api_key, model, "deepseek");
inner.http_client = client;
Self(inner)
Self::new(base_url, api_key, model, 30).with_client(client)
}
}
@@ -72,8 +73,7 @@ impl LlmProvider for DeepSeekProvider {
async fn chat_stream(
&self,
request: MessageRequest,
) -> Result<Pin<Box<dyn Stream<Item = Result<StreamEvent, LlmError>> + Send>>, LlmError>
{
) -> Result<Pin<Box<dyn Stream<Item = Result<StreamEvent, LlmError>> + Send>>, LlmError> {
self.0.chat_stream(request).await
}
@@ -91,7 +91,7 @@ impl LlmProvider for DeepSeekProvider {
pub struct QwenProvider(pub GenericOpenaiProvider);
impl QwenProvider {
pub fn new(base_url: String, api_key: String, model: String) -> Self {
pub fn new(base_url: String, api_key: String, model: String, timeout_secs: u64) -> Self {
let url = if base_url.is_empty() {
"https://dashscope.aliyuncs.com/compatible-mode/v1".to_string()
} else {
@@ -104,31 +104,28 @@ impl QwenProvider {
model,
"qwen",
vec![("X-DashScope-SSE".to_string(), "enable".to_string())],
timeout_secs,
);
Self(inner)
}
/// 替换默认 HTTP Client（用于 timeout 注入等场景）。
pub fn with_client(self, client: reqwest::Client) -> Self {
Self(self.0.with_client(client))
}
/// 测试构造器。
///
/// ponytail: 此处 `30` 是 `timeout_secs` 字段的占位值，仅用于 `map_reqwest_error`
/// 错误消息中的回显。实际请求超时由传入的 `client` 控制（通常测试用的 mock client
/// 无超时），不影响行为。
pub fn new_with_client(
base_url: String,
api_key: String,
model: String,
client: reqwest::Client,
) -> Self {
let url = if base_url.is_empty() {
"https://dashscope.aliyuncs.com/compatible-mode/v1".to_string()
} else {
base_url
};
let mut inner = GenericOpenaiProvider::new_with_name_and_headers(
url,
api_key,
model,
"qwen",
vec![("X-DashScope-SSE".to_string(), "enable".to_string())],
);
inner.http_client = client;
Self(inner)
Self::new(base_url, api_key, model, 30).with_client(client)
}
}
@@ -141,8 +138,7 @@ impl LlmProvider for QwenProvider {
async fn chat_stream(
&self,
request: MessageRequest,
) -> Result<Pin<Box<dyn Stream<Item = Result<StreamEvent, LlmError>> + Send>>, LlmError>
{
) -> Result<Pin<Box<dyn Stream<Item = Result<StreamEvent, LlmError>> + Send>>, LlmError> {
self.0.chat_stream(request).await
}
@@ -156,8 +152,8 @@ impl LlmProvider for QwenProvider {
#[cfg(test)]
mod tests {
use super::*;
use crate::llm::types::request_v2::MessageRequest;
use crate::llm::types::message::Message as IrMessage;
use crate::llm::types::request_v2::MessageRequest;
use serde_json::json;
use wiremock::matchers::{method, path};
use wiremock::{Mock, MockServer, ResponseTemplate};
@@ -182,11 +178,8 @@ mod tests {
.mount(&server)
.await;
let provider = DeepSeekProvider::new(
server.uri(),
"sk-test".into(),
"deepseek-chat".into(),
);
let provider =
DeepSeekProvider::new(server.uri(), "sk-test".into(), "deepseek-chat".into(), 30);
let response = provider
.chat(MessageRequest {
model: "deepseek-chat".into(),
@@ -219,7 +212,7 @@ mod tests {
.mount(&server)
.await;
let provider = QwenProvider::new(server.uri(), "sk-test".into(), "qwen-plus".into());
let provider = QwenProvider::new(server.uri(), "sk-test".into(), "qwen-plus".into(), 30);
let response = provider
.chat(MessageRequest {
model: "qwen-plus".into(),