mirror of
https://github.com/farion1231/cc-switch.git
synced 2026-05-28 09:10:18 +08:00
fix: correct OpenAI ChatCompletion to Anthropic Messages streaming conversion
Rewrite tool call handling in streaming format conversion to properly track multiple concurrent tool blocks with independent Anthropic content indices. Fix block interleaving (thinking/text/tool_use) with correct content_block_start/stop events, buffer tool arguments until both id and name are available, and add tool result message conversion in transform.
This commit is contained in:
Jason
@@ -6,6 +6,7 @@ use bytes::Bytes;
|
||||
use futures::stream::{Stream, StreamExt};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use serde_json::json;
|
||||
use std::collections::{HashMap, HashSet};
|
||||
|
||||
/// OpenAI 流式响应数据结构
|
||||
#[derive(Debug, Deserialize)]
|
||||
@@ -76,6 +77,15 @@ struct PromptTokensDetails {
|
||||
cached_tokens: u32,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
struct ToolBlockState {
|
||||
anthropic_index: u32,
|
||||
id: String,
|
||||
name: String,
|
||||
started: bool,
|
||||
pending_args: String,
|
||||
}
|
||||
|
||||
/// 创建 Anthropic SSE 流
|
||||
pub fn create_anthropic_sse_stream(
|
||||
stream: impl Stream<Item = Result<Bytes, reqwest::Error>> + Send + 'static,
|
||||
@@ -84,10 +94,12 @@ pub fn create_anthropic_sse_stream(
|
||||
let mut buffer = String::new();
|
||||
let mut message_id = None;
|
||||
let mut current_model = None;
|
||||
let mut content_index = 0;
|
||||
let mut next_content_index: u32 = 0;
|
||||
let mut has_sent_message_start = false;
|
||||
let mut current_block_type: Option<String> = None;
|
||||
let mut tool_call_id = None;
|
||||
let mut current_non_tool_block_type: Option<&'static str> = None;
|
||||
let mut current_non_tool_block_index: Option<u32> = None;
|
||||
let mut tool_blocks_by_index: HashMap<usize, ToolBlockState> = HashMap::new();
|
||||
let mut open_tool_block_indices: HashSet<u32> = HashSet::new();
|
||||
|
||||
tokio::pin!(stream);
|
||||
|
||||
@@ -162,10 +174,21 @@ pub fn create_anthropic_sse_stream(
|
||||
|
||||
// 处理 reasoning（thinking）
|
||||
if let Some(reasoning) = &choice.delta.reasoning {
|
||||
if current_block_type.is_none() {
|
||||
if current_non_tool_block_type != Some("thinking") {
|
||||
if let Some(index) = current_non_tool_block_index.take() {
|
||||
let event = json!({
|
||||
"type": "content_block_stop",
|
||||
"index": 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 index = next_content_index;
|
||||
next_content_index += 1;
|
||||
let event = json!({
|
||||
"type": "content_block_start",
|
||||
"index": content_index,
|
||||
"index": index,
|
||||
"content_block": {
|
||||
"type": "thinking",
|
||||
"thinking": ""
|
||||
@@ -174,57 +197,17 @@ pub fn create_anthropic_sse_stream(
|
||||
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());
|
||||
current_non_tool_block_type = Some("thinking");
|
||||
current_non_tool_block_index = Some(index);
|
||||
}
|
||||
|
||||
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());
|
||||
}
|
||||
|
||||
if let Some(index) = current_non_tool_block_index {
|
||||
let event = json!({
|
||||
"type": "content_block_delta",
|
||||
"index": content_index,
|
||||
"index": index,
|
||||
"delta": {
|
||||
"type": "text_delta",
|
||||
"text": content
|
||||
"type": "thinking_delta",
|
||||
"thinking": reasoning
|
||||
}
|
||||
});
|
||||
let sse_data = format!("event: content_block_delta\ndata: {}\n\n",
|
||||
@@ -233,69 +216,272 @@ pub fn create_anthropic_sse_stream(
|
||||
}
|
||||
}
|
||||
|
||||
// 处理工具调用
|
||||
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() {
|
||||
// 处理文本内容
|
||||
if let Some(content) = &choice.delta.content {
|
||||
if !content.is_empty() {
|
||||
if current_non_tool_block_type != Some("text") {
|
||||
if let Some(index) = current_non_tool_block_index.take() {
|
||||
let event = json!({
|
||||
"type": "content_block_stop",
|
||||
"index": content_index
|
||||
"index": 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());
|
||||
let index = next_content_index;
|
||||
next_content_index += 1;
|
||||
let event = json!({
|
||||
"type": "content_block_start",
|
||||
"index": 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_non_tool_block_type = Some("text");
|
||||
current_non_tool_block_index = Some(index);
|
||||
}
|
||||
|
||||
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
|
||||
if let Some(index) = current_non_tool_block_index {
|
||||
let event = json!({
|
||||
"type": "content_block_delta",
|
||||
"index": 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 {
|
||||
if let Some(index) = current_non_tool_block_index.take() {
|
||||
let event = json!({
|
||||
"type": "content_block_stop",
|
||||
"index": 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));
|
||||
}
|
||||
current_non_tool_block_type = None;
|
||||
|
||||
for tool_call in tool_calls {
|
||||
let (
|
||||
anthropic_index,
|
||||
id,
|
||||
name,
|
||||
should_start,
|
||||
pending_after_start,
|
||||
immediate_delta,
|
||||
) = {
|
||||
let state = tool_blocks_by_index
|
||||
.entry(tool_call.index)
|
||||
.or_insert_with(|| {
|
||||
let index = next_content_index;
|
||||
next_content_index += 1;
|
||||
ToolBlockState {
|
||||
anthropic_index: index,
|
||||
id: String::new(),
|
||||
name: String::new(),
|
||||
started: false,
|
||||
pending_args: String::new(),
|
||||
}
|
||||
});
|
||||
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(id) = &tool_call.id {
|
||||
state.id = id.clone();
|
||||
}
|
||||
if let Some(function) = &tool_call.function {
|
||||
if let Some(name) = &function.name {
|
||||
state.name = name.clone();
|
||||
}
|
||||
}
|
||||
|
||||
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));
|
||||
let should_start =
|
||||
!state.started
|
||||
&& !state.id.is_empty()
|
||||
&& !state.name.is_empty();
|
||||
if should_start {
|
||||
state.started = true;
|
||||
}
|
||||
let pending_after_start = if should_start
|
||||
&& !state.pending_args.is_empty()
|
||||
{
|
||||
Some(std::mem::take(&mut state.pending_args))
|
||||
} else {
|
||||
None
|
||||
};
|
||||
let args_delta = tool_call
|
||||
.function
|
||||
.as_ref()
|
||||
.and_then(|f| f.arguments.clone());
|
||||
let immediate_delta = if let Some(args) = args_delta {
|
||||
if state.started {
|
||||
Some(args)
|
||||
} else {
|
||||
state.pending_args.push_str(&args);
|
||||
None
|
||||
}
|
||||
} else {
|
||||
None
|
||||
};
|
||||
(
|
||||
state.anthropic_index,
|
||||
state.id.clone(),
|
||||
state.name.clone(),
|
||||
should_start,
|
||||
pending_after_start,
|
||||
immediate_delta,
|
||||
)
|
||||
};
|
||||
|
||||
if should_start {
|
||||
let event = json!({
|
||||
"type": "content_block_start",
|
||||
"index": anthropic_index,
|
||||
"content_block": {
|
||||
"type": "tool_use",
|
||||
"id": id,
|
||||
"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));
|
||||
open_tool_block_indices.insert(anthropic_index);
|
||||
}
|
||||
|
||||
if let Some(args) = pending_after_start {
|
||||
let event = json!({
|
||||
"type": "content_block_delta",
|
||||
"index": anthropic_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));
|
||||
}
|
||||
|
||||
if let Some(args) = immediate_delta {
|
||||
let event = json!({
|
||||
"type": "content_block_delta",
|
||||
"index": anthropic_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() {
|
||||
if let Some(index) = current_non_tool_block_index.take() {
|
||||
let event = json!({
|
||||
"type": "content_block_stop",
|
||||
"index": content_index
|
||||
"index": 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));
|
||||
}
|
||||
current_non_tool_block_type = None;
|
||||
|
||||
// Late start for blocks that accumulated args before id/name arrived.
|
||||
let mut late_tool_starts: Vec<(u32, String, String, String)> =
|
||||
Vec::new();
|
||||
for (tool_idx, state) in tool_blocks_by_index.iter_mut() {
|
||||
if state.started {
|
||||
continue;
|
||||
}
|
||||
let has_payload = !state.pending_args.is_empty()
|
||||
|| !state.id.is_empty()
|
||||
|| !state.name.is_empty();
|
||||
if !has_payload {
|
||||
continue;
|
||||
}
|
||||
let fallback_id = if state.id.is_empty() {
|
||||
format!("tool_call_{tool_idx}")
|
||||
} else {
|
||||
state.id.clone()
|
||||
};
|
||||
let fallback_name = if state.name.is_empty() {
|
||||
"unknown_tool".to_string()
|
||||
} else {
|
||||
state.name.clone()
|
||||
};
|
||||
state.started = true;
|
||||
let pending = std::mem::take(&mut state.pending_args);
|
||||
late_tool_starts.push((
|
||||
state.anthropic_index,
|
||||
fallback_id,
|
||||
fallback_name,
|
||||
pending,
|
||||
));
|
||||
}
|
||||
late_tool_starts.sort_unstable_by_key(|(index, _, _, _)| *index);
|
||||
for (index, id, name, pending) in late_tool_starts {
|
||||
let event = json!({
|
||||
"type": "content_block_start",
|
||||
"index": index,
|
||||
"content_block": {
|
||||
"type": "tool_use",
|
||||
"id": id,
|
||||
"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));
|
||||
open_tool_block_indices.insert(index);
|
||||
if !pending.is_empty() {
|
||||
let delta_event = json!({
|
||||
"type": "content_block_delta",
|
||||
"index": index,
|
||||
"delta": {
|
||||
"type": "input_json_delta",
|
||||
"partial_json": pending
|
||||
}
|
||||
});
|
||||
let delta_sse = format!("event: content_block_delta\ndata: {}\n\n",
|
||||
serde_json::to_string(&delta_event).unwrap_or_default());
|
||||
yield Ok(Bytes::from(delta_sse));
|
||||
}
|
||||
}
|
||||
|
||||
if !open_tool_block_indices.is_empty() {
|
||||
let mut tool_indices: Vec<u32> =
|
||||
open_tool_block_indices.iter().copied().collect();
|
||||
tool_indices.sort_unstable();
|
||||
for index in tool_indices {
|
||||
let event = json!({
|
||||
"type": "content_block_stop",
|
||||
"index": 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));
|
||||
}
|
||||
open_tool_block_indices.clear();
|
||||
}
|
||||
|
||||
let stop_reason = map_stop_reason(Some(finish_reason));
|
||||
// Build usage with cache token fields
|
||||
@@ -367,11 +553,183 @@ fn extract_cache_read_tokens(usage: &Usage) -> Option<u32> {
|
||||
fn map_stop_reason(finish_reason: Option<&str>) -> Option<String> {
|
||||
finish_reason.map(|r| {
|
||||
match r {
|
||||
"tool_calls" => "tool_use",
|
||||
"tool_calls" | "function_call" => "tool_use",
|
||||
"stop" => "end_turn",
|
||||
"length" => "max_tokens",
|
||||
_ => "end_turn",
|
||||
"content_filter" => "end_turn",
|
||||
other => {
|
||||
log::warn!("[Claude/OpenRouter] Unknown finish_reason in streaming: {other}");
|
||||
"end_turn"
|
||||
}
|
||||
}
|
||||
.to_string()
|
||||
})
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use futures::stream;
|
||||
use futures::StreamExt;
|
||||
use serde_json::Value;
|
||||
use std::collections::HashMap;
|
||||
|
||||
#[test]
|
||||
fn test_map_stop_reason_legacy_and_filtered_values() {
|
||||
assert_eq!(
|
||||
map_stop_reason(Some("function_call")),
|
||||
Some("tool_use".to_string())
|
||||
);
|
||||
assert_eq!(
|
||||
map_stop_reason(Some("content_filter")),
|
||||
Some("end_turn".to_string())
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_streaming_tool_calls_routed_by_index() {
|
||||
let input = concat!(
|
||||
"data: {\"id\":\"chatcmpl_1\",\"model\":\"gpt-4o\",\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"call_0\",\"type\":\"function\",\"function\":{\"name\":\"first_tool\"}}]}}]}\n\n",
|
||||
"data: {\"id\":\"chatcmpl_1\",\"model\":\"gpt-4o\",\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":1,\"id\":\"call_1\",\"type\":\"function\",\"function\":{\"name\":\"second_tool\"}}]}}]}\n\n",
|
||||
"data: {\"id\":\"chatcmpl_1\",\"model\":\"gpt-4o\",\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":1,\"function\":{\"arguments\":\"{\\\"b\\\":2}\"}}]}}]}\n\n",
|
||||
"data: {\"id\":\"chatcmpl_1\",\"model\":\"gpt-4o\",\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"function\":{\"arguments\":\"{\\\"a\\\":1}\"}}]}}]}\n\n",
|
||||
"data: {\"id\":\"chatcmpl_1\",\"model\":\"gpt-4o\",\"choices\":[{\"delta\":{},\"finish_reason\":\"tool_calls\"}],\"usage\":{\"prompt_tokens\":8,\"completion_tokens\":4}}\n\n",
|
||||
"data: [DONE]\n\n"
|
||||
);
|
||||
|
||||
let upstream = stream::iter(vec![Ok(Bytes::from(input.as_bytes().to_vec()))]);
|
||||
let converted = create_anthropic_sse_stream(upstream);
|
||||
let chunks: Vec<_> = converted.collect().await;
|
||||
|
||||
let merged = chunks
|
||||
.into_iter()
|
||||
.map(|chunk| String::from_utf8_lossy(chunk.unwrap().as_ref()).to_string())
|
||||
.collect::<String>();
|
||||
|
||||
let events: Vec<Value> = merged
|
||||
.split("\n\n")
|
||||
.filter_map(|block| {
|
||||
let data = block.lines().find_map(|line| line.strip_prefix("data: "))?;
|
||||
serde_json::from_str::<Value>(data).ok()
|
||||
})
|
||||
.collect();
|
||||
|
||||
let mut tool_index_by_call: HashMap<String, u64> = HashMap::new();
|
||||
for event in &events {
|
||||
if event.get("type").and_then(|v| v.as_str()) == Some("content_block_start")
|
||||
&& event.pointer("/content_block/type").and_then(|v| v.as_str()) == Some("tool_use")
|
||||
{
|
||||
if let (Some(call_id), Some(index)) = (
|
||||
event.pointer("/content_block/id").and_then(|v| v.as_str()),
|
||||
event.get("index").and_then(|v| v.as_u64()),
|
||||
) {
|
||||
tool_index_by_call.insert(call_id.to_string(), index);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
assert_eq!(tool_index_by_call.len(), 2);
|
||||
assert_ne!(
|
||||
tool_index_by_call.get("call_0"),
|
||||
tool_index_by_call.get("call_1")
|
||||
);
|
||||
|
||||
let deltas: Vec<(u64, String)> = events
|
||||
.iter()
|
||||
.filter(|event| {
|
||||
event.get("type").and_then(|v| v.as_str()) == Some("content_block_delta")
|
||||
&& event.pointer("/delta/type").and_then(|v| v.as_str())
|
||||
== Some("input_json_delta")
|
||||
})
|
||||
.filter_map(|event| {
|
||||
let index = event.get("index").and_then(|v| v.as_u64())?;
|
||||
let partial_json = event
|
||||
.pointer("/delta/partial_json")
|
||||
.and_then(|v| v.as_str())?
|
||||
.to_string();
|
||||
Some((index, partial_json))
|
||||
})
|
||||
.collect();
|
||||
|
||||
assert_eq!(deltas.len(), 2);
|
||||
let second_idx = deltas
|
||||
.iter()
|
||||
.find_map(|(index, payload)| (payload == "{\"b\":2}").then_some(*index))
|
||||
.unwrap();
|
||||
let first_idx = deltas
|
||||
.iter()
|
||||
.find_map(|(index, payload)| (payload == "{\"a\":1}").then_some(*index))
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(second_idx, *tool_index_by_call.get("call_1").unwrap());
|
||||
assert_eq!(first_idx, *tool_index_by_call.get("call_0").unwrap());
|
||||
|
||||
assert!(events.iter().any(|event| {
|
||||
event.get("type").and_then(|v| v.as_str()) == Some("message_delta")
|
||||
&& event.pointer("/delta/stop_reason").and_then(|v| v.as_str())
|
||||
== Some("tool_use")
|
||||
}));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_streaming_delays_tool_start_until_id_and_name_ready() {
|
||||
let input = concat!(
|
||||
"data: {\"id\":\"chatcmpl_2\",\"model\":\"gpt-4o\",\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"function\":{\"arguments\":\"{\\\"a\\\":\"}}]}}]}\n\n",
|
||||
"data: {\"id\":\"chatcmpl_2\",\"model\":\"gpt-4o\",\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"call_0\",\"type\":\"function\",\"function\":{\"name\":\"first_tool\"}}]}}]}\n\n",
|
||||
"data: {\"id\":\"chatcmpl_2\",\"model\":\"gpt-4o\",\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"function\":{\"arguments\":\"1}\"}}]}}]}\n\n",
|
||||
"data: {\"id\":\"chatcmpl_2\",\"model\":\"gpt-4o\",\"choices\":[{\"delta\":{},\"finish_reason\":\"tool_calls\"}],\"usage\":{\"prompt_tokens\":6,\"completion_tokens\":2}}\n\n",
|
||||
"data: [DONE]\n\n"
|
||||
);
|
||||
|
||||
let upstream = stream::iter(vec![Ok(Bytes::from(input.as_bytes().to_vec()))]);
|
||||
let converted = create_anthropic_sse_stream(upstream);
|
||||
let chunks: Vec<_> = converted.collect().await;
|
||||
let merged = chunks
|
||||
.into_iter()
|
||||
.map(|chunk| String::from_utf8_lossy(chunk.unwrap().as_ref()).to_string())
|
||||
.collect::<String>();
|
||||
|
||||
let events: Vec<Value> = merged
|
||||
.split("\n\n")
|
||||
.filter_map(|block| {
|
||||
let data = block.lines().find_map(|line| line.strip_prefix("data: "))?;
|
||||
serde_json::from_str::<Value>(data).ok()
|
||||
})
|
||||
.collect();
|
||||
|
||||
let starts: Vec<&Value> = events
|
||||
.iter()
|
||||
.filter(|event| {
|
||||
event.get("type").and_then(|v| v.as_str()) == Some("content_block_start")
|
||||
&& event.pointer("/content_block/type").and_then(|v| v.as_str()) == Some("tool_use")
|
||||
})
|
||||
.collect();
|
||||
assert_eq!(starts.len(), 1);
|
||||
assert_eq!(
|
||||
starts[0]
|
||||
.pointer("/content_block/id")
|
||||
.and_then(|v| v.as_str())
|
||||
.unwrap_or(""),
|
||||
"call_0"
|
||||
);
|
||||
assert_eq!(
|
||||
starts[0]
|
||||
.pointer("/content_block/name")
|
||||
.and_then(|v| v.as_str())
|
||||
.unwrap_or(""),
|
||||
"first_tool"
|
||||
);
|
||||
|
||||
let deltas: Vec<&str> = events
|
||||
.iter()
|
||||
.filter(|event| {
|
||||
event.get("type").and_then(|v| v.as_str()) == Some("content_block_delta")
|
||||
&& event.pointer("/delta/type").and_then(|v| v.as_str())
|
||||
== Some("input_json_delta")
|
||||
})
|
||||
.filter_map(|event| event.pointer("/delta/partial_json").and_then(|v| v.as_str()))
|
||||
.collect();
|
||||
assert!(deltas.contains(&"{\"a\":"));
|
||||
assert!(deltas.contains(&"1}"));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -272,16 +272,49 @@ pub fn openai_to_anthropic(body: Value) -> Result<Value, ProxyError> {
|
||||
.ok_or_else(|| ProxyError::TransformError("No message in choice".to_string()))?;
|
||||
|
||||
let mut content = Vec::new();
|
||||
let mut has_tool_use = false;
|
||||
|
||||
// 文本内容
|
||||
if let Some(text) = message.get("content").and_then(|c| c.as_str()) {
|
||||
if !text.is_empty() {
|
||||
content.push(json!({"type": "text", "text": text}));
|
||||
// 文本/拒绝内容
|
||||
if let Some(msg_content) = message.get("content") {
|
||||
if let Some(text) = msg_content.as_str() {
|
||||
if !text.is_empty() {
|
||||
content.push(json!({"type": "text", "text": text}));
|
||||
}
|
||||
} else if let Some(parts) = msg_content.as_array() {
|
||||
for part in parts {
|
||||
let part_type = part.get("type").and_then(|t| t.as_str()).unwrap_or("");
|
||||
match part_type {
|
||||
"text" | "output_text" => {
|
||||
if let Some(text) = part.get("text").and_then(|t| t.as_str()) {
|
||||
if !text.is_empty() {
|
||||
content.push(json!({"type": "text", "text": text}));
|
||||
}
|
||||
}
|
||||
}
|
||||
"refusal" => {
|
||||
if let Some(refusal) = part.get("refusal").and_then(|r| r.as_str()) {
|
||||
if !refusal.is_empty() {
|
||||
content.push(json!({"type": "text", "text": refusal}));
|
||||
}
|
||||
}
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
// Some providers put refusal at message-level.
|
||||
if let Some(refusal) = message.get("refusal").and_then(|r| r.as_str()) {
|
||||
if !refusal.is_empty() {
|
||||
content.push(json!({"type": "text", "text": refusal}));
|
||||
}
|
||||
}
|
||||
|
||||
// 工具调用
|
||||
// 工具调用（tool_calls）
|
||||
if let Some(tool_calls) = message.get("tool_calls").and_then(|t| t.as_array()) {
|
||||
if !tool_calls.is_empty() {
|
||||
has_tool_use = true;
|
||||
}
|
||||
for tc in tool_calls {
|
||||
let id = tc.get("id").and_then(|i| i.as_str()).unwrap_or("");
|
||||
let empty_obj = json!({});
|
||||
@@ -301,6 +334,33 @@ pub fn openai_to_anthropic(body: Value) -> Result<Value, ProxyError> {
|
||||
}));
|
||||
}
|
||||
}
|
||||
// 兼容旧格式（function_call）
|
||||
if !has_tool_use {
|
||||
if let Some(function_call) = message.get("function_call") {
|
||||
let id = function_call.get("id").and_then(|i| i.as_str()).unwrap_or("");
|
||||
let name = function_call
|
||||
.get("name")
|
||||
.and_then(|n| n.as_str())
|
||||
.unwrap_or("");
|
||||
let has_arguments = function_call.get("arguments").is_some();
|
||||
|
||||
let input = match function_call.get("arguments") {
|
||||
Some(Value::String(s)) => serde_json::from_str(s).unwrap_or(json!({})),
|
||||
Some(v @ Value::Object(_)) | Some(v @ Value::Array(_)) => v.clone(),
|
||||
_ => json!({}),
|
||||
};
|
||||
|
||||
if !name.is_empty() || has_arguments {
|
||||
content.push(json!({
|
||||
"type": "tool_use",
|
||||
"id": id,
|
||||
"name": name,
|
||||
"input": input
|
||||
}));
|
||||
has_tool_use = true;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 映射 finish_reason → stop_reason
|
||||
let stop_reason = choice
|
||||
@@ -309,9 +369,14 @@ pub fn openai_to_anthropic(body: Value) -> Result<Value, ProxyError> {
|
||||
.map(|r| match r {
|
||||
"stop" => "end_turn",
|
||||
"length" => "max_tokens",
|
||||
"tool_calls" => "tool_use",
|
||||
other => other,
|
||||
});
|
||||
"tool_calls" | "function_call" => "tool_use",
|
||||
"content_filter" => "end_turn",
|
||||
other => {
|
||||
log::warn!("[Claude/OpenAI] Unknown finish_reason in non-streaming response: {other}");
|
||||
"end_turn"
|
||||
}
|
||||
})
|
||||
.or(if has_tool_use { Some("tool_use") } else { None });
|
||||
|
||||
// usage — map cache tokens from OpenAI format to Anthropic format
|
||||
let usage = body.get("usage").cloned().unwrap_or(json!({}));
|
||||
@@ -632,4 +697,74 @@ mod tests {
|
||||
assert_eq!(result["usage"]["cache_read_input_tokens"], 60);
|
||||
assert_eq!(result["usage"]["cache_creation_input_tokens"], 20);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_openai_to_anthropic_finish_reason_content_filter_maps_end_turn() {
|
||||
let input = json!({
|
||||
"id": "chatcmpl-123",
|
||||
"model": "gpt-4",
|
||||
"choices": [{
|
||||
"index": 0,
|
||||
"message": {"role": "assistant", "content": "Blocked"},
|
||||
"finish_reason": "content_filter"
|
||||
}],
|
||||
"usage": {"prompt_tokens": 10, "completion_tokens": 1}
|
||||
});
|
||||
|
||||
let result = openai_to_anthropic(input).unwrap();
|
||||
assert_eq!(result["stop_reason"], "end_turn");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_openai_to_anthropic_with_legacy_function_call() {
|
||||
let input = json!({
|
||||
"id": "chatcmpl-123",
|
||||
"model": "gpt-4",
|
||||
"choices": [{
|
||||
"index": 0,
|
||||
"message": {
|
||||
"role": "assistant",
|
||||
"content": null,
|
||||
"function_call": {
|
||||
"name": "get_weather",
|
||||
"arguments": "{\"location\":\"Tokyo\"}"
|
||||
}
|
||||
},
|
||||
"finish_reason": "function_call"
|
||||
}],
|
||||
"usage": {"prompt_tokens": 10, "completion_tokens": 5}
|
||||
});
|
||||
|
||||
let result = openai_to_anthropic(input).unwrap();
|
||||
assert_eq!(result["content"][0]["type"], "tool_use");
|
||||
assert_eq!(result["content"][0]["name"], "get_weather");
|
||||
assert_eq!(result["content"][0]["input"]["location"], "Tokyo");
|
||||
assert_eq!(result["stop_reason"], "tool_use");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_openai_to_anthropic_with_content_parts_and_refusal() {
|
||||
let input = json!({
|
||||
"id": "chatcmpl-123",
|
||||
"model": "gpt-4",
|
||||
"choices": [{
|
||||
"index": 0,
|
||||
"message": {
|
||||
"role": "assistant",
|
||||
"content": [
|
||||
{"type": "text", "text": "Hello"},
|
||||
{"type": "refusal", "refusal": "I can't do that"}
|
||||
]
|
||||
},
|
||||
"finish_reason": "stop"
|
||||
}],
|
||||
"usage": {"prompt_tokens": 10, "completion_tokens": 5}
|
||||
});
|
||||
|
||||
let result = openai_to_anthropic(input).unwrap();
|
||||
assert_eq!(result["content"][0]["type"], "text");
|
||||
assert_eq!(result["content"][0]["text"], "Hello");
|
||||
assert_eq!(result["content"][1]["type"], "text");
|
||||
assert_eq!(result["content"][1]["text"], "I can't do that");
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user