mirror of
https://github.com/farion1231/cc-switch.git
synced 2026-07-25 13:45:03 +08:00
Jason
11f70f676e
- Extract shared map_responses_stop_reason and build_anthropic_usage_from_responses into transform_responses.rs as pub(crate) - Align cache token extraction priority: OpenAI nested details as fallback, direct Anthropic fields as override - Extract resolve_content_index helper to eliminate 3x copy-paste in streaming_responses.rs - Add streaming reasoning/thinking event handlers (response.reasoning.delta/done) - Add explanatory comment to transform_response heuristic detection - Add openai_responses to api_format doc comment and needs_transform test - Add explicit no-op match arms for lifecycle events - Add promptCacheKey to TS ProviderMeta type - Update toast i18n key to be generic for both OpenAI formats (zh/en/ja)
378 lines
20 KiB
Rust
378 lines
20 KiB
Rust
//! 流式响应转换模块
|
||
//!
|
||
//! 实现 OpenAI SSE → Anthropic SSE 格式转换
|
||
|
||
use bytes::Bytes;
|
||
use futures::stream::{Stream, StreamExt};
|
||
use serde::{Deserialize, Serialize};
|
||
use serde_json::json;
|
||
|
||
/// OpenAI 流式响应数据结构
|
||
#[derive(Debug, Deserialize)]
|
||
struct OpenAIStreamChunk {
|
||
id: String,
|
||
model: String,
|
||
choices: Vec<StreamChoice>,
|
||
#[serde(default)]
|
||
usage: Option<Usage>,
|
||
}
|
||
|
||
#[derive(Debug, Deserialize)]
|
||
struct StreamChoice {
|
||
delta: Delta,
|
||
#[serde(default)]
|
||
finish_reason: Option<String>,
|
||
}
|
||
|
||
#[derive(Debug, Deserialize)]
|
||
struct Delta {
|
||
#[serde(default)]
|
||
content: Option<String>,
|
||
#[serde(default)]
|
||
reasoning: Option<String>, // OpenRouter 的推理内容
|
||
#[serde(default)]
|
||
tool_calls: Option<Vec<DeltaToolCall>>,
|
||
}
|
||
|
||
#[derive(Debug, Deserialize, Serialize)]
|
||
struct DeltaToolCall {
|
||
index: usize,
|
||
#[serde(default)]
|
||
id: Option<String>,
|
||
#[serde(rename = "type", default)]
|
||
call_type: Option<String>,
|
||
#[serde(default)]
|
||
function: Option<DeltaFunction>,
|
||
}
|
||
|
||
#[derive(Debug, Deserialize, Serialize)]
|
||
struct DeltaFunction {
|
||
#[serde(default)]
|
||
name: Option<String>,
|
||
#[serde(default)]
|
||
arguments: Option<String>,
|
||
}
|
||
|
||
/// OpenAI 流式响应的 usage 信息（完整版）
|
||
#[derive(Debug, Deserialize)]
|
||
struct Usage {
|
||
#[serde(default)]
|
||
prompt_tokens: u32,
|
||
#[serde(default)]
|
||
completion_tokens: u32,
|
||
#[serde(default)]
|
||
prompt_tokens_details: Option<PromptTokensDetails>,
|
||
/// Some compatible servers return Anthropic-style cache fields directly
|
||
#[serde(default)]
|
||
cache_read_input_tokens: Option<u32>,
|
||
#[serde(default)]
|
||
cache_creation_input_tokens: Option<u32>,
|
||
}
|
||
|
||
/// Nested token details from OpenAI format
|
||
#[derive(Debug, Deserialize)]
|
||
struct PromptTokensDetails {
|
||
#[serde(default)]
|
||
cached_tokens: u32,
|
||
}
|
||
|
||
/// 创建 Anthropic SSE 流
|
||
pub fn create_anthropic_sse_stream(
|
||
stream: impl Stream<Item = Result<Bytes, reqwest::Error>> + Send + 'static,
|
||
) -> impl Stream<Item = Result<Bytes, std::io::Error>> + Send {
|
||
async_stream::stream! {
|
||
let mut buffer = String::new();
|
||
let mut message_id = None;
|
||
let mut current_model = None;
|
||
let mut content_index = 0;
|
||
let mut has_sent_message_start = false;
|
||
let mut current_block_type: Option<String> = None;
|
||
let mut tool_call_id = None;
|
||
|
||
tokio::pin!(stream);
|
||
|
||
while let Some(chunk) = stream.next().await {
|
||
match chunk {
|
||
Ok(bytes) => {
|
||
let text = String::from_utf8_lossy(&bytes);
|
||
buffer.push_str(&text);
|
||
|
||
while let Some(pos) = buffer.find("\n\n") {
|
||
let line = buffer[..pos].to_string();
|
||
buffer = buffer[pos + 2..].to_string();
|
||
|
||
if line.trim().is_empty() {
|
||
continue;
|
||
}
|
||
|
||
for l in line.lines() {
|
||
if let Some(data) = l.strip_prefix("data: ") {
|
||
if data.trim() == "[DONE]" {
|
||
log::debug!("[Claude/OpenRouter] <<< OpenAI SSE: [DONE]");
|
||
let event = json!({"type": "message_stop"});
|
||
let sse_data = format!("event: message_stop\ndata: {}\n\n",
|
||
serde_json::to_string(&event).unwrap_or_default());
|
||
log::debug!("[Claude/OpenRouter] >>> Anthropic SSE: message_stop");
|
||
yield Ok(Bytes::from(sse_data));
|
||
continue;
|
||
}
|
||
|
||
if let Ok(chunk) = serde_json::from_str::<OpenAIStreamChunk>(data) {
|
||
log::debug!("[Claude/OpenRouter] <<< SSE chunk received");
|
||
|
||
if message_id.is_none() {
|
||
message_id = Some(chunk.id.clone());
|
||
}
|
||
if current_model.is_none() {
|
||
current_model = Some(chunk.model.clone());
|
||
}
|
||
|
||
if let Some(choice) = chunk.choices.first() {
|
||
if !has_sent_message_start {
|
||
// Build usage with cache tokens if available from first chunk
|
||
let mut start_usage = json!({
|
||
"input_tokens": 0,
|
||
"output_tokens": 0
|
||
});
|
||
if let Some(u) = &chunk.usage {
|
||
start_usage["input_tokens"] = json!(u.prompt_tokens);
|
||
if let Some(cached) = extract_cache_read_tokens(u) {
|
||
start_usage["cache_read_input_tokens"] = json!(cached);
|
||
}
|
||
if let Some(created) = u.cache_creation_input_tokens {
|
||
start_usage["cache_creation_input_tokens"] = json!(created);
|
||
}
|
||
}
|
||
|
||
let event = json!({
|
||
"type": "message_start",
|
||
"message": {
|
||
"id": message_id.clone().unwrap_or_default(),
|
||
"type": "message",
|
||
"role": "assistant",
|
||
"model": current_model.clone().unwrap_or_default(),
|
||
"usage": start_usage
|
||
}
|
||
});
|
||
let sse_data = format!("event: message_start\ndata: {}\n\n",
|
||
serde_json::to_string(&event).unwrap_or_default());
|
||
yield Ok(Bytes::from(sse_data));
|
||
has_sent_message_start = true;
|
||
}
|
||
|
||
// 处理 reasoning（thinking）
|
||
if let Some(reasoning) = &choice.delta.reasoning {
|
||
if current_block_type.is_none() {
|
||
let event = json!({
|
||
"type": "content_block_start",
|
||
"index": content_index,
|
||
"content_block": {
|
||
"type": "thinking",
|
||
"thinking": ""
|
||
}
|
||
});
|
||
let sse_data = format!("event: content_block_start\ndata: {}\n\n",
|
||
serde_json::to_string(&event).unwrap_or_default());
|
||
yield Ok(Bytes::from(sse_data));
|
||
current_block_type = Some("thinking".to_string());
|
||
}
|
||
|
||
let event = json!({
|
||
"type": "content_block_delta",
|
||
"index": content_index,
|
||
"delta": {
|
||
"type": "thinking_delta",
|
||
"thinking": reasoning
|
||
}
|
||
});
|
||
let sse_data = format!("event: content_block_delta\ndata: {}\n\n",
|
||
serde_json::to_string(&event).unwrap_or_default());
|
||
yield Ok(Bytes::from(sse_data));
|
||
}
|
||
|
||
// 处理文本内容
|
||
if let Some(content) = &choice.delta.content {
|
||
if !content.is_empty() {
|
||
if current_block_type.as_deref() != Some("text") {
|
||
if current_block_type.is_some() {
|
||
let event = json!({
|
||
"type": "content_block_stop",
|
||
"index": content_index
|
||
});
|
||
let sse_data = format!("event: content_block_stop\ndata: {}\n\n",
|
||
serde_json::to_string(&event).unwrap_or_default());
|
||
yield Ok(Bytes::from(sse_data));
|
||
content_index += 1;
|
||
}
|
||
|
||
let event = json!({
|
||
"type": "content_block_start",
|
||
"index": content_index,
|
||
"content_block": {
|
||
"type": "text",
|
||
"text": ""
|
||
}
|
||
});
|
||
let sse_data = format!("event: content_block_start\ndata: {}\n\n",
|
||
serde_json::to_string(&event).unwrap_or_default());
|
||
yield Ok(Bytes::from(sse_data));
|
||
current_block_type = Some("text".to_string());
|
||
}
|
||
|
||
let event = json!({
|
||
"type": "content_block_delta",
|
||
"index": content_index,
|
||
"delta": {
|
||
"type": "text_delta",
|
||
"text": content
|
||
}
|
||
});
|
||
let sse_data = format!("event: content_block_delta\ndata: {}\n\n",
|
||
serde_json::to_string(&event).unwrap_or_default());
|
||
yield Ok(Bytes::from(sse_data));
|
||
}
|
||
}
|
||
|
||
// 处理工具调用
|
||
if let Some(tool_calls) = &choice.delta.tool_calls {
|
||
for tool_call in tool_calls {
|
||
if let Some(id) = &tool_call.id {
|
||
if current_block_type.is_some() {
|
||
let event = json!({
|
||
"type": "content_block_stop",
|
||
"index": content_index
|
||
});
|
||
let sse_data = format!("event: content_block_stop\ndata: {}\n\n",
|
||
serde_json::to_string(&event).unwrap_or_default());
|
||
yield Ok(Bytes::from(sse_data));
|
||
content_index += 1;
|
||
}
|
||
|
||
tool_call_id = Some(id.clone());
|
||
}
|
||
|
||
if let Some(function) = &tool_call.function {
|
||
if let Some(name) = &function.name {
|
||
let event = json!({
|
||
"type": "content_block_start",
|
||
"index": content_index,
|
||
"content_block": {
|
||
"type": "tool_use",
|
||
"id": tool_call_id.clone().unwrap_or_default(),
|
||
"name": name
|
||
}
|
||
});
|
||
let sse_data = format!("event: content_block_start\ndata: {}\n\n",
|
||
serde_json::to_string(&event).unwrap_or_default());
|
||
yield Ok(Bytes::from(sse_data));
|
||
current_block_type = Some("tool_use".to_string());
|
||
}
|
||
|
||
if let Some(args) = &function.arguments {
|
||
let event = json!({
|
||
"type": "content_block_delta",
|
||
"index": content_index,
|
||
"delta": {
|
||
"type": "input_json_delta",
|
||
"partial_json": args
|
||
}
|
||
});
|
||
let sse_data = format!("event: content_block_delta\ndata: {}\n\n",
|
||
serde_json::to_string(&event).unwrap_or_default());
|
||
yield Ok(Bytes::from(sse_data));
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
// 处理 finish_reason
|
||
if let Some(finish_reason) = &choice.finish_reason {
|
||
if current_block_type.is_some() {
|
||
let event = json!({
|
||
"type": "content_block_stop",
|
||
"index": content_index
|
||
});
|
||
let sse_data = format!("event: content_block_stop\ndata: {}\n\n",
|
||
serde_json::to_string(&event).unwrap_or_default());
|
||
yield Ok(Bytes::from(sse_data));
|
||
}
|
||
|
||
let stop_reason = map_stop_reason(Some(finish_reason));
|
||
// Build usage with cache token fields
|
||
let usage_json = chunk.usage.as_ref().map(|u| {
|
||
let mut uj = json!({
|
||
"input_tokens": u.prompt_tokens,
|
||
"output_tokens": u.completion_tokens
|
||
});
|
||
if let Some(cached) = extract_cache_read_tokens(u) {
|
||
uj["cache_read_input_tokens"] = json!(cached);
|
||
}
|
||
if let Some(created) = u.cache_creation_input_tokens {
|
||
uj["cache_creation_input_tokens"] = json!(created);
|
||
}
|
||
uj
|
||
});
|
||
let event = json!({
|
||
"type": "message_delta",
|
||
"delta": {
|
||
"stop_reason": stop_reason,
|
||
"stop_sequence": null
|
||
},
|
||
"usage": usage_json
|
||
});
|
||
let sse_data = format!("event: message_delta\ndata: {}\n\n",
|
||
serde_json::to_string(&event).unwrap_or_default());
|
||
yield Ok(Bytes::from(sse_data));
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
Err(e) => {
|
||
log::error!("Stream error: {e}");
|
||
let error_event = json!({
|
||
"type": "error",
|
||
"error": {
|
||
"type": "stream_error",
|
||
"message": format!("Stream error: {e}")
|
||
}
|
||
});
|
||
let sse_data = format!("event: error\ndata: {}\n\n",
|
||
serde_json::to_string(&error_event).unwrap_or_default());
|
||
yield Ok(Bytes::from(sse_data));
|
||
break;
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
/// Extract cache_read tokens from Usage, checking both direct field and nested details
|
||
fn extract_cache_read_tokens(usage: &Usage) -> Option<u32> {
|
||
// Direct field takes priority (compatible servers)
|
||
if let Some(v) = usage.cache_read_input_tokens {
|
||
return Some(v);
|
||
}
|
||
// OpenAI standard: prompt_tokens_details.cached_tokens
|
||
usage
|
||
.prompt_tokens_details
|
||
.as_ref()
|
||
.map(|d| d.cached_tokens)
|
||
.filter(|&v| v > 0)
|
||
}
|
||
|
||
/// 映射停止原因
|
||
fn map_stop_reason(finish_reason: Option<&str>) -> Option<String> {
|
||
finish_reason.map(|r| {
|
||
match r {
|
||
"tool_calls" => "tool_use",
|
||
"stop" => "end_turn",
|
||
"length" => "max_tokens",
|
||
_ => "end_turn",
|
||
}
|
||
.to_string()
|
||
})
|
||
}
|