xt/agcore
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
Files
main
agcore/design/pdd/10b-phase1-provider-adaptation.md
T
徐涛 28ca43ccb2 chore(docs): 将设计文档从 docs 移至 design 目录 
将 note、pdd、prd、roadmap 四类文档分别归入 `design/` 下对应子目录中，并新增 `.gitkeep` 占位文件
2026-07-23 05:45:53 +08:00

42 KiB
Raw Permalink Blame History

Phase 1 实施计划：Provider 适配

所属方案：10-llm-provider-refinement.md

前置条件：Phase 0 已完成，LlmProvider trait 签名已切换为 chat(MessageRequest) → MessageResponse

产出依赖：Phase 2（LlmCycle 简化）依赖本阶段完成的 Provider 实现

Phase 0 就绪检测

前置条件：Phase 0 引入的 MessageRequest/MessageResponse/StreamEvent 等新类型必须已在代码库中落地。 进入 Phase 1 执行前，必须满足以下全部条件：

检测项 检查方式 预期结果
request_v2.rs / response_v2.rs 存在 ls src/llm/types/ 文件存在
pub type Message = OpenaiChatMessage 别名已移除 grep "pub type Message" src/llm/types/mod.rs 无匹配
pub type ChatRequest = OpenaiChatRequest 别名已移除 grep "pub type ChatRequest" src/llm/types/mod.rs 无匹配
StreamEvent 包含 ContentBlockStart 新变体 grep "ContentBlockStart" src/llm/stream.rs 有匹配（Phase 0 引入）
ContentBlock 是新类型的泛化 enum（非 OpenaiContentPart alias） grep "pub enum ContentBlock" src/llm/types/message.rs 有匹配（Phase 0 引入）
LlmProvider trait 使用新签名 grep "async fn chat" src/llm/provider.rs 签名为 fn chat(&self, request: MessageRequest) -> Result<MessageResponse, LlmError>
cargo build 编译通过 cargo build 2>&1 无错误

以上条件未满足时，禁止启动 Phase 1 开发。

Phase 0 未完成时的降级执行路径

如果 Phase 0 因各种原因延迟，以下工作可独立于 Phase 0 先行完成（不依赖新类型）：

可独立执行的内容 不依赖 Phase 0 的原因 过渡策略
ProviderType enum 扩展（+Anthropic/OpenaiResponse） enum 扩展不涉及 trait 签名变更 先加变体，create_provider 返回 Err(LlmError::Other(...))
AnthropicProvider 的 HTTP 调用层 使用 reqwest::Client 原始调用，不依赖 LlmProvider trait 独立函数（async fn anthropic_chat(...)），Phase 0 完成后包装为 trait impl
SSE 字节解析通用层（SseByteStream） 纯字节处理，无关类型系统 独立模块，Phase 0 完成后挂接到 StreamEvent 映射
wiremock dev-dependency 添加 仅 Cargo.toml 变更 先加依赖，测试代码在 Phase 0 后补齐

当 Phase 0 完成后，上述先行部分直接集成到 Phase 1 主流程，无需重写。

目标

重写 OpenaiProvider（使用新类型），新增 AnthropicProvider。DeepSeek/Qwen 作为 OpenAI-compatible 协议实现一并纳入。本 Phase 结束时，每个 Provider 都有 mock 覆盖的 chat + chat_stream 基本路径测试。

设计决策

ProviderType enum 变更

当前 enum（3 个变体）→ 目标 enum（5 个变体）：

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ProviderType {
    OpenaiChat,     // 原 OpenAI（标准 /v1/chat/completions）
    OpenaiResponse, // OpenAI Response API（新增，独特协议）
    Anthropic,      // Anthropic Messages API（新增）
    DeepSeek,       // DeepSeek（OpenAI-compatible）
    Qwen,           // 通义千问（OpenAI-compatible）
}

名称变更注意：当前 OpenAI 改名为 OpenaiChat，精确区分 Chat API 和 Response API。

OpenAI-compatible 复用策略

推荐方案：参数化 GenericOpenaiProvider

pub struct GenericOpenaiProvider {
    http_client: Client,
    base_url: String,
    api_key: String,
    model: String,
    provider_name: &'static str,
    /// 额外请求头，由 Provider newtype 构造器传入（如 Qwen 的 X-DashScope-SSE: enable）
    extra_headers: Vec<(String, String)>,
}

构造器定义（供 DeepSeek/Qwen newtype 调用）：

impl GenericOpenaiProvider {
    pub fn new_with_name(
        base_url: String, api_key: String, model: String,
        provider_name: &'static str,
    ) -> Self { /* ... */ }

    pub fn new_with_name_and_headers(
        base_url: String, api_key: String, model: String,
        provider_name: &'static str,
        extra_headers: Vec<(String, String)>,
    ) -> Self { /* ... */ }
}

DeepSeek 和 Qwen 共用同一实现，仅配置不同：

  • GenericOpenaiProvider::new_with_name("https://api.deepseek.com", api_key, model, "deepseek")
  • GenericOpenaiProvider::new_with_name("https://dashscope.aliyuncs.com/compatible-mode/v1", api_key, model, "qwen")

extra_headers 注入：在 chat() / chat_stream() 的 HTTP 请求构造中，遍历 self.extra_headers 注入到 reqwest::RequestBuilder：

let mut request_builder = self.http_client.post(&url)
    .header("Authorization", format!("Bearer {}", self.api_key));
for (key, value) in &self.extra_headers {
    request_builder = request_builder.header(key.as_str(), value.as_str());
}

备选方案：如果不想引入 GenericOpenaiProvider 的抽象层，OpenaiChatProvider 直接覆盖所有 OpenAI-compatible 场景，DeepSeek/Qwen 的 Provider 只是调用 OpenaiChatProvider 的不同实例。

实施时决定：先实现 OpenaiChatProvider，确认它足够的通用性后，将 DeepSeek/Qwen 简化为配置别名。

OpenaiResponseProvider 范围

OpenaiResponseProvider 在本 Phase 只覆盖核心对话能力（create response + 流式 + 工具调用），内置工具（web_search, file_search）、previous_response_id 续写、store 等特性通过 MessageRequest.extra 传递。如果资源有限，可延迟到 Phase 2 之后开发，不影响其他 Provider。

HTTP mock 策略

  • 新增 dev-dependency: wiremock
  • 每个 Provider 的测试模块中，用 MockServer 启动 mock 服务端
  • OpenaiChatProvider mock 端点：/chat/completions（SSE 流和 JSON 响应）
  • AnthropicProvider mock 端点：/v1/messages（SSE 事件序列）
  • Provider 构造时 base_url 指向 mock_server.uri()
  • 测试不依赖真实网络

涉及文件

操作 文件 说明
修改 src/llm/provider.rs 扩展 ProviderType enum、更新 create_provider、更新 FromStr
修改 src/llm/provider/openai.rs 重写：接受 MessageRequest，返回 MessageResponse/StreamEvent；新增 GenericOpenaiProvider
新增 src/llm/provider/openai_compat.rs DeepSeek/Qwen 的 newtype 包装 + delegate macro（替代独立 deepseek.rs/qwen.rs）
新增 src/llm/provider/anthropic.rs Anthropic Messages API 实现（含独立错误映射）
新增 src/llm/convert.rs 公共转换模块：from_openai() / to_openai() / content_to_blocks() / blocks_to_content()
修改 src/llm/provider/registry.rs 适配新的 ProviderType 和 trait 签名
修改 Cargo.toml 新增 wiremock dev-dependency

任务 1：扩展 ProviderType enum 和工厂函数

1a：扩展 enum

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ProviderType {
    OpenaiChat,
    OpenaiResponse,
    Anthropic,
    DeepSeek,
    Qwen,
}

1b：更新 FromStr

impl std::str::FromStr for ProviderType {
    type Err = String;
    fn from_str(s: &str) -> Result<Self, Self::Err> {
        match s.to_lowercase().as_str() {
            "openai" | "openaichat" => Ok(ProviderType::OpenaiChat),
            "openai-response" | "response" => Ok(ProviderType::OpenaiResponse),
            "anthropic" | "claude" => Ok(ProviderType::Anthropic),
            "deepseek" => Ok(ProviderType::DeepSeek),
            "qwen" | "dashscope" | "tongyi" => Ok(ProviderType::Qwen),
            _ => Err(format!("未知的 Provider 类型: {}", s)),
        }
    }
}

1c：更新 create_provider 工厂函数

pub fn create_provider(
    provider_type: ProviderType,
    config: ProviderConfig,
) -> Result<Box<dyn LlmProvider>, LlmError> {
    match provider_type {
        ProviderType::OpenaiChat => Ok(Box::new(openai::OpenaiChatProvider::new(
            config.base_url, config.api_key, config.model,
        ))),
        ProviderType::OpenaiResponse => {
            // 注意：此处不返回 unimplemented!()，避免运行时 panic。
            // 返回 Err 让调用方可以优雅地降级。
            Err(LlmError::Other("OpenaiResponse Provider 尚未实现，请使用 OpenaiChat".into()))
        }
        ProviderType::Anthropic => Ok(Box::new(anthropic::AnthropicProvider::new(
            config.base_url, config.api_key, config.model,
        ))),
        ProviderType::DeepSeek => Ok(Box::new(openai_compat::DeepSeekProvider::new(
            config.base_url, config.api_key, config.model,
        ))),
        ProviderType::Qwen => Ok(Box::new(openai_compat::QwenProvider::new(
            config.base_url, config.api_key, config.model,
        ))),
    }
}

任务 2：重写 OpenaiChatProvider

2a：结构体定义

pub struct OpenaiChatProvider {
    http_client: Client,
    base_url: String,
    api_key: String,
    model: String,
}

2b：非流式 chat()

async fn chat(&self, request: MessageRequest) -> Result<MessageResponse, LlmError> {
    let url = format!("{}/chat/completions", self.base_url.trim_end_matches('/'));
    
    // 1. MessageRequest → OpenaiChatRequest
    let openai_req = self.convert_request(request)?;
    
    // 2. HTTP POST
    let response = self.http_client
        .post(&url)
        .header("Authorization", format!("Bearer {}", self.api_key))
        .json(&openai_req)
        .send()
        .await
        .map_err(Self::map_reqwest_error)?;
    
    // 3. 错误处理（复用现有逻辑）
    let status = response.status();
    if !status.is_success() {
        return Err(self.handle_error(response).await);
    }
    
    // 4. 解析 OpenaiChatResponse
    let body_text = response.text().await.unwrap_or_default();
    let chat_response: OpenaiChatResponse = serde_json::from_str(&body_text)
        .map_err(|e| LlmError::Other(format!("响应解析失败: {}", e)))?;
    
    // 5. OpenaiChatResponse → MessageResponse
    self.convert_response(chat_response)
}

2c：流式 chat_stream()

async fn chat_stream(
    &self, request: MessageRequest,
) -> Result<Pin<Box<dyn Stream<Item = Result<StreamEvent, LlmError>> + Send>>, LlmError> {
    // 1. MessageRequest → OpenaiChatRequest + 设置 stream=true
    let mut openai_req = self.convert_request(request)?;
    openai_req.stream = Some(true);
    openai_req.stream_options = Some(StreamOptions {
        include_usage: Some(true),
        include_obfuscation: None,
    });
    
    // 2. HTTP POST 获取 SSE 流
    let url = format!("{}/chat/completions", self.base_url.trim_end_matches('/'));
    let response = self.http_client
        .post(&url)
        .header("Authorization", format!("Bearer {}", self.api_key))
        .json(&openai_req)
        .send()
        .await
        .map_err(Self::map_reqwest_error)?;
    
    let status = response.status();
    if !status.is_success() {
        return Err(self.handle_error(response).await);
    }
    
    // 3. 将 byte stream 转为 StreamEvent 流
    let byte_stream = response.bytes_stream().map(|r| {
        r.map_err(|e| LlmError::Other(format!("流式读取失败: {}", e)))
    });
    
    Ok(Box::pin(OpenaiChunkToEventStream::new(byte_stream)))
}

2d：转换逻辑 convert_request()

MessageRequest → OpenaiChatRequest：

fn convert_request(&self, request: MessageRequest) -> Result<OpenaiChatRequest, LlmError> {
    // messages: Vec<Message> → Vec<OpenaiChatMessage>
    let messages: Vec<OpenaiChatMessage> = request.messages
        .iter()
        .map(|m| self.message_to_chat_message(m))
        .collect();
    
    // 构造 OpenaiChatRequest
    Ok(OpenaiChatRequest {
        model: request.model,
        messages,
        max_tokens: request.max_tokens,
        temperature: request.temperature,
        top_p: request.top_p,
        tools: if request.tools.is_empty() { None } else {
            Some(request.tools.into_iter().map(|t| OpenaiTool::Function {
                function: t,
            }).collect())
        },
        // 假设：MessageRequest.tool_choice 为 ToolChoice（非 Option），
        // 与 OpenaiChatRequest.tool_choice: Option<ToolChoice> 自然匹配。
        // 如果 Phase 0 定义为 Option<ToolChoice>，去掉 Some(...) 包装。
        tool_choice: Some(request.tool_choice),
        stream: None, // 由调用方在 chat_stream 中设置
        stop: Some(StopSequence::Multiple(request.stop_sequences)),
        // 从 extra 读取 OpenAI 特有参数
        frequency_penalty: request.get_extra_opt("frequency_penalty"),
        presence_penalty: request.get_extra_opt("presence_penalty"),
        seed: request.get_extra_opt("seed"),
        response_format: request.get_extra_opt("response_format"),
        parallel_tool_calls: request.get_extra_opt("parallel_tool_calls"),
        // o-series 模型使用 max_completion_tokens 而非 max_tokens
        // 当 request.extra["max_completion_tokens"] 存在时覆盖 max_tokens
        max_completion_tokens: request.get_extra_opt("max_completion_tokens"),
        ..Default::default()
    })
}

**Extra 字段清单**（从 `MessageRequest.extra` 透传到 `OpenaiChatRequest`）：

| Extra Key | 目标字段 | 说明 |
|-----------|---------|------|
| `frequency_penalty` | `frequency_penalty` | 频率惩罚 |
| `presence_penalty` | `presence_penalty` | 存在惩罚 |
| `seed` | `seed` | 随机种子 |
| `response_format` | `response_format` | JSON mode / Structured Outputs（企业高频需求） |
| `parallel_tool_calls` | `parallel_tool_calls` | 是否允许并行工具调用（默认 true） |
| `max_completion_tokens` | `max_completion_tokens` | o-series 模型专用，覆盖 `max_tokens` |
| `openai_organization` | HTTP header `OpenAI-Organization` | 企业组织标识 |
| `openai_project` | HTTP header `OpenAI-Project` | 企业项目标识 |

HTTP header 参数在 `chat()`/`chat_stream()` 中从 extra 提取后注入请求头。

### 公共转换模块

`Message` ↔ `OpenaiChatMessage` 的转换函数在 Phase 0 的 `LlmCycle` 和 Phase 1 的 `OpenaiChatProvider` 中都会用到。
为避免逻辑漂移，应抽取为公共模块：

```rust
// src/llm/convert.rs — 跨 Provider 类型转换模块

/// OpenaiChatMessage → Message
pub fn from_openai(msg: &OpenaiChatMessage) -> Message { ... }

/// Message → OpenaiChatMessage
pub fn to_openai(msg: &Message) -> OpenaiChatMessage { ... }

/// ContentField → Vec<ContentBlock>
pub fn content_to_blocks(field: &ContentField) -> Vec<ContentBlock> { ... }

/// Vec<ContentBlock> → ContentField
pub fn blocks_to_content(blocks: &[ContentBlock]) -> ContentField { ... }

Phase 0 的 cycle.rs 和 Phase 1 的所有 Provider 统一引用此模块。

2e：转换逻辑 convert_response()

OpenaiChatResponse → MessageResponse：

fn convert_response(&self, response: OpenaiChatResponse) -> Result<MessageResponse, LlmError> {
    let choice = response.choices.into_iter().next()
        .ok_or_else(|| LlmError::Other("响应中没有 choices".to_string()))?;
    
    let message = self.chat_message_to_message(&choice.message);
    let stop_reason = match choice.finish_reason {
        Some(FinishReason::Stop) => StopReason::Stop,
        Some(FinishReason::Length) => StopReason::Length,
        Some(FinishReason::ToolCalls) => StopReason::ToolUse,
        Some(FinishReason::ContentFilter) => StopReason::ContentFilter,
        _ => StopReason::Other,
    };
    
    Ok(MessageResponse {
        id: response.id,
        model: response.model,
        message,
        usage: response.usage,
        stop_reason,
        extra: HashMap::new(),
    })
}

2f：转换逻辑 chat_message_to_message()

OpenaiChatMessage → Message（与 Phase 0 中定义的 chat_message_to_message 一致，可复用）：

fn chat_message_to_message(&self, msg: &OpenaiChatMessage) -> Message {
    // 与 Phase 0 中定义的转换函数一致
    match msg {
        OpenaiChatMessage::System { content, .. }
        | OpenaiChatMessage::Developer { content, .. } => {
            Message::System { content: content_to_blocks(content) }
        }
        OpenaiChatMessage::User { content, .. } => {
            Message::User { content: content_to_blocks(content) }
        }
        OpenaiChatMessage::Assistant { content, tool_calls, .. } => {
            let mut blocks = content_to_blocks(content);
            if let Some(calls) = tool_calls {
                for call in calls {
                    if let OpenaiToolCall::Function { id, function } = call {
                        let input: Value = serde_json::from_str(&function.arguments).unwrap_or_default();
                        blocks.push(ContentBlock::ToolUse {
                            id: id.clone(),
                            name: function.name.clone(),
                            input,
                        });
                    }
                }
            }
            Message::Assistant { content: blocks }
        }
        OpenaiChatMessage::Tool { content, tool_call_id } => {
            Message::ToolResult {
                tool_call_id: tool_call_id.clone(),
                content: content_to_blocks(content),
                is_error: false,
            }
        }
        OpenaiChatMessage::Function { content, name } => {
            Message::ToolResult {
                tool_call_id: name.clone(),
                content: content_to_blocks(content),
                is_error: false,
            }
        }
    }
}

2g：转换逻辑 message_to_chat_message()

Message → OpenaiChatMessage（与 Phase 0 一致）：

fn message_to_chat_message(&self, msg: &Message) -> OpenaiChatMessage {
    // 与 Phase 0 中定义的 message_to_chat_message 一致
    // 详见 Phase 0 任务 8k
}

2h：流式 SSE → StreamEvent 转换

OpenaiChunkToEventStream 的流程：

  1. 读取字节流，按 \n 分割 SSE 行
  2. 解析 data: 前缀的 JSON 为 OpenaiChatChunk
  3. 将 OpenaiChatChunk 转换为 StreamEvent

状态机设计

因为 OpenAI 流式没有显式 block 边界，需要内部状态机追踪当前活跃的 block：

状态: Idle | InTextBlock { index } | InToolCall { index, tool_index } | InRefusalBlock { index }

事件处理:
  Idle + choices[0].delta.role == "assistant" 
    → 发出 MessageStart { id, model }，保持 Idle

  Idle/InTextBlock + delta.content 首次出现
    → 发出 ContentBlockStart { index: next_block_index, block_type: Text }
    → 发出 TextDelta { text }
    → 状态: InTextBlock { index }

  InTextBlock + delta.content 后续值
    → 发出 TextDelta { text }

  InTextBlock + finish_reason 出现（且无 tool_calls）
    → 发出 ContentBlockEnd { index }
    → 状态: Idle

  Idle + delta.refusal 首次出现
    → 发出 ContentBlockStart { index: next_block_index, block_type: Refusal }
    → 发出 RefusalDelta { text }
    → 状态: InRefusalBlock { index }

  InRefusalBlock + delta.refusal 后续值
    → 发出 RefusalDelta { text }

  InRefusalBlock + delta.content 或 delta.tool_calls
    → 忽略（refusal block 内部不应有其他内容类型；如确有，按防御性编程忽略）

  InRefusalBlock + finish_reason 出现
    → 发出 ContentBlockEnd { index }
    → 状态: Idle

  Idle/InTextBlock/InToolCall/InRefusalBlock + delta.tool_calls[i] 首次出现
    → 发出 ContentBlockStart { index: next_block_index, block_type: ToolUse { id, name } }
    → 发出 ToolCallArgumentsDelta { index, arguments }
    → 状态: InToolCall { index, tool_index: i }

  InToolCall + delta.tool_calls[i].arguments 后续值
    → 发出 ToolCallArgumentsDelta { index, arguments }

  InToolCall + delta.tool_calls[i] 结束（arguments 为空且 finish_reason 出现）
    → 发出 ToolCallEnd { index }
    → 状态: Idle

  任何状态 + finish_reason 出现（且 tool_calls 已结束）
    → 如有未结束的 TextBlock/RefusalBlock：发出 ContentBlockEnd { index }
    → 状态: Idle

  任何状态 + usage 有值
    → 发出 CostUpdate { usage }

  收到 data: [DONE]
    → 发出 MessageComplete { full_response: partial.finalize()? }

index 管理规则

规则 说明
text block 的 index next_block_index 从 0 开始递增，每个新 text block +1
tool_call 的 index 使用 chunk 中 choices[0].delta.tool_calls[i].index 字段，而非 Provider 自增计数器
并行 tool_calls 多个 tool_call 可能同时出现在同一个 chunk 中，各自有独立 index
index 冲突 text block 和 tool_call 共用一套 index 空间（OpenAI 不混合同一切片）

关键边界场景

场景 处理方式
同一切片同时出现 content + tool_calls 先处理 content 的 TextDelta（或在 InTextBlock 中追加），再处理 tool_calls
refusal 字段 第一次出现时：发出 ContentBlockStart { block_type: Refusal } + RefusalDelta { text }
仅 tool_calls 无内容 skip ContentBlockStart(Text) 直接进入 tool_call 处理
choices 数组为空的中间 chunk 忽略，仅触发 usage 或 finish_reason 解析

MessageComplete 事件的构造： 使用 PartialMessageResponse + apply_to 组装，然后 finalize() 得出 MessageResponse。当收到最后一个 chunk 或 [DONE] 信号时，调用 partial.finalize() 并发出 MessageComplete { full_response }。

任务 3：实现 AnthropicProvider

3a：配置

pub struct AnthropicProvider {
    http_client: Client,
    base_url: String,        // 默认: https://api.anthropic.com
    api_key: String,
    model: String,           // 默认: claude-sonnet-4-20250514
    anthropic_version: String, // 默认: 2023-06-01
}

impl AnthropicProvider {
    pub fn new(base_url: String, api_key: String, model: String) -> Self {
        // 使用 expect 提供有意义的错误信息，避免裸 unwrap 导致难以追踪的 panic
        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))
            .default_headers({
                let mut headers = HeaderMap::new();
                headers.insert("x-api-key", key_header);
                headers.insert("anthropic-version", version_header);
                headers
            })
            .build()
            .expect("创建 HTTP 客户端失败");
        
        Self {
            http_client,
            base_url: if base_url.is_empty() {
                "https://api.anthropic.com".to_string()
            } else {
                base_url
            },
            api_key,
            model,
            anthropic_version: "2023-06-01".to_string(),
        }
    }
}

max_tokens 必填兜底

Anthropic Messages API 要求 max_tokens 为必填字段，而 MessageRequest.max_tokens 的类型是 Option<u32>。 当用户未设置时，Provider 必须提供一个合理的默认值：

fn resolve_max_tokens(&self, request: &MessageRequest) -> u32 {
    request.max_tokens.unwrap_or(4096) // Anthropic 推荐的合理默认值
}

默认值选择：4096 是 Anthropic 官方推荐的无特殊需求时的合理值，覆盖大部分场景。 如果希望更精确，也可以从 capabilities().features.max_context_window / 10 计算（约 20000）， 但对于短对话 20000 可能偏高，4096 更安全。

3b：非流式 chat() — MessageRequest → Anthropic Messages API 请求体

Anthropic Messages API 请求格式：

{
  "model": "claude-sonnet-4-20250514",
  "max_tokens": 1024,
  "system": [{"type": "text", "text": "..."}],
  "messages": [
    {"role": "user", "content": [{"type": "text", "text": "hello"}]},
    {"role": "assistant", "content": [{"type": "text", "text": "hi"}]}
  ],
  "tools": [...]
}

转换要点：

  • Message::System → 提取为请求体的 system 参数（Anthropic 中 system 不在 messages 中）
  • Message::User { content } → { role: "user", content }（ContentBlock 直接映射）
  • Message::UserImage { data, mime_type, detail } → { role: "user", content: [{ type: "image", source: { type: "base64", media_type: mime_type, data } }] }
  • Message::Assistant { content } → { role: "assistant", content }（ContentBlock 直接映射）
  • Message::ToolResult { tool_call_id, content, is_error } → { role: "user", content: [{ type: "tool_result", tool_use_id: tool_call_id, content, is_error }] }
  • ThinkingConfig → 请求体增加 thinking: { type: "enabled", budget_tokens: N }

3c：Anthropic 响应 → MessageResponse

Anthropic Messages API 响应格式：

{
  "id": "msg_01...",
  "type": "message",
  "role": "assistant",
  "content": [
    {"type": "text", "text": "..."},
    {"type": "tool_use", "id": "toolu_01...", "name": "get_weather", "input": {...}},
    {"type": "thinking", "thinking": "..."}
  ],
  "model": "claude-sonnet-4-20250514",
  "stop_reason": "end_turn",
  "usage": {
    "input_tokens": 100,
    "output_tokens": 50
  }
}

转换要点：

  • stop_reason: end_turn → Stop, max_tokens → MaxTokens, tool_use → ToolUse, stop_sequence → StopSequence
  • content 中的 block 直接映射为 ContentBlock（text → Text, tool_use → ToolUse, thinking → Thinking）
  • usage: input_tokens → prompt_tokens, output_tokens → completion_tokens

3g：错误映射

Anthropic API 的错误响应格式与 OpenAI 不同，需要独立的错误解析逻辑：

Anthropic 错误响应格式：

{
  "type": "error",
  "error": {
    "type": "authentication_error",
    "message": "Invalid API key"
  }
}

错误类型映射：

HTTP 状态码 error.type 映射到 LlmError
400 invalid_request_error LlmError::Request { status: 400, body }
401 authentication_error LlmError::Authentication(body)
403 permission_error LlmError::Authentication(body)
404 not_found_error LlmError::Request { status: 404, body }
429 rate_limit_error LlmError::RateLimit { retry_after }
500 api_error LlmError::Request { status: 500, body }
529 overload_error LlmError::RateLimit { retry_after: Some(Duration::from_secs(30)) }

529 特殊处理：Anthropic 在过载时返回 529，应当映射到 RateLimit 而不是 >=500 的 Request， 这样上游的 retry 机制可以正确处理自动重试。

async fn handle_anthropic_error(&self, response: Response) -> LlmError {
    let status = response.status().as_u16();
    let body = response.text().await.unwrap_or_default();
    
    match status {
        401 | 403 => LlmError::Authentication(body),
        429 | 529 => {
            let retry_after = response
                .headers()
                .get("retry-after")
                .and_then(|v| v.to_str().ok())
                .and_then(|v| v.parse::<u64>().ok())
                .map(Duration::from_secs);
            LlmError::RateLimit { retry_after }
        }
        _ => LlmError::Request { status, body },
    }
}

3h：流式 chat_stream()

Anthropic Messages API 的 SSE 事件序列：

event: message_start
data: {"type": "message_start", "message": {"id": "msg_01...", ...}}

event: content_block_start
data: {"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}}

event: content_block_delta
data: {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": "Hello"}}

event: content_block_stop
data: {"type": "content_block_stop", "index": 0}

event: message_delta
data: {"type": "message_delta", "delta": {"stop_reason": "end_turn", "stop_sequence": null}, "usage": {"output_tokens": 50}}

event: message_stop
data: {"type": "message_stop"}

映射到 StreamEvent：

Anthropic Event StreamEvent
message_start MessageStart { id, model }
content_block_start { index, text } ContentBlockStart { index, block_type: Text }
content_block_start { index, tool_use } ContentBlockStart { index, block_type: ToolUse { id, name } }
content_block_start { index, thinking } ContentBlockStart { index, block_type: Thinking }
content_block_delta { text_delta } TextDelta { text }
content_block_delta { thinking_delta } ThinkingDelta { text }
content_block_delta { input_json_delta } ToolCallArgumentsDelta { index, arguments }
content_block_stop ContentBlockEnd { index } 或 ToolCallEnd { index }
message_delta { delta, usage } CostUpdate { usage } + set_thinking_signature() (如 delta.thinking.signature 存在)
message_stop 调用 partial.finalize() → MessageComplete { full_response }
ping 忽略

注意：与 OpenAI 不同，Anthropic 的流式有明确的 block 边界事件，映射到 ContentBlockStart/End 更自然。

3e：thinking_signature 处理

在 message_delta 事件中，如果 delta.thinking?.signature 存在，直接调用 partial.set_thinking_signature(sig) 将其写入内部状态。MessageComplete 事件不再携带 signature 字段。

flow 伪代码：

let mut partial = PartialMessageResponse::new();
partial.id = Some(msg_start.message.id);
partial.model = Some(msg_start.message.model);

while let Some(event) = stream.next().await {
    match event {
        AnthropicEvent::ContentBlockStart { index, content_block } => {
            let block_type = match content_block.type.as_str() { ... };
            StreamEvent::ContentBlockStart { index, block_type }.apply_to(&mut partial);
        }
        AnthropicEvent::ContentBlockDelta { index, delta } => {
            // 根据 delta.type 映射为对应 StreamEvent
            let ir_event = map_delta(index, delta);
            ir_event.apply_to(&mut partial);
        }
        AnthropicEvent::ContentBlockStop { index } => {
            StreamEvent::ContentBlockEnd { index }.apply_to(&mut partial);
        }
        AnthropicEvent::MessageDelta { delta, usage } => {
            // thinking signature: Provider 直接写入内部状态
            if let Some(sig) = delta.thinking.and_then(|t| t.signature) {
                partial.set_thinking_signature(sig);
            }
            // CostUpdate
            let partial_usage = map_usage(usage);
            StreamEvent::CostUpdate { usage: partial_usage }.apply_to(&mut partial);
        }
        AnthropicEvent::MessageStop => {
            let full = partial.finalize()?;
            yield StreamEvent::MessageComplete { full_response: full };
        }
        _ => {}
    }
}

3f：capabilities() 实现

fn capabilities(&self) -> ProviderCapabilities {
    ProviderCapabilities {
        provider_name: "anthropic",
        supported_models: Some(vec![
            "claude-sonnet-4-20250514".into(),
            "claude-3-5-sonnet-20241022".into(),
        ]),
        features: ProviderFeatures {
            streaming: true,
            thinking: true,
            vision: true,
            tool_use: true,
            parallel_tool_calls: true,
            system_prompt_in_messages: false, // Anthropic 使用 system 参数而非 messages 中的 system role
            max_context_window: 200_000,
            ..Default::default()
        },
    }
}

任务 4：实现 DeepSeekProvider / QwenProvider

4a：推荐方案 — newtype 包装 GenericOpenaiProvider

类型别名方案（不推荐）：

pub type DeepSeekProvider = GenericOpenaiProvider; // ❌ 零类型安全

类型别名使 DeepSeekProvider 和 GenericOpenaiProvider 完全等价——编译期无法区分， 也无法为特定 Provider 单独实现 trait（如自定义错误映射）。

推荐方案 — newtype 包装：

将 DeepSeek 和 Qwen 合并在同一个文件 src/llm/provider/openai_compat.rs 中， 使用 newtype 结构体包裹 GenericOpenaiProvider：

// src/llm/provider/openai_compat.rs
use crate::llm::provider::openai::GenericOpenaiProvider;

/// DeepSeek Provider — OpenAI-compatible 协议，仅配置不同。
pub struct DeepSeekProvider(pub GenericOpenaiProvider);

impl DeepSeekProvider {
    pub fn new(base_url: String, api_key: String, model: String) -> Self {
        let url = if base_url.is_empty() {
            "https://api.deepseek.com".to_string()
        } else {
            base_url
        };
        Self(GenericOpenaiProvider::new_with_name(
            url, api_key, model, "deepseek",
        ))
    }
}

/// 通义千问 Provider — OpenAI-compatible 协议。
pub struct QwenProvider(pub GenericOpenaiProvider);

impl QwenProvider {
    pub fn new(base_url: String, api_key: String, model: String) -> Self {
        let url = if base_url.is_empty() {
            "https://dashscope.aliyuncs.com/compatible-mode/v1".to_string()
        } else {
            base_url
        };
        Self(GenericOpenaiProvider::new_with_name(
            url, api_key, model, "qwen",
        ))
    }
}

LlmProvider trait 代理：每个 newtype 需要约 20 行 trait 代理代码。 使用 macro 减少重复：

macro_rules! delegate_openai_compat {
    ($name:ident) => {
        #[async_trait]
        impl LlmProvider for $name {
            async fn chat(&self, req: MessageRequest) -> Result<MessageResponse, LlmError> {
                self.0.chat(req).await
            }
            async fn chat_stream(
                &self, req: MessageRequest,
            ) -> Result<Pin<Box<dyn Stream<Item = Result<StreamEvent, LlmError>> + Send>>, LlmError> {
                self.0.chat_stream(req).await
            }
            fn capabilities(&self) -> ProviderCapabilities {
                self.0.capabilities()
            }
        }
    };
}

delegate_openai_compat!(DeepSeekProvider);
delegate_openai_compat!(QwenProvider);

4b：差异化处理

GenericOpenaiProvider 需要暴露配置参数支持 Provider 间的差异：

pub struct GenericOpenaiProvider {
    http_client: Client,
    base_url: String,
    api_key: String,
    model: String,
    provider_name: &'static str,
    /// 额外请求头，Qwen 需要 X-DashScope-SSE: enable
    extra_headers: Vec<(String, String)>,
}
差异点 DeepSeek Qwen
base_url api.deepseek.com dashscope.aliyuncs.com/compatible-mode/v1
max_tokens 字段名 max_tokens（兼容） max_tokens（兼容）
错误格式 标准 OpenAI 风格 非标准 error body（需 fallback）
模型名 deepseek-chat qwen-plus / qwen-max
额外请求头 无特殊 X-DashScope-SSE: enable（由 extra_headers 传入）
默认模型 deepseek-chat qwen-plus

Qwen 错误格式 fallback：

Qwen 的错误 body 可能是非标准 JSON 格式甚至纯文本。处理策略：

fn handle_error(&self, status: u16, body: &str) -> LlmError {
    // 先尝试按 OpenAI 标准格式解析
    if let Ok(openai_err) = serde_json::from_str::<OpenaiErrorBody>(body) {
        return Self::map_openai_error(status, openai_err);
    }
    // fallback: 将原始文本包装为 Request 错误
    LlmError::Request {
        status,
        body: body.to_string(),
    }
}

实施注意：GenericOpenaiProvider 的所有配置参数由 new_with_name 构造器统一传入， DeepSeek/Qwen 的 newtype 构造器负责填充差异化的参数值。

任务 5：适配 ProviderRegistry

src/llm/provider/registry.rs 的改动较小：

  • register_with_config() 不需要改——它调用 create_provider()，而 create_provider 已适配新 enum
  • get() 返回 Option<&dyn LlmProvider>——trait 签名变更后自动生效
  • 不需要额外修改

任务 6：添加 wiremock 集成测试

6a：Cargo.toml

[dev-dependencies]
wiremock = "0.6"

6b：OpenaiChatProvider 测试

#[cfg(test)]
mod tests {
    use super::*;
    use wiremock::{Mock, MockServer, ResponseTemplate};
    use wiremock::matchers::{method, path};
    
    #[tokio::test]
    async fn test_openai_chat_basic() {
        let mock_server = MockServer::start().await;
        
        // 注册 mock 响应
        Mock::given(method("POST"))
            .and(path("/chat/completions"))
            .respond_with(ResponseTemplate::new(200).set_body_json(json!({
                "id": "chatcmpl-123",
                "object": "chat.completion",
                "created": 1718000000,
                "model": "gpt-4o",
                "choices": [{
                    "index": 0,
                    "message": {
                        "role": "assistant",
                        "content": "Hello!"
                    },
                    "finish_reason": "stop"
                }],
                "usage": {
                    "prompt_tokens": 10,
                    "completion_tokens": 5,
                    "total_tokens": 15
                }
            })))
            .mount(&mock_server)
            .await;
        
        let provider = OpenaiChatProvider::new(
            mock_server.uri(), "sk-test".into(), "gpt-4o".into(),
        );
        
        let request = MessageRequest {
            model: "gpt-4o".into(),
            messages: vec![Message::user_text("Hi")],
            ..Default::default()
        };
        
        let response = provider.chat(request).await.unwrap();
        assert_eq!(response.model, "gpt-4o");
        assert_eq!(response.message.text(), "Hello!");
    }
    
    #[tokio::test]
    async fn test_openai_chat_error() {
        // HTTP 401 → LlmError::Authentication
        // HTTP 429 → LlmError::RateLimit
        // HTTP 500 → LlmError::Request
    }
    
    #[tokio::test]
    async fn test_openai_chat_stream() {
        // SSE 流式响应测试
    }
}

6c：AnthropicProvider 测试

#[tokio::test]
async fn test_anthropic_chat_basic() {
    let mock_server = MockServer::start().await;
    
    // Mock Anthropic Messages API 响应
    Mock::given(method("POST"))
        .and(path("/v1/messages"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "id": "msg_01...",
            "type": "message",
            "role": "assistant",
            "content": [{"type": "text", "text": "Hello from Claude!"}],
            "model": "claude-sonnet-4-20250514",
            "stop_reason": "end_turn",
            "usage": {"input_tokens": 10, "output_tokens": 5}
        })))
        .mount(&mock_server)
        .await;
    
    let provider = AnthropicProvider::new(
        mock_server.uri(), "sk-ant-test".into(), "claude-sonnet-4-20250514".into(),
    );
    
    let request = MessageRequest {
        model: "claude-sonnet-4-20250514".into(),
        messages: vec![Message::user_text("Hi")],
        max_tokens: Some(100),
        ..Default::default()
    };
    
    let response = provider.chat(request).await.unwrap();
    assert_eq!(response.message.text(), "Hello from Claude!");
}

#[tokio::test]
async fn test_anthropic_chat_stream() {
    // SSE 事件序列测试
    // message_start → content_block_start(text) → content_block_delta(text_delta) 
    // → content_block_stop → message_delta → message_stop
}

6d：Provider 双向映射测试

测试 Message → OpenaiChatRequest → MessageResponse 和 Message → AnthropicRequest → MessageResponse 的双向转换一致性。

6e：测试覆盖要求

每个 Provider 的测试至少覆盖以下场景：

基本路径：

  • 基本文本对话（chat + chat_stream）
  • 带工具定义的对话
  • 多轮消息历史（system + user + assistant + tool_result）
  • 空消息列表 → 合理错误

错误路径：

  • HTTP 401 → LlmError::Authentication
  • HTTP 429 → LlmError::RateLimit
  • HTTP 500 → LlmError::Request
  • HTTP 529（Anthropic overloaded）→ LlmError::RateLimit

边界场景：

  • 响应中 choices 数组为空 → convert_response 返回错误而非 panic
  • 流式 choices: [] 的中间 chunk → 忽略，不产生事件
  • SSE 中断后部分数据 → 解析不 panic，返回残数据错误
  • data: [DONE] 之前流中断 → 产出部分响应
  • 并行 tool_calls（多 index 同时出现在同一切片）→ 正确处理 index 分配

Anthropic 特有：

  • message_start → content_block_start(text) → content_block_delta(text_delta) → content_block_stop → message_delta → message_stop 完整序列
  • ping 事件 → 忽略，不中断流
  • message_start 后直接 message_stop（空响应）→ 产出空 MessageResponse

验证方式

  1. 编译检查：cargo build 通过
  2. 单元测试：cargo test 全部通过
  3. Provider 测试：每个 Provider 的 mock 集成测试覆盖基本路径 + 错误路径
  4. clippy 检查：cargo clippy 无新增警告

回滚方案

如果某个 Provider 实现不合理，将该 Provider 回退为 Err(LlmError::Other(...))，不影响其他 Provider。每个 Provider 完成时建议打 tag provider-{name}-v2。

开放事项

  • OpenaiResponseProvider 完整实现：当前返回 Err(LlmError::Other(...))。升级触发条件：Trace 中 Response API 采用率 >20%，或收到 ≥3 个企业客户需求。升级后覆盖核心对话能力（create response + 流式 + 工具调用），内置工具（web_search, file_search）通过 MessageRequest.extra 传递。
  • SSE 解析通用层：当前 OpenaiProvider 有内联 SseChunkStream，AnthropicProvider 也有类似 SSE 解析逻辑。是否在 Phase 1 提取通用 SseByteStream 层（独立于类型系统、纯字节处理），在"Phase 0 未完成时的降级路径"中可作为先行完成项。
  • ContentBlock::Extension 在 OpenAI Response API 内置工具场景的使用
  • Google Gemini Provider：当前不在 Phase 1 范围内。建议的触发条件：3 个客户需求或 Gemini API 流量占项目总 LLM 调用的 5% 以上时进入"Next"阶段。
  • Anthropic anthropic-beta header 与 Prompt Caching：Prompt caching 等特性需通过此 header 启用。当前 Phase 1 不做，当用户明确要求缓存或流量成本显著时启动。
Reference in New Issue View Git Blame Copy Permalink