mirror of
https://github.com/farion1231/cc-switch.git
synced 2026-07-24 12:44:18 +08:00
YoVinchen
1033 lines
54 KiB
Rust
1033 lines
54 KiB
Rust
//! OpenAI Responses API 流式转换模块
|
|
//!
|
|
//! 实现 Responses API SSE → Anthropic SSE 格式转换。
|
|
//!
|
|
//! Responses API 使用命名事件 (named events) 的生命周期模型：
|
|
//! response.created → output_item.added → content_part.added →
|
|
//! output_text.delta → content_part.done → output_item.done → response.completed
|
|
//!
|
|
//! 与 Chat Completions 的 delta chunk 模型完全不同，需要独立的状态机处理。
|
|
|
|
use super::transform_responses::{build_anthropic_usage_from_responses, map_responses_stop_reason};
|
|
use crate::proxy::sse::strip_sse_field;
|
|
use bytes::Bytes;
|
|
use futures::stream::{Stream, StreamExt};
|
|
use serde_json::{json, Value};
|
|
use std::collections::{HashMap, HashSet};
|
|
|
|
#[inline]
|
|
fn response_object_from_event(data: &Value) -> &Value {
|
|
data.get("response").unwrap_or(data)
|
|
}
|
|
|
|
#[inline]
|
|
fn content_part_key(data: &Value) -> Option<String> {
|
|
if let (Some(item_id), Some(content_index)) = (
|
|
data.get("item_id").and_then(|v| v.as_str()),
|
|
data.get("content_index").and_then(|v| v.as_u64()),
|
|
) {
|
|
return Some(format!("part:{item_id}:{content_index}"));
|
|
}
|
|
if let (Some(output_index), Some(content_index)) = (
|
|
data.get("output_index").and_then(|v| v.as_u64()),
|
|
data.get("content_index").and_then(|v| v.as_u64()),
|
|
) {
|
|
return Some(format!("part:out:{output_index}:{content_index}"));
|
|
}
|
|
None
|
|
}
|
|
|
|
#[inline]
|
|
fn tool_item_key_from_added(data: &Value, item: &Value) -> Option<String> {
|
|
if let Some(item_id) = item.get("id").and_then(|v| v.as_str()) {
|
|
return Some(format!("tool:{item_id}"));
|
|
}
|
|
if let Some(item_id) = data.get("item_id").and_then(|v| v.as_str()) {
|
|
return Some(format!("tool:{item_id}"));
|
|
}
|
|
if let Some(output_index) = data.get("output_index").and_then(|v| v.as_u64()) {
|
|
return Some(format!("tool:out:{output_index}"));
|
|
}
|
|
None
|
|
}
|
|
|
|
#[inline]
|
|
fn tool_item_key_from_event(data: &Value) -> Option<String> {
|
|
if let Some(item_id) = data.get("item_id").and_then(|v| v.as_str()) {
|
|
return Some(format!("tool:{item_id}"));
|
|
}
|
|
if let Some(output_index) = data.get("output_index").and_then(|v| v.as_u64()) {
|
|
return Some(format!("tool:out:{output_index}"));
|
|
}
|
|
None
|
|
}
|
|
|
|
/// Resolve content index for a text/refusal content part event.
|
|
///
|
|
/// Uses `content_part_key` to look up or assign a stable index, falling back to
|
|
/// `fallback_open_index` when no key is available.
|
|
#[inline]
|
|
fn resolve_content_index(
|
|
data: &Value,
|
|
next_content_index: &mut u32,
|
|
index_by_key: &mut HashMap<String, u32>,
|
|
fallback_open_index: &mut Option<u32>,
|
|
) -> u32 {
|
|
if let Some(k) = content_part_key(data) {
|
|
if let Some(existing) = index_by_key.get(&k).copied() {
|
|
existing
|
|
} else {
|
|
let assigned = *next_content_index;
|
|
*next_content_index += 1;
|
|
index_by_key.insert(k, assigned);
|
|
assigned
|
|
}
|
|
} else if let Some(existing) = *fallback_open_index {
|
|
existing
|
|
} else {
|
|
let assigned = *next_content_index;
|
|
*next_content_index += 1;
|
|
*fallback_open_index = Some(assigned);
|
|
assigned
|
|
}
|
|
}
|
|
|
|
/// 创建从 Responses API SSE 到 Anthropic SSE 的转换流
|
|
///
|
|
/// 状态机跟踪: message_id, current_model, has_sent_message_start, item/content index map
|
|
/// SSE 解析支持 named events (event: + data: 行)
|
|
pub fn create_anthropic_sse_stream_from_responses<E: std::error::Error + Send + 'static>(
|
|
stream: impl Stream<Item = Result<Bytes, E>> + Send + 'static,
|
|
) -> impl Stream<Item = Result<Bytes, std::io::Error>> + Send {
|
|
async_stream::stream! {
|
|
let mut buffer = String::new();
|
|
let mut message_id: Option<String> = None;
|
|
let mut current_model: Option<String> = None;
|
|
let mut has_sent_message_start = false;
|
|
let mut has_tool_use = false;
|
|
let mut next_content_index: u32 = 0;
|
|
let mut index_by_key: HashMap<String, u32> = HashMap::new();
|
|
let mut open_indices: HashSet<u32> = HashSet::new();
|
|
let mut fallback_open_index: Option<u32> = None;
|
|
let mut current_text_index: Option<u32> = None;
|
|
let mut tool_index_by_item_id: HashMap<String, u32> = HashMap::new();
|
|
let mut last_tool_index: Option<u32> = 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);
|
|
|
|
// SSE 事件由 \n\n 分隔
|
|
while let Some(pos) = buffer.find("\n\n") {
|
|
let block = buffer[..pos].to_string();
|
|
buffer = buffer[pos + 2..].to_string();
|
|
|
|
if block.trim().is_empty() {
|
|
continue;
|
|
}
|
|
|
|
// 解析 SSE 块：提取 event: 和 data: 行
|
|
let mut event_type: Option<String> = None;
|
|
let mut data_parts: Vec<String> = Vec::new();
|
|
|
|
for line in block.lines() {
|
|
if let Some(evt) = strip_sse_field(line, "event") {
|
|
event_type = Some(evt.trim().to_string());
|
|
} else if let Some(d) = strip_sse_field(line, "data") {
|
|
data_parts.push(d.to_string());
|
|
}
|
|
}
|
|
|
|
if data_parts.is_empty() {
|
|
continue;
|
|
}
|
|
|
|
let data_str = data_parts.join("\n");
|
|
let event_name = event_type.as_deref().unwrap_or("");
|
|
|
|
// 解析 JSON 数据
|
|
let data: Value = match serde_json::from_str(&data_str) {
|
|
Ok(v) => v,
|
|
Err(_) => continue,
|
|
};
|
|
|
|
log::debug!("[Claude/Responses] <<< SSE event: {event_name}");
|
|
|
|
match event_name {
|
|
// ================================================
|
|
// response.created → message_start
|
|
// ================================================
|
|
"response.created" => {
|
|
let response_obj = response_object_from_event(&data);
|
|
if let Some(id) = response_obj.get("id").and_then(|i| i.as_str()) {
|
|
message_id = Some(id.to_string());
|
|
}
|
|
if let Some(model) =
|
|
response_obj.get("model").and_then(|m| m.as_str())
|
|
{
|
|
current_model = Some(model.to_string());
|
|
}
|
|
|
|
has_sent_message_start = true;
|
|
// Build usage with cache tokens if available
|
|
let start_usage = build_anthropic_usage_from_responses(
|
|
response_obj.get("usage"),
|
|
);
|
|
|
|
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 = format!("event: message_start\ndata: {}\n\n",
|
|
serde_json::to_string(&event).unwrap_or_default());
|
|
log::debug!("[Claude/Responses] >>> Anthropic SSE: message_start");
|
|
yield Ok(Bytes::from(sse));
|
|
}
|
|
|
|
// ================================================
|
|
// response.content_part.added → content_block_start (text)
|
|
// ================================================
|
|
"response.content_part.added" => {
|
|
// 确保 message_start 已发送
|
|
if !has_sent_message_start {
|
|
let start_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 = format!("event: message_start\ndata: {}\n\n",
|
|
serde_json::to_string(&start_event).unwrap_or_default());
|
|
yield Ok(Bytes::from(sse));
|
|
has_sent_message_start = true;
|
|
}
|
|
|
|
if let Some(part) = data.get("part") {
|
|
let part_type = part.get("type").and_then(|t| t.as_str());
|
|
if matches!(part_type, Some("output_text") | Some("refusal")) {
|
|
let index = if let Some(index) = current_text_index {
|
|
index
|
|
} else {
|
|
let index = resolve_content_index(
|
|
&data,
|
|
&mut next_content_index,
|
|
&mut index_by_key,
|
|
&mut fallback_open_index,
|
|
);
|
|
current_text_index = Some(index);
|
|
index
|
|
};
|
|
|
|
if open_indices.contains(&index) {
|
|
continue;
|
|
}
|
|
|
|
let event = json!({
|
|
"type": "content_block_start",
|
|
"index": index,
|
|
"content_block": {
|
|
"type": "text",
|
|
"text": ""
|
|
}
|
|
});
|
|
let sse = format!("event: content_block_start\ndata: {}\n\n",
|
|
serde_json::to_string(&event).unwrap_or_default());
|
|
yield Ok(Bytes::from(sse));
|
|
open_indices.insert(index);
|
|
}
|
|
}
|
|
}
|
|
|
|
// ================================================
|
|
// response.output_text.delta → content_block_delta (text_delta)
|
|
// ================================================
|
|
"response.output_text.delta" => {
|
|
if let Some(delta) = data.get("delta").and_then(|d| d.as_str()) {
|
|
let index = if let Some(index) = current_text_index {
|
|
index
|
|
} else {
|
|
let index = resolve_content_index(
|
|
&data,
|
|
&mut next_content_index,
|
|
&mut index_by_key,
|
|
&mut fallback_open_index,
|
|
);
|
|
current_text_index = Some(index);
|
|
index
|
|
};
|
|
|
|
if !open_indices.contains(&index) {
|
|
let start_event = json!({
|
|
"type": "content_block_start",
|
|
"index": index,
|
|
"content_block": {
|
|
"type": "text",
|
|
"text": ""
|
|
}
|
|
});
|
|
let start_sse = format!("event: content_block_start\ndata: {}\n\n",
|
|
serde_json::to_string(&start_event).unwrap_or_default());
|
|
yield Ok(Bytes::from(start_sse));
|
|
open_indices.insert(index);
|
|
}
|
|
let event = json!({
|
|
"type": "content_block_delta",
|
|
"index": index,
|
|
"delta": {
|
|
"type": "text_delta",
|
|
"text": delta
|
|
}
|
|
});
|
|
let sse = format!("event: content_block_delta\ndata: {}\n\n",
|
|
serde_json::to_string(&event).unwrap_or_default());
|
|
yield Ok(Bytes::from(sse));
|
|
}
|
|
}
|
|
|
|
// ================================================
|
|
// response.refusal.delta → content_block_delta (text_delta)
|
|
// ================================================
|
|
"response.refusal.delta" => {
|
|
if let Some(delta) = data.get("delta").and_then(|d| d.as_str()) {
|
|
let index = if let Some(index) = current_text_index {
|
|
index
|
|
} else {
|
|
let index = resolve_content_index(
|
|
&data,
|
|
&mut next_content_index,
|
|
&mut index_by_key,
|
|
&mut fallback_open_index,
|
|
);
|
|
current_text_index = Some(index);
|
|
index
|
|
};
|
|
|
|
if !open_indices.contains(&index) {
|
|
let start_event = json!({
|
|
"type": "content_block_start",
|
|
"index": index,
|
|
"content_block": {
|
|
"type": "text",
|
|
"text": ""
|
|
}
|
|
});
|
|
let start_sse = format!("event: content_block_start\ndata: {}\n\n",
|
|
serde_json::to_string(&start_event).unwrap_or_default());
|
|
yield Ok(Bytes::from(start_sse));
|
|
open_indices.insert(index);
|
|
}
|
|
|
|
let event = json!({
|
|
"type": "content_block_delta",
|
|
"index": index,
|
|
"delta": {
|
|
"type": "text_delta",
|
|
"text": delta
|
|
}
|
|
});
|
|
let sse = format!("event: content_block_delta\ndata: {}\n\n",
|
|
serde_json::to_string(&event).unwrap_or_default());
|
|
yield Ok(Bytes::from(sse));
|
|
}
|
|
}
|
|
|
|
// ================================================
|
|
// response.content_part.done → content_block_stop
|
|
// ================================================
|
|
"response.content_part.done" => {}
|
|
|
|
// ================================================
|
|
// response.output_item.added (function_call) → content_block_start (tool_use)
|
|
// ================================================
|
|
"response.output_item.added" => {
|
|
if let Some(item) = data.get("item") {
|
|
let item_type = item.get("type").and_then(|t| t.as_str()).unwrap_or("");
|
|
if item_type == "function_call" {
|
|
has_tool_use = true;
|
|
if let Some(index) = current_text_index.take() {
|
|
if open_indices.remove(&index) {
|
|
let stop_event = json!({
|
|
"type": "content_block_stop",
|
|
"index": index
|
|
});
|
|
let stop_sse = format!("event: content_block_stop\ndata: {}\n\n",
|
|
serde_json::to_string(&stop_event).unwrap_or_default());
|
|
yield Ok(Bytes::from(stop_sse));
|
|
}
|
|
if fallback_open_index == Some(index) {
|
|
fallback_open_index = None;
|
|
}
|
|
}
|
|
// 确保 message_start 已发送
|
|
if !has_sent_message_start {
|
|
let start_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 = format!("event: message_start\ndata: {}\n\n",
|
|
serde_json::to_string(&start_event).unwrap_or_default());
|
|
yield Ok(Bytes::from(sse));
|
|
has_sent_message_start = true;
|
|
}
|
|
|
|
let call_id = item.get("call_id").and_then(|i| i.as_str()).unwrap_or("");
|
|
let name = item.get("name").and_then(|n| n.as_str()).unwrap_or("");
|
|
let index = if let Some(k) = tool_item_key_from_added(&data, item) {
|
|
if let Some(existing) = index_by_key.get(&k).copied() {
|
|
existing
|
|
} else {
|
|
let assigned = next_content_index;
|
|
next_content_index += 1;
|
|
index_by_key.insert(k, assigned);
|
|
assigned
|
|
}
|
|
} else {
|
|
let assigned = next_content_index;
|
|
next_content_index += 1;
|
|
assigned
|
|
};
|
|
if let Some(item_id) = item
|
|
.get("id")
|
|
.and_then(|v| v.as_str())
|
|
.or_else(|| data.get("item_id").and_then(|v| v.as_str()))
|
|
{
|
|
tool_index_by_item_id.insert(item_id.to_string(), index);
|
|
}
|
|
last_tool_index = Some(index);
|
|
|
|
if open_indices.contains(&index) {
|
|
continue;
|
|
}
|
|
|
|
let event = json!({
|
|
"type": "content_block_start",
|
|
"index": index,
|
|
"content_block": {
|
|
"type": "tool_use",
|
|
"id": call_id,
|
|
"name": name
|
|
}
|
|
});
|
|
let sse = format!("event: content_block_start\ndata: {}\n\n",
|
|
serde_json::to_string(&event).unwrap_or_default());
|
|
yield Ok(Bytes::from(sse));
|
|
open_indices.insert(index);
|
|
}
|
|
// message type output_item.added is handled via content_part.added
|
|
}
|
|
}
|
|
|
|
// ================================================
|
|
// response.function_call_arguments.delta → content_block_delta (input_json_delta)
|
|
// ================================================
|
|
"response.function_call_arguments.delta" => {
|
|
if let Some(delta) = data.get("delta").and_then(|d| d.as_str()) {
|
|
let item_id = data.get("item_id").and_then(|v| v.as_str());
|
|
let index = if let Some(id) = item_id {
|
|
tool_index_by_item_id.get(id).copied()
|
|
} else {
|
|
None
|
|
}
|
|
.or_else(|| {
|
|
tool_item_key_from_event(&data)
|
|
.and_then(|k| index_by_key.get(&k).copied())
|
|
})
|
|
.or(last_tool_index)
|
|
.unwrap_or_else(|| {
|
|
let assigned = next_content_index;
|
|
next_content_index += 1;
|
|
assigned
|
|
});
|
|
|
|
if !open_indices.contains(&index) {
|
|
let start_event = json!({
|
|
"type": "content_block_start",
|
|
"index": index,
|
|
"content_block": {
|
|
"type": "tool_use",
|
|
"id": data
|
|
.get("call_id")
|
|
.and_then(|v| v.as_str())
|
|
.or(item_id)
|
|
.unwrap_or(""),
|
|
"name": data
|
|
.get("name")
|
|
.and_then(|v| v.as_str())
|
|
.unwrap_or("")
|
|
}
|
|
});
|
|
let start_sse = format!("event: content_block_start\ndata: {}\n\n",
|
|
serde_json::to_string(&start_event).unwrap_or_default());
|
|
yield Ok(Bytes::from(start_sse));
|
|
open_indices.insert(index);
|
|
}
|
|
|
|
let event = json!({
|
|
"type": "content_block_delta",
|
|
"index": index,
|
|
"delta": {
|
|
"type": "input_json_delta",
|
|
"partial_json": delta
|
|
}
|
|
});
|
|
let sse = format!("event: content_block_delta\ndata: {}\n\n",
|
|
serde_json::to_string(&event).unwrap_or_default());
|
|
yield Ok(Bytes::from(sse));
|
|
}
|
|
}
|
|
|
|
// ================================================
|
|
// response.function_call_arguments.done → content_block_stop
|
|
// ================================================
|
|
"response.function_call_arguments.done" => {
|
|
let item_id = data.get("item_id").and_then(|v| v.as_str());
|
|
let index = if let Some(id) = item_id {
|
|
tool_index_by_item_id.get(id).copied()
|
|
} else {
|
|
None
|
|
}
|
|
.or_else(|| {
|
|
tool_item_key_from_event(&data)
|
|
.and_then(|k| index_by_key.get(&k).copied())
|
|
})
|
|
.or(last_tool_index);
|
|
if let Some(index) = index {
|
|
if !open_indices.remove(&index) {
|
|
continue;
|
|
}
|
|
let event = json!({
|
|
"type": "content_block_stop",
|
|
"index": index
|
|
});
|
|
let sse = format!("event: content_block_stop\ndata: {}\n\n",
|
|
serde_json::to_string(&event).unwrap_or_default());
|
|
yield Ok(Bytes::from(sse));
|
|
if let Some(item_id) = item_id {
|
|
tool_index_by_item_id.remove(item_id);
|
|
}
|
|
}
|
|
}
|
|
|
|
// ================================================
|
|
// response.refusal.done → content_block_stop
|
|
// ================================================
|
|
"response.refusal.done" => {
|
|
let index = current_text_index.take().or_else(|| {
|
|
let key = content_part_key(&data);
|
|
if let Some(k) = key {
|
|
index_by_key.get(&k).copied()
|
|
} else {
|
|
fallback_open_index
|
|
}
|
|
});
|
|
if let Some(index) = index {
|
|
if !open_indices.remove(&index) {
|
|
continue;
|
|
}
|
|
let event = json!({
|
|
"type": "content_block_stop",
|
|
"index": index
|
|
});
|
|
let sse = format!("event: content_block_stop\ndata: {}\n\n",
|
|
serde_json::to_string(&event).unwrap_or_default());
|
|
yield Ok(Bytes::from(sse));
|
|
if fallback_open_index == Some(index) {
|
|
fallback_open_index = None;
|
|
}
|
|
}
|
|
}
|
|
|
|
// ================================================
|
|
// response.reasoning.delta → content_block_delta (thinking_delta)
|
|
// ================================================
|
|
"response.reasoning.delta" => {
|
|
if let Some(delta) = data
|
|
.get("delta")
|
|
.or_else(|| data.get("text"))
|
|
.and_then(|d| d.as_str())
|
|
{
|
|
if let Some(index) = current_text_index.take() {
|
|
if open_indices.remove(&index) {
|
|
let stop_event = json!({
|
|
"type": "content_block_stop",
|
|
"index": index
|
|
});
|
|
let stop_sse = format!("event: content_block_stop\ndata: {}\n\n",
|
|
serde_json::to_string(&stop_event).unwrap_or_default());
|
|
yield Ok(Bytes::from(stop_sse));
|
|
}
|
|
if fallback_open_index == Some(index) {
|
|
fallback_open_index = None;
|
|
}
|
|
}
|
|
let index = resolve_content_index(
|
|
&data,
|
|
&mut next_content_index,
|
|
&mut index_by_key,
|
|
&mut fallback_open_index,
|
|
);
|
|
|
|
if !open_indices.contains(&index) {
|
|
let start_event = json!({
|
|
"type": "content_block_start",
|
|
"index": index,
|
|
"content_block": {
|
|
"type": "thinking",
|
|
"thinking": ""
|
|
}
|
|
});
|
|
let start_sse = format!("event: content_block_start\ndata: {}\n\n",
|
|
serde_json::to_string(&start_event).unwrap_or_default());
|
|
yield Ok(Bytes::from(start_sse));
|
|
open_indices.insert(index);
|
|
}
|
|
|
|
let event = json!({
|
|
"type": "content_block_delta",
|
|
"index": index,
|
|
"delta": {
|
|
"type": "thinking_delta",
|
|
"thinking": delta
|
|
}
|
|
});
|
|
let sse = format!("event: content_block_delta\ndata: {}\n\n",
|
|
serde_json::to_string(&event).unwrap_or_default());
|
|
yield Ok(Bytes::from(sse));
|
|
}
|
|
}
|
|
|
|
// ================================================
|
|
// response.reasoning.done → content_block_stop
|
|
// ================================================
|
|
"response.reasoning.done" => {
|
|
let key = content_part_key(&data);
|
|
let index = if let Some(k) = key {
|
|
index_by_key.get(&k).copied()
|
|
} else {
|
|
fallback_open_index
|
|
};
|
|
if let Some(index) = index {
|
|
if !open_indices.remove(&index) {
|
|
continue;
|
|
}
|
|
let event = json!({
|
|
"type": "content_block_stop",
|
|
"index": index
|
|
});
|
|
let sse = format!("event: content_block_stop\ndata: {}\n\n",
|
|
serde_json::to_string(&event).unwrap_or_default());
|
|
yield Ok(Bytes::from(sse));
|
|
if fallback_open_index == Some(index) {
|
|
fallback_open_index = None;
|
|
}
|
|
}
|
|
}
|
|
|
|
// ================================================
|
|
// response.completed → message_delta + message_stop
|
|
// ================================================
|
|
"response.completed" => {
|
|
let response_obj = response_object_from_event(&data);
|
|
let stop_reason = map_responses_stop_reason(
|
|
response_obj.get("status").and_then(|s| s.as_str()),
|
|
has_tool_use,
|
|
response_obj
|
|
.pointer("/incomplete_details/reason")
|
|
.and_then(|r| r.as_str()),
|
|
);
|
|
|
|
// Best effort: close any dangling blocks before message_delta/message_stop.
|
|
if !open_indices.is_empty() {
|
|
let mut remaining: Vec<u32> = open_indices.iter().copied().collect();
|
|
remaining.sort_unstable();
|
|
for index in remaining {
|
|
let stop_event = json!({
|
|
"type": "content_block_stop",
|
|
"index": index
|
|
});
|
|
let stop_sse = format!("event: content_block_stop\ndata: {}\n\n",
|
|
serde_json::to_string(&stop_event).unwrap_or_default());
|
|
yield Ok(Bytes::from(stop_sse));
|
|
open_indices.remove(&index);
|
|
}
|
|
}
|
|
fallback_open_index = None;
|
|
|
|
let usage_json = response_obj.get("usage").map(|u| {
|
|
build_anthropic_usage_from_responses(Some(u))
|
|
});
|
|
|
|
// Emit message_delta (with usage + stop_reason)
|
|
let delta_event = json!({
|
|
"type": "message_delta",
|
|
"delta": {
|
|
"stop_reason": stop_reason,
|
|
"stop_sequence": null
|
|
},
|
|
"usage": usage_json
|
|
});
|
|
let sse = format!("event: message_delta\ndata: {}\n\n",
|
|
serde_json::to_string(&delta_event).unwrap_or_default());
|
|
log::debug!("[Claude/Responses] >>> Anthropic SSE: message_delta");
|
|
yield Ok(Bytes::from(sse));
|
|
|
|
// Emit message_stop
|
|
let stop_event = json!({"type": "message_stop"});
|
|
let stop_sse = format!("event: message_stop\ndata: {}\n\n",
|
|
serde_json::to_string(&stop_event).unwrap_or_default());
|
|
log::debug!("[Claude/Responses] >>> Anthropic SSE: message_stop");
|
|
yield Ok(Bytes::from(stop_sse));
|
|
}
|
|
|
|
// Lifecycle events that don't need Anthropic counterparts.
|
|
// Listed explicitly so new events trigger a match-completeness review.
|
|
"response.output_text.done" => {
|
|
if let Some(index) = current_text_index.take() {
|
|
if open_indices.remove(&index) {
|
|
let stop_event = json!({
|
|
"type": "content_block_stop",
|
|
"index": index
|
|
});
|
|
let stop_sse = format!("event: content_block_stop\ndata: {}\n\n",
|
|
serde_json::to_string(&stop_event).unwrap_or_default());
|
|
yield Ok(Bytes::from(stop_sse));
|
|
}
|
|
if fallback_open_index == Some(index) {
|
|
fallback_open_index = None;
|
|
}
|
|
}
|
|
}
|
|
"response.output_item.done"
|
|
| "response.in_progress" => {}
|
|
|
|
// Any other unknown/future events — silently skip.
|
|
_ => {}
|
|
}
|
|
}
|
|
}
|
|
Err(e) => {
|
|
log::error!("Responses stream error: {e}");
|
|
let error_event = json!({
|
|
"type": "error",
|
|
"error": {
|
|
"type": "stream_error",
|
|
"message": format!("Stream error: {e}")
|
|
}
|
|
});
|
|
let sse = format!("event: error\ndata: {}\n\n",
|
|
serde_json::to_string(&error_event).unwrap_or_default());
|
|
yield Ok(Bytes::from(sse));
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
use futures::stream;
|
|
use futures::StreamExt;
|
|
use std::collections::HashMap;
|
|
|
|
#[test]
|
|
fn test_map_responses_stop_reason_tool_use() {
|
|
assert_eq!(
|
|
map_responses_stop_reason(Some("completed"), true, None),
|
|
Some("tool_use")
|
|
);
|
|
assert_eq!(
|
|
map_responses_stop_reason(Some("completed"), false, None),
|
|
Some("end_turn")
|
|
);
|
|
assert_eq!(
|
|
map_responses_stop_reason(Some("incomplete"), false, Some("max_output_tokens")),
|
|
Some("max_tokens")
|
|
);
|
|
assert_eq!(
|
|
map_responses_stop_reason(Some("incomplete"), false, Some("content_filter")),
|
|
Some("end_turn")
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn test_response_object_from_event_with_wrapper() {
|
|
let data = json!({
|
|
"type": "response.created",
|
|
"response": {
|
|
"id": "resp_1",
|
|
"model": "gpt-4o"
|
|
}
|
|
});
|
|
let obj = response_object_from_event(&data);
|
|
assert_eq!(obj["id"], "resp_1");
|
|
assert_eq!(obj["model"], "gpt-4o");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_streaming_conversion_with_wrapped_response_events() {
|
|
let input = concat!(
|
|
"event: response.created\n",
|
|
"data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_1\",\"model\":\"gpt-4o\",\"usage\":{\"input_tokens\":12,\"output_tokens\":0}}}\n\n",
|
|
"event: response.output_item.added\n",
|
|
"data: {\"type\":\"response.output_item.added\",\"item\":{\"type\":\"function_call\",\"call_id\":\"call_1\",\"name\":\"get_weather\"}}\n\n",
|
|
"event: response.function_call_arguments.delta\n",
|
|
"data: {\"type\":\"response.function_call_arguments.delta\",\"delta\":\"{\\\"city\\\":\\\"Tokyo\\\"}\"}\n\n",
|
|
"event: response.function_call_arguments.done\n",
|
|
"data: {\"type\":\"response.function_call_arguments.done\"}\n\n",
|
|
"event: response.completed\n",
|
|
"data: {\"type\":\"response.completed\",\"response\":{\"status\":\"completed\",\"usage\":{\"input_tokens\":12,\"output_tokens\":3}}}\n\n"
|
|
);
|
|
|
|
let upstream = stream::iter(vec![Ok::<_, std::io::Error>(Bytes::from(
|
|
input.as_bytes().to_vec(),
|
|
))]);
|
|
let converted = create_anthropic_sse_stream_from_responses(upstream);
|
|
let chunks: Vec<_> = converted.collect().await;
|
|
|
|
let merged = chunks
|
|
.into_iter()
|
|
.map(|c| String::from_utf8_lossy(c.unwrap().as_ref()).to_string())
|
|
.collect::<String>();
|
|
|
|
assert!(merged.contains("\"type\":\"message_start\""));
|
|
assert!(merged.contains("\"id\":\"resp_1\""));
|
|
assert!(merged.contains("\"model\":\"gpt-4o\""));
|
|
assert!(merged.contains("\"type\":\"tool_use\""));
|
|
assert!(merged.contains("\"name\":\"get_weather\""));
|
|
assert!(merged.contains("\"type\":\"input_json_delta\""));
|
|
assert!(merged.contains("\"stop_reason\":\"tool_use\""));
|
|
assert!(merged.contains("\"input_tokens\":12"));
|
|
assert!(merged.contains("\"output_tokens\":3"));
|
|
assert!(merged.contains("\"type\":\"message_stop\""));
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_streaming_conversion_interleaved_tool_deltas_by_item_id() {
|
|
let input = concat!(
|
|
"event: response.created\n",
|
|
"data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_2\",\"model\":\"gpt-4o\"}}\n\n",
|
|
"event: response.output_item.added\n",
|
|
"data: {\"type\":\"response.output_item.added\",\"item\":{\"id\":\"fc_1\",\"type\":\"function_call\",\"call_id\":\"call_1\",\"name\":\"first_tool\"}}\n\n",
|
|
"event: response.output_item.added\n",
|
|
"data: {\"type\":\"response.output_item.added\",\"item\":{\"id\":\"fc_2\",\"type\":\"function_call\",\"call_id\":\"call_2\",\"name\":\"second_tool\"}}\n\n",
|
|
"event: response.function_call_arguments.delta\n",
|
|
"data: {\"type\":\"response.function_call_arguments.delta\",\"item_id\":\"fc_2\",\"delta\":\"{\\\"b\\\":2}\"}\n\n",
|
|
"event: response.function_call_arguments.delta\n",
|
|
"data: {\"type\":\"response.function_call_arguments.delta\",\"item_id\":\"fc_1\",\"delta\":\"{\\\"a\\\":1}\"}\n\n",
|
|
"event: response.function_call_arguments.done\n",
|
|
"data: {\"type\":\"response.function_call_arguments.done\",\"item_id\":\"fc_1\"}\n\n",
|
|
"event: response.function_call_arguments.done\n",
|
|
"data: {\"type\":\"response.function_call_arguments.done\",\"item_id\":\"fc_2\"}\n\n",
|
|
"event: response.completed\n",
|
|
"data: {\"type\":\"response.completed\",\"response\":{\"status\":\"completed\",\"usage\":{\"input_tokens\":8,\"output_tokens\":4}}}\n\n"
|
|
);
|
|
|
|
let upstream = stream::iter(vec![Ok::<_, std::io::Error>(Bytes::from(
|
|
input.as_bytes().to_vec(),
|
|
))]);
|
|
let converted = create_anthropic_sse_stream_from_responses(upstream);
|
|
let chunks: Vec<_> = converted.collect().await;
|
|
let merged = chunks
|
|
.into_iter()
|
|
.map(|c| String::from_utf8_lossy(c.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| strip_sse_field(line, "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") {
|
|
let cb = event.get("content_block");
|
|
if cb.and_then(|v| v.get("type")).and_then(|v| v.as_str()) == Some("tool_use") {
|
|
if let (Some(call_id), Some(index)) = (
|
|
cb.and_then(|v| v.get("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);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
let delta_indices: Vec<u64> = 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.get("index").and_then(|v| v.as_u64()))
|
|
.collect();
|
|
|
|
assert_eq!(delta_indices.len(), 2);
|
|
assert_eq!(delta_indices[0], *tool_index_by_call.get("call_2").unwrap());
|
|
assert_eq!(delta_indices[1], *tool_index_by_call.get("call_1").unwrap());
|
|
assert_ne!(
|
|
tool_index_by_call.get("call_1"),
|
|
tool_index_by_call.get("call_2")
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_streaming_reasoning_delta_emits_thinking_blocks() {
|
|
let input = concat!(
|
|
"event: response.created\n",
|
|
"data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_r\",\"model\":\"o3\",\"usage\":{\"input_tokens\":5,\"output_tokens\":0}}}\n\n",
|
|
"event: response.reasoning.delta\n",
|
|
"data: {\"type\":\"response.reasoning.delta\",\"delta\":\"Let me think...\"}\n\n",
|
|
"event: response.reasoning.done\n",
|
|
"data: {\"type\":\"response.reasoning.done\"}\n\n",
|
|
"event: response.content_part.added\n",
|
|
"data: {\"type\":\"response.content_part.added\",\"part\":{\"type\":\"output_text\",\"text\":\"\"},\"output_index\":0,\"content_index\":0}\n\n",
|
|
"event: response.output_text.delta\n",
|
|
"data: {\"type\":\"response.output_text.delta\",\"delta\":\"42\",\"output_index\":0,\"content_index\":0}\n\n",
|
|
"event: response.content_part.done\n",
|
|
"data: {\"type\":\"response.content_part.done\",\"output_index\":0,\"content_index\":0}\n\n",
|
|
"event: response.completed\n",
|
|
"data: {\"type\":\"response.completed\",\"response\":{\"status\":\"completed\",\"usage\":{\"input_tokens\":5,\"output_tokens\":10}}}\n\n"
|
|
);
|
|
|
|
let upstream = stream::iter(vec![Ok::<_, std::io::Error>(Bytes::from(
|
|
input.as_bytes().to_vec(),
|
|
))]);
|
|
let converted = create_anthropic_sse_stream_from_responses(upstream);
|
|
let chunks: Vec<_> = converted.collect().await;
|
|
let merged = chunks
|
|
.into_iter()
|
|
.map(|c| String::from_utf8_lossy(c.unwrap().as_ref()).to_string())
|
|
.collect::<String>();
|
|
|
|
// Should contain thinking block start, thinking delta, and text content
|
|
assert!(
|
|
merged.contains("\"type\":\"thinking\""),
|
|
"should emit thinking content_block_start"
|
|
);
|
|
assert!(
|
|
merged.contains("\"type\":\"thinking_delta\""),
|
|
"should emit thinking_delta"
|
|
);
|
|
assert!(
|
|
merged.contains("\"thinking\":\"Let me think...\""),
|
|
"should contain thinking text"
|
|
);
|
|
assert!(
|
|
merged.contains("\"type\":\"text_delta\""),
|
|
"should also emit text content"
|
|
);
|
|
assert!(
|
|
merged.contains("\"text\":\"42\""),
|
|
"should contain text delta"
|
|
);
|
|
assert!(merged.contains("\"stop_reason\":\"end_turn\""));
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_streaming_text_parts_are_merged_into_one_text_block() {
|
|
let input = concat!(
|
|
"event: response.created\n",
|
|
"data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_merge\",\"model\":\"gpt-5.4\",\"usage\":{\"input_tokens\":5,\"output_tokens\":0}}}\n\n",
|
|
"event: response.content_part.added\n",
|
|
"data: {\"type\":\"response.content_part.added\",\"part\":{\"type\":\"output_text\",\"text\":\"\"},\"output_index\":0,\"content_index\":0}\n\n",
|
|
"event: response.output_text.delta\n",
|
|
"data: {\"type\":\"response.output_text.delta\",\"delta\":\"你\",\"output_index\":0,\"content_index\":0}\n\n",
|
|
"event: response.content_part.done\n",
|
|
"data: {\"type\":\"response.content_part.done\",\"output_index\":0,\"content_index\":0}\n\n",
|
|
"event: response.content_part.added\n",
|
|
"data: {\"type\":\"response.content_part.added\",\"part\":{\"type\":\"output_text\",\"text\":\"\"},\"output_index\":0,\"content_index\":1}\n\n",
|
|
"event: response.output_text.delta\n",
|
|
"data: {\"type\":\"response.output_text.delta\",\"delta\":\"好\",\"output_index\":0,\"content_index\":1}\n\n",
|
|
"event: response.content_part.done\n",
|
|
"data: {\"type\":\"response.content_part.done\",\"output_index\":0,\"content_index\":1}\n\n",
|
|
"event: response.output_text.done\n",
|
|
"data: {\"type\":\"response.output_text.done\",\"output_index\":0,\"content_index\":1}\n\n",
|
|
"event: response.completed\n",
|
|
"data: {\"type\":\"response.completed\",\"response\":{\"status\":\"completed\",\"usage\":{\"input_tokens\":5,\"output_tokens\":2}}}\n\n"
|
|
);
|
|
|
|
let upstream = stream::iter(vec![Ok::<_, std::io::Error>(Bytes::from(
|
|
input.as_bytes().to_vec(),
|
|
))]);
|
|
let converted = create_anthropic_sse_stream_from_responses(upstream);
|
|
let chunks: Vec<_> = converted.collect().await;
|
|
let events: Vec<Value> = chunks
|
|
.into_iter()
|
|
.flat_map(|chunk| {
|
|
let bytes = chunk.unwrap();
|
|
let text = String::from_utf8_lossy(bytes.as_ref()).to_string();
|
|
text.split("\n\n")
|
|
.filter_map(|block| {
|
|
block.lines().find_map(|line| {
|
|
strip_sse_field(line, "data")
|
|
.and_then(|payload| serde_json::from_str::<Value>(payload).ok())
|
|
})
|
|
})
|
|
.collect::<Vec<_>>()
|
|
})
|
|
.collect();
|
|
|
|
let text_starts = 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("text")
|
|
})
|
|
.count();
|
|
let text_stops = events
|
|
.iter()
|
|
.filter(|event| {
|
|
event.get("type").and_then(|v| v.as_str()) == Some("content_block_stop")
|
|
})
|
|
.count();
|
|
let text_deltas: Vec<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("text_delta")
|
|
})
|
|
.filter_map(|event| {
|
|
event
|
|
.pointer("/delta/text")
|
|
.and_then(|v| v.as_str())
|
|
.map(ToString::to_string)
|
|
})
|
|
.collect();
|
|
|
|
assert_eq!(text_starts, 1);
|
|
assert_eq!(text_stops, 1);
|
|
assert_eq!(text_deltas, vec!["你".to_string(), "好".to_string()]);
|
|
}
|
|
}
|