litellm_rust/providers/
openai_compat.rs

1use crate::config::ProviderConfig;
2use crate::error::{LiteLLMError, Result};
3use crate::http::send_json;
4use crate::providers::{merge_extra_body, resolve_api_key};
5use crate::stream::{parse_sse_stream, ChatStream};
6use crate::types::{
7    ChatRequest, ChatResponse, EmbeddingRequest, EmbeddingResponse, ImageData, ImageRequest,
8    ImageResponse, Reasoning, Usage, VideoRequest, VideoResponse,
9};
10use base64::{engine::general_purpose, Engine as _};
11use reqwest::multipart::Form;
12use reqwest::Client;
13use serde::Deserialize;
14use serde_json::Value;
15use std::time::Duration;
16use tokio::time::sleep;
17
18/// Default maximum polling attempts for video generation (120 * 5s = 10 minutes)
19pub const DEFAULT_VIDEO_MAX_POLL_ATTEMPTS: u32 = 120;
20/// Default polling interval for video generation status checks
21pub const DEFAULT_VIDEO_POLL_INTERVAL_SECS: u64 = 5;
22
23#[derive(Debug, Deserialize)]
24struct OpenAIChatResponse {
25    id: Option<String>,
26    choices: Vec<OpenAIChoice>,
27    usage: Option<OpenAIUsage>,
28}
29
30#[derive(Debug, Deserialize)]
31struct OpenAIChoice {
32    message: OpenAIMessage,
33}
34
35#[derive(Debug, Deserialize)]
36struct OpenAIMessage {
37    content: Option<Value>,
38    reasoning_content: Option<String>,
39    reasoning: Option<String>,
40    reasoning_details: Option<Value>,
41    /// Raw OpenAI-format tool_calls array, surfaced unchanged on `ChatResponse`.
42    /// Optional because non-tool-using replies omit it entirely.
43    #[serde(default)]
44    tool_calls: Option<Value>,
45}
46
47#[derive(Debug, Deserialize)]
48struct OpenAIUsage {
49    prompt_tokens: Option<u64>,
50    completion_tokens: Option<u64>,
51    total_tokens: Option<u64>,
52    cost: Option<Value>,
53    completion_tokens_details: Option<CompletionTokensDetails>,
54}
55
56#[derive(Debug, Deserialize)]
57struct CompletionTokensDetails {
58    reasoning_tokens: Option<u64>,
59}
60
61#[derive(Debug, Deserialize)]
62struct OpenAIEmbeddingResponse {
63    data: Vec<OpenAIEmbeddingItem>,
64    usage: Option<OpenAIUsage>,
65}
66
67#[derive(Debug, Deserialize)]
68struct OpenAIEmbeddingItem {
69    embedding: Vec<f32>,
70}
71
72/// Build the chat request body from a ChatRequest.
73///
74/// This is shared between streaming and non-streaming chat calls.
75fn build_chat_body(req: &ChatRequest, stream: bool) -> Result<Value> {
76    let mut body = serde_json::json!({
77        "model": req.model,
78        "messages": req.messages,
79    });
80
81    if stream {
82        body["stream"] = serde_json::json!(true);
83    }
84
85    if let Some(temp) = req.temperature {
86        body["temperature"] = serde_json::json!(temp);
87    }
88    if let Some(max_tokens) = req.max_tokens {
89        body["max_tokens"] = serde_json::json!(max_tokens);
90    }
91    if let Some(ref fmt) = req.response_format {
92        body["response_format"] = fmt.clone();
93    }
94    if let Some(max_completion_tokens) = req.max_completion_tokens {
95        body["max_completion_tokens"] = serde_json::json!(max_completion_tokens);
96    }
97    if let Some(ref tools) = req.tools {
98        body["tools"] = tools.clone();
99    }
100    if let Some(ref tool_choice) = req.tool_choice {
101        body["tool_choice"] = tool_choice.clone();
102    }
103    if let Some(parallel) = req.parallel_tool_calls {
104        body["parallel_tool_calls"] = serde_json::json!(parallel);
105    }
106    if let Some(ref stop) = req.stop {
107        body["stop"] = stop.clone();
108    }
109    if let Some(top_p) = req.top_p {
110        body["top_p"] = serde_json::json!(top_p);
111    }
112    if let Some(presence) = req.presence_penalty {
113        body["presence_penalty"] = serde_json::json!(presence);
114    }
115    if let Some(frequency) = req.frequency_penalty {
116        body["frequency_penalty"] = serde_json::json!(frequency);
117    }
118    if let Some(seed) = req.seed {
119        body["seed"] = serde_json::json!(seed);
120    }
121    if let Some(ref user) = req.user {
122        body["user"] = serde_json::json!(user);
123    }
124    if let Some(ref metadata) = req.metadata {
125        body["metadata"] = metadata.clone();
126    }
127    if let Some(ref reasoning_effort) = req.reasoning_effort {
128        body["reasoning_effort"] = reasoning_effort.clone();
129    }
130    if let Some(ref thinking) = req.thinking {
131        body["thinking"] = thinking.clone();
132    }
133
134    merge_extra_body(
135        &mut body,
136        req.extra_body.as_ref(),
137        &[
138            "model",
139            "messages",
140            "stream",
141            "temperature",
142            "max_tokens",
143            "response_format",
144            "max_completion_tokens",
145            "tools",
146            "tool_choice",
147            "parallel_tool_calls",
148            "stop",
149            "top_p",
150            "presence_penalty",
151            "frequency_penalty",
152            "seed",
153            "user",
154            "metadata",
155            "reasoning_effort",
156            "thinking",
157        ],
158    )?;
159
160    Ok(body)
161}
162
163pub async fn chat(client: &Client, cfg: &ProviderConfig, req: ChatRequest) -> Result<ChatResponse> {
164    let base = cfg
165        .base_url
166        .clone()
167        .ok_or_else(|| LiteLLMError::Config("base_url required".into()))?;
168    let url = format!("{}/chat/completions", base.trim_end_matches('/'));
169    let key = resolve_api_key(cfg)?;
170
171    let body = build_chat_body(&req, false)?;
172
173    let mut builder = client.post(url).json(&body);
174    if let Some(key) = key {
175        builder = builder.bearer_auth(key);
176    }
177    for (k, v) in &cfg.extra_headers {
178        builder = builder.header(k, v);
179    }
180
181    let (parsed, headers) = send_json::<OpenAIChatResponse>(builder).await?;
182    let first_message = parsed.choices.first().map(|c| &c.message);
183    let content = first_message
184        .and_then(|message| extract_text_value(message.content.as_ref()))
185        .unwrap_or_default();
186    let reasoning = first_message.and_then(extract_reasoning);
187    let tool_calls = first_message
188        .and_then(|m| m.tool_calls.clone())
189        .filter(|v| !matches!(v, Value::Null) && !matches!(v, Value::Array(a) if a.is_empty()));
190    let header_cost = headers
191        .get("x-litellm-response-cost")
192        .and_then(|v| v.to_str().ok())
193        .and_then(|v| v.parse::<f64>().ok());
194    let mut usage = map_usage(parsed.usage);
195    if usage.cost_usd.is_none() {
196        usage.cost_usd = header_cost;
197    }
198
199    Ok(ChatResponse {
200        content,
201        reasoning,
202        tool_calls,
203        usage,
204        response_id: parsed.id,
205        header_cost,
206        raw: None,
207    })
208}
209
210fn extract_reasoning(message: &OpenAIMessage) -> Option<Reasoning> {
211    message
212        .reasoning_details
213        .as_ref()
214        .and_then(|details| details.as_array().cloned())
215        .and_then(Reasoning::from_details)
216        .or_else(|| {
217            message
218                .reasoning_content
219                .as_ref()
220                .or(message.reasoning.as_ref())
221                .cloned()
222                .and_then(Reasoning::from_text)
223        })
224}
225
226fn extract_text_value(value: Option<&Value>) -> Option<String> {
227    let mut out = String::new();
228    collect_text_fragments(value?, &mut out);
229    (!out.trim().is_empty()).then_some(out)
230}
231
232fn collect_text_fragments(value: &Value, out: &mut String) {
233    match value {
234        Value::String(text) => out.push_str(text),
235        Value::Array(items) => {
236            for item in items {
237                collect_text_fragments(item, out);
238            }
239        }
240        Value::Object(map) => {
241            if let Some(text) = map.get("text") {
242                collect_text_fragments(text, out);
243            } else if let Some(content) = map.get("content") {
244                collect_text_fragments(content, out);
245            } else if let Some(value) = map.get("value") {
246                collect_text_fragments(value, out);
247            }
248        }
249        _ => {}
250    }
251}
252
253pub async fn chat_stream(
254    client: &Client,
255    cfg: &ProviderConfig,
256    req: ChatRequest,
257) -> Result<ChatStream> {
258    let base = cfg
259        .base_url
260        .clone()
261        .ok_or_else(|| LiteLLMError::Config("base_url required".into()))?;
262    let url = format!("{}/chat/completions", base.trim_end_matches('/'));
263    let key = resolve_api_key(cfg)?;
264
265    let body = build_chat_body(&req, true)?;
266
267    let mut builder = client.post(url).json(&body);
268    if let Some(key) = key {
269        builder = builder.bearer_auth(key);
270    }
271    for (k, v) in &cfg.extra_headers {
272        builder = builder.header(k, v);
273    }
274
275    let resp = builder.send().await.map_err(LiteLLMError::from)?;
276    let status = resp.status();
277    if !status.is_success() {
278        let text = resp.text().await.map_err(LiteLLMError::from)?;
279        return Err(LiteLLMError::http(format!(
280            "http {}: {}",
281            status.as_u16(),
282            text
283        )));
284    }
285
286    Ok(parse_sse_stream(resp.bytes_stream()))
287}
288
289pub async fn embeddings(
290    client: &Client,
291    cfg: &ProviderConfig,
292    req: EmbeddingRequest,
293) -> Result<EmbeddingResponse> {
294    let base = cfg
295        .base_url
296        .clone()
297        .ok_or_else(|| LiteLLMError::Config("base_url required".into()))?;
298    let url = format!("{}/embeddings", base.trim_end_matches('/'));
299    let key = resolve_api_key(cfg)?;
300
301    let mut body = serde_json::json!({
302        "model": req.model,
303        "input": req.input,
304    });
305    merge_extra_body(&mut body, req.extra_body.as_ref(), &["model", "input"])?;
306
307    let mut builder = client.post(url).json(&body);
308    if let Some(key) = key {
309        builder = builder.bearer_auth(key);
310    }
311    for (k, v) in &cfg.extra_headers {
312        builder = builder.header(k, v);
313    }
314
315    let (parsed, _headers) = send_json::<OpenAIEmbeddingResponse>(builder).await?;
316    let vectors = parsed.data.into_iter().map(|d| d.embedding).collect();
317
318    Ok(EmbeddingResponse {
319        vectors,
320        usage: map_usage(parsed.usage),
321        raw: None,
322    })
323}
324
325pub async fn image_generation(
326    client: &Client,
327    cfg: &ProviderConfig,
328    req: ImageRequest,
329) -> Result<ImageResponse> {
330    let base = cfg
331        .base_url
332        .clone()
333        .ok_or_else(|| LiteLLMError::Config("base_url required".into()))?;
334    let url = format!("{}/images/generations", base.trim_end_matches('/'));
335    let key = resolve_api_key(cfg)?;
336
337    let mut body = serde_json::json!({
338        "model": req.model,
339        "prompt": req.prompt,
340    });
341    if let Some(n) = req.n {
342        body["n"] = serde_json::json!(n);
343    }
344    if let Some(ref size) = req.size {
345        body["size"] = serde_json::json!(size);
346    }
347    if let Some(ref quality) = req.quality {
348        body["quality"] = serde_json::json!(quality);
349    }
350    if let Some(ref background) = req.background {
351        body["background"] = serde_json::json!(background);
352    }
353    merge_extra_body(&mut body, req.extra_body.as_ref(), &["model", "prompt"])?;
354
355    let mut builder = client.post(url).json(&body);
356    if let Some(key) = key {
357        builder = builder.bearer_auth(key);
358    }
359    for (k, v) in &cfg.extra_headers {
360        builder = builder.header(k, v);
361    }
362
363    let (parsed, _headers) = send_json::<Value>(builder).await?;
364    let images = parsed
365        .get("data")
366        .and_then(|v| v.as_array())
367        .map(|arr| {
368            arr.iter()
369                .map(|item| ImageData {
370                    b64_json: item
371                        .get("b64_json")
372                        .and_then(|v| v.as_str())
373                        .map(|s| s.to_string()),
374                    url: item
375                        .get("url")
376                        .and_then(|v| v.as_str())
377                        .map(|s| s.to_string()),
378                    revised_prompt: item
379                        .get("revised_prompt")
380                        .and_then(|v| v.as_str())
381                        .map(|s| s.to_string()),
382                    mime_type: None,
383                })
384                .collect::<Vec<_>>()
385        })
386        .unwrap_or_default();
387
388    Ok(ImageResponse {
389        images,
390        usage: Usage::default(),
391        raw: None,
392    })
393}
394
395/// Video generation options for configurable timeouts.
396#[derive(Debug, Clone)]
397pub struct VideoGenerationOptions {
398    /// Maximum number of polling attempts
399    pub max_poll_attempts: u32,
400    /// Interval between polling attempts in seconds
401    pub poll_interval_secs: u64,
402}
403
404impl Default for VideoGenerationOptions {
405    fn default() -> Self {
406        Self {
407            max_poll_attempts: DEFAULT_VIDEO_MAX_POLL_ATTEMPTS,
408            poll_interval_secs: DEFAULT_VIDEO_POLL_INTERVAL_SECS,
409        }
410    }
411}
412
413pub async fn video_generation(
414    client: &Client,
415    cfg: &ProviderConfig,
416    req: VideoRequest,
417) -> Result<VideoResponse> {
418    video_generation_with_options(client, cfg, req, VideoGenerationOptions::default()).await
419}
420
421pub async fn video_generation_with_options(
422    client: &Client,
423    cfg: &ProviderConfig,
424    req: VideoRequest,
425    options: VideoGenerationOptions,
426) -> Result<VideoResponse> {
427    let base = cfg
428        .base_url
429        .clone()
430        .ok_or_else(|| LiteLLMError::Config("base_url required".into()))?;
431    let url = format!("{}/videos", base.trim_end_matches('/'));
432    let key = resolve_api_key(cfg)?;
433
434    let mut form = Form::new()
435        .text("model", req.model)
436        .text("prompt", req.prompt);
437    if let Some(seconds) = req.seconds {
438        form = form.text("seconds", seconds.to_string());
439    }
440    if let Some(size) = req.size {
441        form = form.text("size", size);
442    }
443
444    let mut builder = client.post(url).multipart(form);
445    if let Some(ref key) = key {
446        builder = builder.bearer_auth(key.clone());
447    }
448    for (k, v) in &cfg.extra_headers {
449        builder = builder.header(k, v);
450    }
451
452    let (parsed, _headers) = send_json::<Value>(builder).await?;
453    let video_id = parsed
454        .get("id")
455        .and_then(|v| v.as_str())
456        .ok_or_else(|| LiteLLMError::Parse("missing video id".into()))?;
457
458    let status_url = format!("{}/videos/{}", base.trim_end_matches('/'), video_id);
459    let poll_interval = Duration::from_secs(options.poll_interval_secs);
460
461    for attempt in 0..options.max_poll_attempts {
462        let mut status_builder = client.get(&status_url);
463        if let Some(ref key) = key {
464            status_builder = status_builder.bearer_auth(key.clone());
465        }
466        let (status_resp, _headers) = send_json::<Value>(status_builder).await?;
467        let status = status_resp
468            .get("status")
469            .and_then(|v| v.as_str())
470            .unwrap_or("unknown");
471
472        match status {
473            "completed" => {
474                return fetch_video_content(client, &base, video_id, key.as_deref()).await;
475            }
476            "failed" => {
477                let msg = status_resp
478                    .get("error")
479                    .and_then(|v| v.as_str())
480                    .unwrap_or("video generation failed");
481                return Err(LiteLLMError::http(msg.to_string()));
482            }
483            _ => {
484                if attempt + 1 >= options.max_poll_attempts {
485                    return Err(LiteLLMError::http(format!(
486                        "video generation timed out after {} attempts",
487                        options.max_poll_attempts
488                    )));
489                }
490                sleep(poll_interval).await;
491            }
492        }
493    }
494
495    Err(LiteLLMError::http("video generation timed out"))
496}
497
498async fn fetch_video_content(
499    client: &Client,
500    base: &str,
501    video_id: &str,
502    key: Option<&str>,
503) -> Result<VideoResponse> {
504    let content_url = format!("{}/videos/{}/content", base.trim_end_matches('/'), video_id);
505    let mut content_builder = client.get(&content_url);
506    if let Some(key) = key {
507        content_builder = content_builder.bearer_auth(key);
508    }
509
510    let bytes = content_builder
511        .send()
512        .await
513        .map_err(LiteLLMError::from)?
514        .bytes()
515        .await
516        .map_err(LiteLLMError::from)?;
517    let b64 = general_purpose::STANDARD.encode(bytes);
518
519    Ok(VideoResponse {
520        video_url: Some(format!("data:video/mp4;base64,{b64}")),
521        raw: None,
522    })
523}
524
525fn map_usage(usage: Option<OpenAIUsage>) -> Usage {
526    usage.map_or_else(Usage::default, |u| Usage {
527        prompt_tokens: u.prompt_tokens,
528        completion_tokens: u.completion_tokens,
529        thoughts_tokens: u.completion_tokens_details.and_then(|d| d.reasoning_tokens),
530        total_tokens: u.total_tokens,
531        cost_usd: parse_cost(u.cost.as_ref()),
532    })
533}
534
535fn parse_cost(value: Option<&Value>) -> Option<f64> {
536    let v = value?;
537    if let Some(n) = v.as_f64() {
538        return Some(n);
539    }
540    if let Some(s) = v.as_str() {
541        return s.parse::<f64>().ok();
542    }
543    None
544}
litellm_rust/providers/openai_compat.rs

litellm_rust/providers/
openai_compat.rs
