- Remove unused import AsyncBufReadExt in compatible.rs - Remove unused mut keywords from response and tx - Remove unused variable 'name' - Prefix unused parameters with _ in traits.rs Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
1131 lines
36 KiB
Rust
1131 lines
36 KiB
Rust
//! Generic OpenAI-compatible provider.
|
|
//! Most LLM APIs follow the same `/v1/chat/completions` format.
|
|
//! This module provides a single implementation that works for all of them.
|
|
|
|
use crate::providers::traits::{
|
|
ChatMessage, ChatRequest as ProviderChatRequest, ChatResponse as ProviderChatResponse,
|
|
Provider, StreamChunk, StreamError, StreamOptions, StreamResult, ToolCall as ProviderToolCall,
|
|
};
|
|
use async_trait::async_trait;
|
|
use futures_util::{stream, StreamExt};
|
|
use reqwest::Client;
|
|
use serde::{Deserialize, Serialize};
|
|
|
|
/// A provider that speaks the OpenAI-compatible chat completions API.
|
|
/// Used by: Venice, Vercel AI Gateway, Cloudflare AI Gateway, Moonshot,
|
|
/// Synthetic, `OpenCode` Zen, `Z.AI`, `GLM`, `MiniMax`, Bedrock, Qianfan, Groq, Mistral, `xAI`, etc.
|
|
pub struct OpenAiCompatibleProvider {
|
|
pub(crate) name: String,
|
|
pub(crate) base_url: String,
|
|
pub(crate) api_key: Option<String>,
|
|
pub(crate) auth_header: AuthStyle,
|
|
/// When false, do not fall back to /v1/responses on chat completions 404.
|
|
/// GLM/Zhipu does not support the responses API.
|
|
supports_responses_fallback: bool,
|
|
client: Client,
|
|
}
|
|
|
|
/// How the provider expects the API key to be sent.
|
|
#[derive(Debug, Clone)]
|
|
pub enum AuthStyle {
|
|
/// `Authorization: Bearer <key>`
|
|
Bearer,
|
|
/// `x-api-key: <key>` (used by some Chinese providers)
|
|
XApiKey,
|
|
/// Custom header name
|
|
Custom(String),
|
|
}
|
|
|
|
impl OpenAiCompatibleProvider {
|
|
pub fn new(name: &str, base_url: &str, api_key: Option<&str>, auth_style: AuthStyle) -> Self {
|
|
Self {
|
|
name: name.to_string(),
|
|
base_url: base_url.trim_end_matches('/').to_string(),
|
|
api_key: api_key.map(ToString::to_string),
|
|
auth_header: auth_style,
|
|
supports_responses_fallback: true,
|
|
client: Client::builder()
|
|
.timeout(std::time::Duration::from_secs(120))
|
|
.connect_timeout(std::time::Duration::from_secs(10))
|
|
.build()
|
|
.unwrap_or_else(|_| Client::new()),
|
|
}
|
|
}
|
|
|
|
/// Same as `new` but skips the /v1/responses fallback on 404.
|
|
/// Use for providers (e.g. GLM) that only support chat completions.
|
|
pub fn new_no_responses_fallback(
|
|
name: &str,
|
|
base_url: &str,
|
|
api_key: Option<&str>,
|
|
auth_style: AuthStyle,
|
|
) -> Self {
|
|
Self {
|
|
name: name.to_string(),
|
|
base_url: base_url.trim_end_matches('/').to_string(),
|
|
api_key: api_key.map(ToString::to_string),
|
|
auth_header: auth_style,
|
|
supports_responses_fallback: false,
|
|
client: Client::builder()
|
|
.timeout(std::time::Duration::from_secs(120))
|
|
.connect_timeout(std::time::Duration::from_secs(10))
|
|
.build()
|
|
.unwrap_or_else(|_| Client::new()),
|
|
}
|
|
}
|
|
|
|
/// Build the full URL for chat completions, detecting if base_url already includes the path.
|
|
/// This allows custom providers with non-standard endpoints (e.g., VolcEngine ARK uses
|
|
/// `/api/coding/v3/chat/completions` instead of `/v1/chat/completions`).
|
|
fn chat_completions_url(&self) -> String {
|
|
let has_full_endpoint = reqwest::Url::parse(&self.base_url)
|
|
.map(|url| {
|
|
url.path()
|
|
.trim_end_matches('/')
|
|
.ends_with("/chat/completions")
|
|
})
|
|
.unwrap_or_else(|_| {
|
|
self.base_url
|
|
.trim_end_matches('/')
|
|
.ends_with("/chat/completions")
|
|
});
|
|
|
|
if has_full_endpoint {
|
|
self.base_url.clone()
|
|
} else {
|
|
format!("{}/chat/completions", self.base_url)
|
|
}
|
|
}
|
|
|
|
fn path_ends_with(&self, suffix: &str) -> bool {
|
|
if let Ok(url) = reqwest::Url::parse(&self.base_url) {
|
|
return url.path().trim_end_matches('/').ends_with(suffix);
|
|
}
|
|
|
|
self.base_url.trim_end_matches('/').ends_with(suffix)
|
|
}
|
|
|
|
fn has_explicit_api_path(&self) -> bool {
|
|
let Ok(url) = reqwest::Url::parse(&self.base_url) else {
|
|
return false;
|
|
};
|
|
|
|
let path = url.path().trim_end_matches('/');
|
|
!path.is_empty() && path != "/"
|
|
}
|
|
|
|
/// Build the full URL for responses API, detecting if base_url already includes the path.
|
|
fn responses_url(&self) -> String {
|
|
if self.path_ends_with("/responses") {
|
|
return self.base_url.clone();
|
|
}
|
|
|
|
let normalized_base = self.base_url.trim_end_matches('/');
|
|
|
|
// If chat endpoint is explicitly configured, derive sibling responses endpoint.
|
|
if let Some(prefix) = normalized_base.strip_suffix("/chat/completions") {
|
|
return format!("{prefix}/responses");
|
|
}
|
|
|
|
// If an explicit API path already exists (e.g. /v1, /openai, /api/coding/v3),
|
|
// append responses directly to avoid duplicate /v1 segments.
|
|
if self.has_explicit_api_path() {
|
|
format!("{normalized_base}/responses")
|
|
} else {
|
|
format!("{normalized_base}/v1/responses")
|
|
}
|
|
}
|
|
}
|
|
|
|
#[derive(Debug, Serialize)]
|
|
struct ChatRequest {
|
|
model: String,
|
|
messages: Vec<Message>,
|
|
temperature: f64,
|
|
#[serde(skip_serializing_if = "Option::is_none")]
|
|
stream: Option<bool>,
|
|
}
|
|
|
|
#[derive(Debug, Serialize)]
|
|
struct Message {
|
|
role: String,
|
|
content: String,
|
|
}
|
|
|
|
#[derive(Debug, Deserialize)]
|
|
struct ApiChatResponse {
|
|
choices: Vec<Choice>,
|
|
}
|
|
|
|
#[derive(Debug, Deserialize)]
|
|
struct Choice {
|
|
message: ResponseMessage,
|
|
}
|
|
|
|
#[derive(Debug, Deserialize, Serialize)]
|
|
struct ResponseMessage {
|
|
#[serde(default)]
|
|
content: Option<String>,
|
|
#[serde(default)]
|
|
tool_calls: Option<Vec<ToolCall>>,
|
|
}
|
|
|
|
#[derive(Debug, Deserialize, Serialize)]
|
|
struct ToolCall {
|
|
#[serde(rename = "type")]
|
|
kind: Option<String>,
|
|
function: Option<Function>,
|
|
}
|
|
|
|
#[derive(Debug, Deserialize, Serialize)]
|
|
struct Function {
|
|
name: Option<String>,
|
|
arguments: Option<String>,
|
|
}
|
|
|
|
#[derive(Debug, Serialize)]
|
|
struct ResponsesRequest {
|
|
model: String,
|
|
input: Vec<ResponsesInput>,
|
|
#[serde(skip_serializing_if = "Option::is_none")]
|
|
instructions: Option<String>,
|
|
#[serde(skip_serializing_if = "Option::is_none")]
|
|
stream: Option<bool>,
|
|
}
|
|
|
|
#[derive(Debug, Serialize)]
|
|
struct ResponsesInput {
|
|
role: String,
|
|
content: String,
|
|
}
|
|
|
|
#[derive(Debug, Deserialize)]
|
|
struct ResponsesResponse {
|
|
#[serde(default)]
|
|
output: Vec<ResponsesOutput>,
|
|
#[serde(default)]
|
|
output_text: Option<String>,
|
|
}
|
|
|
|
#[derive(Debug, Deserialize)]
|
|
struct ResponsesOutput {
|
|
#[serde(default)]
|
|
content: Vec<ResponsesContent>,
|
|
}
|
|
|
|
#[derive(Debug, Deserialize)]
|
|
struct ResponsesContent {
|
|
#[serde(rename = "type")]
|
|
kind: Option<String>,
|
|
text: Option<String>,
|
|
}
|
|
|
|
// ═══════════════════════════════════════════════════════════════
|
|
// Streaming support (SSE parser)
|
|
// ═══════════════════════════════════════════════════════════════
|
|
|
|
/// Server-Sent Event stream chunk for OpenAI-compatible streaming.
|
|
#[derive(Debug, Deserialize)]
|
|
struct StreamChunkResponse {
|
|
choices: Vec<StreamChoice>,
|
|
}
|
|
|
|
#[derive(Debug, Deserialize)]
|
|
struct StreamChoice {
|
|
delta: StreamDelta,
|
|
finish_reason: Option<String>,
|
|
}
|
|
|
|
#[derive(Debug, Deserialize)]
|
|
struct StreamDelta {
|
|
#[serde(default)]
|
|
content: Option<String>,
|
|
}
|
|
|
|
/// Parse SSE (Server-Sent Events) stream from OpenAI-compatible providers.
|
|
/// Handles the `data: {...}` format and `[DONE]` sentinel.
|
|
fn parse_sse_line(line: &str) -> StreamResult<Option<String>> {
|
|
let line = line.trim();
|
|
|
|
// Skip empty lines and comments
|
|
if line.is_empty() || line.starts_with(':') {
|
|
return Ok(None);
|
|
}
|
|
|
|
// SSE format: "data: {...}"
|
|
if let Some(data) = line.strip_prefix("data:") {
|
|
let data = data.trim();
|
|
|
|
// Check for [DONE] sentinel
|
|
if data == "[DONE]" {
|
|
return Ok(None);
|
|
}
|
|
|
|
// Parse JSON delta
|
|
let chunk: StreamChunkResponse = serde_json::from_str(data).map_err(StreamError::Json)?;
|
|
|
|
// Extract content from delta
|
|
if let Some(choice) = chunk.choices.first() {
|
|
if let Some(content) = &choice.delta.content {
|
|
return Ok(Some(content.clone()));
|
|
}
|
|
}
|
|
}
|
|
|
|
Ok(None)
|
|
}
|
|
|
|
/// Convert SSE byte stream to text chunks.
|
|
async fn sse_bytes_to_chunks(
|
|
response: reqwest::Response,
|
|
count_tokens: bool,
|
|
) -> stream::BoxStream<'static, StreamResult<StreamChunk>> {
|
|
// Create a channel to send chunks
|
|
let (tx, rx) = tokio::sync::mpsc::channel::<StreamResult<StreamChunk>>(100);
|
|
|
|
tokio::spawn(async move {
|
|
// Buffer for incomplete lines
|
|
let mut buffer = String::new();
|
|
|
|
// Get response body as bytes stream
|
|
match response.error_for_status_ref() {
|
|
Ok(_) => {}
|
|
Err(e) => {
|
|
let _ = tx.send(Err(StreamError::Http(e))).await;
|
|
return;
|
|
}
|
|
}
|
|
|
|
let mut bytes_stream = response.bytes_stream();
|
|
|
|
while let Some(item) = bytes_stream.next().await {
|
|
match item {
|
|
Ok(bytes) => {
|
|
// Convert bytes to string and process line by line
|
|
let text = match String::from_utf8(bytes.to_vec()) {
|
|
Ok(t) => t,
|
|
Err(e) => {
|
|
let _ = tx
|
|
.send(Err(StreamError::InvalidSse(format!(
|
|
"Invalid UTF-8: {}",
|
|
e
|
|
))))
|
|
.await;
|
|
break;
|
|
}
|
|
};
|
|
|
|
buffer.push_str(&text);
|
|
|
|
// Process complete lines
|
|
while let Some(pos) = buffer.find('\n') {
|
|
let line = buffer.drain(..=pos).collect::<String>();
|
|
buffer = buffer[pos + 1..].to_string();
|
|
|
|
match parse_sse_line(&line) {
|
|
Ok(Some(content)) => {
|
|
let mut chunk = StreamChunk::delta(content);
|
|
if count_tokens {
|
|
chunk = chunk.with_token_estimate();
|
|
}
|
|
if tx.send(Ok(chunk)).await.is_err() {
|
|
return; // Receiver dropped
|
|
}
|
|
}
|
|
Ok(None) => {
|
|
// Empty line or [DONE] sentinel - continue
|
|
continue;
|
|
}
|
|
Err(e) => {
|
|
let _ = tx.send(Err(e)).await;
|
|
return;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
Err(e) => {
|
|
let _ = tx.send(Err(StreamError::Http(e))).await;
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
|
|
// Send final chunk
|
|
let _ = tx.send(Ok(StreamChunk::final_chunk())).await;
|
|
});
|
|
|
|
// Convert channel receiver to stream
|
|
stream::unfold(rx, |mut rx| async {
|
|
match rx.recv().await {
|
|
Some(chunk) => Some((chunk, rx)),
|
|
None => None,
|
|
}
|
|
})
|
|
.boxed()
|
|
}
|
|
|
|
fn first_nonempty(text: Option<&str>) -> Option<String> {
|
|
text.and_then(|value| {
|
|
let trimmed = value.trim();
|
|
if trimmed.is_empty() {
|
|
None
|
|
} else {
|
|
Some(trimmed.to_string())
|
|
}
|
|
})
|
|
}
|
|
|
|
fn extract_responses_text(response: ResponsesResponse) -> Option<String> {
|
|
if let Some(text) = first_nonempty(response.output_text.as_deref()) {
|
|
return Some(text);
|
|
}
|
|
|
|
for item in &response.output {
|
|
for content in &item.content {
|
|
if content.kind.as_deref() == Some("output_text") {
|
|
if let Some(text) = first_nonempty(content.text.as_deref()) {
|
|
return Some(text);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
for item in &response.output {
|
|
for content in &item.content {
|
|
if let Some(text) = first_nonempty(content.text.as_deref()) {
|
|
return Some(text);
|
|
}
|
|
}
|
|
}
|
|
|
|
None
|
|
}
|
|
|
|
impl OpenAiCompatibleProvider {
|
|
fn apply_auth_header(
|
|
&self,
|
|
req: reqwest::RequestBuilder,
|
|
api_key: &str,
|
|
) -> reqwest::RequestBuilder {
|
|
match &self.auth_header {
|
|
AuthStyle::Bearer => req.header("Authorization", format!("Bearer {api_key}")),
|
|
AuthStyle::XApiKey => req.header("x-api-key", api_key),
|
|
AuthStyle::Custom(header) => req.header(header, api_key),
|
|
}
|
|
}
|
|
|
|
async fn chat_via_responses(
|
|
&self,
|
|
api_key: &str,
|
|
system_prompt: Option<&str>,
|
|
message: &str,
|
|
model: &str,
|
|
) -> anyhow::Result<String> {
|
|
let request = ResponsesRequest {
|
|
model: model.to_string(),
|
|
input: vec![ResponsesInput {
|
|
role: "user".to_string(),
|
|
content: message.to_string(),
|
|
}],
|
|
instructions: system_prompt.map(str::to_string),
|
|
stream: Some(false),
|
|
};
|
|
|
|
let url = self.responses_url();
|
|
|
|
let response = self
|
|
.apply_auth_header(self.client.post(&url).json(&request), api_key)
|
|
.send()
|
|
.await?;
|
|
|
|
if !response.status().is_success() {
|
|
let error = response.text().await?;
|
|
anyhow::bail!("{} Responses API error: {error}", self.name);
|
|
}
|
|
|
|
let responses: ResponsesResponse = response.json().await?;
|
|
|
|
extract_responses_text(responses)
|
|
.ok_or_else(|| anyhow::anyhow!("No response from {} Responses API", self.name))
|
|
}
|
|
}
|
|
|
|
#[async_trait]
|
|
impl Provider for OpenAiCompatibleProvider {
|
|
async fn chat_with_system(
|
|
&self,
|
|
system_prompt: Option<&str>,
|
|
message: &str,
|
|
model: &str,
|
|
temperature: f64,
|
|
) -> anyhow::Result<String> {
|
|
let api_key = self.api_key.as_ref().ok_or_else(|| {
|
|
anyhow::anyhow!(
|
|
"{} API key not set. Run `zeroclaw onboard` or set the appropriate env var.",
|
|
self.name
|
|
)
|
|
})?;
|
|
|
|
let mut messages = Vec::new();
|
|
|
|
if let Some(sys) = system_prompt {
|
|
messages.push(Message {
|
|
role: "system".to_string(),
|
|
content: sys.to_string(),
|
|
});
|
|
}
|
|
|
|
messages.push(Message {
|
|
role: "user".to_string(),
|
|
content: message.to_string(),
|
|
});
|
|
|
|
let request = ChatRequest {
|
|
model: model.to_string(),
|
|
messages,
|
|
temperature,
|
|
stream: Some(false),
|
|
};
|
|
|
|
let url = self.chat_completions_url();
|
|
|
|
let response = self
|
|
.apply_auth_header(self.client.post(&url).json(&request), api_key)
|
|
.send()
|
|
.await?;
|
|
|
|
if !response.status().is_success() {
|
|
let status = response.status();
|
|
let error = response.text().await?;
|
|
let sanitized = super::sanitize_api_error(&error);
|
|
|
|
if status == reqwest::StatusCode::NOT_FOUND && self.supports_responses_fallback {
|
|
return self
|
|
.chat_via_responses(api_key, system_prompt, message, model)
|
|
.await
|
|
.map_err(|responses_err| {
|
|
anyhow::anyhow!(
|
|
"{} API error ({status}): {sanitized} (chat completions unavailable; responses fallback failed: {responses_err})",
|
|
self.name
|
|
)
|
|
});
|
|
}
|
|
|
|
anyhow::bail!("{} API error ({status}): {sanitized}", self.name);
|
|
}
|
|
|
|
let chat_response: ApiChatResponse = response.json().await?;
|
|
|
|
chat_response
|
|
.choices
|
|
.into_iter()
|
|
.next()
|
|
.map(|c| {
|
|
// If tool_calls are present, serialize the full message as JSON
|
|
// so parse_tool_calls can handle the OpenAI-style format
|
|
if c.message.tool_calls.is_some()
|
|
&& c.message
|
|
.tool_calls
|
|
.as_ref()
|
|
.map_or(false, |t| !t.is_empty())
|
|
{
|
|
serde_json::to_string(&c.message)
|
|
.unwrap_or_else(|_| c.message.content.unwrap_or_default())
|
|
} else {
|
|
// No tool calls, return content as-is
|
|
c.message.content.unwrap_or_default()
|
|
}
|
|
})
|
|
.ok_or_else(|| anyhow::anyhow!("No response from {}", self.name))
|
|
}
|
|
|
|
async fn chat_with_history(
|
|
&self,
|
|
messages: &[ChatMessage],
|
|
model: &str,
|
|
temperature: f64,
|
|
) -> anyhow::Result<String> {
|
|
let api_key = self.api_key.as_ref().ok_or_else(|| {
|
|
anyhow::anyhow!(
|
|
"{} API key not set. Run `zeroclaw onboard` or set the appropriate env var.",
|
|
self.name
|
|
)
|
|
})?;
|
|
|
|
let api_messages: Vec<Message> = messages
|
|
.iter()
|
|
.map(|m| Message {
|
|
role: m.role.clone(),
|
|
content: m.content.clone(),
|
|
})
|
|
.collect();
|
|
|
|
let request = ChatRequest {
|
|
model: model.to_string(),
|
|
messages: api_messages,
|
|
temperature,
|
|
stream: Some(false),
|
|
};
|
|
|
|
let url = self.chat_completions_url();
|
|
let response = self
|
|
.apply_auth_header(self.client.post(&url).json(&request), api_key)
|
|
.send()
|
|
.await?;
|
|
|
|
if !response.status().is_success() {
|
|
let status = response.status();
|
|
|
|
// Mirror chat_with_system: 404 may mean this provider uses the Responses API
|
|
if status == reqwest::StatusCode::NOT_FOUND && self.supports_responses_fallback {
|
|
// Extract system prompt and last user message for responses fallback
|
|
let system = messages.iter().find(|m| m.role == "system");
|
|
let last_user = messages.iter().rfind(|m| m.role == "user");
|
|
if let Some(user_msg) = last_user {
|
|
return self
|
|
.chat_via_responses(
|
|
api_key,
|
|
system.map(|m| m.content.as_str()),
|
|
&user_msg.content,
|
|
model,
|
|
)
|
|
.await
|
|
.map_err(|responses_err| {
|
|
anyhow::anyhow!(
|
|
"{} API error (chat completions unavailable; responses fallback failed: {responses_err})",
|
|
self.name
|
|
)
|
|
});
|
|
}
|
|
}
|
|
|
|
return Err(super::api_error(&self.name, response).await);
|
|
}
|
|
|
|
let chat_response: ApiChatResponse = response.json().await?;
|
|
|
|
chat_response
|
|
.choices
|
|
.into_iter()
|
|
.next()
|
|
.map(|c| {
|
|
// If tool_calls are present, serialize the full message as JSON
|
|
// so parse_tool_calls can handle the OpenAI-style format
|
|
if c.message.tool_calls.is_some()
|
|
&& c.message
|
|
.tool_calls
|
|
.as_ref()
|
|
.map_or(false, |t| !t.is_empty())
|
|
{
|
|
serde_json::to_string(&c.message)
|
|
.unwrap_or_else(|_| c.message.content.unwrap_or_default())
|
|
} else {
|
|
// No tool calls, return content as-is
|
|
c.message.content.unwrap_or_default()
|
|
}
|
|
})
|
|
.ok_or_else(|| anyhow::anyhow!("No response from {}", self.name))
|
|
}
|
|
|
|
async fn chat(
|
|
&self,
|
|
request: ProviderChatRequest<'_>,
|
|
model: &str,
|
|
temperature: f64,
|
|
) -> anyhow::Result<ProviderChatResponse> {
|
|
let text = self
|
|
.chat_with_history(request.messages, model, temperature)
|
|
.await?;
|
|
|
|
// Backward compatible path: chat_with_history may serialize tool_calls JSON into content.
|
|
if let Ok(message) = serde_json::from_str::<ResponseMessage>(&text) {
|
|
let tool_calls = message
|
|
.tool_calls
|
|
.unwrap_or_default()
|
|
.into_iter()
|
|
.filter_map(|tc| {
|
|
let function = tc.function?;
|
|
let name = function.name?;
|
|
let arguments = function.arguments.unwrap_or_else(|| "{}".to_string());
|
|
Some(ProviderToolCall {
|
|
id: uuid::Uuid::new_v4().to_string(),
|
|
name,
|
|
arguments,
|
|
})
|
|
})
|
|
.collect::<Vec<_>>();
|
|
|
|
return Ok(ProviderChatResponse {
|
|
text: message.content,
|
|
tool_calls,
|
|
});
|
|
}
|
|
|
|
Ok(ProviderChatResponse {
|
|
text: Some(text),
|
|
tool_calls: vec![],
|
|
})
|
|
}
|
|
|
|
fn supports_native_tools(&self) -> bool {
|
|
true
|
|
}
|
|
|
|
fn supports_streaming(&self) -> bool {
|
|
true
|
|
}
|
|
|
|
fn stream_chat_with_system(
|
|
&self,
|
|
system_prompt: Option<&str>,
|
|
message: &str,
|
|
model: &str,
|
|
temperature: f64,
|
|
options: StreamOptions,
|
|
) -> stream::BoxStream<'static, StreamResult<StreamChunk>> {
|
|
let api_key = match self.api_key.as_ref() {
|
|
Some(key) => key.clone(),
|
|
None => {
|
|
let provider_name = self.name.clone();
|
|
return stream::once(async move {
|
|
Err(StreamError::Provider(format!(
|
|
"{} API key not set",
|
|
provider_name
|
|
)))
|
|
})
|
|
.boxed();
|
|
}
|
|
};
|
|
|
|
let mut messages = Vec::new();
|
|
if let Some(sys) = system_prompt {
|
|
messages.push(Message {
|
|
role: "system".to_string(),
|
|
content: sys.to_string(),
|
|
});
|
|
}
|
|
messages.push(Message {
|
|
role: "user".to_string(),
|
|
content: message.to_string(),
|
|
});
|
|
|
|
let request = ChatRequest {
|
|
model: model.to_string(),
|
|
messages,
|
|
temperature,
|
|
stream: Some(options.enabled),
|
|
};
|
|
|
|
let url = self.chat_completions_url();
|
|
let client = self.client.clone();
|
|
let auth_header = self.auth_header.clone();
|
|
|
|
// Use a channel to bridge the async HTTP response to the stream
|
|
let (tx, rx) = tokio::sync::mpsc::channel::<StreamResult<StreamChunk>>(100);
|
|
|
|
tokio::spawn(async move {
|
|
// Build request with auth
|
|
let mut req_builder = client.post(&url).json(&request);
|
|
|
|
// Apply auth header
|
|
req_builder = match &auth_header {
|
|
AuthStyle::Bearer => {
|
|
req_builder.header("Authorization", format!("Bearer {}", api_key))
|
|
}
|
|
AuthStyle::XApiKey => req_builder.header("x-api-key", &api_key),
|
|
AuthStyle::Custom(header) => req_builder.header(header, &api_key),
|
|
};
|
|
|
|
// Set accept header for streaming
|
|
req_builder = req_builder.header("Accept", "text/event-stream");
|
|
|
|
// Send request
|
|
let response = match req_builder.send().await {
|
|
Ok(r) => r,
|
|
Err(e) => {
|
|
let _ = tx.send(Err(StreamError::Http(e))).await;
|
|
return;
|
|
}
|
|
};
|
|
|
|
// Check status
|
|
if !response.status().is_success() {
|
|
let status = response.status();
|
|
let error = match response.text().await {
|
|
Ok(e) => e,
|
|
Err(_) => format!("HTTP error: {}", status),
|
|
};
|
|
let _ = tx
|
|
.send(Err(StreamError::Provider(format!("{}: {}", status, error))))
|
|
.await;
|
|
return;
|
|
}
|
|
|
|
// Convert to chunk stream and forward to channel
|
|
let mut chunk_stream = sse_bytes_to_chunks(response, options.count_tokens).await;
|
|
while let Some(chunk) = chunk_stream.next().await {
|
|
if tx.send(chunk).await.is_err() {
|
|
break; // Receiver dropped
|
|
}
|
|
}
|
|
});
|
|
|
|
// Convert channel receiver to stream
|
|
stream::unfold(rx, |mut rx| async move {
|
|
match rx.recv().await {
|
|
Some(chunk) => Some((chunk, rx)),
|
|
None => None,
|
|
}
|
|
})
|
|
.boxed()
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
|
|
fn make_provider(name: &str, url: &str, key: Option<&str>) -> OpenAiCompatibleProvider {
|
|
OpenAiCompatibleProvider::new(name, url, key, AuthStyle::Bearer)
|
|
}
|
|
|
|
#[test]
|
|
fn creates_with_key() {
|
|
let p = make_provider("venice", "https://api.venice.ai", Some("vn-key"));
|
|
assert_eq!(p.name, "venice");
|
|
assert_eq!(p.base_url, "https://api.venice.ai");
|
|
assert_eq!(p.api_key.as_deref(), Some("vn-key"));
|
|
}
|
|
|
|
#[test]
|
|
fn creates_without_key() {
|
|
let p = make_provider("test", "https://example.com", None);
|
|
assert!(p.api_key.is_none());
|
|
}
|
|
|
|
#[test]
|
|
fn strips_trailing_slash() {
|
|
let p = make_provider("test", "https://example.com/", None);
|
|
assert_eq!(p.base_url, "https://example.com");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn chat_fails_without_key() {
|
|
let p = make_provider("Venice", "https://api.venice.ai", None);
|
|
let result = p
|
|
.chat_with_system(None, "hello", "llama-3.3-70b", 0.7)
|
|
.await;
|
|
assert!(result.is_err());
|
|
assert!(result
|
|
.unwrap_err()
|
|
.to_string()
|
|
.contains("Venice API key not set"));
|
|
}
|
|
|
|
#[test]
|
|
fn request_serializes_correctly() {
|
|
let req = ChatRequest {
|
|
model: "llama-3.3-70b".to_string(),
|
|
messages: vec![
|
|
Message {
|
|
role: "system".to_string(),
|
|
content: "You are ZeroClaw".to_string(),
|
|
},
|
|
Message {
|
|
role: "user".to_string(),
|
|
content: "hello".to_string(),
|
|
},
|
|
],
|
|
temperature: 0.4,
|
|
stream: Some(false),
|
|
};
|
|
let json = serde_json::to_string(&req).unwrap();
|
|
assert!(json.contains("llama-3.3-70b"));
|
|
assert!(json.contains("system"));
|
|
assert!(json.contains("user"));
|
|
}
|
|
|
|
#[test]
|
|
fn response_deserializes() {
|
|
let json = r#"{"choices":[{"message":{"content":"Hello from Venice!"}}]}"#;
|
|
let resp: ApiChatResponse = serde_json::from_str(json).unwrap();
|
|
assert_eq!(
|
|
resp.choices[0].message.content,
|
|
Some("Hello from Venice!".to_string())
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn response_empty_choices() {
|
|
let json = r#"{"choices":[]}"#;
|
|
let resp: ApiChatResponse = serde_json::from_str(json).unwrap();
|
|
assert!(resp.choices.is_empty());
|
|
}
|
|
|
|
#[test]
|
|
fn x_api_key_auth_style() {
|
|
let p = OpenAiCompatibleProvider::new(
|
|
"moonshot",
|
|
"https://api.moonshot.cn",
|
|
Some("ms-key"),
|
|
AuthStyle::XApiKey,
|
|
);
|
|
assert!(matches!(p.auth_header, AuthStyle::XApiKey));
|
|
}
|
|
|
|
#[test]
|
|
fn custom_auth_style() {
|
|
let p = OpenAiCompatibleProvider::new(
|
|
"custom",
|
|
"https://api.example.com",
|
|
Some("key"),
|
|
AuthStyle::Custom("X-Custom-Key".into()),
|
|
);
|
|
assert!(matches!(p.auth_header, AuthStyle::Custom(_)));
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn all_compatible_providers_fail_without_key() {
|
|
let providers = vec![
|
|
make_provider("Venice", "https://api.venice.ai", None),
|
|
make_provider("Moonshot", "https://api.moonshot.cn", None),
|
|
make_provider("GLM", "https://open.bigmodel.cn", None),
|
|
make_provider("MiniMax", "https://api.minimaxi.com/v1", None),
|
|
make_provider("Groq", "https://api.groq.com/openai", None),
|
|
make_provider("Mistral", "https://api.mistral.ai", None),
|
|
make_provider("xAI", "https://api.x.ai", None),
|
|
];
|
|
|
|
for p in providers {
|
|
let result = p.chat_with_system(None, "test", "model", 0.7).await;
|
|
assert!(result.is_err(), "{} should fail without key", p.name);
|
|
assert!(
|
|
result.unwrap_err().to_string().contains("API key not set"),
|
|
"{} error should mention key",
|
|
p.name
|
|
);
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn responses_extracts_top_level_output_text() {
|
|
let json = r#"{"output_text":"Hello from top-level","output":[]}"#;
|
|
let response: ResponsesResponse = serde_json::from_str(json).unwrap();
|
|
assert_eq!(
|
|
extract_responses_text(response).as_deref(),
|
|
Some("Hello from top-level")
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn responses_extracts_nested_output_text() {
|
|
let json =
|
|
r#"{"output":[{"content":[{"type":"output_text","text":"Hello from nested"}]}]}"#;
|
|
let response: ResponsesResponse = serde_json::from_str(json).unwrap();
|
|
assert_eq!(
|
|
extract_responses_text(response).as_deref(),
|
|
Some("Hello from nested")
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn responses_extracts_any_text_as_fallback() {
|
|
let json = r#"{"output":[{"content":[{"type":"message","text":"Fallback text"}]}]}"#;
|
|
let response: ResponsesResponse = serde_json::from_str(json).unwrap();
|
|
assert_eq!(
|
|
extract_responses_text(response).as_deref(),
|
|
Some("Fallback text")
|
|
);
|
|
}
|
|
|
|
// ══════════════════════════════════════════════════════════
|
|
// Custom endpoint path tests (Issue #114)
|
|
// ══════════════════════════════════════════════════════════
|
|
|
|
#[test]
|
|
fn chat_completions_url_standard_openai() {
|
|
// Standard OpenAI-compatible providers get /chat/completions appended
|
|
let p = make_provider("openai", "https://api.openai.com/v1", None);
|
|
assert_eq!(
|
|
p.chat_completions_url(),
|
|
"https://api.openai.com/v1/chat/completions"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn chat_completions_url_trailing_slash() {
|
|
// Trailing slash is stripped, then /chat/completions appended
|
|
let p = make_provider("test", "https://api.example.com/v1/", None);
|
|
assert_eq!(
|
|
p.chat_completions_url(),
|
|
"https://api.example.com/v1/chat/completions"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn chat_completions_url_volcengine_ark() {
|
|
// VolcEngine ARK uses custom path - should use as-is
|
|
let p = make_provider(
|
|
"volcengine",
|
|
"https://ark.cn-beijing.volces.com/api/coding/v3/chat/completions",
|
|
None,
|
|
);
|
|
assert_eq!(
|
|
p.chat_completions_url(),
|
|
"https://ark.cn-beijing.volces.com/api/coding/v3/chat/completions"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn chat_completions_url_custom_full_endpoint() {
|
|
// Custom provider with full endpoint path
|
|
let p = make_provider(
|
|
"custom",
|
|
"https://my-api.example.com/v2/llm/chat/completions",
|
|
None,
|
|
);
|
|
assert_eq!(
|
|
p.chat_completions_url(),
|
|
"https://my-api.example.com/v2/llm/chat/completions"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn chat_completions_url_requires_exact_suffix_match() {
|
|
let p = make_provider(
|
|
"custom",
|
|
"https://my-api.example.com/v2/llm/chat/completions-proxy",
|
|
None,
|
|
);
|
|
assert_eq!(
|
|
p.chat_completions_url(),
|
|
"https://my-api.example.com/v2/llm/chat/completions-proxy/chat/completions"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn responses_url_standard() {
|
|
// Standard providers get /v1/responses appended
|
|
let p = make_provider("test", "https://api.example.com", None);
|
|
assert_eq!(p.responses_url(), "https://api.example.com/v1/responses");
|
|
}
|
|
|
|
#[test]
|
|
fn responses_url_custom_full_endpoint() {
|
|
// Custom provider with full responses endpoint
|
|
let p = make_provider(
|
|
"custom",
|
|
"https://my-api.example.com/api/v2/responses",
|
|
None,
|
|
);
|
|
assert_eq!(
|
|
p.responses_url(),
|
|
"https://my-api.example.com/api/v2/responses"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn responses_url_requires_exact_suffix_match() {
|
|
let p = make_provider(
|
|
"custom",
|
|
"https://my-api.example.com/api/v2/responses-proxy",
|
|
None,
|
|
);
|
|
assert_eq!(
|
|
p.responses_url(),
|
|
"https://my-api.example.com/api/v2/responses-proxy/responses"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn responses_url_derives_from_chat_endpoint() {
|
|
let p = make_provider(
|
|
"custom",
|
|
"https://my-api.example.com/api/v2/chat/completions",
|
|
None,
|
|
);
|
|
assert_eq!(
|
|
p.responses_url(),
|
|
"https://my-api.example.com/api/v2/responses"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn responses_url_base_with_v1_no_duplicate() {
|
|
let p = make_provider("test", "https://api.example.com/v1", None);
|
|
assert_eq!(p.responses_url(), "https://api.example.com/v1/responses");
|
|
}
|
|
|
|
#[test]
|
|
fn responses_url_non_v1_api_path_uses_raw_suffix() {
|
|
let p = make_provider("test", "https://api.example.com/api/coding/v3", None);
|
|
assert_eq!(
|
|
p.responses_url(),
|
|
"https://api.example.com/api/coding/v3/responses"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn chat_completions_url_without_v1() {
|
|
// Provider configured without /v1 in base URL
|
|
let p = make_provider("test", "https://api.example.com", None);
|
|
assert_eq!(
|
|
p.chat_completions_url(),
|
|
"https://api.example.com/chat/completions"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn chat_completions_url_base_with_v1() {
|
|
// Provider configured with /v1 in base URL
|
|
let p = make_provider("test", "https://api.example.com/v1", None);
|
|
assert_eq!(
|
|
p.chat_completions_url(),
|
|
"https://api.example.com/v1/chat/completions"
|
|
);
|
|
}
|
|
|
|
// ══════════════════════════════════════════════════════════
|
|
// Provider-specific endpoint tests (Issue #167)
|
|
// ══════════════════════════════════════════════════════════
|
|
|
|
#[test]
|
|
fn chat_completions_url_zai() {
|
|
// Z.AI uses /api/paas/v4 base path
|
|
let p = make_provider("zai", "https://api.z.ai/api/paas/v4", None);
|
|
assert_eq!(
|
|
p.chat_completions_url(),
|
|
"https://api.z.ai/api/paas/v4/chat/completions"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn chat_completions_url_minimax() {
|
|
// MiniMax OpenAI-compatible endpoint requires /v1 base path.
|
|
let p = make_provider("minimax", "https://api.minimaxi.com/v1", None);
|
|
assert_eq!(
|
|
p.chat_completions_url(),
|
|
"https://api.minimaxi.com/v1/chat/completions"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn chat_completions_url_glm() {
|
|
// GLM (BigModel) uses /api/paas/v4 base path
|
|
let p = make_provider("glm", "https://open.bigmodel.cn/api/paas/v4", None);
|
|
assert_eq!(
|
|
p.chat_completions_url(),
|
|
"https://open.bigmodel.cn/api/paas/v4/chat/completions"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn chat_completions_url_opencode() {
|
|
// OpenCode Zen uses /zen/v1 base path
|
|
let p = make_provider("opencode", "https://opencode.ai/zen/v1", None);
|
|
assert_eq!(
|
|
p.chat_completions_url(),
|
|
"https://opencode.ai/zen/v1/chat/completions"
|
|
);
|
|
}
|
|
}
|