mirror of
https://github.com/farion1231/cc-switch.git
synced 2026-07-26 06:24:32 +08:00
Dex Miller
6dd809701b
* refactor(proxy): simplify logging for better readability - Delete 17 verbose debug logs from handlers, streaming, and response_processor - Convert excessive INFO logs to DEBUG level for internal processing details - Add 2 critical INFO logs in forwarder.rs for failover scenarios: - Log when switching to next provider after failure - Log when all providers have been exhausted - Fix clippy uninlined_format_args warning This reduces log noise while maintaining visibility into key user-facing decisions. * fix: replace unsafe unwrap() calls with proper error handling - database/dao/mcp.rs: Use map_err for serde_json serialization - database/dao/providers.rs: Use map_err for settings_config and meta serialization - commands/misc.rs: Use expect() for compile-time regex pattern - services/prompt.rs: Use unwrap_or_default() for SystemTime - deeplink/provider.rs: Replace unwrap() with is_none_or pattern for Option checks Reduces potential panic points from 26 to 1 (static regex init, safe). * refactor(proxy): simplify verbose logging output - Remove response JSON full output logging in response_processor - Remove per-request INFO logs in provider_router (failover status, provider selection) - Change model mapping log from INFO to DEBUG - Change usage logging failure from INFO to WARN - Remove redundant debug logs for circuit breaker operations Reduces log noise significantly while preserving important warnings and errors. * feat(proxy): add structured log codes for i18n support Add error code system to proxy module logs for multi-language support: - CB-001~006: Circuit breaker state transitions and triggers - SRV-001~004: Proxy server lifecycle events - FWD-001~002: Request forwarding and failover - FO-001~005: Failover switch operations - USG-001~002: Usage logging errors Log format: [CODE] Chinese message Frontend/log tools can map codes to any language. New file: src/proxy/log_codes.rs - centralized code definitions * chore: bump version to 3.9.1 * style: format code with prettier and rustfmt * fix(ui): allow number inputs to be fully cleared before saving - Convert numeric state to string type for controlled inputs - Use isNaN() check instead of || fallback to allow 0 values - Apply fix to ProxyPanel, CircuitBreakerConfigPanel, AutoFailoverConfigPanel, and ModelTestConfigPanel * feat(pricing): support @ separator in model name matching - Refactor model name cleaning into chained method calls - Add @ to - replacement (e.g., gpt-5.2-codex@low → gpt-5.2-codex-low) - Add test case for @ separator matching * fix(proxy): improve validation and error handling in proxy config panels - Add StopTimeout/StopFailed error types for proper stop() error reporting - Replace silent clamp with validation-and-block in config panels - Add listenAddress format validation in ProxyPanel - Use log_codes constants instead of hardcoded strings - Use once_cell::Lazy for regex precompilation * fix(proxy): harden error handling and input validation - Handle RwLock poisoning in settings.rs with unwrap_or_else - Add fallback for dirs::home_dir() in config modules - Normalize localhost to 127.0.0.1 in ProxyPanel - Format IPv6 addresses with brackets for valid URLs - Strict port validation with pure digit regex - Treat NaN as validation failure in config panels - Log warning on cost_multiplier parse failure - Align timeoutSeconds range to [0, 300] across all panels
330 lines
18 KiB
Rust
330 lines
18 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,
|
||
}
|
||
|
||
/// 创建 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) {
|
||
// 仅在 DEBUG 级别简短记录 SSE 事件
|
||
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 {
|
||
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": {
|
||
"input_tokens": 0,
|
||
"output_tokens": 0
|
||
}
|
||
}
|
||
});
|
||
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));
|
||
// 构建 usage 信息，包含 input_tokens 和 output_tokens
|
||
let usage_json = chunk.usage.as_ref().map(|u| json!({
|
||
"input_tokens": u.prompt_tokens,
|
||
"output_tokens": u.completion_tokens
|
||
}));
|
||
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;
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
/// 映射停止原因
|
||
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()
|
||
})
|
||
}
|