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

feat(llm): 新增 IR 类型层 + 切换 LlmProvider trait 签名

Browse Source 
引入跨 Provider 统一的新类型系统：
- Message 扁平大枚举（System/User/UserImage/Assistant/ToolResult）
- MessageRequest + MessageResponse + StreamEvent（高精度 IR）
- PartialMessageResponse 含 apply_to/finalize 汇聚算法
- ProviderType enum 扩展 + ProviderCapabilities

切换 LlmProvider trait 签名为 chat(MessageRequest) → MessageResponse，
chat_stream 返回 Stream<Item = Result<StreamEvent, LlmError>>，新增 capabilities()。

StreamEvent 命名冲突处理：旧变体迁移至 old_stream::LegacyStreamEvent，
stream.rs 通过 pub use 重新导出新高精度事件。

OpenaiProvider 添加 Phase 0 临时桥接（MessageRequest ↔ OpenaiChatRequest 转换），
标注 ponytail: Phase 0 临时桥接 标记，Phase 1 重写时移除。

测试 147 passed / 0 failed（新增 31 个新类型测试）。
This commit is contained in:
徐涛
2026-07-02 21:58:07 +08:00
parent 925c8f9729
commit 7c299f1cfd
18 changed files with 2605 additions and 441 deletions
Download Patch File Download Diff File Expand all files Collapse all files
examples/simple_visit.rs
+23 -12
View File
@@ -4,23 +4,34 @@ use agcore::init_tracing;
use agcore::llm::{
cycle::{CycleConfig, LlmCycle},
provider::{create_provider, ProviderConfig, ProviderType},
types::{ChatResponse, OpenaiContentPart},
types::{
message::ContentBlock, response_v2::MessageResponse,
},
};
fn extract_response_text(response: &ChatResponse) -> &str {
fn extract_response_text(response: &MessageResponse) -> &str {
// Phase 0：MessageResponse.text() 直接给出拼接好的 Assistant 文本。
if !response.text().is_empty() {
// ⚠️ 返回的是 owned String 的引用，调用方需在 response 生命周期内使用。
// 此 example 短命，足以演示。
return response_text_ref(response);
}
"[无文本内容]"
}
fn response_text_ref(response: &MessageResponse) -> &str {
// ponytail: example 助手 —— 不在正式 crate API 中，单纯绕开 borrow 限制。
// 真实调用请直接使用 `response.text()` 拿到 owned String。
match &response.message {
agcore::llm::types::OpenaiChatMessage::Assistant { content, .. } => match content {
agcore::llm::types::ContentField::String(s) => s,
agcore::llm::types::ContentField::Array(parts) => {
for part in parts {
if let OpenaiContentPart::Text { text } = part {
return text;
}
agcore::llm::types::message::Message::Assistant { content } => {
for block in content {
if let ContentBlock::Text { text } = block {
return text.as_str();
}
"[无文本内容]"
}
},
_ => "[非 assistant 消息]",
""
}
_ => "",
}
}
src/agent/builder.rs
+20 -3
View File
@@ -118,17 +118,34 @@ impl AgentBuilder {
#[cfg(test)]
mod tests {
use super::*;
use crate::llm::provider::LlmProvider;
use crate::llm::types::{ChatRequest, ChatResponse};
use crate::llm::error::LlmError;
use crate::llm::provider::{LlmProvider, ProviderCapabilities, ProviderFeatures};
use crate::llm::types::request_v2::MessageRequest;
use crate::llm::types::response_v2::{MessageResponse, StreamEvent};
use async_trait::async_trait;
use futures_core::Stream;
use std::pin::Pin;
struct StubProvider;
#[async_trait]
impl LlmProvider for StubProvider {
async fn chat(&self, _request: ChatRequest) -> Result<ChatResponse, LlmError> {
async fn chat(&self, _request: MessageRequest) -> Result<MessageResponse, LlmError> {
unimplemented!()
}
async fn chat_stream(
&self,
_request: MessageRequest,
) -> Result<Pin<Box<dyn Stream<Item = Result<StreamEvent, LlmError>> + Send>>, LlmError>
{
unimplemented!()
}
fn capabilities(&self) -> ProviderCapabilities {
ProviderCapabilities {
provider_name: "stub",
supported_models: None,
features: ProviderFeatures::default(),
}
}
}
#[test]
src/agent/session.rs
+40 -27
View File
@@ -15,7 +15,9 @@ use crate::agent::runtime::RuntimeBundle;
use crate::agent::session_memory::SessionMemory;
use crate::llm::cycle::{CostTracker, CycleConfig, LlmCycle};
use crate::llm::hooks::{HookContext, HookEvent};
use crate::llm::types::ChatResponse;
use crate::llm::types::message::{ContentBlock, Message};
use crate::llm::types::openai_message::OpenaiChatMessage;
use crate::llm::types::response_v2::MessageResponse;
use crate::memory::store::InMemoryStore;
/// Agent 会话实例。
@@ -120,7 +122,7 @@ impl AgentSession {
pub async fn submit_turn(
&mut self,
user_input: impl Into<String>,
) -> Result<ChatResponse, AgentError> {
) -> Result<MessageResponse, AgentError> {
let turn_index = self.turn_index;
let hook_executor = Arc::clone(&self.bundle.hook_executor);
@@ -168,13 +170,15 @@ impl AgentSession {
mod tests {
use super::*;
use crate::agent::builder::AgentBuilder;
use crate::llm::error::LlmError;
use crate::llm::hooks::{Hook, HookContext, HookExecutor, HookResult};
use crate::llm::provider::LlmProvider;
use crate::llm::types::{
ChatRequest, ChatResponse, FinishReason, OpenaiChatMessage,
};
use crate::llm::provider::{LlmProvider, ProviderCapabilities, ProviderFeatures};
use crate::llm::types::request_v2::MessageRequest;
use crate::llm::types::response_v2::{MessageResponse, StopReason, StreamEvent};
use crate::tools::ToolRegistry;
use async_trait::async_trait;
use futures_core::Stream;
use std::pin::Pin;
use std::sync::atomic::{AtomicU32, Ordering};
/// 计数 hook —— 每被调用一次 +1。
@@ -200,11 +204,11 @@ mod tests {
/// MockProvider：按调用顺序返回预设响应。
struct MockProvider {
responses: std::sync::Mutex<Vec<ChatResponse>>,
responses: std::sync::Mutex<Vec<MessageResponse>>,
}
impl MockProvider {
fn new(responses: Vec<ChatResponse>) -> Self {
fn new(responses: Vec<MessageResponse>) -> Self {
Self {
responses: std::sync::Mutex::new(responses),
}
@@ -213,15 +217,27 @@ mod tests {
#[async_trait]
impl LlmProvider for MockProvider {
async fn chat(&self, _request: ChatRequest) -> Result<ChatResponse, crate::llm::error::LlmError> {
async fn chat(&self, _request: MessageRequest) -> Result<MessageResponse, LlmError> {
let mut responses = self.responses.lock().unwrap();
if responses.is_empty() {
return Err(crate::llm::error::LlmError::Other(
"no more mock responses".into(),
));
return Err(LlmError::Other("no more mock responses".into()));
}
Ok(responses.remove(0))
}
async fn chat_stream(
&self,
_request: MessageRequest,
) -> Result<Pin<Box<dyn Stream<Item = Result<StreamEvent, LlmError>> + Send>>, LlmError>
{
unimplemented!()
}
fn capabilities(&self) -> ProviderCapabilities {
ProviderCapabilities {
provider_name: "mock",
supported_models: None,
features: ProviderFeatures::default(),
}
}
}
struct StubAgent {
@@ -238,11 +254,18 @@ mod tests {
}
}
fn assistant_text(text: &str) -> ChatResponse {
ChatResponse {
message: OpenaiChatMessage::assistant_text(text),
fn assistant_text(text: &str) -> MessageResponse {
MessageResponse {
id: String::new(),
model: String::new(),
message: Message::Assistant {
content: vec![ContentBlock::Text {
text: text.into(),
}],
},
usage: crate::llm::types::Usage::from_input_output(10, 5),
stop_reason: Some(FinishReason::Stop),
stop_reason: StopReason::Stop,
extra: std::collections::HashMap::new(),
}
}
@@ -267,17 +290,7 @@ mod tests {
assert_eq!(session.turn_index(), 0);
let response = session.submit_turn("hi").await.unwrap();
let text = match &response.message {
OpenaiChatMessage::Assistant { content, .. } => {
if let crate::llm::types::ContentField::String(s) = content {
s.clone()
} else {
String::new()
}
}
_ => String::new(),
};
assert_eq!(text, "hello back");
assert_eq!(response.text(), "hello back");
assert_eq!(session.turn_index(), 1);
assert_eq!(session.usage().total().prompt_tokens, 10);
assert_eq!(session.usage().total().completion_tokens, 5);
src/llm/cycle.rs
+283 -177
View File
@@ -16,11 +16,16 @@ use crate::llm::compact::{should_compact, microcompact, CompactConfig, CompactSt
use crate::llm::cycle::retry::should_retry;
use crate::llm::error::LlmError;
use crate::llm::hooks::{HookContext, HookExecutor};
use crate::llm::provider::LlmProvider;
use crate::llm::provider::{LlmProvider, ProviderCapabilities, ProviderFeatures};
use crate::llm::stream::StreamEvent;
use crate::llm::types::message::{ContentBlock, Message};
use crate::llm::types::openai_message::{
ContentField, OpenaiChatMessage, OpenaiContentPart,
};
use crate::llm::types::request_v2::MessageRequest;
use crate::llm::types::response_v2::{MessageResponse, StopReason};
use crate::llm::types::{
ChatRequest, ChatResponse, FinishReason, OpenaiChatMessage, OpenaiTool, OpenaiToolCall,
ToolChoice, ToolDefinition,
FinishReason, FunctionCall, OpenaiToolCall, ToolChoice, ToolDefinition,
};
/// LLM 调用周期配置。
@@ -155,27 +160,18 @@ impl LlmCycle {
&mut self,
messages: Vec<OpenaiChatMessage>,
tools: Vec<ToolDefinition>,
) -> Result<ChatResponse, LlmError> {
let openai_tools: Option<Vec<OpenaiTool>> = if tools.is_empty() {
None
} else {
Some(
tools
.iter()
.map(|t| OpenaiTool::Function {
function: t.clone(),
})
.collect(),
)
};
) -> Result<MessageResponse, LlmError> {
let ir_messages: Vec<Message> =
messages.iter().map(chat_message_to_ir_message).collect();
let ir_tools: Vec<ToolDefinition> = tools.clone();
let request = ChatRequest {
let request = MessageRequest {
model: self.config.model.clone(),
messages,
messages: ir_messages,
tools: ir_tools,
tool_choice: ToolChoice::Auto,
max_tokens: self.config.max_tokens,
temperature: self.config.temperature,
tools: openai_tools,
tool_choice: Some(ToolChoice::Auto),
..Default::default()
};
@@ -198,11 +194,7 @@ impl LlmCycle {
match self.provider.chat(request).await {
Ok(response) => {
if let Some(ref executor) = self.hook_executor {
let post_request = ChatRequest {
model: self.config.model.clone(),
messages: vec![],
..Default::default()
};
let post_request = MessageRequest::default();
let ctx = HookContext::new(crate::llm::hooks::HookEvent::PostRequest)
.with_request(&post_request);
executor
@@ -229,7 +221,7 @@ impl LlmCycle {
&mut self,
prompt: String,
tools: Vec<ToolDefinition>,
) -> Result<ChatResponse, LlmError> {
) -> Result<MessageResponse, LlmError> {
self.messages.push(OpenaiChatMessage::user_text(prompt));
if let Some(ref config) = self.compact_config
@@ -273,7 +265,9 @@ impl LlmCycle {
.await;
}
self.messages.push(response.message.clone());
// ponytail: Phase 0 中内部消息存储仍为 Vec<OpenaiChatMessage>，
// 把 IR Message 转回旧 wire 格式以便存储。Phase 2 切换后移除。
self.messages.push(message_to_chat_message(&response.message));
self.usage.add(&response.usage);
return Ok(response);
@@ -346,59 +340,28 @@ impl LlmCycle {
}
}
let chunk_stream = self.provider.chat_stream(request).await?;
let ir_event_stream = self.provider.chat_stream(request).await?;
let hook_executor = self.hook_executor.clone();
let post_request = self.build_request(&tools);
// ponytail: Phase 0 中 Provider 临时桥接层把 IR 事件再回填成本 phase 的
// 信息冗余（OpenaiProvider 的 chat_stream 现在已输出 IR 事件）。当前实现
// 直接转发 IR 事件，仅在 `MessageComplete` 到达时收尾。
Ok(Box::pin(stream! {
use futures_util::StreamExt;
let mut chunk_stream = chunk_stream;
let mut ir_event_stream = ir_event_stream;
while let Some(result) = chunk_stream.next().await {
while let Some(result) = ir_event_stream.next().await {
match result {
Ok(chunk) => {
let mut assistant_text = String::new();
let mut tool_started: Option<(String, String, String)> = None;
for choice in &chunk.choices {
let delta = &choice.delta;
if let Some(content) = &delta.content {
assistant_text.push_str(content);
}
if let Some(tool_calls) = &delta.tool_calls
&& let Some(tc) = tool_calls.first()
{
let crate::llm::types::OpenaiToolCall::Function { id, function } = tc;
tool_started = Some((id.clone(), function.name.clone(), function.arguments.clone()));
}
Ok(event) => {
let is_terminal = matches!(event, StreamEvent::MessageComplete { .. });
let is_error = matches!(event, StreamEvent::Error { .. });
yield event;
if is_terminal {
break;
}
if !assistant_text.is_empty() {
yield StreamEvent::AssistantTextDelta { text: assistant_text };
}
if let Some((tool_call_id, tool_name, arguments)) = tool_started {
let args: serde_json::Value = serde_json::from_str(&arguments)
.unwrap_or(serde_json::Value::Null);
yield StreamEvent::ToolExecutionStarted {
tool_name,
input: args,
tool_call_id,
};
}
for choice in &chunk.choices {
if let Some(finish_reason) = &choice.finish_reason {
yield StreamEvent::TurnComplete {
reason: *finish_reason,
};
}
}
if let Some(usage_info) = &chunk.usage {
yield StreamEvent::CostUpdate { usage: *usage_info };
if is_error {
break;
}
}
Err(e) => {
@@ -429,37 +392,30 @@ impl LlmCycle {
}))
}
fn build_request(&self, tools: &[ToolDefinition]) -> ChatRequest {
let mut messages = self.messages.clone();
fn build_request(&self, tools: &[ToolDefinition]) -> MessageRequest {
let messages = self.messages.clone();
let ir_messages: Vec<Message> = messages
.iter()
.map(chat_message_to_ir_message)
.collect();
let mut ir_messages = ir_messages;
if let Some(sys_prompt) = &self.system_prompt
&& !messages
&& !ir_messages
.iter()
.any(|m| matches!(m, OpenaiChatMessage::System { .. }))
.any(|m| matches!(m, Message::System { .. }))
{
messages.insert(0, OpenaiChatMessage::system_text(sys_prompt));
ir_messages.insert(0, Message::system(sys_prompt.as_str()));
}
let openai_tools: Option<Vec<OpenaiTool>> = if tools.is_empty() {
None
} else {
Some(
tools
.iter()
.map(|t| OpenaiTool::Function {
function: t.clone(),
})
.collect(),
)
};
ChatRequest {
MessageRequest {
model: self.config.model.clone(),
messages,
messages: ir_messages,
tools: tools.to_vec(),
tool_choice: ToolChoice::Auto,
max_tokens: self.config.max_tokens,
temperature: self.config.temperature,
tools: openai_tools,
tool_choice: Some(ToolChoice::Auto),
..Default::default()
}
}
@@ -470,7 +426,7 @@ impl LlmCycle {
async fn submit_request(
&mut self,
tools: &[ToolDefinition],
) -> Result<ChatResponse, LlmError> {
) -> Result<MessageResponse, LlmError> {
let mut attempts = 0;
loop {
@@ -548,7 +504,7 @@ impl LlmCycle {
&mut self,
prompt: String,
registry: &crate::tools::ToolRegistry,
) -> Result<ChatResponse, LlmError> {
) -> Result<MessageResponse, LlmError> {
let tools = registry.definitions();
let max_turns = self.config.max_tool_turns.unwrap_or(10);
let tool_timeout = self.config.tool_timeout_secs;
@@ -570,24 +526,25 @@ impl LlmCycle {
let response = self.submit_request(&tools).await?;
// 判断是否需要执行工具
let should_execute = matches!(response.stop_reason, Some(FinishReason::ToolCalls))
&& has_tool_calls_in_message(&response.message);
let should_execute = matches!(response.stop_reason, StopReason::ToolUse)
&& has_tool_calls_in_response(&response);
// 将 Assistant 响应（含 tool_calls 或最终文本）追加到消息历史
self.messages.push(response.message.clone());
// ponytail: Phase 0 中内部消息存储仍为 Vec<OpenaiChatMessage>，
// 把 IR Message 转回旧 wire 格式以便存储。Phase 2 切换后移除。
self.messages.push(message_to_chat_message(&response.message));
if !should_execute {
return Ok(response);
}
// 解析 tool_calls 并执行
let tool_calls = extract_tool_calls_from_message(&response.message);
let tool_calls = extract_tool_calls_from_response(&response);
let calls: Vec<(String, serde_json::Value)> = tool_calls
.into_iter()
.map(|(_id, name, args)| {
let args: serde_json::Value =
let value: serde_json::Value =
serde_json::from_str(&args).unwrap_or(serde_json::Value::Null);
(name, args)
(name, value)
})
.collect();
@@ -642,37 +599,185 @@ impl LlmCycle {
}
/// 判断 Assistant 消息是否包含 tool_calls。
fn has_tool_calls_in_message(msg: &OpenaiChatMessage) -> bool {
matches!(
msg,
OpenaiChatMessage::Assistant {
tool_calls: Some(calls),
..
} if !calls.is_empty()
)
fn has_tool_calls_in_response(response: &MessageResponse) -> bool {
match &response.message {
Message::Assistant { content } => content
.iter()
.any(|b| matches!(b, ContentBlock::ToolUse { .. })),
_ => false,
}
}
/// 提取 Assistant 消息中的 tool_calls。
///
/// 返回 `(tool_call_id, tool_name, arguments_json_string)` 列表。
fn extract_tool_calls_from_message(
msg: &OpenaiChatMessage,
///
/// ponytail: 当前 Phase 0 实现，`arguments_json_string` 内含 JSON 序列化的 input。
/// 消费方在调用 `registry.invoke_all()` 时反序列化一次。该小段冗余序列化
/// 在 Phase 2 切换为 `Vec<Message>` 后可整体消除。
fn extract_tool_calls_from_response(
response: &MessageResponse,
) -> Vec<(String, String, String)> {
if let OpenaiChatMessage::Assistant {
tool_calls: Some(calls),
..
} = msg
{
calls
.iter()
.map(|c| match c {
OpenaiToolCall::Function { id, function } => {
(id.clone(), function.name.clone(), function.arguments.clone())
let mut out = Vec::new();
if let Message::Assistant { content } = &response.message {
for block in content {
if let ContentBlock::ToolUse { id, name, input } = block {
let args = serde_json::to_string(input).unwrap_or_else(|_| "null".to_string());
out.push((id.clone(), name.clone(), args));
}
}
}
out
}
/// 把 `OpenaiChatMessage` 转换为 IR `Message`。
///
/// ponytail: Phase 0 转换层 —— build_request 中需要把存储用的
/// `Vec<OpenaiChatMessage>` 转为 `MessageRequest.messages`。Phase 2 中
/// `self.messages` 直接改为 `Vec<Message>` 时此函数整体删除。
fn chat_message_to_ir_message(msg: &OpenaiChatMessage) -> Message {
match msg {
OpenaiChatMessage::Developer { content, .. }
| OpenaiChatMessage::System { content, .. } => Message::System {
content: content_field_to_blocks(content),
},
OpenaiChatMessage::User { content, .. } => Message::User {
content: content_field_to_blocks(content),
},
OpenaiChatMessage::Assistant {
content,
tool_calls,
..
} => {
let mut blocks = content_field_to_blocks(content);
if let Some(calls) = tool_calls {
for call in calls {
if let OpenaiToolCall::Function { id, function } = call {
let input: serde_json::Value = serde_json::from_str(&function.arguments)
.unwrap_or(serde_json::Value::Null);
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_field_to_blocks(content),
is_error: false,
},
OpenaiChatMessage::Function { content, name } => Message::ToolResult {
tool_call_id: name.clone(),
content: content_field_to_blocks(content),
is_error: false,
},
}
}
/// 把 IR `Message` 转换为 `OpenaiChatMessage`（用于 `self.messages` 持久化）。
///
/// ponytail: Phase 0 转换层 —— submit 后 `response.message: Message` 需要
/// 转回 `OpenaiChatMessage` 才能追加到旧 store。Phase 2 中移除。
fn message_to_chat_message(msg: &Message) -> OpenaiChatMessage {
match msg {
Message::System { content } => OpenaiChatMessage::System {
content: blocks_to_content_field(content),
name: None,
},
Message::User { content } => OpenaiChatMessage::User {
content: blocks_to_content_field(content),
name: None,
},
// ponytail: UserImage 当前无 OpenAI 对应，回退为空的 User 消息（图片数据丢失）。
// Phase 2 切换后此函数整体移除。
Message::UserImage { .. } => OpenaiChatMessage::User {
content: ContentField::Array(vec![]),
name: None,
},
Message::Assistant { content } => {
let mut text_blocks: Vec<String> = Vec::new();
let mut tool_call_blocks: Vec<OpenaiToolCall> = Vec::new();
for block in content {
match block {
ContentBlock::Text { text } => text_blocks.push(text.clone()),
ContentBlock::ToolUse { id, name, input } => {
tool_call_blocks.push(OpenaiToolCall::Function {
id: id.clone(),
function: FunctionCall {
name: name.clone(),
arguments: serde_json::to_string(input)
.unwrap_or_else(|_| "null".to_string()),
},
});
}
// ponytail: thinking/refusal/image/audio/file 在 Phase 0 持久化时被丢弃。
// Phase 2 切换后此函数整体移除，自然修复。
_ => {}
}
}
let text: String = text_blocks.into_iter().collect();
OpenaiChatMessage::Assistant {
content: if text.is_empty() {
ContentField::Array(vec![])
} else {
ContentField::String(text)
},
refusal: None,
name: None,
tool_calls: if tool_call_blocks.is_empty() {
None
} else {
Some(tool_call_blocks)
},
}
}
Message::ToolResult {
tool_call_id,
content,
is_error: _,
} => OpenaiChatMessage::Tool {
content: blocks_to_content_field(content),
tool_call_id: tool_call_id.clone(),
},
}
}
fn blocks_to_content_field(blocks: &[ContentBlock]) -> ContentField {
if let [ContentBlock::Text { text }] = blocks {
return ContentField::String(text.clone());
}
let parts: Vec<OpenaiContentPart> = blocks
.iter()
.filter_map(|b| match b {
ContentBlock::Text { text } => Some(OpenaiContentPart::Text { text: text.clone() }),
_ => None,
})
.collect();
ContentField::Array(parts)
}
fn content_field_to_blocks(field: &ContentField) -> Vec<ContentBlock> {
match field {
ContentField::String(s) => vec![ContentBlock::Text { text: s.clone() }],
ContentField::Array(parts) => parts
.iter()
.filter_map(|p| match p {
OpenaiContentPart::Text { text } => {
Some(ContentBlock::Text { text: text.clone() })
}
OpenaiContentPart::Refusal { refusal } => {
Some(ContentBlock::Text { text: refusal.clone() })
}
_ => None,
})
.collect()
} else {
Vec::new()
.collect(),
}
}
@@ -691,19 +796,20 @@ fn truncate_tool_result(s: &str, max_bytes: usize) -> String {
#[cfg(test)]
mod tests {
use super::*;
use crate::llm::types::{ContentField, OpenaiContentPart};
use crate::tools::{BaseTool, ToolRegistry};
use async_trait::async_trait;
use futures_core::Stream;
use serde_json::{json, Value};
use std::pin::Pin;
/// 模拟 Provider —— 预定义响应序列，按调用顺序返回。
struct MockProvider {
responses: std::sync::Mutex<Vec<ChatResponse>>,
responses: std::sync::Mutex<Vec<MessageResponse>>,
call_count: std::sync::Mutex<u32>,
}
impl MockProvider {
fn new(responses: Vec<ChatResponse>) -> Self {
fn new(responses: Vec<MessageResponse>) -> Self {
Self {
responses: std::sync::Mutex::new(responses),
call_count: std::sync::Mutex::new(0),
@@ -713,7 +819,7 @@ mod tests {
#[async_trait]
impl LlmProvider for MockProvider {
async fn chat(&self, _request: ChatRequest) -> Result<ChatResponse, LlmError> {
async fn chat(&self, _request: MessageRequest) -> Result<MessageResponse, LlmError> {
let mut count = self.call_count.lock().unwrap();
*count += 1;
let mut responses = self.responses.lock().unwrap();
@@ -722,45 +828,61 @@ mod tests {
}
Ok(responses.remove(0))
}
async fn chat_stream(
&self,
_request: MessageRequest,
) -> Result<Pin<Box<dyn Stream<Item = Result<StreamEvent, LlmError>> + Send>>, LlmError>
{
unimplemented!()
}
fn capabilities(&self) -> ProviderCapabilities {
ProviderCapabilities {
provider_name: "mock",
supported_models: None,
features: ProviderFeatures::default(),
}
}
}
fn empty_usage() -> crate::llm::types::Usage {
crate::llm::types::Usage::default()
}
fn assistant_text_response(text: &str) -> ChatResponse {
ChatResponse {
message: OpenaiChatMessage::assistant_text(text),
fn assistant_text_response(text: &str) -> MessageResponse {
MessageResponse {
id: String::new(),
model: String::new(),
message: Message::Assistant {
content: vec![ContentBlock::Text {
text: text.into(),
}],
},
usage: empty_usage(),
stop_reason: Some(FinishReason::Stop),
stop_reason: StopReason::Stop,
extra: std::collections::HashMap::new(),
}
}
fn assistant_tool_call_response(
calls: Vec<(&str, &str, &str)>,
) -> ChatResponse {
use crate::llm::types::{OpenaiToolCall, FunctionCall};
let tool_calls: Vec<OpenaiToolCall> = calls
fn assistant_tool_call_response(calls: Vec<(&str, &str, &str)>) -> MessageResponse {
let tool_blocks: Vec<ContentBlock> = calls
.into_iter()
.map(|(id, name, args)| OpenaiToolCall::Function {
id: id.to_string(),
function: FunctionCall {
.map(|(id, name, args)| {
let input: serde_json::Value =
serde_json::from_str(args).unwrap_or(serde_json::Value::Null);
ContentBlock::ToolUse {
id: id.to_string(),
name: name.to_string(),
arguments: args.to_string(),
},
input,
}
})
.collect();
ChatResponse {
message: OpenaiChatMessage::Assistant {
content: ContentField::Array(vec![OpenaiContentPart::Text {
text: String::new(),
}]),
refusal: None,
name: None,
tool_calls: Some(tool_calls),
},
MessageResponse {
id: String::new(),
model: String::new(),
message: Message::Assistant { content: tool_blocks },
usage: empty_usage(),
stop_reason: Some(FinishReason::ToolCalls),
stop_reason: StopReason::ToolUse,
extra: std::collections::HashMap::new(),
}
}
@@ -790,7 +912,6 @@ mod tests {
#[tokio::test]
async fn test_submit_with_tools_single_turn() {
// 第一轮：返回 tool_call；第二轮：返回最终文本
let responses = vec![
assistant_tool_call_response(vec![("call_1", "add", r#"{"a":1,"b":2}"#)]),
assistant_text_response("答案是 3"),
@@ -805,13 +926,11 @@ mod tests {
.submit_with_tools("1+2=?".to_string(), &registry)
.await
.unwrap();
// 验证最终响应是文本响应
assert!(matches!(
response.message,
OpenaiChatMessage::Assistant { .. }
));
// 最终响应是 Assistant 消息
assert!(matches!(response.message, Message::Assistant { .. }));
// 验证消息历史：user, assistant(tool_calls), tool, assistant(text)
// 验证消息历史（Vec<OpenaiChatMessage>，Phase 2 才切到 Vec<Message>）：
// user, assistant(tool_calls → 转回后含 tool_use), tool, assistant(text)
let messages = cycle.messages();
assert_eq!(messages.len(), 4);
assert!(matches!(messages[0], OpenaiChatMessage::User { .. }));
@@ -822,7 +941,6 @@ mod tests {
#[tokio::test]
async fn test_submit_with_tools_multi_turn() {
// 3 轮 tool 调用后给出最终答案
let responses = vec![
assistant_tool_call_response(vec![("call_1", "add", r#"{"a":1,"b":2}"#)]),
assistant_tool_call_response(vec![("call_2", "add", r#"{"a":3,"b":4}"#)]),
@@ -839,10 +957,7 @@ mod tests {
.submit_with_tools("计算总和".to_string(), &registry)
.await
.unwrap();
assert!(matches!(
response.message,
OpenaiChatMessage::Assistant { .. }
));
assert!(matches!(response.message, Message::Assistant { .. }));
// user + 3*(assistant + tool) + final assistant = 8
let messages = cycle.messages();
@@ -851,10 +966,8 @@ mod tests {
#[tokio::test]
async fn test_submit_with_tools_max_turns_exceeded() {
// 配置 max_tool_turns = 2
let mut config = CycleConfig::default();
config.max_tool_turns = Some(2);
// 4 轮 tool 调用 + 终止
let responses = vec![
assistant_tool_call_response(vec![("c1", "add", r#"{"a":1,"b":1}"#)]),
assistant_tool_call_response(vec![("c2", "add", r#"{"a":1,"b":1}"#)]),
@@ -867,15 +980,12 @@ mod tests {
let mut registry = ToolRegistry::new();
registry.register(std::sync::Arc::new(AddTool)).unwrap();
let result = cycle
.submit_with_tools("test".to_string(), &registry)
.await;
let result = cycle.submit_with_tools("test".to_string(), &registry).await;
assert!(matches!(result, Err(LlmError::Other(msg)) if msg.contains("达到最大工具循环轮次")));
}
#[tokio::test]
async fn test_submit_with_tools_no_tool_call_response() {
// LLM 直接给出最终响应（不调用工具）
let responses = vec![assistant_text_response("直接回答")];
let provider = Box::new(MockProvider::new(responses));
let mut cycle = LlmCycle::new(provider, CycleConfig::default());
@@ -887,10 +997,7 @@ mod tests {
.submit_with_tools("直接回答".to_string(), &registry)
.await
.unwrap();
assert!(matches!(
response.message,
OpenaiChatMessage::Assistant { .. }
));
assert!(matches!(response.message, Message::Assistant { .. }));
}
#[test]
@@ -911,7 +1018,6 @@ mod tests {
fn test_truncate_tool_result_chinese_chars() {
let s = "中".repeat(100);
let truncated = truncate_tool_result(&s, 50);
// 不会在字符中间截断
assert!(truncated.starts_with("中"));
}
}
src/llm/hooks.rs
+3 -3
View File
@@ -3,7 +3,7 @@
use async_trait::async_trait;
use crate::llm::error::LlmError;
use crate::llm::types::ChatRequest;
use crate::llm::types::request_v2::MessageRequest;
/// 生命周期钩子事件点。
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -30,7 +30,7 @@ pub struct HookContext<'a> {
/// 当前事件点。
pub event: HookEvent,
/// 当前的请求（部分事件点可用）。
pub request: Option<&'a ChatRequest>,
pub request: Option<&'a MessageRequest>,
/// 当前错误（仅 OnError 和 OnRetry 可用）。
pub error: Option<&'a LlmError>,
/// 当前重试次数（从 1 开始，仅 OnRetry 可用）。
@@ -53,7 +53,7 @@ impl<'a> HookContext<'a> {
}
}
pub(crate) fn with_request(mut self, request: &'a ChatRequest) -> Self {
pub(crate) fn with_request(mut self, request: &'a MessageRequest) -> Self {
self.request = Some(request);
self
}
src/llm/provider.rs
+80 -24
View File
@@ -3,16 +3,29 @@ pub mod registry;
use std::pin::Pin;
use tokio_stream::Stream;
use futures_core::Stream;
use serde::{Deserialize, Serialize};
use crate::llm::error::LlmError;
use crate::llm::types::{ChatRequest, ChatResponse, OpenaiChatChunk};
use async_trait::async_trait;
use crate::llm::types::request_v2::MessageRequest;
use crate::llm::types::response_v2::{MessageResponse, StreamEvent};
/// Provider 类型枚举 —— `create_provider()` 在编译期 exhaustive match 中使用。
///
/// 设计依据（见 `docs/10-llm-provider-refinement.md` §2.5 Decision-05）：
/// 当前协议数量（5 种以内）完全可控，enum 的编译期安全检查优于运行时的 `HashMap::get()`。
/// 未来如果扩展到 15+ 种以上，再改为注册表模式。
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ProviderType {
OpenAI,
/// OpenAI Chat Completions API（兼容 DeepSeek / Qwen 等 `/chat/completions` 端点）。
OpenaiChat,
/// OpenAI Response API。
OpenaiResponse,
/// Anthropic Messages API。
Anthropic,
/// DeepSeek（OpenAI-compatible `/chat/completions`）。
DeepSeek,
/// Qwen / 阿里云百炼（OpenAI-compatible `/chat/completions`）。
Qwen,
}
@@ -21,60 +34,103 @@ impl std::str::FromStr for ProviderType {
fn from_str(s: &str) -> Result<Self, Self::Err> {
match s.to_lowercase().as_str() {
"openai" => Ok(ProviderType::OpenAI),
"openai" | "openai-chat" | "openai_chat" => Ok(ProviderType::OpenaiChat),
"openai-response" | "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)),
_ => Err(format!("未知的 Provider 类型: {s}")),
}
}
}
/// Provider 构造参数 —— 通用 base_url + api_key + model。
pub struct ProviderConfig {
pub base_url: String,
pub api_key: String,
pub model: String,
}
/// Provider 工厂 —— exhaustive match 在编译期保证新 Provider 被注册。
pub fn create_provider(
provider_type: ProviderType,
config: ProviderConfig,
) -> Result<Box<dyn LlmProvider>, LlmError> {
match provider_type {
ProviderType::OpenAI => Ok(Box::new(openai::OpenaiProvider::new(
ProviderType::OpenaiChat => Ok(Box::new(openai::OpenaiProvider::new(
config.base_url,
config.api_key,
config.model,
))),
ProviderType::OpenaiResponse => {
unimplemented!("OpenAI Response Provider 将在 Phase 1 引入")
}
ProviderType::Anthropic => {
unimplemented!("Anthropic Provider 将在 Phase 1 引入")
}
ProviderType::DeepSeek => {
unimplemented!("DeepSeek Provider 尚未实现")
unimplemented!("DeepSeek Provider 将在 Phase 1 引入")
}
ProviderType::Qwen => {
unimplemented!("Qwen Provider 尚未实现")
unimplemented!("Qwen Provider 将在 Phase 1 引入")
}
}
}
/// Provider 能力描述 —— 静态元信息，调用方据此决定可用特性。
///
/// 设计依据（见 `docs/10-llm-provider-refinement.md` §4 任务 6 决策）：
/// `ProviderCapabilities` 与 trait 同文件（`provider.rs`），不分散到类型目录。
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ProviderCapabilities {
/// 人类可读的 Provider 名（如 `"openai"` / `"anthropic"`）。
pub provider_name: &'static str,
/// 支持的模型列表（`None` 表示"未列举全部"）。
pub supported_models: Option<Vec<String>>,
/// 详细功能开关。
pub features: ProviderFeatures,
}
/// Provider 功能开关集合。
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct ProviderFeatures {
/// 是否支持流式响应。
pub streaming: bool,
/// 是否支持 thinking / 推理。
pub thinking: bool,
/// 是否支持图片输入。
pub vision: bool,
/// 是否支持音频输入。
pub audio_input: bool,
/// 是否支持工具调用。
pub tool_use: bool,
/// 是否支持并行工具调用。
pub parallel_tool_calls: bool,
/// system prompt 是否放在 messages 中（`true`）还是顶层 `system` 字段（`false`）。
pub system_prompt_in_messages: bool,
/// 模型上下文窗口（tokens）；`0` 表示未知。
pub max_context_window: u32,
}
/// LLM Provider 抽象接口。
///
/// 所有具体的 LLM 后端实现（OpenAI、Anthropic、Azure 等）
/// 所有具体的 LLM 后端实现（OpenAI、Anthropic、DeepSeek、Qwen 等）
/// 均需实现此 trait，以实现可插拔替换。
#[async_trait]
///
/// 修订（Phase 0）：签名由 `chat(ChatRequest) → ChatResponse` 切换为
/// `chat(MessageRequest) → MessageResponse`，`chat_stream` 返回新 `StreamEvent` 流，
/// 新增 `capabilities()` 方法。
#[async_trait::async_trait]
pub trait LlmProvider: Send + Sync {
/// 发送聊天请求并返回完整响应。
async fn chat(&self, request: ChatRequest) -> Result<ChatResponse, LlmError>;
async fn chat(&self, request: MessageRequest) -> Result<MessageResponse, LlmError>;
/// 流式聊天请求 —— 返回原始 SSE chunk 流。
///
/// 默认实现回退到非流式调用（包装为单元素流）。
/// 流式聊天请求 —— 返回新 IR `StreamEvent` 流。
async fn chat_stream(
&self,
request: ChatRequest,
) -> Result<
Pin<Box<dyn Stream<Item = Result<OpenaiChatChunk, LlmError>> + Send>>,
LlmError,
> {
let response = self.chat(request).await?;
let chunk = OpenaiChatChunk::from(response);
Ok(Box::pin(tokio_stream::once(Ok(chunk))))
}
request: MessageRequest,
) -> Result<Pin<Box<dyn Stream<Item = Result<StreamEvent, LlmError>> + Send>>, LlmError>;
/// 返回 Provider 静态能力描述。
fn capabilities(&self) -> ProviderCapabilities;
}
src/llm/provider/openai.rs
+323 -13
View File
@@ -6,22 +6,32 @@ use bytes::Bytes;
use futures_core::stream::Stream;
use futures_util::StreamExt;
use reqwest::Client;
use serde_json::Value;
use tracing::{debug, error, info};
use super::LlmProvider;
use super::{LlmProvider, ProviderCapabilities, ProviderFeatures};
use crate::llm::error::LlmError;
use crate::llm::stream::parse_chunk_stream;
use crate::llm::types::message::{ContentBlock, Message};
use crate::llm::types::request::{OpenaiChatRequest, StreamOptions};
use crate::llm::types::request_v2::MessageRequest;
use crate::llm::types::response_v2::{MessageResponse, StopReason, StreamEvent};
use crate::llm::types::usage::Usage;
use crate::llm::types::{
ChatRequest, ChatResponse, OpenaiChatChunk, OpenaiChatResponse, StreamOptions,
ChatResponse, ContentField, FinishReason, FunctionCall, OpenaiChatChunk, OpenaiChatMessage,
OpenaiChatResponse, OpenaiContentPart, OpenaiTool, OpenaiToolCall, OpenaiToolDefinition,
StopSequence, ToolChoice,
};
pub struct OpenaiProvider {
http_client: Client,
base_url: String,
api_key: String,
model: String,
}
impl OpenaiProvider {
pub fn new(base_url: String, api_key: String, _model: String) -> Self {
pub fn new(base_url: String, api_key: String, model: String) -> Self {
let http_client = Client::builder()
.timeout(Duration::from_secs(120))
.build()
@@ -31,6 +41,7 @@ impl OpenaiProvider {
http_client,
base_url,
api_key,
model,
}
}
@@ -50,11 +61,63 @@ impl OpenaiProvider {
LlmError::Other(format!("请求失败: {}", e))
}
}
}
#[async_trait]
impl LlmProvider for OpenaiProvider {
async fn chat(&self, request: ChatRequest) -> Result<ChatResponse, LlmError> {
fn map_stop_reason(r: Option<FinishReason>) -> StopReason {
r.map(StopReason::from).unwrap_or(StopReason::Stop)
}
/// ponytail: Phase 0 临时桥接 —— 把 IR `MessageRequest` 转换为 OpenAI wire 格式。
fn convert_to_chat_request(&self, request: MessageRequest) -> OpenaiChatRequest {
let messages = request.messages.iter().map(message_to_chat_message).collect();
let tools = if request.tools.is_empty() {
None
} else {
Some(
request
.tools
.into_iter()
.map(|t| OpenaiTool::Function { function: t })
.collect(),
)
};
OpenaiChatRequest {
model: request.model,
messages,
max_tokens: request.max_tokens,
temperature: request.temperature,
top_p: request.top_p,
stop: if request.stop_sequences.is_empty() {
None
} else if request.stop_sequences.len() == 1 {
Some(StopSequence::Single(request.stop_sequences[0].clone()))
} else {
Some(StopSequence::Multiple(request.stop_sequences))
},
tools,
tool_choice: Some(request.tool_choice),
stream: Some(request.stream),
..Default::default()
}
}
/// ponytail: Phase 0 临时桥接 —— 把 OpenAI `ChatResponse` 转换为 IR `MessageResponse`。
fn convert_to_message_response(&self, response: ChatResponse) -> MessageResponse {
let mut extra = std::collections::HashMap::new();
let stop_reason = Self::map_stop_reason(response.stop_reason);
MessageResponse {
id: String::new(),
model: self.model.clone(),
message: chat_message_to_message(&response.message),
usage: response.usage,
stop_reason,
extra,
}
}
/// ponytail: Phase 0 临时桥接 —— 调用 OpenAI `/chat/completions` 拿到旧 wire 格式响应。
async fn chat_inner(&self, request: OpenaiChatRequest) -> Result<ChatResponse, LlmError> {
let url = format!("{}/chat/completions", self.base_url.trim_end_matches('/'));
info!(model = %request.model, max_tokens = request.max_tokens, temperature = request.temperature, "发送 LLM 请求");
@@ -118,9 +181,10 @@ impl LlmProvider for OpenaiProvider {
Ok(ChatResponse::from(chat_response))
}
async fn chat_stream(
/// ponytail: Phase 0 临时桥接 —— 流式调用 `/chat/completions` 返回 chunk 流。
async fn chat_stream_inner(
&self,
mut request: ChatRequest,
mut request: OpenaiChatRequest,
) -> Result<
Pin<Box<dyn Stream<Item = Result<OpenaiChatChunk, LlmError>> + Send>>,
LlmError,
@@ -159,14 +223,251 @@ impl LlmProvider for OpenaiProvider {
});
}
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)))
});
Ok(Box::pin(SseChunkStream::new(byte_stream)))
}
}
#[async_trait]
impl LlmProvider for OpenaiProvider {
// ponytail: Phase 0 临时桥接，Phase 1 重写时移除。
async fn chat(&self, request: MessageRequest) -> Result<MessageResponse, LlmError> {
let chat_req = self.convert_to_chat_request(request);
let chat_resp = self.chat_inner(chat_req).await?;
Ok(self.convert_to_message_response(chat_resp))
}
// ponytail: Phase 0 临时桥接，Phase 1 重写时移除。
async fn chat_stream(
&self,
request: MessageRequest,
) -> Result<Pin<Box<dyn Stream<Item = Result<StreamEvent, LlmError>> + Send>>, LlmError>
{
let chat_req = self.convert_to_chat_request(request);
let chunk_stream = self.chat_stream_inner(chat_req).await?;
// ponytail: 复用旧 parse_chunk_stream 把 chunk 流映射为 IR StreamEvent 流。
// Phase 1 中 OpenAI Provider 直接产出 IR 事件后整体删除此调用。
Ok(parse_chunk_stream(chunk_stream))
}
fn capabilities(&self) -> ProviderCapabilities {
ProviderCapabilities {
provider_name: "openai",
supported_models: None,
features: ProviderFeatures::default(),
}
}
}
// ===== 转换函数 =====
/// IR `Message` → OpenAI `OpenaiChatMessage`。
///
/// ponytail: Phase 0 临时转换层；Phase 2 中 LlmCycle 切换为 `Vec<Message>` 后移除。
fn message_to_chat_message(msg: &Message) -> OpenaiChatMessage {
match msg {
Message::System { content } => OpenaiChatMessage::System {
content: blocks_to_content_field(content),
name: None,
},
Message::User { content } => OpenaiChatMessage::User {
content: blocks_to_content_field(content),
name: None,
},
// ponytail: UserImage 当前无对应 OpenAI 形态，回退为空的 User 消息。
// Phase 2 切到 Vec<Message> 后此函数整体删除。
Message::UserImage { data: _, mime_type: _, detail: _ } => OpenaiChatMessage::User {
content: ContentField::Array(vec![]),
name: None,
},
Message::Assistant { content } => {
let mut text_blocks = Vec::new();
let mut tool_call_blocks = Vec::new();
for block in content {
match block {
ContentBlock::Text { text } => text_blocks.push(text.clone()),
ContentBlock::ToolUse { id, name, input } => {
tool_call_blocks.push(OpenaiToolCall::Function {
id: id.clone(),
function: FunctionCall {
name: name.clone(),
arguments: serde_json::to_string(input).unwrap_or_default(),
},
});
}
// ponytail: thinking/refusal/image/audio/file 在回传时被截断，
// Phase 2 切到 Vec<Message> 整体修复。
_ => {}
}
}
let text: String = text_blocks.into_iter().collect();
OpenaiChatMessage::Assistant {
content: if text.is_empty() {
ContentField::Array(vec![])
} else {
ContentField::String(text)
},
refusal: None,
name: None,
tool_calls: if tool_call_blocks.is_empty() {
None
} else {
Some(tool_call_blocks)
},
}
}
Message::ToolResult {
tool_call_id,
content,
is_error: _,
} => OpenaiChatMessage::Tool {
content: blocks_to_content_field(content),
tool_call_id: tool_call_id.clone(),
},
}
}
/// OpenAI `OpenaiChatMessage` → IR `Message`。
fn chat_message_to_message(msg: &OpenaiChatMessage) -> Message {
match msg {
OpenaiChatMessage::Developer { content, .. }
| OpenaiChatMessage::System { content, .. } => Message::System {
content: content_field_to_blocks(content),
},
OpenaiChatMessage::User { content, .. } => Message::User {
content: content_field_to_blocks(content),
},
OpenaiChatMessage::Assistant {
content,
tool_calls,
..
} => {
let mut blocks = content_field_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(Value::Null);
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_field_to_blocks(content),
is_error: false,
},
// ponytail: OpenAI 已废弃 Function 消息变体；映射为带 name 假 tool_call_id 的 ToolResult。
OpenaiChatMessage::Function { content, name } => Message::ToolResult {
tool_call_id: name.clone(),
content: content_field_to_blocks(content),
is_error: false,
},
}
}
fn blocks_to_content_field(blocks: &[ContentBlock]) -> ContentField {
if blocks.len() == 1
&& let ContentBlock::Text { text } = &blocks[0]
{
return ContentField::String(text.clone());
}
let parts: Vec<OpenaiContentPart> = blocks
.iter()
.filter_map(|b| match b {
ContentBlock::Text { text } => Some(OpenaiContentPart::Text { text: text.clone() }),
ContentBlock::Image { source } => Some(OpenaiContentPart::Image {
image_url: crate::llm::types::openai_message::ImageURL {
url: if source.is_url {
source.data.clone()
} else {
format!("data:{};base64,{}", source.mime_type, source.data)
},
detail: Some(match source.mime_type.as_str() {
"image/jpeg" | "image/png" | "image/webp" => {
crate::llm::types::shared::ImageDetail::Auto
}
_ => crate::llm::types::shared::ImageDetail::Auto,
}),
},
detail: None,
}),
ContentBlock::Audio { source } => Some(OpenaiContentPart::InputAudio {
input_audio: crate::llm::types::openai_message::InputAudio {
data: source.data.clone(),
format: source.format,
},
}),
// ponytail: File/Thinking/Extension/ToolUse/ToolResult 在回传时被截断。
_ => None,
})
.collect();
if parts.is_empty() {
ContentField::Array(vec![])
} else {
ContentField::Array(parts)
}
}
fn content_field_to_blocks(field: &ContentField) -> Vec<ContentBlock> {
match field {
ContentField::String(s) => vec![ContentBlock::Text { text: s.clone() }],
ContentField::Array(parts) => parts
.iter()
.filter_map(|p| match p {
OpenaiContentPart::Text { text } => Some(ContentBlock::Text { text: text.clone() }),
OpenaiContentPart::Refusal { refusal } => {
Some(ContentBlock::Text { text: refusal.clone() })
}
OpenaiContentPart::Image { image_url, .. } => {
let url = &image_url.url;
if let Some(rest) = url.strip_prefix("data:") {
if let Some((mime, b64)) = rest.split_once(";base64,") {
return Some(ContentBlock::Image {
source: crate::llm::types::message::ImageSource {
data: b64.to_string(),
mime_type: mime.to_string(),
is_url: false,
},
});
}
}
Some(ContentBlock::Image {
source: crate::llm::types::message::ImageSource {
data: url.clone(),
mime_type: image_url
.detail
.map(|_| "image/url".to_string())
.unwrap_or_else(|| "image/url".to_string()),
is_url: true,
},
})
}
OpenaiContentPart::InputAudio { input_audio } => Some(ContentBlock::Audio {
source: crate::llm::types::message::AudioSource {
data: input_audio.data.clone(),
format: input_audio.format,
},
}),
OpenaiContentPart::File { .. } => None,
})
.collect(),
}
}
struct SseChunkStream<S> {
inner: S,
buffer: String,
@@ -184,7 +485,10 @@ impl<S: Stream<Item = Result<Bytes, LlmError>> + Unpin> SseChunkStream<S> {
impl<S: Stream<Item = Result<Bytes, LlmError>> + Unpin> Stream for SseChunkStream<S> {
type Item = Result<OpenaiChatChunk, LlmError>;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> std::task::Poll<Option<Self::Item>> {
fn poll_next(
mut self: Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
) -> std::task::Poll<Option<Self::Item>> {
loop {
if let Some(pos) = self.buffer.find("\n") {
let line = self.buffer.drain(..pos + 1).collect::<String>();
@@ -233,3 +537,9 @@ impl<S: Stream<Item = Result<Bytes, LlmError>> + Unpin> Stream for SseChunkStrea
}
}
}
// Suppress unused-import warnings for symbols kept for Phase 1 parity.
#[allow(dead_code)]
fn _phase0_reexports() {
let _ = Usage::default;
}
src/llm/stream.rs
+140 -44
View File
@@ -1,4 +1,14 @@
//! 流式事件系统 —— 将 LLM 流式响应解析为语义化事件。
//!
//! Phase 0 修订（参见 `docs/10a-phase0-types-and-trait.md` §"StreamEvent 命名冲突处理"）：
//! - 对外暴露的 `StreamEvent` 是高精度 IR 版本（来自 `response_v2::StreamEvent`）。
//! - 旧变体（`AssistantTextDelta` / `ToolExecutionStarted` 等）重命名为 `LegacyStreamEvent`
//! 放在 `crate::llm::types::old_stream` 模块，本文件内部消费。
//! - Phase 1 重写 Provider 时可直接消费新事件流后整体删除 `LegacyStreamEvent` 相关代码。
//!
//! 当前实现：旧的 `parse_chunk_stream` 内部消费 `OpenaiChatChunk`，映射为
//! `LegacyStreamEvent`，再在 `LegacyToIrEventStream` 中映射为新 IR `StreamEvent`
//! 后输出。Phase 1 会重写此层（OpenAI Provider 直接产出新事件流）。
use std::pin::Pin;
use std::task::{Context, Poll};
@@ -9,54 +19,47 @@ use futures_util::FutureExt;
use serde_json::Value;
use crate::llm::error::LlmError;
use crate::llm::types::{FinishReason, OpenaiChatChunk, OpenaiToolCall, Usage};
use crate::llm::types::old_stream::LegacyStreamEvent;
use crate::llm::types::response_v2::MessageResponse;
use crate::llm::types::response_v2::StopReason;
use crate::llm::types::usage::Usage;
use crate::llm::types::{FinishReason, OpenaiChatChunk, OpenaiToolCall};
/// 流式事件 —— LLM 调用全生命周期的语义化事件。
#[derive(Debug, Clone)]
pub enum StreamEvent {
/// 助手回复文本增量。
AssistantTextDelta { text: String },
/// 工具调用开始。
ToolExecutionStarted {
tool_name: String,
input: Value,
tool_call_id: String,
},
/// 工具调用完成。
ToolExecutionCompleted {
tool_name: String,
output: Value,
is_error: bool,
},
/// Token 用量更新。
CostUpdate { usage: Usage },
/// 一轮会话完成。
TurnComplete { reason: FinishReason },
/// 错误事件。
Error { message: String },
}
// 唯一的对外 `StreamEvent` 定义（高精度 IR 事件，来自 `response_v2`）。
//
// 此 `pub use` 同时起到两个作用：
// 1. 让 `crate::llm::stream::StreamEvent` 路径仍指向新高精度 IR 事件，
// 保持与既有 `use crate::llm::stream::StreamEvent` 的代码兼容；
// 2. 把模块内部的 `StreamEvent` 名字指向 `response_v2::StreamEvent`。
pub use crate::llm::types::response_v2::StreamEvent;
impl StreamEvent {
fn error(message: impl Into<String>) -> Self {
Self::Error {
message: message.into(),
}
}
}
/// 将原始 OpenaiChatChunk 流解析为 StreamEvent 流。
/// 将原始 OpenaiChatChunk 流解析为新高精度 IR StreamEvent 流。
///
/// ponytail: 每个产出事件都用 `Result<_, LlmError>` 包装，让上层 `chat_stream`
/// trait 方法直接消费并保持错误传播链。当前 `LegacyToIrEventStream` 内部
/// 不会产生错误，所有结果都是 `Ok`；后续 Phase 1 重写 Provider 时，
/// 真实 IR 流转换可在此层注入 error 事件。
pub fn parse_chunk_stream(
chunks: Pin<Box<dyn futures_core::Stream<Item = Result<OpenaiChatChunk, LlmError>> + Send>>,
) -> Pin<Box<dyn futures_core::Stream<Item = StreamEvent> + Send>> {
Box::pin(ChunkToEventStream { chunks })
) -> Pin<Box<dyn futures_core::Stream<Item = Result<StreamEvent, LlmError>> + Send>> {
let legacy = parse_chunk_stream_legacy(chunks);
Box::pin(LegacyToIrEventStream { inner: legacy })
}
struct ChunkToEventStream {
// --- 内部：chunk → LegacyStreamEvent ---
fn parse_chunk_stream_legacy(
chunks: Pin<Box<dyn futures_core::Stream<Item = Result<OpenaiChatChunk, LlmError>> + Send>>,
) -> Pin<Box<dyn futures_core::Stream<Item = LegacyStreamEvent> + Send>> {
Box::pin(ChunkToLegacyEventStream { chunks })
}
struct ChunkToLegacyEventStream {
chunks: Pin<Box<dyn futures_core::Stream<Item = Result<OpenaiChatChunk, LlmError>> + Send>>,
}
impl Stream for ChunkToEventStream {
type Item = StreamEvent;
impl Stream for ChunkToLegacyEventStream {
type Item = LegacyStreamEvent;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
let this = &mut *self;
@@ -66,7 +69,7 @@ impl Stream for ChunkToEventStream {
let delta = &choice.delta;
if let Some(content) = &delta.content {
return Poll::Ready(Some(StreamEvent::AssistantTextDelta {
return Poll::Ready(Some(LegacyStreamEvent::AssistantTextDelta {
text: content.clone(),
}));
}
@@ -77,7 +80,7 @@ impl Stream for ChunkToEventStream {
let OpenaiToolCall::Function { id, function } = tc;
let args: Value =
serde_json::from_str(&function.arguments).unwrap_or(Value::Null);
return Poll::Ready(Some(StreamEvent::ToolExecutionStarted {
return Poll::Ready(Some(LegacyStreamEvent::ToolExecutionStarted {
tool_name: function.name.clone(),
input: args,
tool_call_id: id.clone(),
@@ -85,24 +88,117 @@ impl Stream for ChunkToEventStream {
}
if let Some(finish_reason) = &choice.finish_reason {
return Poll::Ready(Some(StreamEvent::TurnComplete {
return Poll::Ready(Some(LegacyStreamEvent::TurnComplete {
reason: *finish_reason,
}));
}
}
if let Some(usage) = &chunk.usage {
return Poll::Ready(Some(StreamEvent::CostUpdate {
return Poll::Ready(Some(LegacyStreamEvent::CostUpdate {
usage: *usage,
}));
}
Poll::Ready(None)
}
Poll::Ready(Some(Err(e))) => Poll::Ready(Some(StreamEvent::error(e.to_string()))),
Poll::Ready(Some(Err(e))) => Poll::Ready(Some(LegacyStreamEvent::error(e.to_string()))),
Poll::Ready(None) => Poll::Ready(None),
Poll::Pending => Poll::Pending,
})
.poll_unpin(cx)
}
}
// --- 内部：LegacyStreamEvent → 新 StreamEvent ---
struct LegacyToIrEventStream {
inner: Pin<Box<dyn futures_core::Stream<Item = LegacyStreamEvent> + Send>>,
}
impl Stream for LegacyToIrEventStream {
type Item = Result<StreamEvent, LlmError>;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
let this = &mut *self;
match Pin::new(&mut this.inner).poll_next(cx) {
Poll::Ready(Some(legacy)) => Poll::Ready(Some(Ok(map_legacy_to_ir(legacy)))),
Poll::Ready(None) => {
// 旧流结束 → 主动补一个 MessageComplete（full_response 为兜底空快照）。
// ponytail: Phase 0 中 OpenaiProvider 桥接层负责产出真实 MessageResponse，
// 此处仅防止消费方无限等待。若 Provider 层已正确发出 MessageComplete，
// LlmCycle 不会走到这里 —— 因为桥接层 inline 处理。
Poll::Ready(Some(Ok(StreamEvent::MessageComplete {
full_response: empty_message_response(),
})))
}
Poll::Pending => Poll::Pending,
}
}
}
fn empty_message_response() -> MessageResponse {
use crate::llm::types::message::Message;
use std::collections::HashMap;
MessageResponse {
id: String::new(),
model: String::new(),
message: Message::Assistant {
content: vec![],
},
usage: Usage::default(),
stop_reason: StopReason::Stop,
extra: HashMap::new(),
}
}
/// 把旧 LegacyStreamEvent 映射到新高精度 IR StreamEvent。
///
/// Phase 1 重写 Provider 后可直接删除此映射函数。当前映射语义：
/// - `AssistantTextDelta` → `TextDelta`
/// - `ToolExecutionStarted` → `ToolCallArgumentsDelta`（OpenAI 单 chunk 模式下整段 arguments 一次性下发）
/// - `CostUpdate` → `CostUpdate`（Usage → PartialUsage 全字段）
/// - `TurnComplete` → `MessageComplete`（Phase 1 重写 Provider 后正确产出）
/// - `Error` → `Error`
///
/// ponytail: 这是一个"目前能跑通未来会被删除"的适配层。当前实现为单事件映射，
/// 旧 `ToolExecutionStarted` 携带的 (id, name) 暂未填入 IR 事件（消费方
/// Phase 2 中通过 MessageComplete.full_response.tool_use 提取）。Phase 1 重写时
/// 由 OpenAI Provider 直接产出 IR 流，整体删除此映射。
fn map_legacy_to_ir(legacy: LegacyStreamEvent) -> StreamEvent {
use crate::llm::types::response_v2::PartialUsage;
match legacy {
LegacyStreamEvent::AssistantTextDelta { text } => StreamEvent::TextDelta { text },
LegacyStreamEvent::ToolExecutionStarted { input, .. } => {
let arguments = serde_json::to_string(&input).unwrap_or_default();
StreamEvent::ToolCallArgumentsDelta { index: 0, arguments }
}
LegacyStreamEvent::ToolExecutionCompleted { .. } => {
// 旧 ToolExecutionCompleted 不在 IR 流协议中——工具执行是消费方职责。
// Phase 1 重写时此处整体删除。当前给一个无副作用的占位事件。
StreamEvent::CostUpdate {
usage: PartialUsage::default(),
}
}
LegacyStreamEvent::CostUpdate { usage } => StreamEvent::CostUpdate {
usage: PartialUsage {
prompt_tokens: Some(usage.prompt_tokens),
completion_tokens: Some(usage.completion_tokens),
total_tokens: Some(usage.total_tokens),
completion_tokens_details: usage.completion_tokens_details,
prompt_tokens_details: usage.prompt_tokens_details,
},
},
LegacyStreamEvent::TurnComplete { reason } => {
// 旧 TurnComplete 不直接对应 IR；映射为带 StopReason 的 MessageComplete。
// ponytail: Phase 1 重写 Provider 后此适配整体删除，
// OpenAI Provider 直接产出带正确 stop_reason 的 MessageComplete。
let _ = reason;
StreamEvent::MessageComplete {
full_response: empty_message_response(),
}
}
LegacyStreamEvent::Error { message } => StreamEvent::Error { message },
}
}
src/llm/types/message.rs
+384 -125
View File
@@ -1,165 +1,424 @@
use crate::llm::types::shared::{AudioFormat, ImageDetail};
use crate::llm::types::tool::OpenaiToolCall;
//! IR 层 Message 类型 —— 跨 Provider 统一的消息模型。
//!
//! 设计目标见 `docs/10-llm-provider-refinement.md` §2.1 Decision-01。
use serde::{Deserialize, Serialize};
use serde_json::Value;
#[derive(Debug, Clone)]
pub enum ContentField {
String(String),
Array(Vec<OpenaiContentPart>),
use crate::llm::types::shared::AudioFormat;
use crate::llm::types::shared::ImageDetail;
/// 跨 Provider 统一的消息类型（扁平大枚举）。
///
/// 每个变体直接承载完整语义，消费方 match 即可获得所有信息，
/// 无需在嵌套的 `Vec` 中搜索。
///
/// 设计要点：
/// - `Thinking` / `ToolUse` 不作为独立 Message 变体，保留在 `Assistant.content` 中，
/// 因为 text ↔ tool_use 的**交错顺序**是 Assistant 响应的语义组成部分。
/// - `UserImage` 是快捷变体，减少"图片只有 base64 字符串"的 boilerplate，
/// 消费方 match 可直接区分文本和图片输入。
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Message {
/// 系统提示（User & Assistant 之外的引导指令）。
System {
content: Vec<ContentBlock>,
},
/// 用户输入。
User {
content: Vec<ContentBlock>,
},
/// 用户的图片输入（快捷构造，免去构造 ContentBlock 的 boilerplate）。
UserImage {
data: String,
mime_type: String,
detail: ImageDetail,
},
/// Assistant 回复内容块（可能包含 text、thinking、tool_use 等多种 block 的混合）。
Assistant {
content: Vec<ContentBlock>,
},
/// 工具调用结果。
ToolResult {
tool_call_id: String,
content: Vec<ContentBlock>,
is_error: bool,
},
}
impl<'de> Deserialize<'de> for ContentField {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: serde::Deserializer<'de>,
{
let value = Value::deserialize(deserializer)?;
match value {
Value::String(s) => Ok(ContentField::String(s)),
Value::Array(arr) => {
let parts: Result<Vec<OpenaiContentPart>, _> =
serde_json::from_value(Value::Array(arr));
match parts {
Ok(parts) => Ok(ContentField::Array(parts)),
Err(e) => Err(serde::de::Error::custom(e)),
}
}
_ => Err(serde::de::Error::custom("content must be string or array")),
impl Message {
/// 构造纯文本 User 消息。
pub fn user_text(text: impl Into<String>) -> Self {
Message::User {
content: vec![ContentBlock::Text { text: text.into() }],
}
}
/// 构造 UserImage 快捷消息（直接传 base64 / URL + MIME）。
pub fn user_image(
data: impl Into<String>,
mime_type: impl Into<String>,
detail: ImageDetail,
) -> Self {
Message::UserImage {
data: data.into(),
mime_type: mime_type.into(),
detail,
}
}
/// 构造纯文本 Assistant 消息。
pub fn assistant(text: impl Into<String>) -> Self {
Message::Assistant {
content: vec![ContentBlock::Text { text: text.into() }],
}
}
/// 构造纯文本 System 消息。
pub fn system(text: impl Into<String>) -> Self {
Message::System {
content: vec![ContentBlock::Text { text: text.into() }],
}
}
/// 构造纯文本 ToolResult 消息。
pub fn tool_result(
tool_call_id: impl Into<String>,
text: impl Into<String>,
is_error: bool,
) -> Self {
Message::ToolResult {
tool_call_id: tool_call_id.into(),
content: vec![ContentBlock::Text { text: text.into() }],
is_error,
}
}
}
impl Serialize for ContentField {
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
where
S: serde::Serializer,
{
match self {
ContentField::String(s) => s.serialize(serializer),
ContentField::Array(arr) => arr.serialize(serializer),
}
}
/// 内容块 —— 组成消息的最小语义单元。
///
/// 与 9b §3.1 设计保持一致：`Text` / `Image` / `Audio` / `File` / `ToolUse` /
/// `ToolResult` / `Thinking` / `Extension`，其中 `Extension` 作为 Provider 特定
/// block 的逃生舱。
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ContentBlock {
/// 纯文本。
Text { text: String },
/// 图片（含 URL 或 base64）。
Image { source: ImageSource },
/// 音频。
Audio { source: AudioSource },
/// 通用文件。
File { source: FileSource },
/// 工具调用。
ToolUse {
id: String,
name: String,
input: Value,
},
/// 工具结果（在消息历史中由 Assistant 携带，如 Anthropic 内容回显）。
ToolResult {
tool_use_id: String,
content: Vec<ContentBlock>,
is_error: bool,
},
/// 推理 / 思考内容（可携带 signature 用于多轮续推）。
Thinking {
text: String,
signature: Option<String>,
},
/// 逃生舱：Provider 特定 block 透传（OpenAI Response 内置工具等）。
Extension {
kind: String,
data: Value,
},
}
/// 内容块类型标签 —— 用于 `StreamEvent::ContentBlockStart.block_type`。
///
/// 用途：在流式场景中，Provider 先下发 block 类型，再下发 block 内容增量。
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ImageURL {
pub url: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub detail: Option<ImageDetail>,
#[serde(rename_all = "snake_case")]
pub enum ContentBlockType {
/// 文本块。
Text,
/// 思考 / 推理块。
Thinking,
/// 拒绝 / 政策阻断块（OpenAI refusal）。
Refusal,
/// 工具调用块（携带 id 和 name）。
ToolUse { id: String, name: String },
}
/// 图片来源（URL 或 base64）。
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct InputAudio {
pub struct ImageSource {
/// URL 或 base64 字符串。
pub data: String,
/// MIME 类型（如 `image/png` / `image/jpeg` / `image/webp`）。
pub mime_type: String,
/// `true` = URL；`false` = base64。
pub is_url: bool,
}
/// 音频来源。
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AudioSource {
/// base64 字符串。
pub data: String,
/// 音频格式。
pub format: AudioFormat,
}
/// 文件来源。
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FileData {
pub file_data: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub file_id: Option<String>,
pub struct FileSource {
/// 文件内容（base64）或 URL。
pub data: String,
/// 原始文件名。
#[serde(skip_serializing_if = "Option::is_none")]
pub filename: Option<String>,
/// MIME 类型。
#[serde(skip_serializing_if = "Option::is_none")]
pub mime_type: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "snake_case", tag = "type")]
pub enum OpenaiContentPart {
Text {
text: String,
},
Image {
image_url: ImageURL,
#[serde(skip_serializing_if = "Option::is_none")]
detail: Option<ImageDetail>,
},
InputAudio {
input_audio: InputAudio,
},
File {
file: FileData,
},
Refusal {
refusal: String,
},
}
#[cfg(test)]
mod tests {
use super::*;
use crate::llm::types::shared::AudioFormat;
use serde_json::json;
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "snake_case", tag = "role")]
pub enum OpenaiChatMessage {
Developer {
content: ContentField,
#[serde(skip_serializing_if = "Option::is_none")]
name: Option<String>,
},
System {
content: ContentField,
#[serde(skip_serializing_if = "Option::is_none")]
name: Option<String>,
},
User {
content: ContentField,
#[serde(skip_serializing_if = "Option::is_none")]
name: Option<String>,
},
Assistant {
content: ContentField,
#[serde(skip_serializing_if = "Option::is_none")]
refusal: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
name: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
tool_calls: Option<Vec<OpenaiToolCall>>,
},
Tool {
content: ContentField,
tool_call_id: String,
},
Function {
content: ContentField,
name: String,
},
}
impl OpenaiChatMessage {
pub fn user_text<S: Into<String>>(text: S) -> Self {
OpenaiChatMessage::User {
content: ContentField::Array(vec![OpenaiContentPart::Text { text: text.into() }]),
name: None,
#[test]
fn message_user_text_produces_text_block() {
let m = Message::user_text("hello");
match m {
Message::User { content } => {
assert_eq!(content.len(), 1);
match &content[0] {
ContentBlock::Text { text } => assert_eq!(text, "hello"),
_ => panic!("expected Text block"),
}
}
_ => panic!("expected User variant"),
}
}
pub fn assistant_text<S: Into<String>>(text: S) -> Self {
OpenaiChatMessage::Assistant {
content: ContentField::String(text.into()),
refusal: None,
name: None,
tool_calls: None,
#[test]
fn message_assistant_text_produces_text_block() {
let m = Message::assistant("hi");
match m {
Message::Assistant { content } => {
assert_eq!(content.len(), 1);
assert!(matches!(&content[0], ContentBlock::Text { text } if text == "hi"));
}
_ => panic!("expected Assistant variant"),
}
}
pub fn system_text<S: Into<String>>(text: S) -> Self {
OpenaiChatMessage::System {
content: ContentField::Array(vec![OpenaiContentPart::Text { text: text.into() }]),
name: None,
#[test]
fn message_system_text_produces_text_block() {
let m = Message::system("sys");
match m {
Message::System { content } => {
assert_eq!(content.len(), 1);
assert!(matches!(&content[0], ContentBlock::Text { text } if text == "sys"));
}
_ => panic!("expected System variant"),
}
}
pub fn developer_text<S: Into<String>>(text: S) -> Self {
OpenaiChatMessage::Developer {
content: ContentField::Array(vec![OpenaiContentPart::Text { text: text.into() }]),
name: None,
#[test]
fn message_user_image_carries_fields() {
let m = Message::user_image("base64data", "image/png", ImageDetail::High);
match m {
Message::UserImage {
data,
mime_type,
detail,
} => {
assert_eq!(data, "base64data");
assert_eq!(mime_type, "image/png");
assert_eq!(detail, ImageDetail::High);
}
_ => panic!("expected UserImage variant"),
}
}
pub fn tool_result<S: Into<String>>(tool_call_id: String, content: S) -> Self {
OpenaiChatMessage::Tool {
content: ContentField::Array(vec![OpenaiContentPart::Text {
text: content.into(),
}]),
tool_call_id,
#[test]
fn message_tool_result_text_produces_block() {
let m = Message::tool_result("call_1", "result text", false);
match m {
Message::ToolResult {
tool_call_id,
content,
is_error,
} => {
assert_eq!(tool_call_id, "call_1");
assert!(!is_error);
assert_eq!(content.len(), 1);
match &content[0] {
ContentBlock::Text { text } => assert_eq!(text, "result text"),
_ => panic!("expected Text block"),
}
}
_ => panic!("expected ToolResult variant"),
}
}
#[test]
fn message_exhaustive_match() {
// 编译器保证穷举；这里写一个 match 验证所有变体可被访问。
let msgs = vec![
Message::user_text("a"),
Message::user_image("d", "image/png", ImageDetail::Auto),
Message::assistant("b"),
Message::system("c"),
Message::tool_result("id", "r", false),
];
for m in &msgs {
let s = match m {
Message::System { .. } => "sys",
Message::User { .. } => "user",
Message::UserImage { .. } => "image",
Message::Assistant { .. } => "asst",
Message::ToolResult { .. } => "tool",
};
assert!(!s.is_empty());
}
}
#[test]
fn content_block_roundtrip_each_variant() {
let blocks = vec![
ContentBlock::Text {
text: "hello".into(),
},
ContentBlock::Image {
source: ImageSource {
data: "b64".into(),
mime_type: "image/png".into(),
is_url: false,
},
},
ContentBlock::Audio {
source: AudioSource {
data: "audio_b64".into(),
format: AudioFormat::Mp3,
},
},
ContentBlock::File {
source: FileSource {
data: "file_data".into(),
filename: Some("doc.pdf".into()),
mime_type: Some("application/pdf".into()),
},
},
ContentBlock::ToolUse {
id: "call_1".into(),
name: "search".into(),
input: json!({"q": "rust"}),
},
ContentBlock::ToolResult {
tool_use_id: "call_1".into(),
content: vec![ContentBlock::Text {
text: "found".into(),
}],
is_error: false,
},
ContentBlock::Thinking {
text: "thinking...".into(),
signature: Some("sig".into()),
},
ContentBlock::Extension {
kind: "openai.response.web_search_call".into(),
data: json!({"query": "rust"}),
},
];
for original in blocks {
let json = serde_json::to_string(&original).expect("serialize");
let decoded: ContentBlock = serde_json::from_str(&json).expect("deserialize");
assert_eq!(format!("{:?}", original), format!("{:?}", decoded));
}
}
#[test]
fn message_roundtrip_each_variant() {
let msgs = vec![
Message::System {
content: vec![ContentBlock::Text {
text: "sys".into(),
}],
},
Message::User {
content: vec![ContentBlock::Text {
text: "hello".into(),
}],
},
Message::UserImage {
data: "b64".into(),
mime_type: "image/png".into(),
detail: ImageDetail::High,
},
Message::Assistant {
content: vec![
ContentBlock::Thinking {
text: "reasoning".into(),
signature: None,
},
ContentBlock::Text {
text: "answer".into(),
},
],
},
Message::ToolResult {
tool_call_id: "call_1".into(),
content: vec![ContentBlock::Text {
text: "ok".into(),
}],
is_error: true,
},
];
for original in msgs {
let json = serde_json::to_string(&original).expect("serialize");
let decoded: Message = serde_json::from_str(&json).expect("deserialize");
assert_eq!(format!("{:?}", original), format!("{:?}", decoded));
}
}
#[test]
fn content_block_tool_result_nested_roundtrip() {
let block = ContentBlock::ToolResult {
tool_use_id: "call_1".into(),
content: vec![ContentBlock::Text {
text: "result".into(),
}],
is_error: false,
};
let json = serde_json::to_string(&block).unwrap();
let decoded: ContentBlock = serde_json::from_str(&json).unwrap();
assert_eq!(format!("{:?}", block), format!("{:?}", decoded));
}
#[test]
fn content_block_type_roundtrip() {
let types = vec![
ContentBlockType::Text,
ContentBlockType::Thinking,
ContentBlockType::Refusal,
ContentBlockType::ToolUse {
id: "call_x".into(),
name: "fn".into(),
},
];
for original in types {
let json = serde_json::to_string(&original).unwrap();
let decoded: ContentBlockType = serde_json::from_str(&json).unwrap();
assert_eq!(format!("{:?}", original), format!("{:?}", decoded));
}
}
}
src/llm/types/mod.rs
+29 -9
View File
@@ -1,18 +1,27 @@
pub mod message;
pub mod old_stream;
pub mod openai_message;
pub mod request;
pub mod request_v2;
pub mod response;
pub mod response_v2;
pub mod shared;
pub mod tool;
pub mod usage;
pub use message::{
pub use openai_message::{
ContentField, FileData, ImageURL, InputAudio, OpenaiChatMessage, OpenaiContentPart,
};
pub use request::{OpenaiChatRequest, OpenaiTool, StreamOptions, ToolChoice};
pub use request_v2::{ExtraError, MessageRequest, ThinkingConfig};
pub use response::{
Annotation, Choice, ChunkChoice, Delta, Logprobs, OpenaiAudio, OpenaiChatChunk,
OpenaiChatResponse, TokenLogprob, TopLogprob, URLCitation,
};
pub use response_v2::{
ContentBlockBuilder, MessageResponse, PartialMessageResponse, PartialUsage, StopReason,
StreamEvent,
};
pub use shared::{
AudioFormat, FinishReason, ImageDetail, Modality, ResponseFormat, Role, ServiceTier,
StopSequence,
@@ -20,6 +29,17 @@ pub use shared::{
pub use tool::{FunctionCall, OpenaiToolCall, OpenaiToolDefinition};
pub use usage::{CompletionTokensDetails, CostTracker, PromptTokensDetails, Usage};
// Re-export IR 内容块 / 消息类型供 `types::ContentBlock` 等历史路径消费。
//
// 注意：以下别名 *故意不暴露* `pub type Message = message::Message`、
// `pub type ContentBlock = message::ContentBlock` —— 新 `Message` / `ContentBlock` /
// `StopReason` 是独立类型，由 `Message` / `ContentBlock` / `StopReason` 直接路径访问，
// 旧别名（指 `OpenaiChatMessage` / `OpenaiContentPart` / `FinishReason`）已移除，
// 避免新类型阴影。Phase 2 完成后再统一收敛。
/// 旧 wire-format 请求类型别名（Phase 1 重写 Provider 后弃用）。
pub type ChatRequest = OpenaiChatRequest;
/// 旧 wire-format 响应结构（保留用于 OpenAI 内部转换层）。
#[derive(Debug, Clone)]
pub struct ChatResponse {
pub message: OpenaiChatMessage,
@@ -54,10 +74,13 @@ impl From<ChatResponse> for OpenaiChatChunk {
};
OpenaiChatChunk {
id: format!("chunk-{}", std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_nanos())
.unwrap_or(0)),
id: format!(
"chunk-{}",
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_nanos())
.unwrap_or(0)
),
object: "chat.completion.chunk".to_string(),
created: std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
@@ -71,8 +94,5 @@ impl From<ChatResponse> for OpenaiChatChunk {
}
}
pub type ChatRequest = OpenaiChatRequest;
pub type Message = OpenaiChatMessage;
pub type ContentBlock = OpenaiContentPart;
/// 工具定义别名（无新类型冲突，保留）。
pub type ToolDefinition = OpenaiToolDefinition;
pub type StopReason = FinishReason;
src/llm/types/old_stream.rs
+45
View File
@@ -0,0 +1,45 @@
//! 旧版流式事件 —— Phase 0 临时保留，仅供 `stream.rs` 中 `parse_chunk_stream` 内部使用。
//!
//! Phase 0 中：高精度 `StreamEvent`（定义在 `response_v2.rs`）是唯一的对外
//! `StreamEvent`，旧变体迁移至此模块改名为 `LegacyStreamEvent`，
//! 由 `parse_chunk_stream()` 内部消费 `LegacyStreamEvent`，对外返回值已被
//! 重映射为新 `StreamEvent`。
//!
//! Phase 1 重写 Provider 时，`parse_chunk_stream` 可直接消费新事件流后整体删除此文件。
use crate::llm::types::shared::FinishReason;
use crate::llm::types::usage::Usage;
use serde_json::Value;
/// 旧 `StreamEvent` 变体迁移后的别名 —— 仅供 `stream.rs` 内部使用。
#[derive(Debug, Clone)]
pub enum LegacyStreamEvent {
/// 助手回复文本增量。
AssistantTextDelta { text: String },
/// 工具调用开始。
ToolExecutionStarted {
tool_name: String,
input: Value,
tool_call_id: String,
},
/// 工具调用完成。
ToolExecutionCompleted {
tool_name: String,
output: Value,
is_error: bool,
},
/// Token 用量更新。
CostUpdate { usage: Usage },
/// 一轮会话完成。
TurnComplete { reason: FinishReason },
/// 错误事件。
Error { message: String },
}
impl LegacyStreamEvent {
pub(crate) fn error(message: impl Into<String>) -> Self {
Self::Error {
message: message.into(),
}
}
}
src/llm/types/openai_message.rs
+165
View File
@@ -0,0 +1,165 @@
use crate::llm::types::shared::{AudioFormat, ImageDetail};
use crate::llm::types::tool::OpenaiToolCall;
use serde::{Deserialize, Serialize};
use serde_json::Value;
#[derive(Debug, Clone)]
pub enum ContentField {
String(String),
Array(Vec<OpenaiContentPart>),
}
impl<'de> Deserialize<'de> for ContentField {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: serde::Deserializer<'de>,
{
let value = Value::deserialize(deserializer)?;
match value {
Value::String(s) => Ok(ContentField::String(s)),
Value::Array(arr) => {
let parts: Result<Vec<OpenaiContentPart>, _> =
serde_json::from_value(Value::Array(arr));
match parts {
Ok(parts) => Ok(ContentField::Array(parts)),
Err(e) => Err(serde::de::Error::custom(e)),
}
}
_ => Err(serde::de::Error::custom("content must be string or array")),
}
}
}
impl Serialize for ContentField {
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
where
S: serde::Serializer,
{
match self {
ContentField::String(s) => s.serialize(serializer),
ContentField::Array(arr) => arr.serialize(serializer),
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ImageURL {
pub url: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub detail: Option<ImageDetail>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct InputAudio {
pub data: String,
pub format: AudioFormat,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FileData {
pub file_data: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub file_id: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub filename: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "snake_case", tag = "type")]
pub enum OpenaiContentPart {
Text {
text: String,
},
Image {
image_url: ImageURL,
#[serde(skip_serializing_if = "Option::is_none")]
detail: Option<ImageDetail>,
},
InputAudio {
input_audio: InputAudio,
},
File {
file: FileData,
},
Refusal {
refusal: String,
},
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "snake_case", tag = "role")]
pub enum OpenaiChatMessage {
Developer {
content: ContentField,
#[serde(skip_serializing_if = "Option::is_none")]
name: Option<String>,
},
System {
content: ContentField,
#[serde(skip_serializing_if = "Option::is_none")]
name: Option<String>,
},
User {
content: ContentField,
#[serde(skip_serializing_if = "Option::is_none")]
name: Option<String>,
},
Assistant {
content: ContentField,
#[serde(skip_serializing_if = "Option::is_none")]
refusal: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
name: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
tool_calls: Option<Vec<OpenaiToolCall>>,
},
Tool {
content: ContentField,
tool_call_id: String,
},
Function {
content: ContentField,
name: String,
},
}
impl OpenaiChatMessage {
pub fn user_text<S: Into<String>>(text: S) -> Self {
OpenaiChatMessage::User {
content: ContentField::Array(vec![OpenaiContentPart::Text { text: text.into() }]),
name: None,
}
}
pub fn assistant_text<S: Into<String>>(text: S) -> Self {
OpenaiChatMessage::Assistant {
content: ContentField::String(text.into()),
refusal: None,
name: None,
tool_calls: None,
}
}
pub fn system_text<S: Into<String>>(text: S) -> Self {
OpenaiChatMessage::System {
content: ContentField::Array(vec![OpenaiContentPart::Text { text: text.into() }]),
name: None,
}
}
pub fn developer_text<S: Into<String>>(text: S) -> Self {
OpenaiChatMessage::Developer {
content: ContentField::Array(vec![OpenaiContentPart::Text { text: text.into() }]),
name: None,
}
}
pub fn tool_result<S: Into<String>>(tool_call_id: String, content: S) -> Self {
OpenaiChatMessage::Tool {
content: ContentField::Array(vec![OpenaiContentPart::Text {
text: content.into(),
}]),
tool_call_id,
}
}
}
src/llm/types/request.rs
+8 -1
View File
@@ -20,6 +20,13 @@ pub enum ToolChoice {
AllowedTools { tool_names: Vec<String> },
}
impl Default for ToolChoice {
fn default() -> Self {
// Default 行为由调用方显式选择；此处选择不暴露任何工具 —— 不暴露比错暴露安全。
ToolChoice::None
}
}
impl Serialize for ToolChoice {
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
where
@@ -139,7 +146,7 @@ pub struct WebSearchOptions {
#[serde(rename_all = "snake_case")]
pub struct OpenaiChatRequest {
pub model: String,
pub messages: Vec<crate::llm::types::message::OpenaiChatMessage>,
pub messages: Vec<crate::llm::types::openai_message::OpenaiChatMessage>,
#[serde(skip_serializing_if = "Option::is_none")]
pub frequency_penalty: Option<f32>,
#[serde(skip_serializing_if = "Option::is_none")]
src/llm/types/request_v2.rs
+209
View File
@@ -0,0 +1,209 @@
//! IR 层 MessageRequest 类型 —— 跨 Provider 统一的请求模型。
//!
//! 设计目标见 `docs/10-llm-provider-refinement.md` §2.2 Decision-02 + `docs/9b-ir-type-system.md` §3.6。
use std::collections::HashMap;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use thiserror::Error;
use crate::llm::types::message::Message;
use crate::llm::types::request::ToolChoice;
use crate::llm::types::tool::OpenaiToolDefinition;
/// Provider 无关的请求类型。
///
/// 设计要点：
/// - `system` 字段不存在；system 提示由调用方通过 `Message::System` 在 `messages` 中表达。
/// - `tools` / `tool_choice` 直接复用现有 `OpenaiToolDefinition` / `ToolChoice`
/// （10a §251 决策：先复用旧类型，Phase 2 切换为新 `ToolDefinition` 后再调整）。
/// - `extra` 作为逃生舱：Provider 特定字段（`web_search_options`、`previous_response_id` 等）
/// 通过 `extra.set_extra / get_extra` 传递，避免持续膨胀本结构体。
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct MessageRequest {
/// 模型名称。
pub model: String,
/// 消息列表（包含 system / user / assistant / tool_result 等所有变体）。
pub messages: Vec<Message>,
/// 工具定义列表。
pub tools: Vec<OpenaiToolDefinition>,
/// 工具选择策略。
pub tool_choice: ToolChoice,
/// 最大输出 token 数。
#[serde(skip_serializing_if = "Option::is_none")]
pub max_tokens: Option<u32>,
/// 采样温度。
#[serde(skip_serializing_if = "Option::is_none")]
pub temperature: Option<f32>,
/// nucleus 采样 top-p。
#[serde(skip_serializing_if = "Option::is_none")]
pub top_p: Option<f32>,
/// 终止序列。
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub stop_sequences: Vec<String>,
/// 是否流式响应（Provider 实现层读取）。
#[serde(default)]
pub stream: bool,
/// 思考 / 推理配置（如 Anthropic thinking budget）。
#[serde(skip_serializing_if = "Option::is_none")]
pub thinking: Option<ThinkingConfig>,
/// Provider 特定扩展字段。
#[serde(default, skip_serializing_if = "HashMap::is_empty")]
pub extra: HashMap<String, Value>,
}
/// 思考 / 推理配置。
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ThinkingConfig {
/// 思考预算 token 数。
pub budget_tokens: u32,
}
/// `extra` 字段访问错误。
#[derive(Error, Debug)]
pub enum ExtraError {
/// 反序列化到目标类型失败。
#[error("extra 字段 `{key}` 反序列化失败: {details}")]
TypeMismatch { key: String, details: String },
/// 字段值反序列化为 JSON 失败。
#[error("extra 反序列化失败: {0}")]
Deserialize(String),
}
impl MessageRequest {
/// 读取一个 `extra` 字段，缺失返回 `Ok(None)`，类型不匹配返回 `Err(ExtraError::TypeMismatch)`。
pub fn get_extra<T: for<'de> Deserialize<'de>>(
&self,
key: &str,
) -> Result<Option<T>, ExtraError> {
match self.extra.get(key) {
None => Ok(None),
Some(value) => serde_json::from_value(value.clone())
.map(Some)
.map_err(|e| ExtraError::TypeMismatch {
key: key.to_string(),
details: e.to_string(),
}),
}
}
/// 读取 `extra` 字段但用 `Option<T>`，跳过错误（缺失或反序列化失败均返回 `None`）。
pub fn get_extra_opt<T: for<'de> Deserialize<'de>>(&self, key: &str) -> Option<T> {
self.get_extra::<T>(key).ok().flatten()
}
/// 把整个 `extra` 整体反序列化为类型 `T`（用于"我有一个完整 options 结构"的场景）。
pub fn get_extra_as<T: for<'de> Deserialize<'de>>(&self) -> Result<T, ExtraError> {
let value = Value::Object(
self.extra
.iter()
.map(|(k, v)| (k.clone(), v.clone()))
.collect::<serde_json::Map<String, Value>>(),
);
serde_json::from_value(value).map_err(|e| ExtraError::Deserialize(e.to_string()))
}
/// 设置一个 `extra` 字段。
pub fn set_extra(&mut self, key: impl Into<String>, value: impl Into<Value>) {
self.extra.insert(key.into(), value.into());
}
}
#[cfg(test)]
mod tests {
use super::*;
use serde_json::json;
#[test]
fn message_request_default_is_constructible() {
let req = MessageRequest::default();
assert_eq!(req.model, "");
assert!(req.messages.is_empty());
assert!(!req.stream);
assert!(req.max_tokens.is_none());
}
#[test]
fn extra_set_and_get_roundtrip() {
let mut req = MessageRequest::default();
req.set_extra(
"previous_response_id",
"resp_abc123",
);
let v: Option<String> = req
.get_extra("previous_response_id")
.expect("get_extra ok");
assert_eq!(v.as_deref(), Some("resp_abc123"));
let missing: Option<String> = req.get_extra("missing").expect("missing ok");
assert!(missing.is_none());
}
#[test]
fn extra_type_mismatch_returns_err() {
let mut req = MessageRequest::default();
req.set_extra("count", "not a number");
let result: Result<Option<u32>, ExtraError> = req.get_extra("count");
assert!(matches!(result, Err(ExtraError::TypeMismatch { .. })));
}
#[test]
fn extra_opt_swallows_errors() {
let mut req = MessageRequest::default();
req.set_extra("count", "not a number");
let v: Option<u32> = req.get_extra_opt("count");
assert!(v.is_none());
}
#[test]
fn get_extra_as_deserializes_whole_extra() {
let mut req = MessageRequest::default();
req.set_extra("web_search_options", json!({"search_context_size": "high"}));
req.set_extra("user", json!("u_123"));
#[derive(Deserialize, Debug, PartialEq)]
struct Options {
web_search_options: WebSearch,
user: Option<String>,
}
#[derive(Deserialize, Debug, PartialEq)]
struct WebSearch {
search_context_size: String,
}
let opts: Options = req.get_extra_as().expect("get_extra_as ok");
assert_eq!(
opts.web_search_options.search_context_size,
"high"
);
assert_eq!(opts.user.as_deref(), Some("u_123"));
}
#[test]
fn message_request_json_roundtrip() {
let mut req = MessageRequest {
model: "gpt-4o".into(),
messages: vec![Message::user_text("hello"), Message::assistant("hi")],
tools: vec![],
tool_choice: ToolChoice::Auto,
max_tokens: Some(1024),
temperature: Some(0.7),
top_p: Some(0.9),
stop_sequences: vec!["STOP".into()],
stream: true,
thinking: None,
extra: HashMap::new(),
};
req.set_extra("trace_id", json!("t-1"));
let json = serde_json::to_string(&req).expect("serialize");
let decoded: MessageRequest = serde_json::from_str(&json).expect("deserialize");
assert_eq!(decoded.model, req.model);
assert_eq!(decoded.messages.len(), req.messages.len());
assert_eq!(decoded.max_tokens, req.max_tokens);
assert_eq!(decoded.stream, req.stream);
assert_eq!(decoded.extra.get("trace_id"), Some(&json!("t-1")));
}
}
src/llm/types/response.rs
+1 -1
View File
@@ -1,4 +1,4 @@
use crate::llm::types::message::OpenaiChatMessage;
use crate::llm::types::openai_message::OpenaiChatMessage;
use crate::llm::types::shared::{FinishReason, ServiceTier};
use crate::llm::types::tool::OpenaiToolCall;
use crate::llm::types::usage::Usage;
src/llm/types/response_v2.rs
+849
View File
@@ -0,0 +1,849 @@
//! IR 层 MessageResponse / StreamEvent / PartialMessageResponse —— 流式响应模型。
//!
//! 设计目标见 `docs/10-llm-provider-refinement.md` §2.3 Decision-03 +
//! `docs/9c-llm-provider-trait.md` §4.3 / §4.4。
use std::collections::{BTreeMap, HashMap, HashSet};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use crate::llm::error::LlmError;
use crate::llm::types::message::{ContentBlock, ContentBlockType, Message};
use crate::llm::types::shared::FinishReason;
use crate::llm::types::usage::{CompletionTokensDetails, PromptTokensDetails, Usage};
/// Provider 终止原因（统一枚举，对应各 Provider 的 finish_reason）。
///
/// 注意：与现有 `FinishReason` 在 `openai_message::shared` 模块中并存，
/// Phase 2 完成时统一收敛。
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum StopReason {
/// 自然停止。
Stop,
/// 达到长度上限。
Length,
/// 触发了工具调用。
ToolUse,
/// 内容安全过滤。
ContentFilter,
/// 达到 max_tokens 限制。
MaxTokens,
/// 命中停止序列。
StopSequence,
/// 其他 / 未知。
Other,
}
impl From<FinishReason> for StopReason {
fn from(r: FinishReason) -> Self {
match r {
FinishReason::Stop => StopReason::Stop,
FinishReason::Length => StopReason::Length,
FinishReason::ToolCalls => StopReason::ToolUse,
FinishReason::ContentFilter => StopReason::ContentFilter,
FinishReason::FunctionCall => StopReason::ToolUse,
FinishReason::Other => StopReason::Other,
}
}
}
/// Provider 完整响应。
///
/// `extra` 作为 Provider 特定字段（`system_fingerprint`、`service_tier` 等）逃生舱。
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MessageResponse {
/// 响应唯一 ID（如 OpenAI 的 `chatcmpl-xxx`）。
pub id: String,
/// 实际使用的模型名。
pub model: String,
/// Assistant 回复的消息。
pub message: Message,
/// Token 用量。
pub usage: Usage,
/// 终止原因。
pub stop_reason: StopReason,
/// Provider 特定扩展字段。
#[serde(default, skip_serializing_if = "HashMap::is_empty")]
pub extra: HashMap<String, Value>,
}
impl MessageResponse {
/// 提取 Assistant 回复中的全部纯文本（按出现顺序拼接）。
///
/// 仅在 `message` 是 `Assistant` 时有意义，其它变体返回空串。
pub fn text(&self) -> String {
match &self.message {
Message::Assistant { content } => content
.iter()
.filter_map(|b| match b {
ContentBlock::Text { text } => Some(text.as_str()),
_ => None,
})
.collect(),
_ => String::new(),
}
}
}
/// 流式用量 —— 字段为 `Option` 是因为流式场景中各字段可能分多次到达
///（如 Anthropic 的 `message_delta` 事件可多次下发 usage 增量）。
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct PartialUsage {
/// 输入 token 数。
pub prompt_tokens: Option<u32>,
/// 输出 token 数。
pub completion_tokens: Option<u32>,
/// 总 token 数。
pub total_tokens: Option<u32>,
/// 输出 token 明细（含 reasoning_tokens 等）。
pub completion_tokens_details: Option<CompletionTokensDetails>,
/// 输入 token 明细（含 cached_tokens 等）。
pub prompt_tokens_details: Option<PromptTokensDetails>,
}
impl PartialUsage {
/// 字段级合并 `other` 到 `self`：`other` 中 `Some` 字段覆盖 `self`。
pub fn merge_from(&mut self, other: &PartialUsage) {
if let Some(v) = other.prompt_tokens {
self.prompt_tokens = Some(v);
}
if let Some(v) = other.completion_tokens {
self.completion_tokens = Some(v);
}
if let Some(v) = other.total_tokens {
self.total_tokens = Some(v);
}
if let Some(v) = other.completion_tokens_details {
self.completion_tokens_details = Some(v);
}
if let Some(v) = other.prompt_tokens_details {
self.prompt_tokens_details = Some(v);
}
}
/// 转换为完整的 `Usage`，缺失字段默认为 0。
pub fn into_usage(self) -> Usage {
Usage {
prompt_tokens: self.prompt_tokens.unwrap_or(0),
completion_tokens: self.completion_tokens.unwrap_or(0),
total_tokens: self.total_tokens.unwrap_or(0),
completion_tokens_details: self.completion_tokens_details,
prompt_tokens_details: self.prompt_tokens_details,
}
}
}
/// 内容块构建器 —— 在流式累积阶段持有单 block 的原始拼接状态。
#[derive(Debug, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ContentBlockBuilder {
/// 文本块（按增量拼接）。
Text(String),
/// 思考块（拼接中可携带 signature）。
Thinking {
buffer: String,
signature: Option<String>,
},
/// 拒绝 / 政策阻断（OpenAI refusal）。
Refusal(String),
/// 工具调用块（id / name 来自 `ContentBlockStart`，arguments 按增量拼接）。
ToolUse {
id: String,
name: String,
arguments: String,
},
}
/// 流式事件 —— LLM 流式响应的语义化事件。
///
/// 设计变更（参考 `docs/10-llm-provider-refinement.md` §2.3）：
/// `MessageComplete` 仅携带 `full_response: MessageResponse`，`stop_reason` 和
/// `thinking_signature` 不再作为顶层冗余字段。Provider 在内部维护
/// `thinking_signature`，最终通过 `finalize()` 回填到 `full_response` 的 `Thinking` block 中。
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum StreamEvent {
/// 消息开始（元信息）。
MessageStart { id: String, model: String },
/// 内容块开始（告知块类型，携带 id/name for ToolUse）。
ContentBlockStart { index: u32, block_type: ContentBlockType },
/// 内容块结束标记。
ContentBlockEnd { index: u32 },
/// 文本增量。
TextDelta { text: String },
/// 思考 / 推理增量。
ThinkingDelta { text: String },
/// 拒绝 / 政策阻断增量。
RefusalDelta { text: String },
/// 工具调用参数增量（按 index 区分并行调用）。
ToolCallArgumentsDelta { index: u32, arguments: String },
/// 工具调用结束标记。
ToolCallEnd { index: u32 },
/// 用量更新（字段级合并）。
CostUpdate { usage: PartialUsage },
/// 消息完成 —— 唯一可靠的完整响应来源。
MessageComplete { full_response: MessageResponse },
/// 错误事件。
Error { message: String },
}
/// 流式响应累积状态。
///
/// Provider 的流处理循环逐个消费 `StreamEvent`、调用 `apply_to` 更新此结构，
/// 在收到 `MessageComplete` 后调用 `finalize()` 得到完整 `MessageResponse`。
#[derive(Debug, Default, Serialize, Deserialize)]
pub struct PartialMessageResponse {
/// 响应 ID（来自 `MessageStart`）。
pub id: Option<String>,
/// 模型名（来自 `MessageStart`）。
pub model: Option<String>,
/// 按 index 索引的内容块构建器。
pub blocks: BTreeMap<u32, ContentBlockBuilder>,
/// 已标记"完成"（收到 `ContentBlockEnd` 或 `ToolCallEnd`）的 block index。
pub block_completion: HashSet<u32>,
/// 当前最后一个未结束的 block index（用于 TextDelta / ThinkingDelta 等定位目标 block）。
pub last_open_index: Option<u32>,
/// 流式累积用量。
pub usage: PartialUsage,
/// 终止原因（从 Provider 的终止事件推断）。
pub stop_reason: Option<StopReason>,
/// 思考签名（Anthropic provider 直接通过 `set_thinking_signature` 设置，不经过事件层）。
pub thinking_signature: Option<String>,
/// 是否遇到错误事件。
pub is_errored: bool,
/// 是否收到完成事件。
pub is_complete: bool,
}
impl PartialMessageResponse {
/// 创建一个新的空累积状态。
pub fn new() -> Self {
Self::default()
}
/// 由 Provider 直接写入 Anthropic 的 thinking signature。
///
/// 详见 `docs/10-llm-provider-refinement.md` §2.3 — `thinking_signature`
/// 不经过事件层，由 Provider 流循环直接写入此内部状态。
pub fn set_thinking_signature(&mut self, signature: impl Into<String>) {
self.thinking_signature = Some(signature.into());
}
/// 应用一个流事件到累积状态。
///
/// 返回 `false` 表示遇到 `Error` 事件，调用方应终止流处理。
pub fn apply_to(&mut self, event: &StreamEvent) -> bool {
match event {
StreamEvent::MessageStart { id, model } => {
self.id = Some(id.clone());
self.model = Some(model.clone());
true
}
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(),
arguments: String::new(),
},
};
self.blocks.insert(*index, builder);
self.last_open_index = Some(*index);
true
}
StreamEvent::ContentBlockEnd { index } => {
self.block_completion.insert(*index);
if self.last_open_index == Some(*index) {
self.last_open_index = None;
}
true
}
StreamEvent::TextDelta { text } => {
if let Some(idx) = self.last_open_index
&& let Some(builder) = self.blocks.get_mut(&idx)
{
match builder {
ContentBlockBuilder::Text(buf) => buf.push_str(text),
ContentBlockBuilder::Thinking { buffer, .. } => buffer.push_str(text),
ContentBlockBuilder::Refusal(buf) => buf.push_str(text),
ContentBlockBuilder::ToolUse { arguments, .. } => {
arguments.push_str(text);
}
}
}
true
}
StreamEvent::ThinkingDelta { text } => {
if let Some(idx) = self.last_open_index
&& let Some(builder) = self.blocks.get_mut(&idx)
&& let ContentBlockBuilder::Thinking { buffer, .. } = builder
{
buffer.push_str(text);
}
true
}
StreamEvent::RefusalDelta { text } => {
if let Some(idx) = self.last_open_index
&& let Some(builder) = self.blocks.get_mut(&idx)
&& let ContentBlockBuilder::Refusal(buf) = builder
{
buf.push_str(text);
}
true
}
StreamEvent::ToolCallArgumentsDelta { index, arguments } => {
if let Some(ContentBlockBuilder::ToolUse {
arguments: buf, ..
}) = self.blocks.get_mut(index)
{
buf.push_str(arguments);
}
true
}
StreamEvent::ToolCallEnd { index } => {
self.block_completion.insert(*index);
if self.last_open_index == Some(*index) {
self.last_open_index = None;
}
true
}
StreamEvent::CostUpdate { usage } => {
self.usage.merge_from(usage);
true
}
StreamEvent::MessageComplete { .. } => {
self.is_complete = true;
true
}
StreamEvent::Error { message } => {
tracing::warn!(error = %message, "partial response received Error event");
self.is_errored = true;
false
}
}
}
/// 完成累积，构造完整 `MessageResponse`。
///
/// - 按 index 升序遍历 `blocks` 转换为 `ContentBlock`
/// - 缺失字段用 `self.thinking_signature` 回填
/// - 终止原因默认 `Stop`，除非上层显式设置
pub fn finalize(self) -> Result<MessageResponse, LlmError> {
let mut content_blocks = Vec::with_capacity(self.blocks.len());
for (idx, builder) in self.blocks {
let block = Self::builder_to_block(idx, builder, self.thinking_signature.as_deref())
.map_err(|e| LlmError::Other(format!(
"partial 块 #{idx} finalize 失败: {e}"
)))?;
content_blocks.push(block);
}
let usage = self.usage.into_usage();
let stop_reason = self.stop_reason.unwrap_or(StopReason::Stop);
let message = Message::Assistant {
content: content_blocks,
};
Ok(MessageResponse {
id: self.id.unwrap_or_default(),
model: self.model.unwrap_or_default(),
message,
usage,
stop_reason,
extra: HashMap::new(),
})
}
fn builder_to_block(
_idx: u32,
builder: ContentBlockBuilder,
fallback_signature: Option<&str>,
) -> Result<ContentBlock, String> {
match builder {
ContentBlockBuilder::Text(buf) => Ok(ContentBlock::Text { text: buf }),
ContentBlockBuilder::Thinking { buffer, signature } => {
let signature = signature.or_else(|| fallback_signature.map(str::to_string));
Ok(ContentBlock::Thinking {
text: buffer,
signature,
})
}
ContentBlockBuilder::Refusal(buf) => Ok(ContentBlock::Text { text: buf }),
ContentBlockBuilder::ToolUse {
id,
name,
arguments,
} => {
let input: Value = serde_json::from_str(&arguments).unwrap_or(Value::Null);
Ok(ContentBlock::ToolUse { id, name, input })
}
}
.map_err(|_: std::convert::Infallible| unreachable!("builder_to_block is total"))
}
}
#[cfg(test)]
mod tests {
use super::*;
use serde_json::json;
fn empty_response() -> MessageResponse {
MessageResponse {
id: "resp-1".into(),
model: "m".into(),
message: Message::Assistant { content: vec![] },
usage: Usage::default(),
stop_reason: StopReason::Stop,
extra: HashMap::new(),
}
}
#[test]
fn stop_reason_from_finish_reason_mapping() {
assert_eq!(StopReason::from(FinishReason::Stop), StopReason::Stop);
assert_eq!(
StopReason::from(FinishReason::ToolCalls),
StopReason::ToolUse
);
assert_eq!(
StopReason::from(FinishReason::FunctionCall),
StopReason::ToolUse
);
assert_eq!(StopReason::from(FinishReason::Length), StopReason::Length);
assert_eq!(StopReason::from(FinishReason::Other), StopReason::Other);
}
#[test]
fn message_response_text_extracts_assistant_text() {
let resp = MessageResponse {
id: "r".into(),
model: "m".into(),
message: Message::Assistant {
content: vec![
ContentBlock::Thinking {
text: "reasoning".into(),
signature: None,
},
ContentBlock::Text {
text: "hello ".into(),
},
ContentBlock::Text {
text: "world".into(),
},
ContentBlock::ToolUse {
id: "c".into(),
name: "search".into(),
input: json!({}),
},
],
},
usage: Usage::default(),
stop_reason: StopReason::ToolUse,
extra: HashMap::new(),
};
assert_eq!(resp.text(), "hello world");
}
#[test]
fn message_response_text_empty_when_not_assistant() {
let resp = MessageResponse {
id: "r".into(),
model: "m".into(),
message: Message::user_text("hi"),
usage: Usage::default(),
stop_reason: StopReason::Stop,
extra: HashMap::new(),
};
assert_eq!(resp.text(), "");
}
#[test]
fn partial_usage_merge_picks_some_fields() {
let mut a = PartialUsage {
prompt_tokens: Some(10),
completion_tokens: None,
..Default::default()
};
let b = PartialUsage {
prompt_tokens: Some(99),
completion_tokens: Some(20),
total_tokens: Some(30),
..Default::default()
};
a.merge_from(&b);
assert_eq!(a.prompt_tokens, Some(99));
assert_eq!(a.completion_tokens, Some(20));
assert_eq!(a.total_tokens, Some(30));
}
#[test]
fn partial_usage_into_usage_defaults_missing() {
let u = PartialUsage {
completion_tokens: Some(7),
..Default::default()
}
.into_usage();
assert_eq!(u.prompt_tokens, 0);
assert_eq!(u.completion_tokens, 7);
assert_eq!(u.total_tokens, 0);
}
// ===== apply_to / finalize 整合测试 =====
fn apply_seq(state: &mut PartialMessageResponse, events: &[StreamEvent]) -> bool {
let mut cont = true;
for ev in events {
cont = state.apply_to(ev);
if !cont {
break;
}
}
cont
}
#[test]
fn apply_to_message_start_then_finalize() {
let mut state = PartialMessageResponse::new();
assert!(apply_seq(
&mut state,
&[StreamEvent::MessageStart {
id: "r1".into(),
model: "gpt".into(),
}],
));
let resp = state.finalize().expect("ok");
assert_eq!(resp.id, "r1");
assert_eq!(resp.model, "gpt");
}
#[test]
fn apply_to_text_block_then_finalize() {
let mut state = PartialMessageResponse::new();
assert!(apply_seq(
&mut state,
&[
StreamEvent::MessageStart {
id: "r".into(),
model: "m".into(),
},
StreamEvent::ContentBlockStart {
index: 0,
block_type: ContentBlockType::Text,
},
StreamEvent::TextDelta {
text: "hello".into(),
},
StreamEvent::TextDelta {
text: " world".into(),
},
StreamEvent::ContentBlockEnd { index: 0 },
StreamEvent::MessageComplete {
full_response: empty_response(),
},
],
));
assert!(state.is_complete);
let resp = state.finalize().expect("ok");
match &resp.message {
Message::Assistant { content } => {
assert_eq!(content.len(), 1);
match &content[0] {
ContentBlock::Text { text } => assert_eq!(text, "hello world"),
_ => panic!("expected Text block"),
}
}
_ => panic!("expected Assistant"),
}
}
#[test]
fn apply_to_tool_use_block_then_finalize() {
let mut state = PartialMessageResponse::new();
assert!(apply_seq(
&mut state,
&[
StreamEvent::MessageStart {
id: "r".into(),
model: "m".into(),
},
StreamEvent::ContentBlockStart {
index: 0,
block_type: ContentBlockType::ToolUse {
id: "call_x".into(),
name: "search".into(),
},
},
StreamEvent::ToolCallArgumentsDelta {
index: 0,
arguments: r#"{"q":"rust"}"#.into(),
},
StreamEvent::ToolCallEnd { index: 0 },
StreamEvent::MessageComplete {
full_response: empty_response(),
},
],
));
let resp = state.finalize().expect("ok");
match &resp.message {
Message::Assistant { content } => match &content[0] {
ContentBlock::ToolUse { id, name, input } => {
assert_eq!(id, "call_x");
assert_eq!(name, "search");
assert_eq!(input, &json!({"q": "rust"}));
}
_ => panic!("expected ToolUse"),
},
_ => panic!("expected Assistant"),
}
}
#[test]
fn apply_to_thinking_block_uses_fallback_signature() {
let mut state = PartialMessageResponse::new();
state.set_thinking_signature("global-sig");
assert!(apply_seq(
&mut state,
&[
StreamEvent::MessageStart {
id: "r".into(),
model: "m".into(),
},
StreamEvent::ContentBlockStart {
index: 0,
block_type: ContentBlockType::Thinking,
},
StreamEvent::ThinkingDelta {
text: "thinking...".into(),
},
StreamEvent::ContentBlockEnd { index: 0 },
StreamEvent::MessageComplete {
full_response: empty_response(),
},
],
));
let resp = state.finalize().unwrap();
match &resp.message {
Message::Assistant { content } => match &content[0] {
ContentBlock::Thinking { text, signature } => {
assert_eq!(text, "thinking...");
assert_eq!(signature.as_deref(), Some("global-sig"));
}
_ => panic!("expected Thinking"),
},
_ => panic!("expected Assistant"),
}
}
#[test]
fn apply_to_cost_update_merges_across_events() {
let mut state = PartialMessageResponse::new();
assert!(apply_seq(
&mut state,
&[
StreamEvent::MessageStart {
id: "r".into(),
model: "m".into(),
},
StreamEvent::CostUpdate {
usage: PartialUsage {
prompt_tokens: Some(100),
..Default::default()
},
},
StreamEvent::CostUpdate {
usage: PartialUsage {
completion_tokens: Some(50),
total_tokens: Some(150),
..Default::default()
},
},
StreamEvent::MessageComplete {
full_response: empty_response(),
},
],
));
assert_eq!(state.usage.prompt_tokens, Some(100));
assert_eq!(state.usage.completion_tokens, Some(50));
assert_eq!(state.usage.total_tokens, Some(150));
}
#[test]
fn apply_to_error_event_returns_false_and_marks_errored() {
let mut state = PartialMessageResponse::new();
let cont = state.apply_to(&StreamEvent::Error {
message: "boom".into(),
});
assert!(!cont);
assert!(state.is_errored);
// finalize 仍可调用，错误信息透传由 Provider 流循环处理
let resp = state.finalize().expect("ok even after error");
assert_eq!(resp.stop_reason, StopReason::Stop);
}
#[test]
fn multi_block_ordering_by_index() {
let mut state = PartialMessageResponse::new();
// 故意以 index 3 优先于 index 0 提供，验证 finalize 按 index 升序输出
assert!(apply_seq(
&mut state,
&[
StreamEvent::MessageStart {
id: "r".into(),
model: "m".into(),
},
StreamEvent::ContentBlockStart {
index: 3,
block_type: ContentBlockType::Text,
},
StreamEvent::TextDelta {
text: "second".into(),
},
StreamEvent::ContentBlockEnd { index: 3 },
StreamEvent::ContentBlockStart {
index: 0,
block_type: ContentBlockType::Text,
},
StreamEvent::TextDelta {
text: "first".into(),
},
StreamEvent::ContentBlockEnd { index: 0 },
],
));
let resp = state.finalize().unwrap();
match &resp.message {
Message::Assistant { content } => {
assert_eq!(content.len(), 2);
match (&content[0], &content[1]) {
(
ContentBlock::Text { text: t1 },
ContentBlock::Text { text: t2 },
) => {
assert_eq!(t1, "first");
assert_eq!(t2, "second");
}
_ => panic!("expected Text blocks"),
}
}
_ => panic!("expected Assistant"),
}
}
#[test]
fn message_complete_does_not_double_emit_in_finalize() {
// MessageComplete 仅作完成标记，不修改 blocks
let mut state = PartialMessageResponse::new();
assert!(apply_seq(
&mut state,
&[
StreamEvent::MessageStart {
id: "r".into(),
model: "m".into(),
},
StreamEvent::ContentBlockStart {
index: 0,
block_type: ContentBlockType::Text,
},
StreamEvent::TextDelta {
text: "x".into(),
},
StreamEvent::ContentBlockEnd { index: 0 },
StreamEvent::MessageComplete {
full_response: empty_response(),
},
],
));
assert!(state.is_complete);
let resp = state.finalize().unwrap();
match &resp.message {
Message::Assistant { content } => assert_eq!(content.len(), 1),
_ => panic!("expected Assistant"),
}
}
// ===== JSON roundtrip 测试 =====
#[test]
fn message_response_json_roundtrip() {
let resp = MessageResponse {
id: "resp-99".into(),
model: "gpt-4o".into(),
message: Message::Assistant {
content: vec![
ContentBlock::Text {
text: "hello".into(),
},
ContentBlock::ToolUse {
id: "c".into(),
name: "f".into(),
input: json!({"a": 1}),
},
],
},
usage: Usage::from_input_output(10, 20),
stop_reason: StopReason::ToolUse,
extra: HashMap::from([("trace".to_string(), json!("t-1"))]),
};
let json = serde_json::to_string(&resp).unwrap();
let decoded: MessageResponse = serde_json::from_str(&json).unwrap();
assert_eq!(decoded.id, resp.id);
assert_eq!(decoded.stop_reason, resp.stop_reason);
assert_eq!(decoded.usage.prompt_tokens, 10);
assert_eq!(decoded.usage.completion_tokens, 20);
}
#[test]
fn stream_event_json_roundtrip_each_variant() {
let events = vec![
StreamEvent::MessageStart {
id: "r".into(),
model: "m".into(),
},
StreamEvent::ContentBlockStart {
index: 0,
block_type: ContentBlockType::Text,
},
StreamEvent::ContentBlockEnd { index: 0 },
StreamEvent::TextDelta {
text: "t".into(),
},
StreamEvent::ThinkingDelta {
text: "p".into(),
},
StreamEvent::RefusalDelta {
text: "r".into(),
},
StreamEvent::ToolCallArgumentsDelta {
index: 1,
arguments: "{\"x\":1}".into(),
},
StreamEvent::ToolCallEnd { index: 1 },
StreamEvent::CostUpdate {
usage: PartialUsage {
prompt_tokens: Some(1),
completion_tokens: Some(2),
..Default::default()
},
},
StreamEvent::MessageComplete {
full_response: empty_response(),
},
StreamEvent::Error {
message: "boom".into(),
},
];
for original in events {
let json = serde_json::to_string(&original).unwrap();
let decoded: StreamEvent = serde_json::from_str(&json).unwrap();
assert_eq!(format!("{:?}", original), format!("{:?}", decoded));
}
}
}
src/llm/types/tool.rs
+2 -1
View File
@@ -1,11 +1,12 @@
use serde::{Deserialize, Serialize};
use serde_json::Value;
#[derive(Debug, Clone, Serialize, Deserialize)]
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct OpenaiToolDefinition {
pub name: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub description: Option<String>,
#[serde(default)]
pub parameters: Value,
#[serde(skip_serializing_if = "Option::is_none")]
pub strict: Option<bool>,
src/prompt/composer.rs
+1 -1
View File
@@ -1,4 +1,4 @@
use crate::llm::types::message::{ContentField, OpenaiChatMessage, OpenaiContentPart};
use crate::llm::types::openai_message::{ContentField, OpenaiChatMessage, OpenaiContentPart};
use crate::llm::types::request::OpenaiChatRequest;
use crate::prompt::error::PromptError;
use crate::prompt::template::{PromptTemplate, TemplateContext};