agcore/design/pdd/9c-llm-provider-trait.md

LlmProvider Trait 设计

本文档从 9-llm-provider-unified-interface.md 拆分而来，包含 §4 LlmProvider Trait 设计。

相关文件：

• 9b-ir-type-system.md — IR 类型定义（MessageRequest、MessageResponse、StreamEvent 等）
• 9d-provider-implementations.md — 各 Provider 的具体实现策略
• 9f-edge-cases.md — Thinking signature 等边界情况（与 StreamEvent 设计关联）

4. LlmProvider Trait 设计

4.1 核心接口

#[async_trait]
pub trait LlmProvider: Send + Sync {
    /// 发送消息请求，获取完整响应。
    async fn chat(&self, request: MessageRequest) -> Result<MessageResponse, LlmError>;

    /// 流式消息请求，返回语义化事件流。
    async fn chat_stream(
        &self,
        request: MessageRequest,
    ) -> Result<
        Pin<Box<dyn Stream<Item = Result<StreamEvent, LlmError>> + Send>>,
        LlmError,
    >;

    /// 返回 Provider 的能力描述。
    fn capabilities(&self) -> ProviderCapabilities;
}

4.2 ProviderCapabilities

#[derive(Debug, Clone)]
pub struct ProviderCapabilities {
    /// Provider 标识名称
    pub provider_name: &'static str,
    /// 支持的模型列表（None = 不限制）
    pub supported_models: Option<Vec<String>>,
    /// 特性标记
    pub features: ProviderFeatures,
}

#[derive(Debug, Clone, Default)]
pub struct ProviderFeatures {
    pub streaming: bool,
    pub thinking: bool,
    pub vision: bool,
    pub audio_input: bool,
    pub tool_use: bool,
    pub parallel_tool_calls: bool,
    /// true=OpenAI风格(system in messages), false=Anthropic风格(system as param)
    pub system_prompt_in_messages: bool,
    pub max_context_window: u32,
}

capabilities() 的使用场景

1. LlmCycle：根据 features.thinking 决定是否启用 thinking 模式
2. AgentBuilder：在构建时校验模型是否支持所需特性
3. UI/CLI：展示 Provider 的能力矩阵
4. 智能路由：根据能力自动选择最佳 Provider

4.3 流式事件 StreamEvent

/// 流式事件 —— 流式响应的语义化增量构建过程。
/// 一组 StreamEvent 最终可汇聚为一个完整的 MessageResponse。
#[derive(Debug, Clone)]
pub enum StreamEvent {
    // ── Meta ──
    /// 消息开始
    MessageStart { id: String, model: String },

    // ── Content Block 边界 ──
    /// Content block 开始（index = 全局 content block 序号）
    ContentBlockStart { index: u32, block_type: ContentBlockType },
    /// Content block 结束
    ContentBlockEnd { index: u32 },

    // ── 块内增量（无 index，隐含属于当前活跃 block）──
    /// 文本增量
    TextDelta { text: String },
    /// 思考增量（Anthropic thinking block）
    ThinkingDelta { text: String },
    /// 拒绝增量（OpenAI refusal，汇聚为 ContentBlock::Text）
    RefusalDelta { text: String },

    // ── Tool Call 参数（index = block index）──
    /// 工具参数增量
    ToolCallArgumentsDelta { index: u32, arguments: String },
    /// 工具参数结束，可尝试解析 JSON
    ToolCallEnd { index: u32 },

    // ── 汇总 ──
    /// Token 用量（字段级合并，见 PartialUsage）
    CostUpdate { usage: PartialUsage },
    /// 消息完成（thinking_signature 仅 Anthropic 场景，回填到最后的 Thinking block）
    MessageComplete { stop_reason: StopReason, thinking_signature: Option<String> },

    // ── 错误 ──
    Error { message: String },
}

/// ContentBlock 的类型标识，嵌入在 ContentBlockStart 事件中。
#[derive(Debug, Clone)]
pub enum ContentBlockType {
    Text,
    Thinking,
    Refusal,
    ToolUse { id: String, name: String },
}

/// CostUpdate 中携带的用量数据——只含本次更新中真正下发的字段。
///
/// 汇聚算法做字段级合并（覆盖各字段的 Some 值），而非全量覆盖。
/// 解决 Provider 分多次下发 Usage 的问题：
/// - OpenAI: 一次全量下发（所有字段都有值）
/// - Anthropic: message_start 时下发 input_tokens，message_delta 时下发 output_tokens
#[derive(Debug, Clone, Default)]
pub struct PartialUsage {
    pub prompt_tokens: Option<u32>,
    pub completion_tokens: Option<u32>,
    pub total_tokens: Option<u32>,
    pub completion_tokens_details: Option<CompletionTokensDetails>,
    pub prompt_tokens_details: Option<PromptTokensDetails>,
}

设计决策：为什么移除 ToolCallStart

ToolCallStart { index, id, name } 的语义本质是一个 ContentBlock 的开始（类型为 ToolUse）， 没有单独存在的必要。合并到 ContentBlockStart.block_type = ToolUse { id, name } 后：

• index 语义统一：ContentBlockStart.index 是全文档唯一的 content block 序号
• 事件减少：每次 tool_use block 开始少发一个事件，Anthropic 映射更自然（content_block_start → ContentBlockStart）
• #4议题自动解决：index 在 IR 层面始终为全局序号，OpenAI Provider 内部的局部→全局映射完全封装在 Provider 层， 不在 IR 中暴露差异

设计决策：为什么 TextDelta 不带 index

• 单 SSE 连接内 TCP 保证字节序，TextDelta 必然属于最后一个 ContentBlockStart 开始的 block
• 不携带 index 可减少事件体积
• 如果未来 LlmCycle 需要跨 Provider 合并流，可向后兼容地添加 index 字段（降级策略：无 index 时默认归属当前活跃 block）

设计决策：CostUpdate 使用 PartialUsage 做字段级合并

覆盖策略（取最新值）在 Anthropic 场景下出错——Anthropic 分两次下发：

CostUpdate #1: { prompt_tokens: Some(100), completion_tokens: None,   total_tokens: Some(100) }
CostUpdate #2: { prompt_tokens: None,   completion_tokens: Some(50),  total_tokens: Some(150) }

如果直接 Usage 全量覆盖，第一次的 prompt_tokens 会被第二次的 None 清零。 改为 PartialUsage（字段级 Option）后，汇聚算法只更新 Some 的字段，避免清零。

与当前 StreamEvent 的对比

当前 StreamEvent	新 StreamEvent	理由
AssistantTextDelta { text }	TextDelta { text }	简洁化
—	ThinkingDelta { text }	Anthropic 需要
—	RefusalDelta { text }	OpenAI 需要
ToolExecutionStarted { tool_name, input, tool_call_id }	ContentBlockStart { index, ToolUse { id, name } } + ToolCallArgumentsDelta { index, arguments }	拆分为 block 边界 + 参数增量
ToolExecutionCompleted	移除	LlmCycle 层事件，非 Provider 层
—	ContentBlockStart / ContentBlockEnd	显式 block 边界标记
—	BlockContentType	ContentBlock 类型标识
CostUpdate { usage: Usage }	CostUpdate { usage: PartialUsage }	字段级合并，兼容多 Provider
TurnComplete { reason }	MessageComplete { stop_reason, thinking_signature: Option<String> }	语义更准确 + thinking 签名回填
Error { message }	保留	✅
—	MessageStart { id, model }	Anthropic 需要
—	ToolCallEnd { index }	明确参数完整时间点
ToolCallStart { index, id, name }	移除	合并到 ContentBlockStart.ToolUse

---

4.4 PartialMessageResponse —— 流式事件的汇聚算法

✅ 推演结论（2026-06-17）：采用方案 B（显式边界）—— ContentBlockStart/End 声明 block 边界，BTreeMap 按 index 分桶组装

核心思路： 放弃"类型切换推断 block 边界"的隐含方案。为 StreamEvent 增加 ContentBlockStart 和 ContentBlockEnd 事件，使每个 block 的开始和结束都有精确的事件标记。汇聚算法由"线性累积 + 隐含 flush + 延迟排序"改为 "按 index 分桶 + 排序组装"，彻底消除对事件到达顺序的依赖。

决策理由：

1. 类型切换推断在 ToolCallEnd 后跟 TextDelta 的场景下导致顺序错乱（ToolUse 被延迟插入到 Text 之后）
2. 显式边界将位置信息绑定到 index，不依赖事件到达时序，容错性更强
3. Anthropic 的 content_block_start/stop 天然映射为 ContentBlockStart/End
4. OpenAI 的无边界流式由 Provider 内部分析 delta 类型来合成边界，封装在 Provider 层
5. #4（ToolCallStart index 归一化）自动解决——index 统一为全局 content block 序号

PartialMessageResponse 结构体

use std::collections::{BTreeMap, HashMap, HashSet};
use serde_json::Value;

/// 流式事件的累积状态 —— 按 index 分桶，逐个构建 ContentBlock。
#[derive(Debug, Default)]
pub struct PartialMessageResponse {
    // ── 来自 MessageStart ──
    pub id: Option<String>,
    pub model: Option<String>,

    /// 按 index 分桶的 ContentBlock 构建器（BTreeMap 保证升序遍历）
    pub blocks: BTreeMap<u32, ContentBlockBuilder>,

    /// Tool call 参数累积（按 block index 关联）
    pub tool_call_args: HashMap<u32, String>,

    /// 已收到 ContentBlockEnd 的 block index 集合
    pub block_completion: HashSet<u32>,

    /// 当前最后打开的 block index（供无 index 的 TextDelta 定位）
    pub last_open_index: Option<u32>,

    /// Token 用量（字段级合并后的最终值）
    pub usage: Usage,

    /// 结束状态
    pub stop_reason: Option<StopReason>,

    /// MessageComplete 携带的 thinking signature（留待 finalize 回填）
    pub thinking_signature: Option<String>,

    pub is_errored: bool,
    pub is_complete: bool,
}

/// 按 index 分桶的 ContentBlock 构建器。
#[derive(Debug)]
pub enum ContentBlockBuilder {
    Text(String),
    Thinking { buffer: String, signature: Option<String> },
    Refusal(String),
    ToolUse { id: String, name: String },
}

apply_to 算法

impl StreamEvent {
    /// 将当前事件应用到 PartialMessageResponse 上。
    /// 返回 true 表示正常处理，false 表示应停止处理后续事件。
    pub fn apply_to(&self, state: &mut PartialMessageResponse) -> bool {
        match self {
            // ──────────────── Meta ────────────────
            StreamEvent::MessageStart { id, model } => {
                state.id = Some(id.clone());
                state.model = Some(model.clone());
                true
            }

            // ──────────────── Block 边界 ────────────────
            StreamEvent::ContentBlockStart { index, block_type } => {
                let builder = match block_type {
                    ContentBlockType::Text => ContentBlockBuilder::Text(String::new()),
                    ContentBlockType::Thinking => ContentBlockBuilder::Thinking {
                        buffer: String::new(),
                        signature: None,
                    },
                    ContentBlockType::Refusal => ContentBlockBuilder::Refusal(String::new()),
                    ContentBlockType::ToolUse { id, name } =>
                        ContentBlockBuilder::ToolUse { id: id.clone(), name: name.clone() },
                };
                state.blocks.entry(*index).or_insert(builder);
                state.last_open_index = Some(*index);
                true
            }

            StreamEvent::ContentBlockEnd { index } => {
                state.block_completion.insert(*index);
                true
            }

            // ──────────────── 块内增量 ────────────────
            StreamEvent::TextDelta { text } => {
                // 定位到最后打开的 block
                if let Some(idx) = state.last_open_index {
                    if let Some(ContentBlockBuilder::Text(ref mut buf)) = state.blocks.get_mut(&idx) {
                        buf.push_str(text);
                    }
                } else {
                    tracing::warn!("TextDelta 到达时无活跃 block");
                }
                true
            }

            StreamEvent::ThinkingDelta { text } => {
                if let Some(idx) = state.last_open_index {
                    if let Some(ContentBlockBuilder::Thinking { ref mut buffer, .. }) =
                        state.blocks.get_mut(&idx)
                    {
                        buffer.push_str(text);
                    }
                } else {
                    tracing::warn!("ThinkingDelta 到达时无活跃 block");
                }
                true
            }

            StreamEvent::RefusalDelta { text } => {
                if let Some(idx) = state.last_open_index {
                    if let Some(ContentBlockBuilder::Refusal(ref mut buf)) =
                        state.blocks.get_mut(&idx)
                    {
                        buf.push_str(text);
                    }
                } else {
                    tracing::warn!("RefusalDelta 到达时无活跃 block");
                }
                true
            }

            // ──────────────── Tool Call 参数 ────────────────
            StreamEvent::ToolCallArgumentsDelta { index, arguments } => {
                state
                    .tool_call_args
                    .entry(*index)
                    .or_default()
                    .push_str(arguments);
                true
            }

            StreamEvent::ToolCallEnd { index } => {
                // ToolCallEnd 只做标记，参数在 finalize 时解析
                state.block_completion.insert(*index);
                true
            }

            // ──────────────── 汇总 ────────────────
            StreamEvent::CostUpdate { usage } => {
                Self::apply_cost_update(state, usage);
                true // 即使 is_complete 后 CostUpdate 仍然可以到达
            }

            StreamEvent::MessageComplete {
                stop_reason,
                thinking_signature,
            } => {
                state.stop_reason = Some(*stop_reason);
                state.thinking_signature = thinking_signature.clone();
                state.is_complete = true;
                true
            }

            // ──────────────── 错误 ────────────────
            StreamEvent::Error { message } => {
                state.is_errored = true;
                tracing::error!("流式处理中发生错误: {}", message);
                false // 终止处理
            }
        }
    }

    /// 字段级合并 CostUpdate（仅覆盖 Some 的字段）。
    fn apply_cost_update(state: &mut PartialMessageResponse, update: &PartialUsage) {
        if let Some(v) = update.prompt_tokens {
            state.usage.prompt_tokens = v;
        }
        if let Some(v) = update.completion_tokens {
            state.usage.completion_tokens = v;
        }
        if let Some(v) = update.total_tokens {
            state.usage.total_tokens = v;
        }
        if update.completion_tokens_details.is_some() {
            state.usage.completion_tokens_details = update.completion_tokens_details;
        }
        if update.prompt_tokens_details.is_some() {
            state.usage.prompt_tokens_details = update.prompt_tokens_details;
        }
    }
}

finalize —— 完成逻辑

impl PartialMessageResponse {
    /// 将所有累积状态转化为最终的 MessageResponse。
    pub fn finalize(mut self) -> Result<MessageResponse, LlmError> {
        let id = self.id.unwrap_or_else(|| "stream-unknown".to_string());
        let model = self.model.unwrap_or_else(|| "unknown".to_string());
        let stop_reason = self.stop_reason.unwrap_or(StopReason::Other);

        // 1. 按 index 升序遍历所有 blocks，转换为 ContentBlock
        let mut content: Vec<ContentBlock> = Vec::new();
        for (_index, builder) in std::mem::take(&mut self.blocks).into_iter() {
            let block = match builder {
                ContentBlockBuilder::Text(text) => ContentBlock::Text { text },
                ContentBlockBuilder::Thinking { buffer, mut signature } => {
                    // 如果 Thinking block 的 signature 尚未填充，用 MessageComplete 的回填
                    if signature.is_none() {
                        signature = self.thinking_signature.clone();
                    }
                    ContentBlock::Thinking {
                        text: buffer,
                        signature,
                    }
                }
                ContentBlockBuilder::Refusal(text) => ContentBlock::Text { text },
                ContentBlockBuilder::ToolUse { id, name } => {
                    let arguments = self.tool_call_args.remove(&(_index as u32))
                        .unwrap_or_default();
                    let input: Value = serde_json::from_str(&arguments)
                        .unwrap_or(Value::Null);
                    ContentBlock::ToolUse { id, name, input }
                }
            };
            content.push(block);
        }

        Ok(MessageResponse {
            id,
            model,
            message: Message::Assistant { content },
            usage: self.usage,
            stop_reason,
            extra: HashMap::new(),
        })
    }

    /// 检查是否已收到足以构造"有意义"响应的数据。
    pub fn is_meaningful(&self) -> bool {
        self.id.is_some() && (self.is_complete || self.is_errored)
    }
}

边界情况处理

边界场景	处理策略	理由
MessageStart 前收到 TextDelta	warn log，忽略增量	不合法流，防御性处理
ToolCallArgumentsDelta 乱序	HashMap key 天然容忍	index 作为关联 key，到达顺序不影响最终累积
多次 CostUpdate	字段级合并（PartialUsage）	OpenAI 一次全量，Anthropic 分两次，OpenAI Response 可能多次
同一 index 收到多次 ContentBlockStart	entry(*index).or_insert() 幂等	第二次不覆盖已有内容
MessageComplete 后 CostUpdate 才到	不阻断，仍然合并	Usage 最终值可能最后才到
ContentBlockEnd 缺失	finalize 时仍然输出内容	不完整的 block 也是内容
无任何 ContentBlockStart	finalize 返回空的 Assistant 消息	边界场景，不阻断
Error 后收到其他事件	Error 返回 false，上层停止调用 apply_to	一旦出错，不再处理后续事件
thinking_signature 未填	Thinking block 的 signature 为 None	安全降级