820 lines
33 KiB
Rust
820 lines
33 KiB
Rust
//! OpenAI provider：Responses API 与 Chat Completions 双协议（SSE 流式）。
|
||
//! OpenAI provider: both the Responses API and Chat Completions (SSE streaming).
|
||
//!
|
||
//! 两套协议共享同一套核心类型翻译（messages / tools / usage / stop_reason），
|
||
//! 仅请求格式与流式事件不同；通过 [`OpenAiProtocol`] 切换。
|
||
//! Both protocols share the same core-type translation (messages / tools /
|
||
//! usage / stop_reason); only the request format and stream events differ,
|
||
//! selected via [`OpenAiProtocol`].
|
||
|
||
use crate::common;
|
||
use crate::config::{split_base_url, ProviderConfig};
|
||
use focus_core::model::*;
|
||
use focus_core::provider::{
|
||
ProviderEventReducer, ProviderRequest, StreamEvent, StreamProvider, StreamResult,
|
||
};
|
||
use focus_core::CoreError;
|
||
use focus_json::JsonValue;
|
||
use focus_transport::{HttpRequest, SseEvent, SseParser, Transport};
|
||
use std::collections::HashSet;
|
||
use std::fmt;
|
||
use std::sync::mpsc::SyncSender;
|
||
use std::sync::Arc;
|
||
|
||
/// OpenAI 默认 API 基址。
|
||
/// The default OpenAI API base.
|
||
const DEFAULT_BASE_URL: &str = "https://api.openai.com";
|
||
|
||
/// 两套 OpenAI 协议。
|
||
/// The two OpenAI protocols.
|
||
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
|
||
pub enum OpenAiProtocol {
|
||
/// 最新的 Responses API（`/v1/responses`）。
|
||
/// The latest Responses API (`/v1/responses`).
|
||
#[default]
|
||
Responses,
|
||
/// 传统 Chat Completions（`/v1/chat/completions`），兼容绝大多数第三方端点。
|
||
/// Legacy Chat Completions (`/v1/chat/completions`), compatible with most
|
||
/// third-party endpoints.
|
||
ChatCompletions,
|
||
}
|
||
|
||
impl OpenAiProtocol {
|
||
fn endpoint_path(&self, prefix: &str) -> String {
|
||
match self {
|
||
OpenAiProtocol::Responses => format!("{}/v1/responses", prefix),
|
||
OpenAiProtocol::ChatCompletions => format!("{}/v1/chat/completions", prefix),
|
||
}
|
||
}
|
||
}
|
||
|
||
/// 实现 OpenAI API 的 provider。
|
||
/// A provider implementing the OpenAI API.
|
||
#[derive(Clone)]
|
||
pub struct OpenAiProvider {
|
||
config: ProviderConfig,
|
||
protocol: OpenAiProtocol,
|
||
transport: Arc<dyn Transport>,
|
||
}
|
||
|
||
impl fmt::Debug for OpenAiProvider {
|
||
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||
f.debug_struct("OpenAiProvider")
|
||
.field("config", &self.config)
|
||
.field("protocol", &self.protocol)
|
||
.field("transport", &"<Arc<dyn Transport>>")
|
||
.finish()
|
||
}
|
||
}
|
||
|
||
impl OpenAiProvider {
|
||
/// 使用默认 HTTP 传输构造 provider。
|
||
/// Build a provider with the default HTTP transport.
|
||
pub fn new(config: ProviderConfig, protocol: OpenAiProtocol) -> Self {
|
||
Self::with_transport(
|
||
config,
|
||
protocol,
|
||
Arc::new(focus_transport::HttpTransport::new()),
|
||
)
|
||
}
|
||
|
||
/// 注入自定义传输（测试用 mock）构造 provider。
|
||
/// Build a provider with an injected transport (e.g. a mock in tests).
|
||
pub fn with_transport(
|
||
config: ProviderConfig,
|
||
protocol: OpenAiProtocol,
|
||
transport: Arc<dyn Transport>,
|
||
) -> Self {
|
||
Self {
|
||
config,
|
||
protocol,
|
||
transport,
|
||
}
|
||
}
|
||
|
||
/// 依据配置解析目标 (host, port, path)。
|
||
/// Resolve the target (host, port, path) from the config.
|
||
fn endpoint(&self) -> Result<(String, u16, String), String> {
|
||
let base = self
|
||
.config
|
||
.base_url
|
||
.clone()
|
||
.unwrap_or_else(|| DEFAULT_BASE_URL.to_string());
|
||
let (host, port, prefix) = split_base_url(&base)?;
|
||
Ok((host, port, self.protocol.endpoint_path(&prefix)))
|
||
}
|
||
}
|
||
|
||
impl StreamProvider for OpenAiProvider {
|
||
fn stream(&self, request: &ProviderRequest) -> StreamResult {
|
||
let (host, port, path) = self.endpoint().map_err(CoreError::Provider)?;
|
||
let body = match self.protocol {
|
||
OpenAiProtocol::Responses => build_responses_body(request),
|
||
OpenAiProtocol::ChatCompletions => build_chat_body(request),
|
||
};
|
||
let body = focus_json::to_string(&body);
|
||
let model = request.model.clone();
|
||
let req = HttpRequest::post(&host, &path)
|
||
.port(port)
|
||
.header("Authorization", &format!("Bearer {}", self.config.api_key))
|
||
.header("accept", "text/event-stream")
|
||
.json_body(&body);
|
||
let transport = self.transport.clone();
|
||
let timeout = self.config.timeout;
|
||
// 先取出协议（Copy 类型），避免闭包借用 self。
|
||
// Copy the protocol out first so the closure doesn't borrow self.
|
||
let protocol = self.protocol;
|
||
|
||
common::run_stream(move |tx| {
|
||
Box::pin(async move {
|
||
let model = model.clone();
|
||
let result: Result<(), String> = async {
|
||
let mut chunks = transport
|
||
.stream_request(&req, timeout)
|
||
.await
|
||
.map_err(|e| e.to_string())?;
|
||
let mut sse = SseParser::new();
|
||
let mut turn = match protocol {
|
||
OpenAiProtocol::Responses => OpenAiTurn::responses(&model),
|
||
OpenAiProtocol::ChatCompletions => OpenAiTurn::chat(&model),
|
||
};
|
||
loop {
|
||
let chunk = chunks.next_chunk().await.map_err(|e| e.to_string())?;
|
||
match chunk {
|
||
Some(bytes) => {
|
||
common::debug_chunk(&bytes);
|
||
sse.feed(&bytes);
|
||
while let Some(ev) = sse.next_event() {
|
||
match turn.handle_sse(&ev, &tx) {
|
||
Ok(true) => {}
|
||
Ok(false) => return Ok(()), // 消费者已离开 / consumer gone
|
||
Err(msg) => {
|
||
let _ = tx.send(StreamEvent::Error {
|
||
error: common::error_message(&model, &msg),
|
||
});
|
||
return Ok(());
|
||
}
|
||
}
|
||
if turn.done {
|
||
return Ok(());
|
||
}
|
||
}
|
||
}
|
||
None => break,
|
||
}
|
||
}
|
||
if !turn.done {
|
||
return Err("stream ended without a terminal event".to_string());
|
||
}
|
||
Ok(())
|
||
}
|
||
.await;
|
||
if let Err(msg) = result {
|
||
let _ = tx.send(StreamEvent::Error {
|
||
error: common::error_message(&model, &msg),
|
||
});
|
||
}
|
||
})
|
||
})
|
||
}
|
||
}
|
||
|
||
// ---- 请求体构造 ----
|
||
// ---- request-body construction -------------------------------------------
|
||
|
||
/// Chat Completions 请求体。
|
||
/// Chat Completions request body.
|
||
fn build_chat_body(request: &ProviderRequest) -> JsonValue {
|
||
let mut body = JsonValue::obj();
|
||
body.insert("model", request.model.clone().into()).ok();
|
||
let mut messages = JsonValue::arr();
|
||
// 系统提示作为第一条 system 消息。
|
||
// The system prompt becomes the first system message.
|
||
if !request.system_prompt.is_empty() {
|
||
let mut sys = JsonValue::obj();
|
||
sys.insert("role", "system".into()).ok();
|
||
sys.insert("content", request.system_prompt.clone().into())
|
||
.ok();
|
||
messages.push(sys).ok();
|
||
}
|
||
for m in &request.messages {
|
||
messages.push(chat_message(m)).ok();
|
||
}
|
||
body.insert("messages", messages).ok();
|
||
if has_tools(&request.tools) {
|
||
body.insert("tools", tools_to_chat(&request.tools)).ok();
|
||
body.insert("tool_choice", "auto".into()).ok();
|
||
}
|
||
if let Some(mt) = request.max_tokens {
|
||
// 新模型使用 max_completion_tokens（旧字段 max_tokens 已被弃用）。
|
||
// Newer models use max_completion_tokens (max_tokens is deprecated).
|
||
body.insert("max_completion_tokens", (mt as f64).into())
|
||
.ok();
|
||
}
|
||
if let Some(t) = request.temperature {
|
||
body.insert("temperature", t.into()).ok();
|
||
}
|
||
body.insert("stream", true.into()).ok();
|
||
// 请求结束 chunk 携带用量。
|
||
// Ask for usage in the final chunk.
|
||
let mut so = JsonValue::obj();
|
||
so.insert("include_usage", true.into()).ok();
|
||
body.insert("stream_options", so).ok();
|
||
body
|
||
}
|
||
|
||
/// Responses API 请求体。
|
||
/// Responses API request body.
|
||
fn build_responses_body(request: &ProviderRequest) -> JsonValue {
|
||
let mut body = JsonValue::obj();
|
||
body.insert("model", request.model.clone().into()).ok();
|
||
if !request.system_prompt.is_empty() {
|
||
body.insert("instructions", request.system_prompt.clone().into())
|
||
.ok();
|
||
}
|
||
body.insert("input", messages_to_responses(&request.messages))
|
||
.ok();
|
||
if has_tools(&request.tools) {
|
||
body.insert("tools", tools_to_responses(&request.tools))
|
||
.ok();
|
||
}
|
||
if let Some(mt) = request.max_tokens {
|
||
body.insert("max_output_tokens", (mt as f64).into()).ok();
|
||
}
|
||
if let Some(t) = request.temperature {
|
||
body.insert("temperature", t.into()).ok();
|
||
}
|
||
body.insert("stream", true.into()).ok();
|
||
body
|
||
}
|
||
|
||
/// 一条核心消息 → Chat Completions 消息。
|
||
/// One core message → a Chat Completions message.
|
||
fn chat_message(m: &Message) -> JsonValue {
|
||
match m {
|
||
Message::User(u) => {
|
||
let mut o = JsonValue::obj();
|
||
o.insert("role", "user".into()).ok();
|
||
let text = join_text(&u.content);
|
||
let images: Vec<&ImageContent> = u
|
||
.content
|
||
.iter()
|
||
.filter_map(|c| match c {
|
||
ContentBlock::Image(i) => Some(i),
|
||
_ => None,
|
||
})
|
||
.collect();
|
||
if images.is_empty() {
|
||
o.insert("content", text.into()).ok();
|
||
} else {
|
||
let mut parts = JsonValue::arr();
|
||
if !text.is_empty() {
|
||
let mut p = JsonValue::obj();
|
||
p.insert("type", "text".into()).ok();
|
||
p.insert("text", text.clone().into()).ok();
|
||
parts.push(p).ok();
|
||
}
|
||
for img in images {
|
||
let mut p = JsonValue::obj();
|
||
p.insert("type", "image_url".into()).ok();
|
||
let mut url = JsonValue::obj();
|
||
url.insert(
|
||
"url",
|
||
format!("data:{};base64,{}", img.mime_type, img.data).into(),
|
||
)
|
||
.ok();
|
||
p.insert("image_url", url).ok();
|
||
parts.push(p).ok();
|
||
}
|
||
o.insert("content", parts).ok();
|
||
}
|
||
o
|
||
}
|
||
Message::Assistant(a) => {
|
||
let mut o = JsonValue::obj();
|
||
o.insert("role", "assistant".into()).ok();
|
||
o.insert("content", join_text(&a.content).into()).ok();
|
||
// 推理 API（如 DeepSeek thinking 模式）要求把思考内容作为
|
||
// reasoning_content 原样回传；OpenAI 官方不会产生思考块，因此
|
||
// 仅在存在思考块时发送该字段，安全兼容两者。
|
||
// Reasoning-capable APIs (e.g. DeepSeek thinking mode) require the
|
||
// thinking text to be echoed back as reasoning_content; OpenAI
|
||
// official never produces thinking blocks, so sending the field
|
||
// only when thinking exists stays safe for both.
|
||
let reasoning: String = a
|
||
.content
|
||
.iter()
|
||
.filter_map(|c| match c {
|
||
ContentBlock::Thinking(t) => Some(t.thinking.clone()),
|
||
_ => None,
|
||
})
|
||
.collect();
|
||
if !reasoning.is_empty() {
|
||
// 同时写 reasoning_content（DeepSeek 官方）与 reasoning_text
|
||
//（部分兼容层）；推理端点普遍接受这两个字段名。
|
||
// Write both reasoning_content (DeepSeek official) and
|
||
// reasoning_text (some compatible layers); reasoning-capable
|
||
// endpoints generally accept either name.
|
||
o.insert("reasoning_content", reasoning.clone().into()).ok();
|
||
o.insert("reasoning_text", reasoning.into()).ok();
|
||
}
|
||
let calls: Vec<&ToolCall> = a.content.iter().filter_map(|c| c.as_tool_call()).collect();
|
||
if !calls.is_empty() {
|
||
let mut arr = JsonValue::arr();
|
||
for tc in calls {
|
||
let mut c = JsonValue::obj();
|
||
c.insert("id", tc.id.clone().into()).ok();
|
||
c.insert("type", "function".into()).ok();
|
||
let mut f = JsonValue::obj();
|
||
f.insert("name", tc.name.clone().into()).ok();
|
||
f.insert("arguments", focus_json::to_string(&tc.arguments).into())
|
||
.ok();
|
||
c.insert("function", f).ok();
|
||
arr.push(c).ok();
|
||
}
|
||
o.insert("tool_calls", arr).ok();
|
||
}
|
||
o
|
||
}
|
||
Message::ToolResult(t) => {
|
||
let mut o = JsonValue::obj();
|
||
o.insert("role", "tool".into()).ok();
|
||
o.insert("tool_call_id", t.tool_call_id.clone().into()).ok();
|
||
o.insert("content", join_text(&t.content).into()).ok();
|
||
o
|
||
}
|
||
}
|
||
}
|
||
|
||
/// 核心消息列表 → Responses API input items。
|
||
/// Core messages → Responses API input items.
|
||
fn messages_to_responses(messages: &[Message]) -> JsonValue {
|
||
let mut items = JsonValue::arr();
|
||
for m in messages {
|
||
match m {
|
||
Message::User(u) => {
|
||
let mut item = JsonValue::obj();
|
||
item.insert("role", "user".into()).ok();
|
||
let mut content = JsonValue::arr();
|
||
let text = join_text(&u.content);
|
||
if !text.is_empty() {
|
||
let mut part = JsonValue::obj();
|
||
part.insert("type", "input_text".into()).ok();
|
||
part.insert("text", text.into()).ok();
|
||
content.push(part).ok();
|
||
}
|
||
item.insert("content", content).ok();
|
||
items.push(item).ok();
|
||
}
|
||
Message::Assistant(a) => {
|
||
// 推理内容必须回传为 reasoning item（thinking 模式 API 校验）。
|
||
// Thinking must be replayed as a reasoning item (thinking-mode
|
||
// APIs validate this).
|
||
let reasoning: String = a
|
||
.content
|
||
.iter()
|
||
.filter_map(|c| match c {
|
||
ContentBlock::Thinking(t) => Some(t.thinking.clone()),
|
||
_ => None,
|
||
})
|
||
.collect();
|
||
if !reasoning.is_empty() {
|
||
let mut ri = JsonValue::obj();
|
||
ri.insert("type", "reasoning".into()).ok();
|
||
let mut part = JsonValue::obj();
|
||
part.insert("type", "reasoning_text".into()).ok();
|
||
part.insert("text", reasoning.into()).ok();
|
||
let mut content = JsonValue::arr();
|
||
content.push(part).ok();
|
||
ri.insert("content", content).ok();
|
||
ri.insert("summary", JsonValue::arr()).ok();
|
||
items.push(ri).ok();
|
||
}
|
||
let mut item = JsonValue::obj();
|
||
item.insert("role", "assistant".into()).ok();
|
||
let mut content = JsonValue::arr();
|
||
let text = join_text(&a.content);
|
||
if !text.is_empty() {
|
||
let mut part = JsonValue::obj();
|
||
part.insert("type", "output_text".into()).ok();
|
||
part.insert("text", text.into()).ok();
|
||
content.push(part).ok();
|
||
}
|
||
item.insert("content", content).ok();
|
||
items.push(item).ok();
|
||
// 工具调用作为独立的 function_call items（function_call_output
|
||
// 的 call_id 必须对应它们）。
|
||
// Tool calls become standalone function_call items (the
|
||
// function_call_output call_ids must match them).
|
||
for tc in a.content.iter().filter_map(|c| c.as_tool_call()) {
|
||
let mut fc = JsonValue::obj();
|
||
fc.insert("type", "function_call".into()).ok();
|
||
fc.insert("call_id", tc.id.clone().into()).ok();
|
||
fc.insert("name", tc.name.clone().into()).ok();
|
||
fc.insert("arguments", focus_json::to_string(&tc.arguments).into())
|
||
.ok();
|
||
items.push(fc).ok();
|
||
}
|
||
}
|
||
Message::ToolResult(t) => {
|
||
let mut item = JsonValue::obj();
|
||
item.insert("type", "function_call_output".into()).ok();
|
||
item.insert("call_id", t.tool_call_id.clone().into()).ok();
|
||
item.insert("output", join_text(&t.content).into()).ok();
|
||
items.push(item).ok();
|
||
}
|
||
}
|
||
}
|
||
items
|
||
}
|
||
|
||
/// 核心工具定义 → Chat Completions tools（function 格式）。
|
||
/// Core tool definitions → Chat Completions tools (function format).
|
||
fn tools_to_chat(tools: &JsonValue) -> JsonValue {
|
||
match tools {
|
||
JsonValue::Arr(items) => {
|
||
let mut arr = JsonValue::arr();
|
||
for t in items {
|
||
let mut o = JsonValue::obj();
|
||
o.insert("type", "function".into()).ok();
|
||
let mut f = JsonValue::obj();
|
||
if let Some(n) = t.get_str("name") {
|
||
f.insert("name", n.into()).ok();
|
||
}
|
||
if let Some(d) = t.get_str("description") {
|
||
f.insert("description", d.into()).ok();
|
||
}
|
||
if let Some(p) = t.get("parameters") {
|
||
f.insert("parameters", p.clone()).ok();
|
||
}
|
||
o.insert("function", f).ok();
|
||
arr.push(o).ok();
|
||
}
|
||
arr
|
||
}
|
||
_ => JsonValue::arr(),
|
||
}
|
||
}
|
||
|
||
/// 核心工具定义 → Responses API tools。
|
||
/// Core tool definitions → Responses API tools.
|
||
fn tools_to_responses(tools: &JsonValue) -> JsonValue {
|
||
match tools {
|
||
JsonValue::Arr(items) => {
|
||
let mut arr = JsonValue::arr();
|
||
for t in items {
|
||
let mut o = JsonValue::obj();
|
||
o.insert("type", "function".into()).ok();
|
||
if let Some(n) = t.get_str("name") {
|
||
o.insert("name", n.into()).ok();
|
||
}
|
||
if let Some(d) = t.get_str("description") {
|
||
o.insert("description", d.into()).ok();
|
||
}
|
||
if let Some(p) = t.get("parameters") {
|
||
o.insert("parameters", p.clone()).ok();
|
||
}
|
||
arr.push(o).ok();
|
||
}
|
||
arr
|
||
}
|
||
_ => JsonValue::arr(),
|
||
}
|
||
}
|
||
|
||
fn has_tools(tools: &JsonValue) -> bool {
|
||
matches!(tools, JsonValue::Arr(items) if !items.is_empty())
|
||
}
|
||
|
||
/// 拼接文本块；思考块不计入（OpenAI 无法消费 reasoning 内容）。
|
||
/// Join text blocks; thinking is excluded (OpenAI cannot consume reasoning).
|
||
fn join_text(blocks: &[ContentBlock]) -> String {
|
||
blocks
|
||
.iter()
|
||
.filter_map(|c| match c {
|
||
ContentBlock::Text(t) => Some(t.text.clone()),
|
||
_ => None,
|
||
})
|
||
.collect::<Vec<_>>()
|
||
.join("\n")
|
||
}
|
||
|
||
/// 把 Chat Completions 的 finish_reason 映射到核心类型。
|
||
/// Map a Chat Completions finish_reason onto the core type.
|
||
fn map_chat_finish(s: &str) -> StopReason {
|
||
match s {
|
||
"length" => StopReason::Length,
|
||
"tool_calls" | "function_call" => StopReason::ToolUse,
|
||
_ => StopReason::Stop, // stop / content_filter
|
||
}
|
||
}
|
||
|
||
/// 两种协议共享的回合累积状态（基于 [`ProviderEventReducer`]）。
|
||
/// Per-turn accumulator shared by both protocols (built on
|
||
/// [`ProviderEventReducer`]).
|
||
struct OpenAiTurn {
|
||
reducer: ProviderEventReducer,
|
||
model: String,
|
||
protocol: OpenAiProtocol,
|
||
done: bool,
|
||
/// 已打开的工具调用（provider 局部索引），用于去重。
|
||
/// Tool calls already opened (provider-local index), for dedup.
|
||
seen_calls: HashSet<usize>,
|
||
/// 输入 / 输出 token（最后一块 usage 提供）。
|
||
/// Input / output tokens (from the final usage chunk).
|
||
input_tokens: u64,
|
||
output_tokens: u64,
|
||
cached_tokens: u64,
|
||
}
|
||
|
||
impl OpenAiTurn {
|
||
fn responses(model: &str) -> Self {
|
||
Self::new(model, OpenAiProtocol::Responses)
|
||
}
|
||
|
||
fn chat(model: &str) -> Self {
|
||
Self::new(model, OpenAiProtocol::ChatCompletions)
|
||
}
|
||
|
||
fn new(model: &str, protocol: OpenAiProtocol) -> Self {
|
||
Self {
|
||
reducer: ProviderEventReducer::new(model),
|
||
model: model.to_string(),
|
||
protocol,
|
||
done: false,
|
||
seen_calls: HashSet::new(),
|
||
input_tokens: 0,
|
||
output_tokens: 0,
|
||
cached_tokens: 0,
|
||
}
|
||
}
|
||
|
||
/// 收尾：关闭所有未闭合块并发送 Done。
|
||
/// Finalize: close all open blocks and send Done.
|
||
fn finalize(&mut self, tx: &SyncSender<StreamEvent>) -> Result<bool, String> {
|
||
for ev in self.reducer.finalize_events() {
|
||
if !common::send(tx, ev) {
|
||
return Ok(false);
|
||
}
|
||
}
|
||
let mut message = self
|
||
.reducer
|
||
.finish()
|
||
.ok_or_else(|| "reducer produced no final message".to_string())?;
|
||
message.usage = Usage {
|
||
input_tokens: self.input_tokens,
|
||
output_tokens: self.output_tokens,
|
||
cache_read_tokens: self.cached_tokens,
|
||
cache_write_tokens: 0,
|
||
};
|
||
self.done = true;
|
||
Ok(common::send(tx, StreamEvent::Done { message }))
|
||
}
|
||
|
||
/// 处理一个 SSE 事件。
|
||
/// Handle one SSE event.
|
||
fn handle_sse(&mut self, ev: &SseEvent, tx: &SyncSender<StreamEvent>) -> Result<bool, String> {
|
||
let data = ev.data.trim();
|
||
// Chat Completions 的流结束标记不是合法 JSON，必须特判。
|
||
// Chat Completions' end-of-stream marker is not valid JSON; special-case it.
|
||
if data == "[DONE]" {
|
||
return self.finalize(tx);
|
||
}
|
||
let json: JsonValue = if data.is_empty() {
|
||
JsonValue::obj()
|
||
} else {
|
||
focus_json::parse(data).map_err(|e| format!("invalid SSE JSON: {}", e))?
|
||
};
|
||
match self.protocol {
|
||
OpenAiProtocol::ChatCompletions => self.handle_chat_event(&json, tx),
|
||
OpenAiProtocol::Responses => self.handle_responses_event(ev, &json, tx),
|
||
}
|
||
}
|
||
|
||
fn handle_chat_event(
|
||
&mut self,
|
||
json: &JsonValue,
|
||
tx: &SyncSender<StreamEvent>,
|
||
) -> Result<bool, String> {
|
||
// API 错误以 `{"error": {...}}` chunk 出现。
|
||
// API errors arrive as a `{"error": {...}}` chunk.
|
||
if let Some(err) = json.get("error") {
|
||
let ty = err.get_str("type").unwrap_or("api_error");
|
||
let message = err.get_str("message").unwrap_or("unknown error");
|
||
self.done = true;
|
||
return Ok(common::send(
|
||
tx,
|
||
StreamEvent::Error {
|
||
error: common::error_message(&self.model, &format!("{}: {}", ty, message)),
|
||
},
|
||
));
|
||
}
|
||
if let Some(choices) = json.get_arr("choices") {
|
||
for choice in choices {
|
||
if let Some(delta) = choice.get("delta") {
|
||
if let Some(content) = delta.get_str("content") {
|
||
if !content.is_empty() {
|
||
for ev in self.reducer.text_delta(content) {
|
||
if !common::send(tx, ev) {
|
||
return Ok(false);
|
||
}
|
||
}
|
||
}
|
||
}
|
||
// 推理内容 → 思考块。不同端点字段名不同（DeepSeek 官方用
|
||
// reasoning_content，部分兼容层用 reasoning_text / reasoning /
|
||
// thinking），逐个尝试，取第一个非空。
|
||
// Reasoning → thinking block. Field names differ across
|
||
// endpoints (DeepSeek official uses reasoning_content; some
|
||
// compatible layers use reasoning_text / reasoning /
|
||
// thinking); try each in order, keep the first non-empty.
|
||
const REASONING_FIELDS: [&str; 4] = [
|
||
"reasoning_content",
|
||
"reasoning_text",
|
||
"reasoning",
|
||
"thinking",
|
||
];
|
||
if let Some(reasoning) = REASONING_FIELDS
|
||
.iter()
|
||
.find_map(|k| delta.get_str(k).filter(|v| !v.is_empty()))
|
||
{
|
||
for ev in self.reducer.thinking_delta(reasoning) {
|
||
if !common::send(tx, ev) {
|
||
return Ok(false);
|
||
}
|
||
}
|
||
}
|
||
if let Some(tcs) = delta.get_arr("tool_calls") {
|
||
for tc in tcs {
|
||
let idx = tc.get_num("index").unwrap_or(0.0) as usize;
|
||
if let Some(f) = tc.get("function") {
|
||
if let Some(name) = f.get_str("name") {
|
||
if !name.is_empty() && self.seen_calls.insert(idx) {
|
||
for ev in self.reducer.tool_call_start(idx, name) {
|
||
if !common::send(tx, ev) {
|
||
return Ok(false);
|
||
}
|
||
}
|
||
}
|
||
}
|
||
if let Some(args) = f.get_str("arguments") {
|
||
if !args.is_empty() {
|
||
for ev in self.reducer.tool_call_delta(idx, args) {
|
||
if !common::send(tx, ev) {
|
||
return Ok(false);
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
if let Some(fr) = choice.get_str("finish_reason") {
|
||
if !fr.is_empty() {
|
||
self.reducer.set_stop(map_chat_finish(fr));
|
||
}
|
||
}
|
||
}
|
||
}
|
||
if let Some(usage) = json.get("usage") {
|
||
self.input_tokens = usage.get_num("prompt_tokens").unwrap_or(0.0) as u64;
|
||
self.output_tokens = usage.get_num("completion_tokens").unwrap_or(0.0) as u64;
|
||
if let Some(details) = usage.get("prompt_tokens_details") {
|
||
self.cached_tokens = details.get_num("cached_tokens").unwrap_or(0.0) as u64;
|
||
}
|
||
}
|
||
Ok(true)
|
||
}
|
||
|
||
fn handle_responses_event(
|
||
&mut self,
|
||
ev: &SseEvent,
|
||
json: &JsonValue,
|
||
tx: &SyncSender<StreamEvent>,
|
||
) -> Result<bool, String> {
|
||
match ev.event.as_str() {
|
||
"response.output_item.added" => {
|
||
if let Some(item) = json.get("item") {
|
||
match item.get_str("type").unwrap_or("") {
|
||
"message" => {
|
||
// 打开一个空的文本块，随后的 output_text.delta 填充。
|
||
// Open an empty text block; output_text.delta fills it.
|
||
for e in self.reducer.text_delta("") {
|
||
if !common::send(tx, e) {
|
||
return Ok(false);
|
||
}
|
||
}
|
||
}
|
||
"function_call" => {
|
||
let idx = json.get_num("output_index").unwrap_or(0.0) as usize;
|
||
let name = item.get_str("name").unwrap_or("").to_string();
|
||
if !name.is_empty() {
|
||
for e in self.reducer.tool_call_start(idx, &name) {
|
||
if !common::send(tx, e) {
|
||
return Ok(false);
|
||
}
|
||
}
|
||
}
|
||
}
|
||
"reasoning" => {
|
||
// 推理条目：打开思考块，随后的 reasoning_text.delta 填充。
|
||
// A reasoning item: open a thinking block; the
|
||
// following reasoning_text.delta events fill it.
|
||
for e in self.reducer.thinking_delta("") {
|
||
if !common::send(tx, e) {
|
||
return Ok(false);
|
||
}
|
||
}
|
||
}
|
||
_ => {}
|
||
}
|
||
}
|
||
Ok(true)
|
||
}
|
||
"response.output_text.delta" => {
|
||
if let Some(delta) = json.get_str("delta") {
|
||
if !delta.is_empty() {
|
||
for e in self.reducer.text_delta(delta) {
|
||
if !common::send(tx, e) {
|
||
return Ok(false);
|
||
}
|
||
}
|
||
}
|
||
}
|
||
Ok(true)
|
||
}
|
||
"response.function_call_arguments.delta" => {
|
||
if let Some(delta) = json.get_str("delta") {
|
||
if !delta.is_empty() {
|
||
let idx = json.get_num("output_index").unwrap_or(0.0) as usize;
|
||
for e in self.reducer.tool_call_delta(idx, delta) {
|
||
if !common::send(tx, e) {
|
||
return Ok(false);
|
||
}
|
||
}
|
||
}
|
||
}
|
||
Ok(true)
|
||
}
|
||
"response.reasoning_text.delta" => {
|
||
// 推理增量（DeepSeek 等 thinking 模式）→ 思考块。
|
||
// Reasoning deltas (thinking mode, e.g. DeepSeek) → thinking.
|
||
if let Some(delta) = json.get_str("delta") {
|
||
if !delta.is_empty() {
|
||
for e in self.reducer.thinking_delta(delta) {
|
||
if !common::send(tx, e) {
|
||
return Ok(false);
|
||
}
|
||
}
|
||
}
|
||
}
|
||
Ok(true)
|
||
}
|
||
"response.completed" => {
|
||
if let Some(resp) = json.get("response") {
|
||
match resp.get_str("status").unwrap_or("completed") {
|
||
"completed" => self.reducer.set_stop(StopReason::Stop),
|
||
"incomplete" => {
|
||
let reason = resp
|
||
.get("incomplete_details")
|
||
.and_then(|d| d.get_str("reason"))
|
||
.unwrap_or("unknown");
|
||
self.reducer.set_stop(if reason == "max_output_tokens" {
|
||
StopReason::Length
|
||
} else {
|
||
StopReason::Stop
|
||
});
|
||
}
|
||
_ => {}
|
||
}
|
||
if let Some(usage) = resp.get("usage") {
|
||
self.input_tokens = usage.get_num("input_tokens").unwrap_or(0.0) as u64;
|
||
self.output_tokens = usage.get_num("output_tokens").unwrap_or(0.0) as u64;
|
||
if let Some(details) = usage.get("input_tokens_details") {
|
||
self.cached_tokens =
|
||
details.get_num("cached_tokens").unwrap_or(0.0) as u64;
|
||
}
|
||
}
|
||
}
|
||
self.finalize(tx)
|
||
}
|
||
"response.failed" | "error" => {
|
||
let code = json.get_str("code").unwrap_or("response_failed");
|
||
let message = json.get_str("message").unwrap_or("unknown error");
|
||
self.done = true;
|
||
Ok(common::send(
|
||
tx,
|
||
StreamEvent::Error {
|
||
error: common::error_message(
|
||
&self.model,
|
||
&format!("{}: {}", code, message),
|
||
),
|
||
},
|
||
))
|
||
}
|
||
_ => Ok(true), // created / in_progress / output_item.done 等忽略
|
||
}
|
||
}
|
||
}
|