scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-26 06:03:19 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 35

feat(gateway): full openai chat request translation for anthropic and bedrock (#47088)

Browse source 
* feat(gateway): share router cooldowns and follow proxy key invalidations through Redis

Gateway replicas now share state through the coordination Redis the
Python proxy uses (general_settings.coordination_redis, a redis
response cache, or REDIS_*):

- A deployment one replica cools down is written under Python's
  deployment:<id>:cooldown key and applied by every other replica's
  background poll (one MGET a second), so the fleet stops sending to a
  failing deployment together. allowed_fails counts failures across
  replicas, as Python's v1 rule does; the v2 failure-rate rule stays
  per replica, as in Python. Deployments are named by model_info.id, or
  an id derived from the config that is the same on every replica.
- The virtual key cache follows the proxy's
  litellm_proxy.auth_cache_invalidation channel, so a key blocked,
  deleted or edited through the admin API (or its team or user) is
  re-read on every replica at once instead of after up to 60 s.
  general_settings.user_api_key_cache_ttl sets the key TTL.

Nothing is added to the path of a successful request. Without Redis,
or while it is down, each replica keeps its own state as before.
Redis Cluster and Sentinel are refused at startup for now.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0138knkqKpkoSr6PApenZan5

* feat(gateway): hooks and observers around inference requests, with usage and cost

Adds the litellm-gateway-hooks crate: request facts built only when something
reads them, on-path hooks that declare what they read and what they may
change and run in config order, off-path observers on bounded queues, a
usage tap for SSE and JSON responses, and per-deployment prices from the
model cost map overlaid by config prices. The messages and chat handlers run
through it when anything is installed; with nothing installed requests take
the old path.

The key store now also returns the key alias, user, team alias and key and
team metadata, and router attempts carry the deployment id.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0138knkqKpkoSr6PApenZan5

* feat(gateway): write spend to the proxy's database from a native observer

Adds a spend writer observer that records every gateway request in the
proxy's own Postgres tables with the same statements the Python proxy
uses: spend logs, key/user/team/team member/end user spend, and the
daily user, team and end-user activity tables. Writes are batched on the
proxy's timers and are additive, so replicas and Python proxies can
write side by side. The database pool is now shared between virtual
keys and the writer.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0138knkqKpkoSr6PApenZan5

* feat(gateway): enforce key, team, user, member and end-user budgets on shared spend counters

Adds a built-in budgets hook that runs the proxy's budget checks, in its
order and with its messages, on the proxy's Redis spend:* counters. Each
request reads and seeds the counters, checks them and reserves its
maximum cost in one Redis script, then settles the reservation to its
real cost off the request path, so replicas and Python proxies sharing
the Redis enforce one budget. Without Redis the counters are kept per
replica. The key store now reads user, team member and end-user budgets,
and auth no longer checks budgets against stored spend.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0138knkqKpkoSr6PApenZan5

* feat(gateway): enforce key, user, team, member and end-user rate limits on shared counters

Adds a built-in rate limits hook with the proxy's v3 limiter semantics:
requests per window, tokens per window and in-flight requests per key,
on the proxy's own {scope:id}:window|requests|tokens|max_parallel_requests
Redis keys and with its scripts, so replicas and Python proxies sharing
the Redis enforce one limit. Tokens are reserved from the proxy's
estimate and settled to usage when the response ends. Refusals carry the
proxy's message, body and headers on each endpoint, and complete
responses carry its x-ratelimit-* headers. Without Redis the counters are
kept per replica. The key store now reads key, user, team and member
rpm/tpm limits and the key's max_parallel_requests, and end users' budget
limits are read with their budgets; keys with plain rate limits are no
longer refused.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0138knkqKpkoSr6PApenZan5

* feat(gateway): configurable header, spend metadata and provider metadata hooks

Adds four built-in hooks a config lists under litellm_settings.hooks and
runs in that order: forward_headers copies named client headers to the
provider, strip_headers removes headers from the provider request
(including the gateway's default pass-through), spend_metadata adds
header and key-metadata values to the spend log's spend_logs_metadata,
and provider_metadata sets Messages body fields and headers from
templates over request facts. Hooks' upstream header edits now reach the
provider on both endpoints, and body edits can set nested fields.
Credentials and signer headers are never forwarded or set, and a
misconfigured hook stops the gateway at startup.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0138knkqKpkoSr6PApenZan5

* feat(gateway): model capacity hook, the proxy's dynamic rate limiter

Adds the model_capacity hook, turned on by listing dynamic_rate_limiter_v3
(or model_capacity) in litellm_settings.callbacks. It caps each model
group at the sum of its deployments' rpm and tpm across every replica,
and with priority_reservation holds each priority's pool to its share
once the group's saturation reaches the threshold, counting it below.
It uses the proxy's model_saturation_check and priority_model keys and
scripts, its refusal messages and headers, and adds a successful
response's tokens to the group and the pool admission chose. The rate
limit counters gain a general check-and-add over request and token
pairs, and the throttling refusal shape is shared.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0138knkqKpkoSr6PApenZan5

* test(gateway-auth): expect the narrowed unsupported rate limit message

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0138knkqKpkoSr6PApenZan5

* feat(gateway): prometheus observer with the proxy's metric and label names

Adds the prometheus observer, turned on by listing prometheus in
litellm_settings.callbacks, success_callback or failure_callback. Each
finished request updates the proxy's request, spend, token, latency and
deployment metrics under its names and labels, and /metrics serves them
after the gateway's own. prometheus_metrics_config, the exclude lists
and prometheus_latency_buckets are read as the proxy reads them. The
metrics the gateway already served under the same names come from the
observer alone when it is on. Outcomes now carry when the attempt was
sent upstream, for the LLM API latency and time to first token.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0138knkqKpkoSr6PApenZan5

* feat(gateway): s3_v2 observer writing the proxy's logging payload to S3

Adds the s3_v2 observer, turned on by listing s3_v2 (or s3_log) in
litellm_settings.callbacks, success_callback or failure_callback. Each
finished request becomes the proxy's StandardLoggingPayload, with the
messages from the request body and the response as the client got it,
streams put back together. It is queued and written every flush interval
under the proxy's key layout, with its Content-MD5, Content-Disposition
and encryption headers, one PUT per request or one NDJSON object per
folder with s3_batch_file_upload. s3_callback_params is read as the
proxy reads it, os.environ references included. Uploads are retried,
dropped and kept for the next flush by the proxy's rules, and what is
queued is written at shutdown. auth-aws gains an S3 signer that signs
the path as sent and every header, as botocore does.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0138knkqKpkoSr6PApenZan5

* feat(gateway): Azure OpenAI chat completions through the OpenAI passthrough

Adds azure/<deployment> chat completions, streaming and not, on the same
passthrough the openai/ route uses. The request goes to the deployment
path with the api-version query, or to /openai/v1/ for the v1, latest and
preview versions, with api_base, api_key and api_version falling back to
the AZURE_* variables and os.environ references resolved as the proxy
does. A key is sent as api-key, and an azure_ad_token or an Entra ID
token from auth-azure goes as a Bearer token. The body follows the
proxy's gpt, o-series and gpt-5 param rules, picked by base_model or the
deployment name and reading model capabilities from the cost map, so an
unsupported param is refused unless drop_params allows it to be dropped.
Spend is priced as the proxy prices Azure, by azure/<base_model> when one
is set and otherwise by the model the response names, and a content
filter error keeps Azure's inner error for the client

* feat(gateway): Anthropic and Bedrock chat completions streaming

Adds "stream": true to /v1/chat/completions for anthropic/ and bedrock/
deployments, which were refused with 501 before. The request is the
non-streaming one with "stream": true added for Anthropic and sent to
/converse-stream for Bedrock, and the provider's events are decoded the
way the proxy's chunk parsers decode them: text, tool call deltas with
empty arguments sent as "{}", reasoning_content and thinking_blocks with
their index and signature, citations and the mapped finish_reason. A
wrapper in the shape of the proxy's stream wrapper gives every chunk one
id and the client's model name, sends role on the first chunk, adds the
empty finish chunk, strips usage from the chunks and, when
stream_options.include_usage is set, ends with the usage chunk before
[DONE]. The gateway waits for the first chunk before answering, so an
upstream error up to there is retried or falls back to another
deployment, and an error after it ends the stream with the OpenAI error
frame. Usage, cache reads and writes included, is handed to the meter
through the response whether or not the client asked for it, and
dropping the client stream drops the upstream request

* feat(gateway): full OpenAI chat request translation for Anthropic and Bedrock

Anthropic and Bedrock Converse deployments now take the whole OpenAI chat request the
Python proxy accepts instead of a narrow allowlist. Messages translate as Python builds
them: system as a string, parts or several messages, text, image_url (URL or data URL),
file and document parts, assistant tool_calls, the tool and function roles, consecutive
role merging, empty content, cache_control on Anthropic and cachePoint on Bedrock.
Params map per provider: max_tokens and max_completion_tokens with Anthropic's model
default, temperature, top_p, stop, tools and tool_choice, parallel_tool_calls,
response_format as a json tool or native output format, reasoning_effort and thinking,
user as metadata, with n and any other unsupported param refused with a 400 unless
drop_params is set globally, on the deployment or on the request, and
allowed_openai_params and additional_drop_params honoured. Responses carry tool calls,
thinking blocks and response_format answers back in OpenAI shape, streamed and not. The
route downloads media URLs a provider cannot take by reference, merges the beta headers
a request needs with forwarded ones, and the gateway hands the cost map to both
providers for the capability rules that read it.

* fix(rust-bridge): read Python's model_cost for chat translation capabilities

Chat translation for Anthropic and Bedrock reads capability flags from the
cost map. The Python bridge passed none, so every SDK chat call through the
Rust path failed. It now reads litellm.model_cost live, so lookups match
Python's, runtime-registered models included. ModelInfo::entry returns a Cow
so a map read from Python can hand over a copy.

Two tests sent temperature != 1 to a model that removed sampling params,
which Python itself refuses; they now send temperature=1.

* fix(model-catalog): read token counts the cost map writes as whole floats

The published cost map now has prompt_cache_min_tokens: 4096.0, which the
typed u64 fields rejected, so the gateway dropped the whole map at startup
and priced every request at zero. Counts accept any whole non-negative
number, as Python reads any JSON number, and a test keeps the checked-in
map parsing.

* feat(gateway): Azure OpenAI chat completions through the OpenAI passthrough (#47074)

Adds azure/<deployment> chat completions, streaming and not, on the same
passthrough the openai/ route uses. The request goes to the deployment
path with the api-version query, or to /openai/v1/ for the v1, latest and
preview versions, with api_base, api_key and api_version falling back to
the AZURE_* variables and os.environ references resolved as the proxy
does. A key is sent as api-key, and an azure_ad_token or an Entra ID
token from auth-azure goes as a Bearer token. The body follows the
proxy's gpt, o-series and gpt-5 param rules, picked by base_model or the
deployment name and reading model capabilities from the cost map, so an
unsupported param is refused unless drop_params allows it to be dropped.
Spend is priced as the proxy prices Azure, by azure/<base_model> when one
is set and otherwise by the model the response names, and a content
filter error keeps Azure's inner error for the client

* feat(gateway): Anthropic and Bedrock chat completions streaming (#47077)

Adds "stream": true to /v1/chat/completions for anthropic/ and bedrock/
deployments, which were refused with 501 before. The request is the
non-streaming one with "stream": true added for Anthropic and sent to
/converse-stream for Bedrock, and the provider's events are decoded the
way the proxy's chunk parsers decode them: text, tool call deltas with
empty arguments sent as "{}", reasoning_content and thinking_blocks with
their index and signature, citations and the mapped finish_reason. A
wrapper in the shape of the proxy's stream wrapper gives every chunk one
id and the client's model name, sends role on the first chunk, adds the
empty finish chunk, strips usage from the chunks and, when
stream_options.include_usage is set, ends with the usage chunk before
[DONE]. The gateway waits for the first chunk before answering, so an
upstream error up to there is retried or falls back to another
deployment, and an error after it ends the stream with the OpenAI error
frame. Usage, cache reads and writes included, is handed to the meter
through the response whether or not the client asked for it, and
dropping the client stream drops the upstream request

---------

Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Nate Armstrong 2026-10-19 10:44:12 -07:00 • committed by GitHub
parent f7323dc681
commit 3d976a5791
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
42 changed files with 8944 additions and 933 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
litellm-rust/Cargo.lock generated
View file

@ -4055,6 +4055,7 @@ dependencies = [
name = "litellm-inference-chat"
version = "0.1.0"
dependencies = [
"base64 0.22.1",
"bytes",
"futures-util",
"litellm-auth",
@ -4235,11 +4236,13 @@ dependencies = [
"serde_json",
"serde_path_to_error",
"serde_with",
"sha2 0.10.9",
"strum",
"thiserror 2.0.19",
"time",
"tokio",
"url",
"uuid",
]
[[package]]

20
litellm-rust/crates/gateway-inference/src/chat_completions.rs
View file

@ -146,6 +146,8 @@ async fn handle(
custom_llm_provider: deployment.custom_llm_provider.clone(),
extra_headers: extra_headers.clone(),
timeout: deployment.timeout,
drop_params: Some(drop_params(gateway, deployment)),
model_info: gateway.model_info(),
};
if stream {
return translated_stream(route, call, name).await;
@ -197,6 +199,17 @@ fn provider_params(deployment: &Deployment) -> Map<String, Value> {
provider_params
}
/// `litellm.drop_params` or the deployment's own: either one set drops what the provider
/// cannot take, as Python's `get_optional_params` reads them.
fn drop_params(gateway: &Gateway, deployment: &Deployment) -> bool {
gateway.drop_params()
|| deployment
.litellm_params
.drop_params
.as_ref()
.is_some_and(|flag| flag.is_set())
}
async fn openai(
gateway: &Gateway,
deployment: &Deployment,
@ -213,12 +226,7 @@ async fn openai(
extra_headers,
timeout: deployment.timeout,
litellm_params: provider_params(deployment),
drop_params: gateway.drop_params()
|| deployment
.litellm_params
.drop_params
.as_ref()
.is_some_and(|flag| flag.is_set()),
drop_params: drop_params(gateway, deployment),
model_info: gateway.model_info(),
},
None,

12
litellm-rust/crates/gateway-inference/src/lib.rs
View file

@ -18,7 +18,10 @@ mod responses;
use std::sync::{Arc, OnceLock};
use axum::{Router, routing::post};
use litellm_http::{ClientVariant, HttpClientConfig, media::UrlPolicy};
use litellm_http::{
ClientVariant, HttpClientConfig,
media::{MediaFetcher, UrlPolicy},
};
use litellm_inference::resources::CoreResources;
use litellm_inference_chat::ChatCompletionsRoute;
use litellm_inference_messages::MessagesRoute;
@ -138,7 +141,12 @@ impl Gateway {
provider.clone(),
auth.clone(),
secrets.clone(),
),
)
.with_media(MediaFetcher::new(
&resources.pool,
&http,
UrlPolicy::default(),
)?),
messages: MessagesRoute::new(provider.clone(), auth.clone(), secrets.clone()),
responses: ResponsesRoute::new(provider, auth.clone(), secrets.clone()),
ocr: OcrRoute::new(OcrClient::new(

2
litellm-rust/crates/gateway-inference/tests/routes.rs
View file

@ -105,6 +105,8 @@ async fn chat_errors_come_from_core(
custom_llm_provider: None,
extra_headers: None,
timeout: None,
drop_params: Some(false),
model_info: None,
},
&(),
None,

30
litellm-rust/crates/gateway/src/hooks.rs
View file

@ -1,7 +1,7 @@
//! Builds the hooks and observers inference requests run through, and the per-deployment prices
//! observers that read cost need.
use std::{collections::HashSet, sync::Arc};
use std::{borrow::Cow, collections::HashSet, sync::Arc};
use litellm_config::{Config, Model};
use litellm_gateway_hooks::{
@ -30,19 +30,19 @@ pub async fn install(
hooks: Vec<Arc<dyn Hook>>,
observers: Vec<Arc<dyn Observer>>,
) -> Result<(), Error> {
let serves_azure = config.model_list.iter().any(is_azure);
if hooks.is_empty() && observers.is_empty() && !serves_azure {
let reads_model_info = config.model_list.iter().any(reads_model_info);
if hooks.is_empty() && observers.is_empty() && !reads_model_info {
return Ok(());
}
let needs_cost = observers
.iter()
.any(|observer| observer.reads().contains(Facts::COST))
|| hooks.iter().any(|hook| hook.reads().contains(Facts::COST));
let catalog = match needs_cost || serves_azure {
let catalog = match needs_cost || reads_model_info {
true => load_catalog(inference).await.map(Arc::new),
false => None,
};
if let Some(catalog) = catalog.as_ref().filter(|_| serves_azure) {
if let Some(catalog) = catalog.as_ref().filter(|_| reads_model_info) {
inference.install_model_info(Arc::new(CostMap::new(catalog.clone())));
}
if hooks.is_empty() && observers.is_empty() {
@ -204,7 +204,19 @@ fn is_azure(model: &Model) -> bool {
}
}
/// The model cost map as the capabilities Azure's param rules read.
/// Whether the deployment's chat param rules read model capabilities from the cost map:
/// Azure, and the Anthropic and Bedrock chat translations.
fn reads_model_info(model: &Model) -> bool {
is_azure(model)
|| litellm_inference::provider::resolve_llm_provider(
&model.litellm_params.model,
model.litellm_params.custom_llm_provider.as_deref(),
"chat completions",
)
.is_ok_and(|resolved| matches!(<&str>::from(resolved.provider), "anthropic" | "bedrock"))
}
/// The model cost map as the capabilities the chat param rules read.
struct CostMap {
catalog: Arc<Catalog>,
keys: HashSet<String>,
@ -222,8 +234,10 @@ impl CostMap {
}
impl ModelInfo for CostMap {
fn entry(&self, model: &str) -> Option<&Map<String, Value>> {
self.catalog.lookup(model).map(|found| found.entry.fields())
fn entry(&self, model: &str) -> Option<Cow<'_, Map<String, Value>>> {
self.catalog
.lookup(model)
.map(|found| Cow::Borrowed(found.entry.fields()))
}
fn declares(&self, key: &str) -> bool {

722
litellm-rust/crates/gateway/tests/chat_translation.rs Normal file
View file

@ -0,0 +1,722 @@
//! Chat completions for deployments the gateway translates (Anthropic, Bedrock Converse): the
//! OpenAI request widened to tools, images and the provider's own limits, the provider body
//! the upstream sees, and the OpenAI answer the client gets back, streamed and not.
use std::{
io::Write,
sync::{Arc, Mutex, Once},
};
use axum::{
Router,
body::{Body, Bytes, to_bytes},
http::{Request, header::CONTENT_TYPE},
response::IntoResponse,
};
use litellm_config::Config;
use litellm_framer::aws_event_stream::{AwsEventStreamCodec, Header, HeaderValue, Message};
use rstest::rstest;
use serde_json::{Value, json};
use tokio::net::TcpListener;
use tokio_util::codec::Encoder;
use tower::ServiceExt;
const ANTHROPIC_MODEL: &str = "claude-x";
const BEDROCK_MODEL: &str = "anthropic.claude-x-v1:0";
/// A one-pixel PNG.
const PNG: &str = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNkYPhfDwAChwGA60e6kgAAAABJRU5ErkJggg==";
/// The gateway reads its cost map from this variable; every test sets it before building a
/// gateway, so no test reads the environment while it changes.
fn use_local_cost_map() {
static ONCE: Once = Once::new();
ONCE.call_once(|| {
let capabilities = json!({
"mode": "chat",
"max_tokens": 8192,
"max_output_tokens": 8192,
"supports_tool_choice": true,
"supports_function_calling": true,
"supports_vision": true,
"supports_pdf_input": true,
"supports_prompt_caching": true,
"supports_response_schema": true,
});
let entry = |provider: &str| {
let mut entry = capabilities.clone();
entry["litellm_provider"] = json!(provider);
entry
};
let map = json!({
ANTHROPIC_MODEL: entry("anthropic"),
BEDROCK_MODEL: entry("bedrock_converse"),
});
let mut file = tempfile::NamedTempFile::new().unwrap();
file.write_all(map.to_string().as_bytes()).unwrap();
let (_, path) = file.keep().unwrap();
// SAFETY: set once, before any test builds a gateway or spawns a thread that reads the
// environment.
unsafe { std::env::set_var("LITELLM_MODEL_COST_MAP_PATH", path) };
});
}
fn anthropic_events(events: &[Value]) -> String {
events
.iter()
.map(|event| {
format!(
"event: {}\ndata: {event}\n\n",
event["type"].as_str().unwrap()
)
})
.collect()
}
fn anthropic_tool_message() -> Value {
json!({
"id": "msg_1", "type": "message", "role": "assistant", "model": ANTHROPIC_MODEL,
"content": [
{"type": "text", "text": "Checking."},
{"type": "tool_use", "id": "toolu_1", "name": "get_weather", "input": {"city": "Paris"}}
],
"stop_reason": "tool_use", "stop_sequence": null,
"usage": {"input_tokens": 10, "output_tokens": 5}
})
}
/// The answer to a `response_format` request, given as the tool it became.
fn anthropic_json_message() -> Value {
json!({
"id": "msg_3", "type": "message", "role": "assistant", "model": ANTHROPIC_MODEL,
"content": [{"type": "tool_use", "id": "toolu_2", "name": "json_tool_call", "input": {"city": "Paris"}}],
"stop_reason": "tool_use", "stop_sequence": null,
"usage": {"input_tokens": 10, "output_tokens": 5}
})
}
fn anthropic_text_message() -> Value {
json!({
"id": "msg_2", "type": "message", "role": "assistant", "model": ANTHROPIC_MODEL,
"content": [{"type": "text", "text": "Sunny."}],
"stop_reason": "end_turn", "stop_sequence": null,
"usage": {"input_tokens": 10, "output_tokens": 2}
})
}
fn anthropic_tool_stream() -> String {
anthropic_events(&[
json!({"type": "message_start", "message": {"id": "msg_1", "type": "message", "role": "assistant", "model": ANTHROPIC_MODEL, "content": [], "stop_reason": null, "stop_sequence": null, "usage": {"input_tokens": 10, "output_tokens": 1}}}),
json!({"type": "content_block_start", "index": 0, "content_block": {"type": "tool_use", "id": "toolu_1", "name": "get_weather", "input": {}}}),
json!({"type": "content_block_delta", "index": 0, "delta": {"type": "input_json_delta", "partial_json": "{\"city\": "}}),
json!({"type": "content_block_delta", "index": 0, "delta": {"type": "input_json_delta", "partial_json": "\"Paris\"}"}}),
json!({"type": "content_block_stop", "index": 0}),
json!({"type": "message_delta", "delta": {"stop_reason": "tool_use", "stop_sequence": null}, "usage": {"output_tokens": 5}}),
json!({"type": "message_stop"}),
])
}
fn converse_tool_response() -> Value {
json!({
"output": {"message": {"role": "assistant", "content": [
{"text": "Checking."},
{"toolUse": {"toolUseId": "tooluse_1", "name": "get_weather", "input": {"city": "Paris"}}}
]}},
"stopReason": "tool_use",
"usage": {"inputTokens": 10, "outputTokens": 5, "totalTokens": 15}
})
}
fn converse_text_response() -> Value {
json!({
"output": {"message": {"role": "assistant", "content": [{"text": "Sunny."}]}},
"stopReason": "end_turn",
"usage": {"inputTokens": 10, "outputTokens": 2, "totalTokens": 12}
})
}
fn bedrock_frame(name: &str, payload: Value, wire: &mut bytes::BytesMut) {
let message = Message::new(Bytes::from(payload.to_string()))
.add_header(Header::new(
":message-type",
HeaderValue::String("event".to_owned().into()),
))
.add_header(Header::new(
":event-type",
HeaderValue::String(name.to_owned().into()),
));
AwsEventStreamCodec.encode(message, wire).unwrap();
}
fn converse_tool_stream() -> Bytes {
let mut wire = bytes::BytesMut::new();
for (name, payload) in [
("messageStart", json!({"role": "assistant"})),
(
"contentBlockStart",
json!({"contentBlockIndex": 0, "start": {"toolUse": {"toolUseId": "tooluse_1", "name": "get_weather"}}}),
),
(
"contentBlockDelta",
json!({"contentBlockIndex": 0, "delta": {"toolUse": {"input": "{\"city\": "}}}),
),
(
"contentBlockDelta",
json!({"contentBlockIndex": 0, "delta": {"toolUse": {"input": "\"Paris\"}"}}}),
),
("contentBlockStop", json!({"contentBlockIndex": 0})),
("messageStop", json!({"stopReason": "tool_use"})),
(
"metadata",
json!({"usage": {"inputTokens": 10, "outputTokens": 5, "totalTokens": 15}, "metrics": {"latencyMs": 9}}),
),
] {
bedrock_frame(name, payload, &mut wire);
}
wire.freeze()
}
#[derive(Clone, Debug)]
struct Sent {
path: String,
headers: Vec<(String, String)>,
body: Value,
}
impl Sent {
fn header(&self, name: &str) -> Option<&str> {
self.headers
.iter()
.find(|(key, _)| key == name)
.map(|(_, value)| value.as_str())
}
}
type Calls = Arc<Mutex<Vec<Sent>>>;
/// Answers each provider by path: a tool call while the conversation has no tool result yet,
/// the final answer once it does.
async fn upstream(calls: Calls) -> String {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let address = listener.local_addr().unwrap();
let app = Router::new().fallback(move |request: Request<Body>| {
let calls = calls.clone();
async move {
let (parts, body) = request.into_parts();
let body: Value =
serde_json::from_slice(&to_bytes(body, 1 << 22).await.unwrap()).unwrap();
let path = parts.uri.path().to_owned();
calls.lock().unwrap().push(Sent {
path: path.clone(),
headers: parts
.headers
.iter()
.map(|(name, value)| {
(
name.to_string(),
value.to_str().unwrap_or_default().to_owned(),
)
})
.collect(),
body: body.clone(),
});
let answered =
body.to_string().contains("tool_result") || body.to_string().contains("toolResult");
let json = [(CONTENT_TYPE, "application/json")];
if path.ends_with("/converse-stream") {
return (
[(CONTENT_TYPE, "application/vnd.amazon.eventstream")],
converse_tool_stream(),
)
.into_response();
}
if path.ends_with("/converse") {
let answer = match answered {
true => converse_text_response(),
false => converse_tool_response(),
};
return (json, answer.to_string()).into_response();
}
if body["stream"] == json!(true) {
return (
[(CONTENT_TYPE, "text/event-stream")],
anthropic_tool_stream(),
)
.into_response();
}
let answer = match answered {
_ if body["tools"][0]["name"] == "json_tool_call" => anthropic_json_message(),
true => anthropic_text_message(),
false => anthropic_tool_message(),
};
(json, answer.to_string()).into_response()
}
});
tokio::spawn(async move { axum::serve(listener, app).await });
format!("http://{address}")
}
struct Harness {
app: Router,
calls: Calls,
}
async fn harness() -> Harness {
use_local_cost_map();
let calls = Calls::default();
let upstream = upstream(calls.clone()).await;
let config = Config::from_yaml(&format!(
"model_list:
- model_name: claude
litellm_params:
model: anthropic/{ANTHROPIC_MODEL}
api_base: {upstream}
api_key: sk-anthropic
- model_name: bedrock
litellm_params:
model: bedrock/{BEDROCK_MODEL}
api_base: {upstream}
api_key: bedrock-token
- model_name: claude-dropping
litellm_params:
model: anthropic/{ANTHROPIC_MODEL}
api_base: {upstream}
api_key: sk-anthropic
drop_params: true
router_settings:
num_retries: 0
general_settings:
master_key: sk-master
"
))
.unwrap();
let inference = litellm_gateway::build_inference(&config).unwrap();
litellm_gateway::hooks::install(&inference, &config, Vec::new(), Vec::new())
.await
.unwrap();
let app = litellm_gateway::router(inference, &config, None, None);
Harness { app, calls }
}
impl Harness {
async fn send(&self, body: Value) -> (u16, String) {
let request = Request::post("/v1/chat/completions")
.header("content-type", "application/json")
.header("authorization", "Bearer sk-master")
.body(Body::from(body.to_string()))
.unwrap();
let response = self.app.clone().oneshot(request).await.unwrap();
let status = response.status().as_u16();
let body = to_bytes(response.into_body(), 1 << 22).await.unwrap();
(status, String::from_utf8(body.to_vec()).unwrap())
}
async fn json(&self, body: Value) -> Value {
let (status, text) = self.send(body).await;
assert_eq!(status, 200, "{text}");
serde_json::from_str(&text).unwrap()
}
fn sent(&self) -> Vec<Sent> {
self.calls.lock().unwrap().clone()
}
}
fn weather_tool() -> Value {
json!({"type": "function", "function": {
"name": "get_weather",
"description": "The weather in a city",
"parameters": {"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"]}
}})
}
fn question() -> Value {
json!({"role": "user", "content": "Weather in Paris?"})
}
/// The conversation after the model asked for the tool and the tool answered.
fn answered(call_id: &str) -> Value {
json!([
question(),
{"role": "assistant", "content": null, "tool_calls": [
{"id": call_id, "type": "function", "function": {"name": "get_weather", "arguments": "{\"city\": \"Paris\"}"}}
]},
{"role": "tool", "tool_call_id": call_id, "content": "sunny, 21C"}
])
}
/// The streamed chunks' JSON and whether the stream ended with `[DONE]`.
fn frames(body: &str) -> (Vec<Value>, bool) {
let mut done = false;
let frames = body
.split("\n\n")
.filter_map(|frame| frame.strip_prefix("data: "))
.filter(|data| {
let last = *data == "[DONE]";
done |= last;
!last
})
.map(|data| serde_json::from_str(data).unwrap())
.collect();
(frames, done)
}
/// The tool calls the chunks stream, joined: (id, name, arguments) per index.
fn streamed_tool_calls(chunks: &[Value]) -> Vec<(String, String, String)> {
let mut calls: Vec<(String, String, String)> = Vec::new();
for chunk in chunks {
let Some(deltas) = chunk["choices"][0]["delta"]["tool_calls"].as_array() else {
continue;
};
for delta in deltas {
let index = delta["index"].as_u64().unwrap() as usize;
if calls.len() <= index {
calls.resize(index + 1, Default::default());
}
let call = &mut calls[index];
if let Some(id) = delta["id"].as_str() {
call.0 = id.to_owned();
}
if let Some(name) = delta["function"]["name"].as_str() {
call.1 = name.to_owned();
}
call.2
.push_str(delta["function"]["arguments"].as_str().unwrap_or_default());
}
}
calls
}
fn finish_reason(chunks: &[Value]) -> Option<String> {
chunks
.iter()
.find_map(|chunk| chunk["choices"][0]["finish_reason"].as_str())
.map(str::to_owned)
}
#[tokio::test]
async fn anthropic_tool_round_trip() {
let harness = harness().await;
let asked = harness
.json(json!({
"model": "claude",
"messages": [question()],
"tools": [weather_tool()],
"tool_choice": "required",
"max_tokens": 64,
}))
.await;
let message = &asked["choices"][0]["message"];
assert_eq!(asked["choices"][0]["finish_reason"], "tool_calls");
assert_eq!(message["content"], "Checking.");
assert_eq!(
message["tool_calls"],
json!([{"id": "toolu_1", "type": "function", "index": 1,
"function": {"name": "get_weather", "arguments": "{\"city\": \"Paris\"}"}}])
);
let done = harness
.json(json!({
"model": "claude",
"messages": answered("toolu_1"),
"tools": [weather_tool()],
"max_tokens": 64,
}))
.await;
assert_eq!(done["choices"][0]["message"]["content"], "Sunny.");
assert_eq!(done["choices"][0]["finish_reason"], "stop");
let sent = harness.sent();
assert_eq!(sent[0].path, "/v1/messages");
assert_eq!(
sent[0].body,
json!({
"model": ANTHROPIC_MODEL,
"messages": [{"role": "user", "content": [{"type": "text", "text": "Weather in Paris?"}]}],
"max_tokens": 64,
"tools": [{"name": "get_weather", "type": "custom", "description": "The weather in a city",
"input_schema": {"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"]}}],
"tool_choice": {"type": "any"}
})
);
assert_eq!(
sent[1].body["messages"],
json!([
{"role": "user", "content": [{"type": "text", "text": "Weather in Paris?"}]},
{"role": "assistant", "content": [{"type": "tool_use", "id": "toolu_1", "name": "get_weather", "input": {"city": "Paris"}}]},
{"role": "user", "content": [{"type": "tool_result", "tool_use_id": "toolu_1", "content": "sunny, 21C"}]}
])
);
}
#[tokio::test]
async fn bedrock_tool_round_trip() {
let harness = harness().await;
let asked = harness
.json(json!({
"model": "bedrock",
"messages": [question()],
"tools": [weather_tool()],
"tool_choice": {"type": "function", "function": {"name": "get_weather"}},
"max_tokens": 64,
}))
.await;
assert_eq!(asked["choices"][0]["finish_reason"], "tool_calls");
assert_eq!(
asked["choices"][0]["message"]["tool_calls"],
json!([{"id": "tooluse_1", "type": "function", "index": 1,
"function": {"name": "get_weather", "arguments": "{\"city\": \"Paris\"}"}}])
);
let done = harness
.json(json!({
"model": "bedrock",
"messages": answered("tooluse_1"),
"tools": [weather_tool()],
"max_tokens": 64,
}))
.await;
assert_eq!(done["choices"][0]["message"]["content"], "Sunny.");
assert_eq!(done["choices"][0]["finish_reason"], "stop");
let sent = harness.sent();
assert_eq!(sent[0].path, format!("/model/{BEDROCK_MODEL}/converse"));
assert_eq!(
sent[0].body["toolConfig"],
json!({
"tools": [{"toolSpec": {"name": "get_weather", "description": "The weather in a city",
"inputSchema": {"json": {"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"]}}}}],
"toolChoice": {"tool": {"name": "get_weather"}}
})
);
assert_eq!(
sent[1].body["messages"],
json!([
{"role": "user", "content": [{"text": "Weather in Paris?"}]},
{"role": "assistant", "content": [{"toolUse": {"toolUseId": "tooluse_1", "name": "get_weather", "input": {"city": "Paris"}}}]},
{"role": "user", "content": [{"toolResult": {"toolUseId": "tooluse_1", "content": [{"text": "sunny, 21C"}]}}]}
])
);
}
#[rstest]
#[case::anthropic("claude", "toolu_1")]
#[case::bedrock("bedrock", "tooluse_1")]
#[tokio::test]
async fn a_streamed_tool_call_arrives_as_openai_tool_call_chunks(
#[case] model: &str,
#[case] id: &str,
) {
let harness = harness().await;
let (status, body) = harness
.send(json!({
"model": model,
"messages": [question()],
"tools": [weather_tool()],
"max_tokens": 64,
"stream": true,
}))
.await;
assert_eq!(status, 200, "{body}");
let (chunks, done) = frames(&body);
assert!(done, "{body}");
assert_eq!(
streamed_tool_calls(&chunks),
[(
id.to_owned(),
"get_weather".to_owned(),
"{\"city\": \"Paris\"}".to_owned()
)]
);
assert_eq!(finish_reason(&chunks).as_deref(), Some("tool_calls"));
let sent = &harness.sent()[0];
let tools = match model {
"claude" => sent.body["tools"][0]["name"].clone(),
_ => sent.body["toolConfig"]["tools"][0]["toolSpec"]["name"].clone(),
};
assert_eq!(tools, "get_weather");
}
#[rstest]
#[case::max_tokens("max_tokens")]
#[case::max_completion_tokens("max_completion_tokens")]
#[tokio::test]
async fn bedrock_takes_the_token_limit_in_its_inference_config(#[case] param: &str) {
let harness = harness().await;
harness
.json(json!({
"model": "bedrock",
"messages": [{"role": "system", "content": "Be brief."}, question()],
param: 32,
"temperature": 0.5,
"stop": ["END"],
}))
.await;
let sent = &harness.sent()[0];
assert_eq!(
sent.body["inferenceConfig"],
json!({"maxTokens": 32, "temperature": 0.5, "stopSequences": ["END"]})
);
assert_eq!(sent.body["system"], json!([{"text": "Be brief."}]));
}
#[tokio::test]
async fn anthropic_gets_the_model_default_max_tokens_when_the_caller_names_none() {
let harness = harness().await;
harness
.json(json!({"model": "claude", "messages": [question()]}))
.await;
assert_eq!(harness.sent()[0].body["max_tokens"], 8192);
}
#[tokio::test]
async fn an_image_data_url_reaches_each_provider_inline() {
let harness = harness().await;
let messages = json!([{"role": "user", "content": [
{"type": "text", "text": "What is this?"},
{"type": "image_url", "image_url": {"url": format!("data:image/png;base64,{PNG}")}}
]}]);
harness
.json(json!({"model": "claude", "messages": messages, "max_tokens": 16}))
.await;
harness
.json(json!({"model": "bedrock", "messages": messages, "max_tokens": 16}))
.await;
let sent = harness.sent();
assert_eq!(
sent[0].body["messages"][0]["content"][1],
json!({"type": "image", "source": {"type": "base64", "media_type": "image/png", "data": PNG}})
);
assert_eq!(
sent[1].body["messages"][0]["content"][1],
json!({"image": {"format": "png", "source": {"bytes": PNG}}})
);
}
#[tokio::test]
async fn anthropic_response_format_comes_back_as_the_answer() {
let harness = harness().await;
let schema =
json!({"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"]});
let answer = harness
.json(json!({
"model": "claude",
"messages": [question()],
"max_tokens": 64,
"response_format": {"type": "json_schema", "json_schema": {"name": "place", "schema": schema}},
"user": "end-user-1",
}))
.await;
assert_eq!(
answer["choices"][0]["message"]["content"],
"{\"city\": \"Paris\"}"
);
assert_eq!(answer["choices"][0]["message"]["tool_calls"], Value::Null);
assert_eq!(answer["choices"][0]["finish_reason"], "stop");
let sent = &harness.sent()[0];
assert_eq!(sent.body["metadata"], json!({"user_id": "end-user-1"}));
assert_eq!(sent.body["tools"][0]["name"], "json_tool_call");
assert_eq!(sent.body["tools"][0]["input_schema"], schema);
assert_eq!(
sent.body["tool_choice"],
json!({"name": "json_tool_call", "type": "tool"})
);
}
#[rstest]
#[case::anthropic("claude", json!({"logprobs": true}))]
#[case::bedrock("bedrock", json!({"tool_choice": "none", "tools": [weather_tool()]}))]
#[case::n_choices("claude", json!({"n": 2}))]
#[tokio::test]
async fn an_unsupported_param_is_refused_and_dropped_with_drop_params(
#[case] model: &str,
#[case] extra: Value,
) {
let harness = harness().await;
let mut body = json!({"model": model, "messages": [question()], "max_tokens": 16});
body.as_object_mut()
.unwrap()
.extend(extra.as_object().unwrap().clone());
let (status, refused) = harness.send(body.clone()).await;
assert_eq!(status, 400, "{refused}");
assert!(
harness.sent().is_empty(),
"a refused request never goes out"
);
body["drop_params"] = json!(true);
harness.json(body).await;
let sent = &harness.sent()[0];
for name in extra.as_object().unwrap().keys() {
if name == "tools" {
continue;
}
assert!(
sent.body.get(name).is_none(),
"{name} reached {}",
sent.body
);
}
assert!(
sent.body
.get("toolConfig")
.and_then(|config| config.get("toolChoice"))
.is_none()
);
assert!(sent.body.get("drop_params").is_none());
}
#[tokio::test]
async fn a_deployment_that_drops_params_drops_them_without_the_request_asking() {
let harness = harness().await;
harness
.json(json!({
"model": "claude-dropping",
"messages": [question()],
"max_tokens": 16,
"logprobs": true,
}))
.await;
assert!(harness.sent()[0].body.get("logprobs").is_none());
}
#[tokio::test]
async fn a_cache_control_part_reaches_anthropic_and_bedrock() {
let harness = harness().await;
let messages = json!([
{"role": "system", "content": [{"type": "text", "text": "Long instructions.", "cache_control": {"type": "ephemeral"}}]},
question()
]);
harness
.json(json!({"model": "claude", "messages": messages, "max_tokens": 16}))
.await;
harness
.json(json!({"model": "bedrock", "messages": messages, "max_tokens": 16}))
.await;
let sent = harness.sent();
assert_eq!(
sent[0].body["system"],
json!([{"type": "text", "text": "Long instructions.", "cache_control": {"type": "ephemeral"}}])
);
assert_eq!(
sent[1].body["system"],
json!([{"text": "Long instructions."}, {"cachePoint": {"type": "default"}}])
);
assert!(sent[0].header("x-api-key").is_some());
}

1
litellm-rust/crates/inference-chat/Cargo.toml
View file

@ -6,6 +6,7 @@ license.workspace = true
repository.workspace = true
[dependencies]
base64.workspace = true
bytes.workspace = true
futures-util.workspace = true
litellm-auth = { workspace = true, features = ["aws", "azure", "gcp"] }

6
litellm-rust/crates/inference-chat/src/handler.rs
View file

@ -29,7 +29,7 @@ pub(super) async fn execute(
config,
url,
body,
stream_shape: _,
stream_shape,
optional_params,
environment,
secrets,
@ -122,7 +122,7 @@ pub(super) async fn execute(
))
})?;
config
.transform_response(&model, ProviderChatResponseData { body })
.transform_response(&model, ProviderChatResponseData { body }, &stream_shape)
.map_err(Error::from)
.map_err(as_response_error)
},
@ -229,6 +229,8 @@ mod tests {
custom_llm_provider: None,
extra_headers: None,
timeout: None,
drop_params: Some(false),
model_info: None,
})
.unwrap(),
std::sync::Arc::new(|_: &str| None),

15
litellm-rust/crates/inference-chat/src/lib.rs
View file

@ -8,6 +8,7 @@ pub use litellm_inference::RouteError as Error;
mod common_utils;
pub mod constants;
pub(crate) mod handler;
mod media;
pub mod openai;
mod prepare;
pub mod stream;
@ -25,6 +26,7 @@ pub struct ChatCompletionsRoute {
auth: Arc<AuthServices>,
secrets: Arc<dyn SecretSource>,
cache: Option<litellm_cache_response::ScopedCache>,
media: Option<litellm_http::media::MediaFetcher>,
}
impl ChatCompletionsRoute {
@ -38,6 +40,16 @@ impl ChatCompletionsRoute {
auth,
secrets,
cache: None,
media: None,
}
}
/// Downloads the media URLs a provider cannot take by reference. Without one such a
/// request is refused as unsupported.
pub fn with_media(self, media: litellm_http::media::MediaFetcher) -> Self {
Self {
media: Some(media),
..self
}
}
@ -77,7 +89,8 @@ impl ChatCompletionsRoute {
interceptors: &impl litellm_host::interceptors::Interceptors<Error>,
observers: Option<&ObservationSender>,
) -> Result<ChatCompletionsResponse, Error> {
let resolved = resolve_request(request)?;
let mut resolved = resolve_request(request)?;
self.inline_remote_media(&mut resolved).await?;
let snapshot = self
.secrets
.resolve(&resolved.config.secret_names())

158
litellm-rust/crates/inference-chat/src/media.rs Normal file
View file

@ -0,0 +1,158 @@
//! The media a translated request needs downloaded first: an image or file URL the provider
//! cannot take by reference, which Python fetches while it builds the request.
use std::{collections::HashMap, time::Duration};
use base64::Engine;
use futures_util::future::try_join_all;
use litellm_http::media::DownloadPolicy;
use litellm_llms::base_llm::chat::media::{
MAX_MEDIA_DOWNLOAD_BYTES, downloaded_media_type, inline_media,
};
use crate::{ChatCompletionsRoute, Error, types::ResolvedChatCompletionsRequest};
const MEDIA_DOWNLOAD: DownloadPolicy = DownloadPolicy {
timeout: Duration::from_secs(60),
max_bytes: MAX_MEDIA_DOWNLOAD_BYTES,
max_redirects: 10,
};
impl ChatCompletionsRoute {
/// Downloads every URL the config names and inlines each as a data URL. A route with no
/// fetcher leaves such a request to the host; a URL that cannot be fetched is the
/// caller's error, as Python's image fetch failure is.
pub(crate) async fn inline_remote_media(
&self,
request: &mut ResolvedChatCompletionsRequest<'_>,
) -> Result<(), Error> {
let urls = request.config.remote_media(&request.messages);
if urls.is_empty() {
return Ok(());
}
let fetcher = self
.media
.as_ref()
.ok_or(Error::Unsupported("message media URL"))?;
let fetch = |url: String| async move {
let unable = |reason: String| {
Error::InvalidRequest(
format!("Unable to fetch media from url {url}: {reason}").into(),
)
};
let parsed = reqwest::Url::parse(&url).map_err(|err| unable(err.to_string()))?;
let media = fetcher
.fetch(parsed, MEDIA_DOWNLOAD)
.await
.map_err(|err| unable(err.to_string()))?;
let data = format!(
"data:{};base64,{}",
downloaded_media_type(&url, &media.content_type),
base64::engine::general_purpose::STANDARD.encode(media.bytes)
);
Ok::<_, Error>((url, data))
};
let downloaded: HashMap<String, String> = try_join_all(urls.into_iter().map(fetch))
.await?
.into_iter()
.collect();
inline_media(&mut request.messages, &downloaded);
Ok(())
}
}
#[cfg(test)]
mod tests {
use litellm_http::{Client, media::MediaFetcher};
use rstest::rstest;
use serde_json::{Value, json};
use wiremock::{
Mock, MockServer, ResponseTemplate,
matchers::{method, path},
};
use crate::{
ChatCompletionsRoute, Error, prepare::resolve_request, types::ChatCompletionsRequest,
};
fn route(media: bool) -> ChatCompletionsRoute {
let route = crate::test_support::route();
match media {
true => route.with_media(MediaFetcher::for_test(Client::plain_for_test())),
false => route,
}
}
fn request(url: &str) -> ChatCompletionsRequest<'static> {
ChatCompletionsRequest {
model: "anthropic/claude-sonnet-4-5",
messages: json!([{"role": "user", "content": [
{"type": "image_url", "image_url": {"url": url}}
]}]),
optional_params: serde_json::Map::new(),
api_key: Some("sk-test"),
api_base: None,
custom_llm_provider: None,
extra_headers: None,
timeout: None,
drop_params: Some(false),
model_info: None,
}
}
fn image(request: &crate::types::ResolvedChatCompletionsRequest<'_>) -> Value {
serde_json::to_value(&request.messages[0].content).unwrap()[0]["image_url"]["url"].clone()
}
#[rstest]
#[tokio::test]
async fn a_remote_image_is_downloaded_and_inlined() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/cat.png"))
.respond_with(ResponseTemplate::new(200).set_body_raw(b"png".to_vec(), "image/png"))
.mount(&server)
.await;
let url = format!("{}/cat.png", server.uri());
let url: &'static str = Box::leak(url.into_boxed_str());
let mut resolved = resolve_request(request(url)).unwrap();
route(true)
.inline_remote_media(&mut resolved)
.await
.unwrap();
assert_eq!(image(&resolved), "data:image/png;base64,cG5n");
}
#[rstest]
#[tokio::test]
async fn a_media_url_that_cannot_be_fetched_is_the_callers_error() {
let server = MockServer::start().await;
let url = format!("{}/missing.png", server.uri());
let url: &'static str = Box::leak(url.into_boxed_str());
let mut resolved = resolve_request(request(url)).unwrap();
let error = route(true)
.inline_remote_media(&mut resolved)
.await
.unwrap_err();
assert!(matches!(error, Error::InvalidRequest(_)), "{error:?}");
}
#[rstest]
#[tokio::test]
async fn without_a_fetcher_a_media_url_is_unsupported_and_a_data_url_is_left_alone() {
let mut remote = resolve_request(request("http://example.com/cat.png")).unwrap();
assert!(matches!(
route(false).inline_remote_media(&mut remote).await,
Err(Error::Unsupported(_))
));
let mut inline = resolve_request(request("data:image/png;base64,cG5n")).unwrap();
route(false).inline_remote_media(&mut inline).await.unwrap();
assert_eq!(image(&inline), "data:image/png;base64,cG5n");
}
}

76
litellm-rust/crates/inference-chat/src/prepare.rs
View file

@ -1,7 +1,10 @@
use litellm_auth::SecretValue;
use litellm_core_utils::settings::Lookup;
use litellm_http::request::with_default_headers;
use litellm_llms::base_llm::{auth::ValidatedEnvironment, chat::transformation::BaseConfig};
use litellm_llms::base_llm::{
auth::{Headers, ValidatedEnvironment},
chat::{params::ChatRequestContext, transformation::BaseConfig},
};
use litellm_llms_types::formats::chat_completions::ChatMessage;
use litellm_secrets::source::Secrets;
use serde_json::Value;
@ -72,9 +75,38 @@ pub(super) fn resolve_request(
api_base: request.api_base,
extra_headers: request.extra_headers,
timeout: request.timeout,
drop_params: request.drop_params,
model_info: request.model_info,
})
}
/// The headers a transformation asked for, added to the ones already going out. A header
/// already there (a forwarded `anthropic-beta`, say) keeps its values and gains the new ones,
/// comma separated, the way Python unions beta headers.
fn merge_headers(headers: &mut Headers, added: Headers) {
for (name, value) in added {
match headers
.iter_mut()
.find(|(existing, _)| existing.eq_ignore_ascii_case(&name))
{
Some((_, existing)) => {
let mut values: Vec<&str> = existing
.split(',')
.map(str::trim)
.filter(|value| !value.is_empty())
.collect();
for value in value.split(',').map(str::trim) {
if !value.is_empty() && !values.contains(&value) {
values.push(value);
}
}
*existing = values.join(", ");
}
None => headers.push((name, value)),
}
}
}
fn validate_environment(
request: &ResolvedChatCompletionsRequest<'_>,
model: &str,
@ -101,7 +133,7 @@ pub(super) fn prepare_provider_request(
request: ResolvedChatCompletionsRequest<'_>,
secrets: Secrets,
) -> Result<ProviderChatCompletionsRequest, Error> {
let environment =
let mut environment =
validate_environment(&request, &request.model, request.config, secrets.as_ref())?;
let model = request.model;
let config = request.config;
@ -112,8 +144,17 @@ pub(super) fn prepare_provider_request(
&request.optional_params,
&env_lookup,
)?;
let transformed =
config.transform_request(&model, request.messages, request.optional_params.clone())?;
let context = ChatRequestContext {
drop_params: request.drop_params,
model_info: request.model_info.as_deref(),
};
let transformed = config.transform_request(
&model,
request.messages,
request.optional_params.clone(),
&context,
)?;
merge_headers(&mut environment.headers, transformed.headers);
Ok(ProviderChatCompletionsRequest {
model,
@ -136,7 +177,7 @@ mod tests {
use litellm_llms::base_llm::auth::{AuthScheme, resolve_auth};
use serde_json::{Map, Value, json};
use super::{prepare_provider_request, resolve_request};
use super::{merge_headers, prepare_provider_request, resolve_request};
use crate::{
Error,
types::{ChatCompletionsRequest, ProviderChatCompletionsRequest},
@ -183,6 +224,8 @@ mod tests {
custom_llm_provider: provider,
extra_headers: None,
timeout: None,
drop_params: Some(false),
model_info: None,
}
}
@ -636,4 +679,27 @@ mod tests {
"prepare did not carry the bearer token"
);
}
#[test]
fn transformation_headers_union_with_the_forwarded_ones() {
let mut headers = vec![
("Anthropic-Beta".to_owned(), "a,b".to_owned()),
("x-other".to_owned(), "1".to_owned()),
];
merge_headers(
&mut headers,
vec![
("anthropic-beta".to_owned(), "b, c".to_owned()),
("x-new".to_owned(), "2".to_owned()),
],
);
assert_eq!(
headers,
[
("Anthropic-Beta".to_owned(), "a, b, c".to_owned()),
("x-other".to_owned(), "1".to_owned()),
("x-new".to_owned(), "2".to_owned()),
]
);
}
}

2
litellm-rust/crates/inference-chat/src/route.rs
View file

@ -69,6 +69,8 @@ impl ChatCompletionsRoute {
custom_llm_provider: call.custom_llm_provider.as_deref(),
extra_headers: call.extra_headers,
timeout: call.timeout,
drop_params: call.drop_params,
model_info: call.model_info,
};
self.run(request, cache_options, interceptors, observers)
.await

7
litellm-rust/crates/inference-chat/src/stream.rs
View file

@ -140,7 +140,7 @@ impl ChatCompletionsRoute {
let timeout = call
.timeout
.unwrap_or(Duration::from_secs(CHAT_COMPLETIONS_TIMEOUT_SECS));
let resolved = resolve_request(ChatCompletionsRequest {
let mut resolved = resolve_request(ChatCompletionsRequest {
model: &call.model,
messages: call.messages,
optional_params,
@ -149,7 +149,10 @@ impl ChatCompletionsRoute {
custom_llm_provider: call.custom_llm_provider.as_deref(),
extra_headers: call.extra_headers,
timeout: call.timeout,
drop_params: call.drop_params,
model_info: call.model_info,
})?;
self.inline_remote_media(&mut resolved).await?;
let snapshot = self
.secrets
.resolve(&resolved.config.secret_names())
@ -1019,6 +1022,8 @@ mod tests {
custom_llm_provider: None,
extra_headers: None,
timeout: Some(Duration::from_secs(5)),
drop_params: Some(false),
model_info: None,
},
response_model: "public/model".into(),
usage: ReportedUsage::default(),

27
litellm-rust/crates/inference-chat/src/types.rs
View file

@ -1,20 +1,20 @@
use litellm_secrets::source::Secrets;
use std::time::Duration;
use std::{sync::Arc, time::Duration};
use litellm_auth::SecretValue;
use litellm_llms::base_llm::{
auth::ValidatedEnvironment,
chat::{streaming::StreamShape, transformation::BaseConfig},
model_info::ModelInfo,
};
use litellm_llms_types::formats::chat_completions::ChatMessage;
use serde_json::{Map, Value};
/// A `/chat/completions` call as it crosses into the core.
///
/// `optional_params` arrives already mapped to the provider's own parameter
/// names by the host, exactly as the messages route receives an already
/// Anthropic-shaped body. The core owns the conversation translation, the
/// provider call, and the response normalization.
/// `optional_params` are the OpenAI request's own params; the provider config maps them,
/// refusing what the provider cannot take unless `drop_params` says to drop it. The core
/// owns the conversation translation, the provider call, and the response normalization.
pub struct ChatCompletionsRequest<'a> {
pub model: &'a str,
pub messages: Value,
@ -24,6 +24,11 @@ pub struct ChatCompletionsRequest<'a> {
pub custom_llm_provider: Option<&'a str>,
pub extra_headers: Option<Map<String, Value>>,
pub timeout: Option<Duration>,
/// `drop_params` from the deployment or the proxy settings, `None` when neither says. A
/// request's own `drop_params` rides in `optional_params`.
pub drop_params: Option<bool>,
/// The model cost map, for the param rules that read model capabilities.
pub model_info: Option<Arc<dyn ModelInfo>>,
}
pub struct ChatCompletionsCall {
@ -35,6 +40,11 @@ pub struct ChatCompletionsCall {
pub custom_llm_provider: Option<String>,
pub extra_headers: Option<Map<String, Value>>,
pub timeout: Option<Duration>,
/// `drop_params` from the deployment or the proxy settings, `None` when neither says. A
/// request's own `drop_params` rides in `optional_params`.
pub drop_params: Option<bool>,
/// The model cost map, for the param rules that read model capabilities.
pub model_info: Option<Arc<dyn ModelInfo>>,
}
impl From<ChatCompletionsRequest<'_>> for ChatCompletionsCall {
@ -48,6 +58,8 @@ impl From<ChatCompletionsRequest<'_>> for ChatCompletionsCall {
custom_llm_provider: request.custom_llm_provider.map(str::to_owned),
extra_headers: request.extra_headers,
timeout: request.timeout,
drop_params: request.drop_params,
model_info: request.model_info,
}
}
}
@ -62,6 +74,11 @@ pub struct ResolvedChatCompletionsRequest<'a> {
pub api_base: Option<&'a str>,
pub extra_headers: Option<Map<String, Value>>,
pub timeout: Option<Duration>,
/// `drop_params` from the deployment or the proxy settings, `None` when neither says. A
/// request's own `drop_params` rides in `optional_params`.
pub drop_params: Option<bool>,
/// The model cost map, for the param rules that read model capabilities.
pub model_info: Option<Arc<dyn ModelInfo>>,
}
pub struct ProviderChatCompletionsRequest {

6
litellm-rust/crates/inference-chat/tests/caching.rs
View file

@ -84,6 +84,8 @@ async fn the_same_route_entrypoint_reports_facts_with_or_without_caching(
custom_llm_provider: None,
extra_headers: None,
timeout: None,
drop_params: Some(false),
model_info: None,
},
&(),
Some(observer.clone()),
@ -254,6 +256,8 @@ async fn chat_cache_identity_follows_resolved_configuration_and_request_callback
custom_llm_provider: None,
extra_headers: None,
timeout: None,
drop_params: Some(false),
model_info: None,
},
&hooks,
None,
@ -319,6 +323,8 @@ async fn signed_requests_bypass_response_caching(cache: Arc<dyn ResponseCacheSer
messages:json!([{"role":"user","content":"hello"}]),
optional_params:json!({"aws_access_key_id":"test-access","aws_secret_access_key":"test-secret","aws_region_name":"eu-west-1"}).as_object().unwrap().clone(),
api_key:None,api_base:Some(&upstream.uri()),custom_llm_provider:None,extra_headers:None,timeout:None,
drop_params: Some(false),
model_info: None,
}, &hooks, None).await.unwrap();
assert_eq!(
serde_json::to_value(response).unwrap()["usage"]["total_tokens"],

6
litellm-rust/crates/inference-chat/tests/chat_completions.rs
View file

@ -47,6 +47,8 @@ fn request() -> ChatCompletionsRequest<'static> {
custom_llm_provider: None,
extra_headers: None,
timeout: Some(Duration::from_secs(10)),
drop_params: Some(false),
model_info: None,
}
}
@ -163,7 +165,7 @@ async fn bedrock_round_trip_is_signed_and_normalized(request: ChatCompletionsReq
#[case::missing_usage(
r#"{"model":"m","content":[{"type":"text","text":"hi"}],"stop_reason":"end_turn"}"#
)]
#[case::tool_use_block(r#"{"model":"m","content":[{"type":"tool_use","id":"t","name":"f","input":{}}],"stop_reason":"tool_use","usage":{"input_tokens":1,"output_tokens":1}}"#)]
#[case::malformed_token_count(r#"{"model":"m","content":[{"type":"text","text":"hi"}],"stop_reason":"end_turn","usage":{"input_tokens":"1","output_tokens":1}}"#)]
#[case::not_json("not json")]
#[tokio::test]
async fn a_response_it_cannot_normalize_is_reported_as_already_sent(
@ -294,6 +296,8 @@ async fn direct_and_hosted_calls_share_hooks_and_lifecycle(
custom_llm_provider: call.custom_llm_provider.as_deref(),
extra_headers: call.extra_headers,
timeout: call.timeout,
drop_params: Some(false),
model_info: None,
},
&host,
Some(host.events.0.sender.clone()),

9
litellm-rust/crates/llms-types/src/formats/chat_completions/response.rs
View file

@ -87,6 +87,7 @@ pub struct ChatCompletionsUsage {
}
#[macro_rules_attribute::apply(crate::wire_type)]
#[derive(Default)]
pub struct ChatCompletionsChoiceMessage {
pub role: String,
// Whether an empty turn is `None` or `""` is the provider's choice, not a
@ -94,6 +95,14 @@ pub struct ChatCompletionsChoiceMessage {
// while Converse assigns the joined string unconditionally. Each config
// mirrors its own, so keep this optional and serialize it even when None.
pub content: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub tool_calls: Option<Vec<ChatCompletionToolCallChunk>>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub reasoning_content: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub thinking_blocks: Option<Vec<ChatCompletionThinkingBlock>>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub provider_specific_fields: Option<Map<String, Value>>,
}
#[macro_rules_attribute::apply(crate::wire_type)]

2
litellm-rust/crates/llms/Cargo.toml
View file

@ -22,6 +22,8 @@ litellm-secrets.workspace = true
litellm-python-compat.workspace = true
litellm-router-types.workspace = true
base64.workspace = true
sha2.workspace = true
uuid.workspace = true
bytes.workspace = true
data-url = "0.3.2"
futures-util.workspace = true

86
litellm-rust/crates/llms/src/anthropic/chat/handler.rs
View file

@ -4,8 +4,9 @@
//! Each event becomes one provider chunk, empty ones included; the host's stream wrapper
//! (Python's `CustomStreamWrapper`) decides which of them the client sees.
//!
//! DIVERGENCE: not built here, since the chat route never asks for what produces them:
//! `json_mode` / `response_format` tool conversion, `web_search_calls` and
//! DIVERGENCE: a `response_format` tool's answer is held until its block ends and sent as
//! one piece of content, where Python streams it as it decides the envelope; an answer
//! spread over several blocks is sent per block. Not built: `web_search_calls` and
//! `code_interpreter_results` summaries, and a tool call's `caller`. The reasoning token
//! estimate Python makes from streamed thinking text (with its tokenizer) is not made, so
//! such a stream reports `completion_tokens_details` without counts unless Anthropic sends
@ -28,7 +29,14 @@ use serde_json::{Map, Value, json};
use crate::{
Error,
base_llm::{base_model_iterator::StreamTransformer, chat::streaming::StreamShape},
base_llm::{
base_model_iterator::StreamTransformer,
chat::{
params::{RESPONSE_FORMAT_TOOL_NAME, python_dumps},
response_format::{envelope_contents, json_tool_answer},
streaming::StreamShape,
},
},
};
/// Anthropic sends no status with an `error` event; Python raises it as a 500.
@ -51,6 +59,10 @@ pub struct ModelResponseIterator {
web_search_results: Vec<Value>,
tool_results: Vec<Value>,
compaction_blocks: Vec<Value>,
/// Whether the tool block in progress is the `response_format` tool.
is_response_format_tool: bool,
converted_response_format_tool: bool,
response_format_buffer: String,
}
/// What one event contributes to its chunk.
@ -79,9 +91,58 @@ impl ModelResponseIterator {
web_search_results: Vec::new(),
tool_results: Vec::new(),
compaction_blocks: Vec::new(),
is_response_format_tool: false,
converted_response_format_tool: false,
response_format_buffer: String::new(),
}
}
/// `_handle_json_mode_chunk`: the `response_format` tool's arguments become content.
fn handle_json_mode(&mut self, parsed: &mut Parsed) {
if !self.shape.json_mode {
return;
}
let Some(tool_use) = parsed.tool_use.take() else {
return;
};
if tool_use.id.is_some() {
self.is_response_format_tool =
tool_use.function.name.as_deref() == Some(RESPONSE_FORMAT_TOOL_NAME);
if self.is_response_format_tool {
self.tool_index -= 1;
}
}
if !self.is_response_format_tool {
parsed.tool_use = Some(tool_use);
return;
}
self.converted_response_format_tool = true;
if self.shape.response_format_schema.is_some() {
match tool_use.id {
Some(_) => self.response_format_buffer.clear(),
None => self
.response_format_buffer
.push_str(&tool_use.function.arguments),
}
parsed.text = String::new();
return;
}
parsed.text = json_tool_answer(&[tool_use.function.arguments], None);
}
/// `_held_answer`: the held arguments, or what their envelope wraps.
fn flush_response_format_buffer(&mut self) -> String {
let held = std::mem::take(&mut self.response_format_buffer);
let Some(schema) = &self.shape.response_format_schema else {
return String::new();
};
serde_json::from_str::<Value>(&held)
.ok()
.and_then(|payload| envelope_contents(&payload, schema, RESPONSE_FORMAT_TOOL_NAME))
.map(|contents| python_dumps(&contents))
.unwrap_or(held)
}
/// Whether the tool block so far streamed only empty argument fragments.
pub fn check_empty_tool_call_args(&self) -> bool {
match self.content_blocks.first() {
@ -273,6 +334,10 @@ impl ModelResponseIterator {
if in_tool && self.check_empty_tool_call_args() {
parsed.tool_use = Some(self.tool_call(None, None, "{}".into()));
}
if self.is_response_format_tool {
parsed.text = self.flush_response_format_buffer();
}
self.is_response_format_tool = false;
self.current_content_block_type = None;
parsed
}
@ -282,10 +347,16 @@ impl ModelResponseIterator {
delta: MessagesDelta,
usage: Option<MessagesStreamUsage>,
) -> Parsed {
let provider_finish_reason =
finish_reason_for(delta.stop_reason.as_deref().unwrap_or("stop")).to_string();
let called_only_the_response_format_tool = provider_finish_reason == "tool_calls"
&& self.converted_response_format_tool
&& self.tool_index < 0;
let mut parsed = Parsed {
finish_reason: Some(
finish_reason_for(delta.stop_reason.as_deref().unwrap_or("stop")).to_string(),
),
finish_reason: Some(match called_only_the_response_format_tool {
true => "stop".to_owned(),
false => provider_finish_reason,
}),
usage: usage.map(|usage| self.handle_usage(&usage)),
..Parsed::default()
};
@ -314,7 +385,7 @@ impl ModelResponseIterator {
&mut self,
event: MessagesStreamEvent,
) -> Result<ChatCompletionChunk, Error> {
let parsed = match event {
let mut parsed = match event {
MessagesStreamEvent::ContentBlockDelta { delta, .. } => self.content_block_delta(delta),
MessagesStreamEvent::ContentBlockStart { content_block, .. } => {
self.content_block_start(content_block)
@ -341,6 +412,7 @@ impl ModelResponseIterator {
});
}
};
self.handle_json_mode(&mut parsed);
Ok(self.chunk(parsed))
}

777
litellm-rust/crates/llms/src/anthropic/chat/messages.rs Normal file
View file

@ -0,0 +1,777 @@
//! Chat messages as the Messages API takes them: Python's `translate_system_message`,
//! `place_mid_conversation_system` and `anthropic_messages_pt`.
use std::collections::HashSet;
use base64::Engine;
use serde_json::{Map, Value, json};
use crate::{
Error, ErrorDetail,
anthropic::common_utils::ENCRYPTED_REASONING_SIGNATURE_PREFIX,
base_llm::chat::{
media::{data_url, image_url},
params::{parse_tool_arguments, truthy},
},
};
pub const EMPTY_TEXT_PLACEHOLDER: &str =
"[System: Empty message content sanitised to satisfy protocol]";
pub const CONVERTED_SYSTEM_NOTE: &str = "Operator note (not from the user): the following was originally a mid-conversation system-role reminder.";
const TOOL_USE_ID_FALLBACK: &str = "tool_use_id";
fn bad_request(message: impl Into<String>) -> Error {
Error::InvalidRequest(ErrorDetail::Message(message.into()))
}
pub fn role(message: &Value) -> &str {
match message.get("role").and_then(Value::as_str) {
Some("developer") => "system",
Some(role) => role,
None => "",
}
}
/// Python's truthiness of a message's `content`.
fn content_is_empty(message: &Value) -> bool {
match message.get("content") {
None | Some(Value::Null) => true,
Some(Value::String(text)) => text.is_empty(),
Some(Value::Array(items)) => items.is_empty(),
Some(_) => false,
}
}
/// `[^a-zA-Z0-9_-]` -> `_`, an empty id becoming `tool_use_id`.
pub fn sanitize_tool_use_id(id: &str) -> String {
let sanitized: String = id
.chars()
.map(
|c| match c.is_ascii_alphanumeric() || c == '_' || c == '-' {
true => c,
false => '_',
},
)
.collect();
match sanitized.is_empty() {
true => TOOL_USE_ID_FALLBACK.to_owned(),
false => sanitized,
}
}
/// `add_cache_control_to_content`: a dict `cache_control` on the source is copied over.
fn with_cache_control(mut block: Value, source: &Value) -> Value {
if let (Some(cache_control @ Value::Object(_)), Value::Object(fields)) =
(source.get("cache_control"), &mut block)
{
fields.insert("cache_control".into(), cache_control.clone());
}
block
}
/// `_cache_control` of the mid-conversation module: only an ephemeral one, rebuilt.
fn normalized_cache_control(source: &Value) -> Option<Value> {
let cache_control = source.get("cache_control")?.as_object()?;
if cache_control.get("type").and_then(Value::as_str) != Some("ephemeral") {
return None;
}
Some(match cache_control.get("ttl").and_then(Value::as_str) {
Some(ttl @ ("1h" | "5m")) => json!({"type": "ephemeral", "ttl": ttl}),
_ => json!({"type": "ephemeral"}),
})
}
/// The leading run of system messages, and the rest.
pub fn split_leading_system_run(messages: Vec<Value>) -> (Vec<Value>, Vec<Value>) {
let leading = messages
.iter()
.position(|message| role(message) != "system")
.unwrap_or(messages.len());
let mut messages = messages;
let rest = messages.split_off(leading);
(messages, rest)
}
/// `translate_system_message` for the leading run.
pub fn system_blocks(run: &[Value]) -> Vec<Value> {
let mut blocks = Vec::new();
for message in run {
match message.get("content") {
Some(Value::String(text)) => {
if text.is_empty() {
continue;
}
let mut block = Map::from_iter([
("type".to_owned(), json!("text")),
("text".to_owned(), json!(text)),
]);
if let Some(cache_control) = message.get("cache_control") {
block.insert("cache_control".into(), cache_control.clone());
}
blocks.push(Value::Object(block));
}
Some(Value::Array(parts)) => {
for part in parts {
let text = part.get("text").cloned().unwrap_or(Value::Null);
let kind = part.get("type").cloned().unwrap_or(Value::Null);
let empty = match &text {
Value::String(text) => text.is_empty(),
Value::Null => true,
_ => false,
};
if kind.as_str() == Some("text") && empty {
continue;
}
let mut block =
Map::from_iter([("type".to_owned(), kind), ("text".to_owned(), text)]);
if let Some(cache_control) = part.get("cache_control") {
block.insert("cache_control".into(), cache_control.clone());
}
blocks.push(Value::Object(block));
}
}
_ => {}
}
}
blocks
}
/// The non-empty text parts of a system message, each with its normalized `cache_control`.
fn system_text_parts(message: &Value) -> Vec<Value> {
let block = |text: &str, cache_control: Option<Value>| {
let mut block = Map::from_iter([
("type".to_owned(), json!("text")),
("text".to_owned(), json!(text)),
]);
if let Some(cache_control) = cache_control {
block.insert("cache_control".into(), cache_control);
}
Value::Object(block)
};
match message.get("content") {
Some(Value::String(text)) if !text.is_empty() => {
vec![block(text, normalized_cache_control(message))]
}
Some(Value::Array(parts)) => parts
.iter()
.filter(|part| part.get("type").and_then(Value::as_str) == Some("text"))
.filter_map(|part| {
let text = part.get("text")?.as_str().filter(|text| !text.is_empty())?;
Some(block(text, normalized_cache_control(part)))
})
.collect(),
_ => Vec::new(),
}
}
/// `system_message_as_user` for each message of a run that carries text.
pub fn converted_user_turns(run: &[Value]) -> Vec<Value> {
run.iter()
.filter_map(|message| {
let parts = system_text_parts(message);
if parts.is_empty() {
return None;
}
let mut content = vec![json!({"type": "text", "text": CONVERTED_SYSTEM_NOTE})];
content.extend(parts);
Some(json!({"role": "user", "content": content}))
})
.collect()
}
#[derive(Clone, Copy, PartialEq, Eq)]
enum Kind {
System,
Tool,
User,
Other,
}
fn kind(message: &Value) -> Kind {
match role(message) {
"system" => Kind::System,
"tool" | "function" => Kind::Tool,
"user" => Kind::User,
_ => Kind::Other,
}
}
/// `_converted_for_unflagged_model`: every later system message becomes a user turn in
/// place, a run caught before a tool result moving behind it.
pub fn place_mid_conversation_system(messages: Vec<Value>) -> Vec<Value> {
if !messages.iter().any(|message| kind(message) == Kind::System) {
return messages;
}
let mut runs: Vec<(Kind, Vec<Value>)> = Vec::new();
for message in messages {
let message_kind = kind(&message);
match runs.last_mut() {
Some((last, run)) if *last == message_kind => run.push(message),
_ => runs.push((message_kind, vec![message])),
}
}
let mut placed = Vec::new();
for index in 0..runs.len() {
let (run_kind, run) = &runs[index];
let follower = runs.get(index + 1).map(|(kind, _)| *kind);
match run_kind {
Kind::System => {
if follower != Some(Kind::Tool) {
placed.extend(converted_user_turns(run));
}
}
Kind::Tool if index > 0 && runs[index - 1].0 == Kind::System => {
placed.extend(run.iter().cloned());
placed.extend(converted_user_turns(&runs[index - 1].1));
}
_ => placed.extend(run.iter().cloned()),
}
}
placed
}
/// `_sanitize_empty_text_content`.
fn sanitize_empty_text(mut message: Value) -> Value {
if !matches!(role(&message), "user" | "assistant") {
return message;
}
match message.get_mut("content") {
Some(Value::String(text)) if text.trim().is_empty() => {
*text = EMPTY_TEXT_PLACEHOLDER.to_owned();
}
Some(Value::Array(parts)) => {
for part in parts {
if part.get("type").and_then(Value::as_str) != Some("text") {
continue;
}
let blank = part
.get("text")
.and_then(Value::as_str)
.is_none_or(|text| text.trim().is_empty());
if blank && let Value::Object(fields) = part {
fields.insert("text".into(), json!(EMPTY_TEXT_PLACEHOLDER));
}
}
}
_ => {}
}
message
}
/// `convert_to_anthropic_image_obj`: a data URL as its media type and payload.
fn image_source(url: &str, format: Option<&str>) -> Result<(String, String), Error> {
let (media_type, data) = data_url(url).ok_or_else(|| {
bad_request(format!(
"Image url not in expected format. Example Expected input - \"image_url\": \"data:image/jpeg;base64,{{base64_image}}\". Supported formats - ['image/jpeg', 'image/png', 'image/gif', 'image/webp']. Error: {url:.40}"
))
})?;
let media_type = match format {
Some(format) => format.to_owned(),
None => media_type.replace("\\/", "/"),
};
Ok((media_type, data.to_owned()))
}
/// `create_anthropic_image_param`.
fn image_block(part: &Value) -> Result<Value, Error> {
let url = image_url(part).unwrap_or_default();
let format = part
.get("image_url")
.and_then(|image| image.get("format"))
.and_then(Value::as_str);
if url.starts_with("https://") {
return Ok(json!({"type": "image", "source": {"type": "url", "url": url}}));
}
if url.starts_with("http") {
return Err(Error::Unsupported(
"an http:// image the host did not download",
));
}
let (media_type, data) = image_source(url, format)?;
Ok(
json!({"type": "image", "source": {"type": "base64", "media_type": media_type, "data": data}}),
)
}
/// `anthropic_process_openai_file_message`.
fn file_block(part: &Value) -> Result<Value, Error> {
let file = part.get("file").ok_or_else(|| {
bad_request("Content block has type='file' but is missing the required 'file' field")
})?;
let format = file.get("format").and_then(Value::as_str);
if let Some(data) = file
.get("file_data")
.and_then(Value::as_str)
.filter(|data| !data.is_empty())
{
if data.starts_with("http") {
return Err(Error::Unsupported(
"a file_data URL the host did not download",
));
}
let (media_type, data) = image_source(data, format)?;
return Ok(
json!({"type": "document", "source": {"type": "base64", "media_type": media_type, "data": data}}),
);
}
match file
.get("file_id")
.and_then(Value::as_str)
.filter(|id| !id.is_empty())
{
Some(id) if format.is_none() && id.starts_with("http") => {
Ok(json!({"type": "document", "source": {"type": "url", "url": id}}))
}
// Uploaded files need the Files API betas and, for a container upload, the code
// execution tool.
Some(_) => Err(Error::Unsupported("an Anthropic Files API file_id")),
None => Err(bad_request(format!(
"Either file_data or file_id must be present in the file message: {part}"
))),
}
}
/// `_is_anthropic_document_data_uri`.
fn is_document_data_uri(url: &str) -> bool {
url.strip_prefix("data:")
.map(|rest| rest.split([';', ',']).next().unwrap_or_default())
.is_some_and(|media_type| matches!(media_type, "application/pdf" | "text/plain"))
}
/// `convert_to_anthropic_tool_result`.
fn tool_result(message: &Value) -> Result<Value, Error> {
let content = match message.get("content") {
Some(Value::String(text)) => json!(text),
Some(Value::Array(parts)) => {
let mut blocks = Vec::new();
for part in parts {
match part.get("type").and_then(Value::as_str) {
Some("text") => {
let mut block = Map::from_iter([
("type".to_owned(), json!("text")),
(
"text".to_owned(),
part.get("text").cloned().unwrap_or(Value::Null),
),
]);
if let Some(cache_control) =
part.get("cache_control").filter(|value| !value.is_null())
{
block.insert("cache_control".into(), cache_control.clone());
}
blocks.push(Value::Object(block));
}
Some("image_url") => {
let url = image_url(part).unwrap_or_default();
let block = match is_document_data_uri(url) {
true => {
file_block(&json!({"type": "file", "file": {"file_data": url}}))?
}
false => image_block(part)?,
};
blocks.push(with_cache_control(block, part));
}
Some("tool_reference") => blocks.push(json!({
"type": "tool_reference",
"tool_name": part.get("tool_name").cloned().unwrap_or(Value::Null),
})),
Some("file") => blocks.push(with_cache_control(file_block(part)?, part)),
_ => {}
}
}
Value::Array(blocks)
}
_ => json!(""),
};
let id = match role(message) {
"tool" => message
.get("tool_call_id")
.and_then(Value::as_str)
.map(str::to_owned)
.ok_or_else(|| bad_request("A tool message needs a tool_call_id."))?,
_ => message
.get("tool_call_id")
.and_then(Value::as_str)
.filter(|id| !id.is_empty())
.map(str::to_owned)
.unwrap_or_else(|| uuid::Uuid::new_v4().to_string()),
};
let mut result = Map::from_iter([
("type".to_owned(), json!("tool_result")),
("tool_use_id".to_owned(), json!(sanitize_tool_use_id(&id))),
("content".to_owned(), content),
]);
if let Some(cache_control) = message
.get("cache_control")
.filter(|value| !value.is_null())
{
result.insert("cache_control".into(), cache_control.clone());
}
Ok(Value::Object(result))
}
pub fn is_encrypted_reasoning(block: &Value) -> bool {
let field = match block.get("type").and_then(Value::as_str) {
Some("thinking") => block.get("signature"),
Some("redacted_thinking") => block.get("data"),
_ => None,
};
field
.and_then(Value::as_str)
.is_some_and(|value| value.starts_with(ENCRYPTED_REASONING_SIGNATURE_PREFIX))
}
/// `is_unsignable_thinking_block`.
pub fn is_unsignable_thinking_block(block: &Value) -> bool {
if is_encrypted_reasoning(block) {
return true;
}
block.get("type").and_then(Value::as_str) == Some("thinking")
&& block
.get("signature")
.and_then(Value::as_str)
.is_none_or(str::is_empty)
}
/// `native_redacted_thinking_data`: the base64 wrapper Converse adds, removed.
pub fn native_redacted_thinking_data(data: &str) -> String {
let engine = base64::engine::general_purpose::STANDARD;
engine
.decode(data)
.ok()
.and_then(|bytes| String::from_utf8(bytes).ok())
.filter(|payload| payload.is_ascii() && engine.decode(payload).is_ok())
.unwrap_or_else(|| data.to_owned())
}
/// `_replayable_thinking_blocks`.
fn replayable_thinking_blocks(blocks: &[Value]) -> Vec<Value> {
blocks
.iter()
.filter(|block| !is_unsignable_thinking_block(block))
.map(|block| {
let mut block = block.clone();
if let Value::Object(fields) = &mut block {
if fields.get("type").and_then(Value::as_str) == Some("redacted_thinking")
&& let Some(Value::String(data)) = fields.get("data")
{
let native = native_redacted_thinking_data(data);
fields.insert("data".into(), json!(native));
}
fields.remove("index");
}
block
})
.collect()
}
/// The tool calls of an assistant message, as `tool_use` blocks.
fn tool_uses(tool_calls: &[Value], seen: &mut HashSet<String>) -> Result<Vec<Value>, Error> {
let mut blocks = Vec::new();
for call in tool_calls {
if call.get("type").and_then(Value::as_str) != Some("function") {
continue;
}
let id = call.get("id").and_then(Value::as_str).unwrap_or_default();
if id.starts_with("srvtoolu_") {
return Err(Error::Unsupported(
"replaying an Anthropic server tool call",
));
}
let function = call.get("function");
let name = function
.and_then(|function| function.get("name"))
.cloned()
.unwrap_or(Value::Null);
let input = parse_tool_arguments(function.and_then(|function| function.get("arguments")))?;
let id = sanitize_tool_use_id(id);
let block = with_cache_control(
json!({"type": "tool_use", "id": id, "name": name, "input": input}),
call,
);
if !seen.insert(id) {
continue;
}
blocks.push(block);
}
Ok(blocks)
}
/// `anthropic_messages_pt` over messages whose later system messages are already placed.
pub fn anthropic_messages_pt(messages: Vec<Value>) -> Result<Vec<Value>, Error> {
let messages: Vec<Value> = messages.into_iter().map(sanitize_empty_text).collect();
if messages.is_empty() {
return Err(bad_request(
"Anthropic requires at least one non-system message. Either provide one, or set `litellm.modify_params = True` // `litellm_settings::modify_params: True` to add the dummy user message - {'role': 'user', 'content': 'Please continue.'}.",
));
}
let user_types = |message: &Value| matches!(role(message), "user" | "tool" | "function");
// `_tool_result_later_in_turn`: whether a client tool result follows in the same turn.
let mut tool_result_later = vec![false; messages.len()];
let mut later = false;
for index in (0..messages.len()).rev() {
tool_result_later[index] = later;
let message = &messages[index];
later = user_types(message) && (later || matches!(role(message), "tool" | "function"));
}
let mut out: Vec<Value> = Vec::new();
let mut index = 0;
while index < messages.len() {
let start = index;
let mut user_content: Vec<Value> = Vec::new();
while index < messages.len() && user_types(&messages[index]) {
let message = &messages[index];
if role(message) == "user" {
match message.get("content") {
_ if content_is_empty(message) => {
if !tool_result_later[index]
&& message.get("cache_control").is_some_and(truthy)
{
user_content.push(with_cache_control(
json!({"type": "text", "text": EMPTY_TEXT_PLACEHOLDER}),
message,
));
}
}
Some(Value::Array(parts)) => {
for part in parts {
match part.get("type").and_then(Value::as_str) {
Some("image_url") => {
user_content.push(with_cache_control(image_block(part)?, part))
}
Some("text") => user_content.push(with_cache_control(
json!({"type": "text", "text": part.get("text").cloned().unwrap_or(Value::Null)}),
part,
)),
Some("document") => {
user_content.push(with_cache_control(part.clone(), part))
}
Some("file") => {
user_content.push(with_cache_control(file_block(part)?, part))
}
_ => {}
}
}
}
Some(Value::String(text)) => user_content.push(with_cache_control(
json!({"type": "text", "text": text}),
message,
)),
_ => {}
}
} else {
user_content.push(tool_result(message)?);
}
index += 1;
}
if !user_content.is_empty() {
out.push(json!({"role": "user", "content": user_content}));
} else if let Some(empty) = messages[start..index]
.iter()
.rev()
.find(|message| role(message) == "user" && content_is_empty(message))
{
out.push(json!({"role": "user", "content": [with_cache_control(
json!({"type": "text", "text": EMPTY_TEXT_PLACEHOLDER}),
empty,
)]}));
}
let mut seen_tool_ids: HashSet<String> = HashSet::new();
let mut assistant_content: Vec<Value> = Vec::new();
while index < messages.len() && role(&messages[index]) == "assistant" {
let message = &messages[index];
let provider_fields = message.get("provider_specific_fields");
if let Some(Value::Array(blocks)) =
provider_fields.and_then(|fields| fields.get("compaction_blocks"))
{
assistant_content.extend(blocks.iter().cloned());
}
if provider_fields.is_some_and(|fields| {
fields
.get("web_search_results")
.is_some_and(|v| !v.is_null())
|| fields.get("tool_results").is_some_and(|v| !v.is_null())
}) && message
.get("tool_calls")
.and_then(Value::as_array)
.is_some_and(|calls| !calls.is_empty())
{
return Err(Error::Unsupported(
"replaying Anthropic server tool results",
));
}
let thinking_blocks = match message.get("thinking_blocks") {
Some(Value::Array(blocks)) => Some(replayable_thinking_blocks(blocks)),
_ => None,
};
let parts = message.get("content").and_then(Value::as_array);
let inline_thinking = parts.is_some_and(|parts| {
parts.iter().any(|part| {
matches!(
part.get("type").and_then(Value::as_str),
Some("thinking" | "redacted_thinking")
)
})
});
if let Some(blocks) = thinking_blocks
&& !inline_thinking
{
assistant_content.extend(blocks);
}
match message.get("content") {
Some(Value::Array(parts)) => {
for part in parts.iter().filter(|part| part.is_object()) {
let kind = part.get("type").and_then(Value::as_str).unwrap_or_default();
let text = part.get("text").and_then(Value::as_str).unwrap_or_default();
if kind == "thinking" && !is_unsignable_thinking_block(part) {
assistant_content.push(part.clone());
} else if kind == "text" && !text.is_empty() {
assistant_content.push(with_cache_control(
json!({"type": "text", "text": text}),
part,
));
} else if kind == "server_tool_use" || kind.ends_with("_tool_result") {
assistant_content.push(part.clone());
}
}
}
Some(Value::String(text)) if !text.is_empty() => {
assistant_content.push(with_cache_control(
json!({"type": "text", "text": text}),
message,
));
}
_ => {}
}
if let Some(Value::Array(calls)) = message.get("tool_calls") {
assistant_content.extend(tool_uses(calls, &mut seen_tool_ids)?);
}
if let Some(call) = message.get("function_call").filter(|call| !call.is_null()) {
let input = parse_tool_arguments(call.get("arguments"))?;
assistant_content.push(json!({
"type": "tool_use",
"id": uuid::Uuid::new_v4().to_string(),
"name": call.get("name").filter(|name| !name.is_null()).cloned().unwrap_or_else(|| json!("")),
"input": input,
}));
}
index += 1;
}
if !assistant_content.is_empty() {
out.push(json!({"role": "assistant", "content": assistant_content}));
}
if index == start {
return Err(bad_request(format!(
"Invalid Message passed in {}",
messages[index]
)));
}
}
if let Some(last) = out.last_mut()
&& last["role"] == "assistant"
&& let Some(Value::Array(blocks)) = last.get_mut("content")
{
for block in blocks {
if block.get("type").and_then(Value::as_str) == Some("text")
&& let Some(Value::String(text)) = block.get_mut("text")
{
*text = text.trim_end().to_owned();
}
}
}
Ok(out)
}
#[cfg(test)]
mod tests {
use rstest::rstest;
use super::*;
fn list(value: Value) -> Vec<Value> {
value.as_array().unwrap().clone()
}
#[rstest]
fn tool_round_trip_merges_turns() {
let messages = anthropic_messages_pt(list(json!([
{"role": "user", "content": "weather?"},
{"role": "assistant", "content": null, "tool_calls": [{"id": "call.1", "type": "function", "function": {"name": "get", "arguments": "{\"c\": \"P\"}"}}]},
{"role": "tool", "tool_call_id": "call.1", "content": "sunny"},
{"role": "user", "content": [{"type": "text", "text": "thanks", "cache_control": {"type": "ephemeral"}}]}
])))
.unwrap();
assert_eq!(
messages,
list(json!([
{"role": "user", "content": [{"type": "text", "text": "weather?"}]},
{"role": "assistant", "content": [{"type": "tool_use", "id": "call_1", "name": "get", "input": {"c": "P"}}]},
{"role": "user", "content": [
{"type": "tool_result", "tool_use_id": "call_1", "content": "sunny"},
{"type": "text", "text": "thanks", "cache_control": {"type": "ephemeral"}}
]}
]))
);
}
#[rstest]
fn images_empty_text_and_final_assistant_whitespace() {
let messages = anthropic_messages_pt(list(json!([
{"role": "user", "content": [
{"type": "image_url", "image_url": {"url": "data:image/png;base64,iVBO"}},
{"type": "image_url", "image_url": "https://x/y.png"},
{"type": "text", "text": " "}
]},
{"role": "assistant", "content": "Sure "}
])))
.unwrap();
assert_eq!(
messages[0]["content"],
json!([
{"type": "image", "source": {"type": "base64", "media_type": "image/png", "data": "iVBO"}},
{"type": "image", "source": {"type": "url", "url": "https://x/y.png"}},
{"type": "text", "text": EMPTY_TEXT_PLACEHOLDER}
])
);
assert_eq!(
messages[1]["content"],
json!([{"type": "text", "text": "Sure"}])
);
}
#[rstest]
fn later_system_messages_become_operator_notes() {
let placed = place_mid_conversation_system(list(json!([
{"role": "user", "content": "a"},
{"role": "assistant", "content": "b", "tool_calls": [{"id": "t", "type": "function", "function": {"name": "f", "arguments": "{}"}}]},
{"role": "system", "content": "remember", "cache_control": {"type": "ephemeral", "ttl": "1h", "x": 1}},
{"role": "tool", "tool_call_id": "t", "content": "r"}
])));
assert_eq!(placed[2]["role"], "tool");
assert_eq!(
placed[3],
json!({"role": "user", "content": [
{"type": "text", "text": CONVERTED_SYSTEM_NOTE},
{"type": "text", "text": "remember", "cache_control": {"type": "ephemeral", "ttl": "1h"}}
]})
);
}
#[rstest]
fn unsigned_thinking_is_dropped_and_signed_thinking_leads() {
let messages = anthropic_messages_pt(list(json!([
{"role": "user", "content": "q"},
{"role": "assistant", "content": "a", "thinking_blocks": [
{"type": "thinking", "thinking": "t", "signature": "s", "index": 0},
{"type": "thinking", "thinking": "u"}
]}
])))
.unwrap();
assert_eq!(
messages[1]["content"],
json!([{"type": "thinking", "thinking": "t", "signature": "s"}, {"type": "text", "text": "a"}])
);
}
}

2
litellm-rust/crates/llms/src/anthropic/chat/mod.rs
View file

@ -1,2 +1,4 @@
pub mod handler;
pub mod messages;
pub mod params;
pub mod transformation;

1293
litellm-rust/crates/llms/src/anthropic/chat/params.rs Normal file
View file

File diff suppressed because it is too large Load diff

1632
litellm-rust/crates/llms/src/anthropic/chat/transformation.rs
View file

File diff suppressed because it is too large Load diff

104
litellm-rust/crates/llms/src/azure/chat/transformation.rs
View file

@ -8,56 +8,15 @@ use serde_json::{Map, Value};
use crate::{
Error, ErrorDetail,
base_llm::model_info::{self, ModelInfo},
base_llm::{
chat::params::{
OPENAI_PARAMS, UNVALIDATED_PARAMS, ordered, request_drop_params, string_list, truthy,
},
model_info::{self, ModelInfo},
},
openai::chat::transformation::{is_litellm_owned, take_extra_body},
};
/// The keys of `DEFAULT_CHAT_COMPLETION_PARAM_VALUES` a request carries as OpenAI params, in
/// that table's order. Every other key goes to Azure through `extra_body`.
const OPENAI_PARAMS: &[&str] = &[
"functions",
"function_call",
"temperature",
"top_p",
"n",
"stream",
"stream_options",
"stop",
"max_tokens",
"max_completion_tokens",
"modalities",
"prediction",
"audio",
"presence_penalty",
"frequency_penalty",
"logit_bias",
"user",
"response_format",
"seed",
"tools",
"tool_choice",
"max_retries",
"logprobs",
"top_logprobs",
"extra_headers",
"parallel_tool_calls",
"reasoning_effort",
"verbosity",
"thinking",
"web_search_options",
"include_server_side_tool_invocations",
"service_tier",
"safety_identifier",
"prompt_cache_key",
"prompt_cache_retention",
"store",
"metadata",
"context_management",
];
/// `PROVIDER_UNVALIDATED_PARAMS`: never refused, only left out when unsupported.
const UNVALIDATED_PARAMS: &[&str] = &["user", "stream_options", "stream", "max_retries"];
/// `AzureOpenAIConfig.get_supported_openai_params`.
const AZURE_GPT_PARAMS: &[&str] = &[
"temperature",
@ -410,6 +369,7 @@ impl AzureChatRequest<'_> {
let model = self.deployment;
if model.starts_with(RESPONSES_ROUTE)
|| model_info::field(self.info(), &format!("azure/{model}"), "mode")
.as_ref()
.and_then(Value::as_str)
== Some("responses")
{
@ -830,7 +790,9 @@ fn openai_gpt_params(model: &str, info: Option<&dyn ModelInfo>) -> Vec<&'static
.split_once(RESPONSES_ROUTE)
.map_or(model, |(_, rest)| rest);
if matches!(
model_info::field(info, catalog_name, "litellm_provider").and_then(Value::as_str),
model_info::field(info, catalog_name, "litellm_provider")
.as_ref()
.and_then(Value::as_str),
Some("openai" | "text-completion-openai")
) {
params.push("user");
@ -913,14 +875,6 @@ fn azure_messages(messages: Value) -> Value {
Value::Array(messages)
}
/// The request's OpenAI params in `DEFAULT_CHAT_COMPLETION_PARAM_VALUES` order, the order
/// Python maps them in.
fn ordered(params: &Map<String, Value>) -> impl Iterator<Item = (&'static str, &Value)> {
OPENAI_PARAMS
.iter()
.filter_map(|name| params.get(*name).map(|value| (*name, value)))
}
/// `_map_openai_params`: every supported param still unmapped goes as sent.
fn copy_supported(
non_default: &mut Map<String, Value>,
@ -936,40 +890,6 @@ fn copy_supported(
}
}
/// Python's truthiness for a JSON value.
fn truthy(value: &Value) -> bool {
match value {
Value::Null => false,
Value::Bool(flag) => *flag,
Value::Number(number) => number.as_f64() != Some(0.0),
Value::String(text) => !text.is_empty(),
Value::Array(items) => !items.is_empty(),
Value::Object(object) => !object.is_empty(),
}
}
/// `normalize_drop_params`: a boolean, or a config's `"true"`.
fn request_drop_params(value: Option<Value>) -> bool {
match value {
Some(Value::Bool(flag)) => flag,
Some(Value::String(text)) => text.trim().eq_ignore_ascii_case("true"),
_ => false,
}
}
fn string_list(value: Option<Value>) -> Vec<String> {
match value {
Some(Value::Array(items)) => items
.into_iter()
.filter_map(|item| match item {
Value::String(text) => Some(text),
_ => None,
})
.collect(),
_ => Vec::new(),
}
}
#[cfg(test)]
mod tests {
use std::collections::HashMap;
@ -997,8 +917,8 @@ mod tests {
}
impl ModelInfo for Entries {
fn entry(&self, model: &str) -> Option<&Map<String, Value>> {
self.0.get(model)
fn entry(&self, model: &str) -> Option<std::borrow::Cow<'_, Map<String, Value>>> {
self.0.get(model).map(std::borrow::Cow::Borrowed)
}
fn declares(&self, key: &str) -> bool {

209
litellm-rust/crates/llms/src/base_llm/chat/media.rs Normal file
View file

@ -0,0 +1,209 @@
//! Media a chat message points at. Python downloads an image or file URL a provider cannot
//! take by reference while it builds the request; here the host downloads them first, from
//! the list a config names, and hands the messages back with each one inlined as a data URL.
use std::collections::HashMap;
use litellm_llms_types::formats::chat_completions::{ChatMessage, ChatMessageContent};
use serde_json::Value;
/// Python's `MAX_IMAGE_URL_DOWNLOAD_SIZE_MB`.
pub const MAX_MEDIA_DOWNLOAD_BYTES: u64 = 50 * 1024 * 1024;
/// A part's `image_url`, given as a string or as `{"url": ...}`.
pub fn image_url(part: &Value) -> Option<&str> {
let image_url = part.get("image_url")?;
image_url
.as_str()
.or_else(|| image_url.get("url").and_then(Value::as_str))
}
/// `data:<meta>;base64,<data>` split into its media type and payload, the way Python's
/// `split("data:")[1].split(";base64,")` reads it.
pub fn data_url(url: &str) -> Option<(&str, &str)> {
let (_, rest) = url.split_once("data:")?;
let (media_type, data) = rest.split_once(";base64,")?;
(!data.contains(";base64,")).then_some((media_type, data))
}
fn parts(message: &ChatMessage) -> &[Value] {
match &message.content {
Some(ChatMessageContent::Parts(parts)) => parts,
_ => &[],
}
}
/// The URLs of `image_url` parts (and, with `files`, of `file.file_id` / `file.file_data`)
/// in the messages `roles` names that `remote` says must be downloaded, in order, each once.
pub fn remote_urls(
messages: &[ChatMessage],
roles: &[&str],
files: bool,
remote: impl Fn(&str) -> bool,
) -> Vec<String> {
let mut urls: Vec<String> = Vec::new();
let mut add = |url: &str| {
if remote(url) && !urls.iter().any(|seen| seen == url) {
urls.push(url.to_owned());
}
};
for message in messages
.iter()
.filter(|message| roles.contains(&message.role.as_str()))
{
for part in parts(message) {
match part.get("type").and_then(Value::as_str) {
Some("image_url") => {
if let Some(url) = image_url(part) {
add(url);
}
}
Some("file") if files => {
let file = part.get("file");
let source = ["file_id", "file_data"].into_iter().find_map(|key| {
file.and_then(|file| file.get(key)).and_then(Value::as_str)
});
if let Some(url) = source {
add(url);
}
}
_ => {}
}
}
}
urls
}
/// Every reference to a downloaded URL replaced by its data URL.
pub fn inline_media(messages: &mut [ChatMessage], downloaded: &HashMap<String, String>) {
if downloaded.is_empty() {
return;
}
for message in messages {
let Some(ChatMessageContent::Parts(parts)) = &mut message.content else {
continue;
};
for part in parts {
match part.get("type").and_then(Value::as_str) {
Some("image_url") => match part.get_mut("image_url") {
Some(Value::String(url)) => {
if let Some(data) = downloaded.get(url.as_str()) {
*url = data.clone();
}
}
Some(Value::Object(image)) => {
if let Some(Value::String(url)) = image.get_mut("url")
&& let Some(data) = downloaded.get(url.as_str())
{
*url = data.clone();
}
}
_ => {}
},
Some("file") => {
if let Some(Value::Object(file)) = part.get_mut("file") {
for key in ["file_id", "file_data"] {
if let Some(Value::String(url)) = file.get_mut(key)
&& let Some(data) = downloaded.get(url.as_str())
{
*url = data.clone();
}
}
}
}
_ => {}
}
}
}
}
/// The media type a download is inlined with: the response's own, or one guessed from the
/// URL's extension the way Python's `infer_content_type_from_url_and_content` falls back.
pub fn downloaded_media_type(url: &str, content_type: &str) -> String {
let declared = content_type
.split(';')
.next()
.unwrap_or_default()
.trim()
.to_ascii_lowercase();
if !declared.is_empty() && declared != "application/octet-stream" && declared.contains('/') {
return declared;
}
let path = url
.split(['?', '#'])
.next()
.unwrap_or(url)
.to_ascii_lowercase();
let extension = path.rsplit('.').next().unwrap_or_default();
match extension {
"png" => "image/png",
"jpg" | "jpeg" => "image/jpeg",
"gif" => "image/gif",
"webp" => "image/webp",
"pdf" => "application/pdf",
"txt" => "text/plain",
"csv" => "text/csv",
"html" | "htm" => "text/html",
"md" => "text/markdown",
"mp4" => "video/mp4",
_ => "image/jpeg",
}
.to_owned()
}
#[cfg(test)]
mod tests {
use rstest::rstest;
use serde_json::json;
use super::*;
fn messages(value: Value) -> Vec<ChatMessage> {
serde_json::from_value(value).unwrap()
}
#[rstest]
fn remote_urls_are_listed_once_and_inlined() {
let mut list = messages(json!([
{"role": "user", "content": [
{"type": "image_url", "image_url": "http://a/x.png"},
{"type": "image_url", "image_url": {"url": "http://a/x.png"}},
{"type": "image_url", "image_url": {"url": "data:image/png;base64,AA=="}},
{"type": "file", "file": {"file_id": "https://b/doc.pdf"}}
]},
{"role": "assistant", "content": [{"type": "image_url", "image_url": "http://c"}]}
]));
let urls = remote_urls(&list, &["user", "tool"], true, |url| {
url.starts_with("http")
});
assert_eq!(urls, vec!["http://a/x.png", "https://b/doc.pdf"]);
let downloaded = HashMap::from([(
"http://a/x.png".to_owned(),
"data:image/png;base64,Zm8=".to_owned(),
)]);
inline_media(&mut list, &downloaded);
let content = serde_json::to_value(&list[0].content).unwrap();
assert_eq!(content[0]["image_url"], "data:image/png;base64,Zm8=");
assert_eq!(content[1]["image_url"]["url"], "data:image/png;base64,Zm8=");
assert_eq!(content[3]["file"]["file_id"], "https://b/doc.pdf");
}
#[rstest]
#[case("data:image/png;base64,AAAA", Some(("image/png", "AAAA")))]
#[case("https://x", None)]
fn data_urls_split(#[case] url: &str, #[case] expected: Option<(&str, &str)>) {
assert_eq!(data_url(url), expected);
}
#[rstest]
#[case("https://x/a.PNG?x=1", "", "image/png")]
#[case("https://x/a", "image/webp; charset=x", "image/webp")]
#[case("https://x/a.pdf", "application/octet-stream", "application/pdf")]
fn media_types_fall_back_to_the_extension(
#[case] url: &str,
#[case] content_type: &str,
#[case] expected: &str,
) {
assert_eq!(downloaded_media_type(url, content_type), expected);
}
}

3
litellm-rust/crates/llms/src/base_llm/chat/mod.rs
View file

@ -1,2 +1,5 @@
pub mod media;
pub mod params;
pub mod response_format;
pub mod streaming;
pub mod transformation;

434
litellm-rust/crates/llms/src/base_llm/chat/params.rs Normal file
View file

@ -0,0 +1,434 @@
//! Python's `get_optional_params` for the providers chat translates: the request's keys are
//! sorted into OpenAI params, keys LiteLLM consumes itself, and provider-specific keys that
//! pass through; an OpenAI param the provider does not take is dropped under `drop_params`
//! and refused otherwise.
use serde_json::{Map, Value};
use crate::{
Error, ErrorDetail, base_llm::model_info::ModelInfo,
openai::chat::transformation::is_litellm_owned,
};
/// The keys of `DEFAULT_CHAT_COMPLETION_PARAM_VALUES` a request carries as OpenAI params, in
/// that table's order, which is the order Python maps them in.
pub const OPENAI_PARAMS: &[&str] = &[
"functions",
"function_call",
"temperature",
"top_p",
"n",
"stream",
"stream_options",
"stop",
"max_tokens",
"max_completion_tokens",
"modalities",
"prediction",
"audio",
"presence_penalty",
"frequency_penalty",
"logit_bias",
"user",
"response_format",
"seed",
"tools",
"tool_choice",
"max_retries",
"logprobs",
"top_logprobs",
"extra_headers",
"parallel_tool_calls",
"reasoning_effort",
"verbosity",
"thinking",
"web_search_options",
"include_server_side_tool_invocations",
"service_tier",
"safety_identifier",
"prompt_cache_key",
"prompt_cache_retention",
"store",
"metadata",
"context_management",
];
/// `PROVIDER_UNVALIDATED_PARAMS`: never refused, only left out when unsupported.
pub const UNVALIDATED_PARAMS: &[&str] = &["user", "stream_options", "stream", "max_retries"];
/// The tool Python adds when `response_format` asks for JSON and the model has no native
/// structured output.
pub const RESPONSE_FORMAT_TOOL_NAME: &str = "json_tool_call";
/// Python's `DEFAULT_MAX_TOKENS`, added to a thinking budget when the caller set no limit.
pub const DEFAULT_MAX_TOKENS: u64 = 4096;
/// The request's OpenAI params in `DEFAULT_CHAT_COMPLETION_PARAM_VALUES` order.
pub fn ordered(params: &Map<String, Value>) -> impl Iterator<Item = (&'static str, &Value)> {
OPENAI_PARAMS
.iter()
.filter_map(|name| params.get(*name).map(|value| (*name, value)))
}
/// Python's truthiness for a JSON value.
pub fn truthy(value: &Value) -> bool {
match value {
Value::Null => false,
Value::Bool(flag) => *flag,
Value::Number(number) => number.as_f64() != Some(0.0),
Value::String(text) => !text.is_empty(),
Value::Array(items) => !items.is_empty(),
Value::Object(object) => !object.is_empty(),
}
}
/// `normalize_drop_params`: a boolean, or a config's `"true"`.
pub fn request_drop_params(value: Option<Value>) -> bool {
match value {
Some(Value::Bool(flag)) => flag,
Some(Value::String(text)) => text.trim().eq_ignore_ascii_case("true"),
_ => false,
}
}
pub fn string_list(value: Option<Value>) -> Vec<String> {
match value {
Some(Value::Array(items)) => items
.into_iter()
.filter_map(|item| match item {
Value::String(text) => Some(text),
_ => None,
})
.collect(),
_ => Vec::new(),
}
}
/// What the host tells a config about the request beyond its params.
#[derive(Clone, Copy, Default)]
pub struct ChatRequestContext<'a> {
/// `drop_params` from the proxy settings or the deployment. `None` when the host cannot
/// say, as when Python hosts the call: a request Python would drop or refuse a param of
/// then goes back to Python instead.
pub drop_params: Option<bool>,
/// The model cost map, which several param rules read.
pub model_info: Option<&'a dyn ModelInfo>,
}
/// Whether a param the provider does not take is dropped, refused, or left to Python.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct DropPolicy(Option<bool>);
impl DropPolicy {
pub fn new(drop_params: Option<bool>) -> Self {
Self(drop_params)
}
/// Whether unsupported params are dropped.
pub fn drops(self) -> bool {
self.0 == Some(true)
}
/// Whether the host said what `drop_params` is.
pub fn known(self) -> bool {
self.0.is_some()
}
/// A param Python drops under `drop_params` and refuses otherwise: `Ok` means drop it.
pub fn unsupported(self, refusal: impl FnOnce() -> Error) -> Result<(), Error> {
match self.0 {
Some(true) => Ok(()),
Some(false) => Err(refusal()),
None => Err(Error::Unsupported(
"a param Python drops or refuses by drop_params",
)),
}
}
}
/// Python's `UnsupportedParamsError` for one param with its own message.
pub fn unsupported_value(message: impl Into<String>) -> Error {
Error::InvalidRequest(ErrorDetail::Message(message.into()))
}
/// A request's params as `get_optional_params` sorts them.
#[derive(Debug, Default)]
pub struct SortedParams {
/// The OpenAI params, refused or dropped ones removed.
pub openai: Map<String, Value>,
/// Every other key, sent to the provider as given.
pub provider: Map<String, Value>,
/// Keys the config consumes as call configuration.
pub config: Map<String, Value>,
/// `allowed_openai_params`: sent as given even when the provider does not list them.
pub allowed: Vec<String>,
pub drop: DropPolicy,
}
/// Sorts `params` and refuses or drops the OpenAI params `supported` leaves out.
pub fn sort_params(
params: Map<String, Value>,
config_params: &[&str],
context: &ChatRequestContext<'_>,
supported: impl FnOnce(&Map<String, Value>) -> Result<Vec<&'static str>, Error>,
provider: &'static str,
model: &str,
) -> Result<SortedParams, Error> {
let mut params = params;
let request_drop = request_drop_params(params.remove("drop_params"));
let drop = DropPolicy::new(match context.drop_params {
Some(configured) => Some(configured || request_drop),
None if request_drop => Some(true),
None => None,
});
let allowed = string_list(params.remove("allowed_openai_params"));
let additional_drop = string_list(params.remove("additional_drop_params"));
let mut sorted = SortedParams {
allowed,
drop,
..SortedParams::default()
};
for (name, value) in params {
if config_params.contains(&name.as_str()) {
sorted.config.insert(name, value);
continue;
}
if value.is_null() || is_litellm_owned(&name) || additional_drop.contains(&name) {
continue;
}
match OPENAI_PARAMS.contains(&name.as_str()) {
true => sorted.openai.insert(name, value),
false => sorted.provider.insert(name, value),
};
}
let supported = supported(&sorted.openai)?;
let unsupported: Vec<String> = sorted
.openai
.iter()
.filter(|(name, _)| {
!supported.contains(&name.as_str()) && !sorted.allowed.iter().any(|a| a == *name)
})
.map(|(name, _)| name.clone())
.collect();
let refused: Vec<String> = unsupported
.iter()
.filter(|name| {
!UNVALIDATED_PARAMS.contains(&name.as_str())
&& !(name.as_str() == "n"
&& sorted.openai.get(name.as_str()).and_then(Value::as_f64) == Some(1.0))
})
.cloned()
.collect();
if !refused.is_empty() {
drop.unsupported(|| {
Error::InvalidRequest(ErrorDetail::UnsupportedParams {
provider,
params: refused.clone(),
model: model.to_string(),
})
})?;
}
for name in unsupported {
sorted.openai.remove(&name);
}
Ok(sorted)
}
/// A capability flag the cost map sets for the first of `candidates` it knows.
///
/// Without a cost map Python's own lookups have nothing to go on either, so the answer is
/// `None`, except when the host could still fall back to Python, which does have one.
pub fn capability(
context: &ChatRequestContext<'_>,
candidates: &[String],
key: &str,
) -> Result<Option<bool>, Error> {
let Some(info) = context.model_info else {
return match context.drop_params {
None => Err(Error::Unsupported(
"a request whose translation reads the model cost map",
)),
Some(_) => Ok(None),
};
};
Ok(candidates
.iter()
.filter_map(|candidate| info.entry(candidate))
.find_map(|entry| entry.get(key).and_then(Value::as_bool)))
}
/// A numeric field of the first of `candidates` the cost map knows.
pub fn model_field(
context: &ChatRequestContext<'_>,
candidates: &[String],
key: &str,
) -> Option<Value> {
let info = context.model_info?;
candidates
.iter()
.filter_map(|candidate| info.entry(candidate))
.find_map(|entry| entry.get(key).filter(|value| !value.is_null()).cloned())
}
/// Whether the cost map has an entry for any of `candidates`.
pub fn model_known(context: &ChatRequestContext<'_>, candidates: &[String]) -> bool {
context
.model_info
.is_some_and(|info| candidates.iter().any(|c| info.entry(c).is_some()))
}
/// `json.dumps(value)` with Python's default separators and ASCII escaping.
pub fn python_dumps(value: &Value) -> String {
litellm_python_compat::json::dumps(&litellm_python_compat::json::from_json(value.clone()))
.unwrap_or_else(|_| value.to_string())
}
/// `json.loads` of tool call arguments, as `parse_tool_call_arguments` reads them: empty
/// arguments are `{}`.
pub fn parse_tool_arguments(arguments: Option<&Value>) -> Result<Value, Error> {
match arguments {
None | Some(Value::Null) => Ok(Value::Object(Map::new())),
Some(Value::String(text)) if text.trim().is_empty() => Ok(Value::Object(Map::new())),
Some(Value::String(text)) => serde_json::from_str(text).map_err(|error| {
Error::InvalidRequest(ErrorDetail::Message(format!(
"tool call arguments are not valid JSON: {error}"
)))
}),
Some(other) => Ok(other.clone()),
}
}
/// `is_thinking_enabled` over a params map.
pub fn is_thinking_enabled(params: &Map<String, Value>) -> bool {
let thinking = params.get("thinking");
thinking == Some(&Value::Bool(true))
|| thinking
.and_then(|thinking| thinking.get("type"))
.and_then(Value::as_str)
== Some("enabled")
|| params
.get("reasoning_effort")
.is_some_and(|effort| !effort.is_null())
}
/// The JSON schema a `response_format` names: `response_schema`, or `json_schema.schema`.
pub fn response_format_schema(value: &Value) -> Option<Value> {
if let Some(schema) = value.get("response_schema") {
return Some(schema.clone());
}
value
.get("json_schema")
.and_then(|json_schema| json_schema.get("schema"))
.cloned()
}
#[cfg(test)]
mod tests {
use rstest::rstest;
use serde_json::json;
use super::*;
fn object(value: Value) -> Map<String, Value> {
match value {
Value::Object(map) => map,
other => panic!("expected an object, got {other}"),
}
}
fn sort(params: Value, drop_params: Option<bool>) -> Result<SortedParams, Error> {
sort_params(
object(params),
&["aws_region_name"],
&ChatRequestContext {
drop_params,
model_info: None,
},
|_| Ok(vec!["temperature", "max_tokens"]),
"anthropic",
"claude",
)
}
#[rstest]
fn params_sort_into_openai_provider_config_and_owned() {
let sorted = sort(
json!({"temperature": 0.5, "top_k": 3, "aws_region_name": "us-east-1",
"metadata": {"x": 1}, "mock_response": "hi", "_litellm_x": 1, "user": "u", "seed": null}),
Some(false),
)
.unwrap();
assert_eq!(sorted.openai, object(json!({"temperature": 0.5})));
assert_eq!(sorted.provider, object(json!({"top_k": 3})));
assert_eq!(
sorted.config,
object(json!({"aws_region_name": "us-east-1"}))
);
}
#[rstest]
#[case::refused(Some(false), false)]
#[case::dropped(Some(true), true)]
#[case::request_drop(None, true)]
fn an_unsupported_param_is_refused_or_dropped(
#[case] drop_params: Option<bool>,
#[case] dropped: bool,
) {
let mut body = json!({"temperature": 0.5, "seed": 3});
if drop_params.is_none() {
body["drop_params"] = json!(true);
}
let result = sort(body, drop_params);
match dropped {
true => assert_eq!(result.unwrap().openai, object(json!({"temperature": 0.5}))),
false => assert!(matches!(
result,
Err(Error::InvalidRequest(ErrorDetail::UnsupportedParams { ref params, .. }))
if params == &vec!["seed".to_string()]
)),
}
}
#[rstest]
fn an_unknown_drop_policy_falls_back() {
assert!(matches!(
sort(json!({"seed": 3}), None),
Err(Error::Unsupported(_))
));
}
#[rstest]
fn allowed_and_additional_drop_params_are_honoured() {
let sorted = sort(
json!({"seed": 3, "top_k": 2, "allowed_openai_params": ["seed"], "additional_drop_params": ["top_k"]}),
Some(false),
)
.unwrap();
assert_eq!(sorted.openai, object(json!({"seed": 3})));
assert!(sorted.provider.is_empty());
assert_eq!(sorted.allowed, vec!["seed".to_string()]);
}
#[rstest]
#[case::n_of_one(json!({"n": 1}), true)]
#[case::n_of_two(json!({"n": 2}), false)]
#[case::user(json!({"user": "u"}), true)]
fn unvalidated_params_and_one_choice_are_never_refused(
#[case] body: Value,
#[case] accepted: bool,
) {
let result = sort(body, Some(false));
assert_eq!(result.is_ok(), accepted, "{result:?}");
if let Ok(sorted) = result {
assert!(sorted.openai.is_empty());
}
}
#[rstest]
fn python_dumps_matches_json_dumps() {
assert_eq!(
python_dumps(&json!({"b": "é", "a": [1, 2.5]})),
r#"{"b": "\u00e9", "a": [1, 2.5]}"#
);
}
}

445
litellm-rust/crates/llms/src/base_llm/chat/response_format.rs Normal file
View file

@ -0,0 +1,445 @@
//! `response_format` answers delivered through the `json_tool_call` tool, as Python's
//! `base_utils.py` reads them back: the schema's shape, the envelopes a model wraps the
//! answer in, and an answer split across several tool calls.
use std::collections::BTreeSet;
use serde_json::{Map, Value};
use crate::base_llm::chat::params::python_dumps;
/// `ResponseFormatSchema`: what the requested schema declares, enough to tell an answer
/// from an envelope around it.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct ResponseFormatSchema {
pub property_names: BTreeSet<String>,
pub required_names: BTreeSet<String>,
pub allows_undeclared_keys: bool,
}
impl ResponseFormatSchema {
/// `response_format_schema`: `None` unless `value` is an object whose `properties`, if
/// present, is one too.
pub fn from_schema(value: &Value) -> Option<Self> {
let document = value.as_object()?;
let properties = match document.get("properties") {
None => Map::new(),
Some(Value::Object(properties)) => properties.clone(),
Some(_) => return None,
};
let required_names = match document.get("required") {
Some(Value::Array(items)) if items.iter().all(Value::is_string) => items
.iter()
.filter_map(Value::as_str)
.map(str::to_owned)
.collect(),
_ => BTreeSet::new(),
};
let additional = document.get("additionalProperties");
let allows_undeclared_keys = additional == Some(&Value::Bool(true))
|| additional.is_some_and(Value::is_object)
|| document
.get("patternProperties")
.and_then(Value::as_object)
.is_some_and(|patterns| !patterns.is_empty());
Some(Self {
property_names: properties.keys().cloned().collect(),
required_names,
allows_undeclared_keys,
})
}
fn accepts_a_lone_undeclared_key(&self) -> bool {
self.allows_undeclared_keys && self.required_names.is_empty()
}
}
const ENVELOPE_NAME_KEYS: [&str; 2] = ["name", "tool_name"];
/// `_json_object_or_encoded_object`.
fn object_or_encoded_object(value: &Value) -> Option<Map<String, Value>> {
match value {
Value::String(text) => match serde_json::from_str(text) {
Ok(Value::Object(object)) => Some(object),
_ => None,
},
Value::Object(object) => Some(object.clone()),
_ => None,
}
}
struct EnvelopeLayer {
inner: Map<String, Value>,
names_the_tool: bool,
}
fn envelope_layer(
payload: &Map<String, Value>,
schema: &ResponseFormatSchema,
tool_name: &str,
) -> Option<EnvelopeLayer> {
let name_keys: BTreeSet<&str> = ENVELOPE_NAME_KEYS
.into_iter()
.filter(|key| payload.get(*key).and_then(Value::as_str) == Some(tool_name))
.collect();
let label_keys: BTreeSet<&str> = match (name_keys.is_empty(), schema.required_names.is_empty())
{
(false, _) => name_keys.clone(),
(true, true) => BTreeSet::new(),
(true, false) => payload
.iter()
.filter(|(key, value)| {
value.is_string()
&& !schema.property_names.contains(*key)
&& !schema.required_names.contains(*key)
})
.map(|(key, _)| key.as_str())
.collect(),
};
let wrapper_keys: Vec<&String> = payload
.keys()
.filter(|key| !label_keys.contains(key.as_str()))
.collect();
let [wrapper] = wrapper_keys.as_slice() else {
return None;
};
if schema.property_names.contains(*wrapper) {
return None;
}
let wrapped = &payload[wrapper.as_str()];
let inner = match name_keys.is_empty() {
false => object_or_encoded_object(wrapped),
true => wrapped.as_object().cloned(),
}?;
Some(EnvelopeLayer {
inner,
names_the_tool: !name_keys.is_empty(),
})
}
fn holds_the_schema_object(layer: &EnvelopeLayer, schema: &ResponseFormatSchema) -> bool {
layer.names_the_tool
|| (!schema.accepts_a_lone_undeclared_key()
&& layer
.inner
.keys()
.any(|key| schema.property_names.contains(key)))
}
fn schema_object_inside_envelope(
payload: &Map<String, Value>,
schema: &ResponseFormatSchema,
tool_name: &str,
) -> Option<Map<String, Value>> {
let mut layer = envelope_layer(payload, schema, tool_name);
let mut first = true;
while let Some(current) = layer {
if !first && schema.property_names.is_empty() {
return None;
}
first = false;
if holds_the_schema_object(&current, schema) {
return Some(current.inner);
}
layer = envelope_layer(&current.inner, schema, tool_name);
}
None
}
/// `response_format_envelope_contents`.
pub fn envelope_contents(
payload: &Value,
schema: &ResponseFormatSchema,
tool_name: &str,
) -> Option<Value> {
let object = payload.as_object()?;
if let Some(inner) = schema_object_inside_envelope(object, schema, tool_name) {
return Some(Value::Object(inner));
}
if schema.property_names.contains("values") {
return None;
}
object
.get("values")
.filter(|values| !values.is_null())
.cloned()
}
/// `unwrap_response_format_envelope`.
pub fn unwrap_envelope(payload: Value, schema: &ResponseFormatSchema, tool_name: &str) -> Value {
envelope_contents(&payload, schema, tool_name).unwrap_or(payload)
}
fn field_names(
blocks: &[Map<String, Value>],
key: &str,
schema: &ResponseFormatSchema,
) -> Option<Vec<String>> {
let names: Vec<String> = blocks
.iter()
.filter_map(|block| block[key].as_str())
.filter(|name| schema.property_names.contains(*name))
.map(str::to_owned)
.collect();
let distinct: BTreeSet<&String> = names.iter().collect();
(distinct.len() == blocks.len()).then_some(names)
}
fn is_a_label_every_block_repeats(blocks: &[Map<String, Value>], key: &str) -> bool {
blocks[0][key].is_string() && blocks.iter().all(|block| block[key] == blocks[0][key])
}
fn value_key(blocks: &[Map<String, Value>], keys: &[&String]) -> Option<String> {
if let [only] = keys {
return Some((*only).clone());
}
let varying: Vec<&&String> = keys
.iter()
.filter(|key| !is_a_label_every_block_repeats(blocks, key))
.collect();
match varying.as_slice() {
[only] => Some((**only).clone()),
_ => None,
}
}
fn split_answer_fields(
blocks: &[Map<String, Value>],
schema: &ResponseFormatSchema,
) -> Option<Map<String, Value>> {
let keys: BTreeSet<&String> = blocks[0].keys().collect();
if keys.iter().any(|key| schema.property_names.contains(*key))
|| blocks
.iter()
.any(|block| block.keys().collect::<BTreeSet<_>>() != keys)
{
return None;
}
// Python iterates a frozenset here; the outcome only depends on there being one match.
let named: Vec<(&String, Vec<String>)> = keys
.iter()
.filter_map(|key| Some((*key, field_names(blocks, key, schema)?)))
.collect();
let [(name_key, names)] = named.as_slice() else {
return None;
};
let rest: Vec<&String> = keys.iter().copied().filter(|key| key != name_key).collect();
let value_key = value_key(blocks, &rest)?;
Some(
names
.iter()
.cloned()
.zip(blocks.iter().map(|block| block[value_key.as_str()].clone()))
.collect(),
)
}
fn merged_partial_answers(
blocks: &[Map<String, Value>],
schema: &ResponseFormatSchema,
) -> Option<Map<String, Value>> {
let keys: BTreeSet<&String> = blocks.iter().flat_map(Map::keys).collect();
if !keys.iter().all(|key| schema.property_names.contains(*key))
|| keys.len() != blocks.iter().map(Map::len).sum::<usize>()
{
return None;
}
Some(
blocks
.iter()
.flat_map(|block| block.iter().map(|(k, v)| (k.clone(), v.clone())))
.collect(),
)
}
/// `split_response_format_answer`: one answer spread over several tool calls.
pub fn split_answer(
encoded_blocks: &[String],
schema: &ResponseFormatSchema,
) -> Option<Map<String, Value>> {
let blocks: Vec<Map<String, Value>> = encoded_blocks
.iter()
.map(|encoded| object_or_encoded_object(&Value::String(encoded.clone())))
.collect::<Option<_>>()?;
if blocks.len() < 2 {
return None;
}
split_answer_fields(&blocks, schema).or_else(|| merged_partial_answers(&blocks, schema))
}
/// Anthropic's `_convert_tool_response_to_message`: the message content the `json_tool_call`
/// calls' arguments become.
pub fn json_tool_answer(arguments: &[String], schema: Option<&ResponseFormatSchema>) -> String {
let first = arguments.first().cloned().unwrap_or_default();
if let Some(split) = schema.and_then(|schema| split_answer(arguments, schema)) {
return python_dumps(&Value::Object(split));
}
let Ok(payload) = serde_json::from_str::<Value>(&first) else {
return first;
};
match schema {
Some(schema) => python_dumps(&unwrap_envelope(
payload,
schema,
crate::base_llm::chat::params::RESPONSE_FORMAT_TOOL_NAME,
)),
None => {
let values = payload
.as_object()
.and_then(|object| object.get("values"))
.filter(|values| !values.is_null())
.cloned();
python_dumps(values.as_ref().unwrap_or(&payload))
}
}
}
/// `unpack_defs`: every `$ref` into `$defs` / `definitions` replaced by a copy of its
/// target, nested definitions shadowing outer ones, a circular reference left as it is.
pub fn unpack_defs(schema: &mut Value, defs: &Map<String, Value>) {
fn scoped(node: &Map<String, Value>, outer: &Map<String, Value>) -> Map<String, Value> {
let mut defs = outer.clone();
for key in ["$defs", "definitions"] {
if let Some(Value::Object(local)) = node.get(key) {
defs.extend(local.iter().map(|(k, v)| (k.clone(), v.clone())));
}
}
defs
}
fn walk(node: &mut Value, defs: &Map<String, Value>, chain: &BTreeSet<String>) {
match node {
Value::Object(object) => {
if let Some(reference) = object.get("$ref").and_then(Value::as_str) {
let name = reference.rsplit('/').next().unwrap_or(reference).to_owned();
if chain.contains(&name) {
return;
}
let Some(target) = defs.get(&name).cloned() else {
return;
};
let child_defs = match &target {
Value::Object(target) => scoped(target, defs),
_ => defs.clone(),
};
*node = target;
let mut chain = chain.clone();
chain.insert(name);
walk(node, &child_defs, &chain);
return;
}
if object.contains_key("$ref") {
return;
}
let current = scoped(object, defs);
for value in object.values_mut() {
walk(value, &current, chain);
}
}
Value::Array(items) => {
for item in items {
walk(item, defs, chain);
}
}
_ => {}
}
}
let root = match schema {
Value::Object(object) => scoped(object, defs),
_ => defs.clone(),
};
walk(schema, &root, &BTreeSet::new());
}
/// `unpack_legacy_defs`: refs into `definitions` or OpenAPI `components.schemas` inlined,
/// those blocks removed. `$defs` is left alone.
pub fn unpack_legacy_defs(schema: &mut Map<String, Value>) {
let has_components = schema
.get("components")
.and_then(|components| components.get("schemas"))
.is_some();
if !schema.contains_key("definitions") && !has_components {
return;
}
let mut defs = match schema.remove("components") {
Some(Value::Object(mut components)) => match components.remove("schemas") {
Some(Value::Object(schemas)) => schemas,
_ => Map::new(),
},
_ => Map::new(),
};
if let Some(Value::Object(definitions)) = schema.remove("definitions") {
defs.extend(definitions);
}
let mut value = Value::Object(std::mem::take(schema));
unpack_defs(&mut value, &defs);
if let Value::Object(object) = value {
*schema = object;
}
}
#[cfg(test)]
mod tests {
use rstest::rstest;
use serde_json::json;
use super::*;
fn schema(value: Value) -> ResponseFormatSchema {
ResponseFormatSchema::from_schema(&value).unwrap()
}
#[rstest]
#[case::bare(json!({"name": "a"}), json!({"name": "a"}))]
#[case::named_envelope(json!({"name": "json_tool_call", "arguments": {"name": "a"}}), json!({"name": "a"}))]
#[case::encoded_envelope(json!({"tool_name": "json_tool_call", "parameters": "{\"name\": \"a\"}"}), json!({"name": "a"}))]
#[case::wrapper(json!({"answer": {"name": "a"}}), json!({"name": "a"}))]
#[case::values(json!({"values": {"x": 1}}), json!({"x": 1}))]
fn envelopes_are_unwrapped(#[case] payload: Value, #[case] expected: Value) {
let schema = schema(
json!({"type": "object", "properties": {"name": {"type": "string"}}, "required": ["name"]}),
);
assert_eq!(
unwrap_envelope(payload, &schema, "json_tool_call"),
expected
);
}
#[rstest]
fn an_answer_split_over_named_fields_is_joined() {
let schema = schema(json!({"properties": {"a": {}, "b": {}}}));
let blocks = vec![
r#"{"field": "a", "value": 1}"#.to_owned(),
r#"{"field": "b", "value": 2}"#.to_owned(),
];
assert_eq!(
split_answer(&blocks, &schema),
Some(json!({"a": 1, "b": 2}).as_object().unwrap().clone())
);
let partial = vec![r#"{"a": 1}"#.to_owned(), r#"{"b": 2}"#.to_owned()];
assert_eq!(
json_tool_answer(&partial, Some(&schema)),
r#"{"a": 1, "b": 2}"#
);
}
#[rstest]
fn an_undecodable_answer_is_returned_as_sent() {
assert_eq!(json_tool_answer(&["{bad".to_owned()], None), "{bad");
assert_eq!(
json_tool_answer(&[r#"{"values": [1]}"#.to_owned()], None),
"[1]"
);
}
#[rstest]
fn refs_are_inlined_and_cycles_left_alone() {
let mut value =
json!({"properties": {"a": {"$ref": "#/$defs/A"}, "n": {"$ref": "#/$defs/Node"}}});
let defs = json!({"A": {"type": "string"}, "Node": {"properties": {"next": {"$ref": "#/$defs/Node"}}}});
unpack_defs(&mut value, defs.as_object().unwrap());
assert_eq!(
value,
json!({"properties": {"a": {"type": "string"}, "n": {"properties": {"next": {"$ref": "#/$defs/Node"}}}}})
);
}
}

8
litellm-rust/crates/llms/src/base_llm/chat/streaming.rs
View file

@ -1,5 +1,7 @@
use std::collections::HashMap;
use crate::base_llm::chat::response_format::ResponseFormatSchema;
use bytes::Bytes;
use futures_util::{StreamExt, stream::BoxStream};
use litellm_llms_types::formats::chat_completions::ChatCompletionChunk;
@ -35,6 +37,12 @@ pub struct StreamShape {
pub json_mode: bool,
pub speed: Option<String>,
pub tool_name_reverse_map: HashMap<String, String>,
/// The schema of the `json_tool_call` tool a `response_format` added, which tells its
/// answer apart from an envelope around it.
pub response_format_schema: Option<ResponseFormatSchema>,
/// The text of a trailing assistant message sent with `prefix: true`, which Anthropic
/// continues and Python puts back in front of the answer.
pub prefix_prompt: Option<String>,
}
/// A wire decoder paired with the iterator that turns its events into chat chunks.

23
litellm-rust/crates/llms/src/base_llm/chat/transformation.rs
View file

@ -5,7 +5,10 @@ use serde_json::{Map, Value};
use crate::{
Error,
base_llm::chat::streaming::{ChatStream, StreamShape},
base_llm::chat::{
params::ChatRequestContext,
streaming::{ChatStream, StreamShape},
},
};
/// The provider-shaped request body a config produces. Named rather than a bare
@ -14,6 +17,8 @@ use crate::{
pub struct ProviderChatRequestData {
pub body: Value,
pub stream_shape: StreamShape,
/// Headers the request needs beyond the config's defaults, such as `anthropic-beta`.
pub headers: Vec<(String, String)>,
}
/// The raw provider response body handed back to a config for normalization.
@ -42,8 +47,11 @@ pub struct Unsupported(pub &'static str);
pub trait BaseConfig: Sync {
fn secret_names(&self) -> Vec<&'static str>;
/// Supported OpenAI parameter names paired with their provider names.
fn supported_openai_param_mappings(&self) -> &'static [(&'static str, &'static str)];
/// Supported OpenAI parameter names paired with their provider names, for a config
/// whose requests the host checks against an allowlist.
fn supported_openai_param_mappings(&self) -> &'static [(&'static str, &'static str)] {
&[]
}
fn get_complete_url(
&self,
@ -58,14 +66,23 @@ pub trait BaseConfig: Sync {
model: &str,
messages: Vec<ChatMessage>,
optional_params: Map<String, Value>,
context: &ChatRequestContext<'_>,
) -> Result<ProviderChatRequestData, Error>;
/// `shape` is what the request decided about its answer, the same one a stream reads.
fn transform_response(
&self,
model: &str,
response: ProviderChatResponseData,
shape: &StreamShape,
) -> Result<ChatCompletionsResponse, Error>;
/// The media URLs the messages point at that Python downloads while building this
/// provider's request. The host fetches them and inlines each as a data URL first.
fn remote_media(&self, _messages: &[ChatMessage]) -> Vec<String> {
Vec::new()
}
/// `None` means this config has no streaming path yet, so the host keeps the request.
fn model_response_iterator(&self, _shape: StreamShape) -> Option<ChatStream> {
None

14
litellm-rust/crates/llms/src/base_llm/model_info.rs
View file

@ -1,11 +1,14 @@
//! What the model cost map says about a model, for configs whose request rules depend on it
//! the way Python's `supports_*` helpers read `litellm.model_cost`.
use std::borrow::Cow;
use serde_json::{Map, Value};
pub trait ModelInfo: Send + Sync {
/// The cost map entry `model` names, if there is one.
fn entry(&self, model: &str) -> Option<&Map<String, Value>>;
/// The cost map entry `model` names, if there is one. A map held in memory lends it; one
/// read from elsewhere, such as Python's `litellm.model_cost`, hands over a copy.
fn entry(&self, model: &str) -> Option<Cow<'_, Map<String, Value>>>;
/// Whether any entry of the map sets `key`.
fn declares(&self, key: &str) -> bool;
@ -14,21 +17,22 @@ pub trait ModelInfo: Send + Sync {
/// Python's `_supports_factory`: the entry sets `key` to `true`. An unknown model supports
/// nothing.
pub fn supports(info: Option<&dyn ModelInfo>, model: &str, key: &str) -> bool {
field(info, model, key).and_then(Value::as_bool) == Some(true)
field(info, model, key).and_then(|value| value.as_bool()) == Some(true)
}
/// Python's `is_explicitly_disabled_factory`: the entry sets `key` to `false`. An unknown
/// model or an unset key is not disabled.
pub fn explicitly_disabled(info: Option<&dyn ModelInfo>, model: &str, key: &str) -> bool {
field(info, model, key).and_then(Value::as_bool) == Some(false)
field(info, model, key).and_then(|value| value.as_bool()) == Some(false)
}
/// The value the entry for `model` sets for `key`.
pub fn field<'a>(info: Option<&'a dyn ModelInfo>, model: &str, key: &str) -> Option<&'a Value> {
pub fn field(info: Option<&dyn ModelInfo>, model: &str, key: &str) -> Option<Value> {
info?
.entry(model)?
.get(key)
.filter(|value| !value.is_null())
.cloned()
}
/// Whether the map has an entry for `model`.

1343
litellm-rust/crates/llms/src/bedrock/chat/converse_messages.rs Normal file
View file

File diff suppressed because it is too large Load diff

1039
litellm-rust/crates/llms/src/bedrock/chat/converse_params.rs Normal file
View file

File diff suppressed because it is too large Load diff

1151
litellm-rust/crates/llms/src/bedrock/chat/converse_transformation.rs
View file

File diff suppressed because it is too large Load diff

82
litellm-rust/crates/llms/src/bedrock/chat/invoke_handler.rs
View file

@ -23,9 +23,13 @@ use crate::{
anthropic::chat::handler::ModelResponseIterator,
base_llm::{
base_model_iterator::StreamTransformer,
chat::streaming::{ChatStream, StreamShape},
chat::{
params::RESPONSE_FORMAT_TOOL_NAME,
streaming::{ChatStream, StreamShape},
},
messages::streaming::{ByteStream, EventStream},
},
bedrock::chat::converse_transformation::unwrap_bedrock_properties,
};
#[derive(Deserialize)]
@ -200,7 +204,13 @@ pub fn converse_event_stream(bytes: ByteStream) -> BoxStream<'static, Result<Val
/// ConverseStream events to Chat Completions chunks, as Python's `AWSEventStreamDecoder`
/// `converse_chunk_parser` builds them.
///
/// DIVERGENCE: `json_mode` tool conversion is not done (the route never asks for it),
/// With `json_mode` the `json_tool_call` tool a `response_format` became is streamed back as
/// text: its argument fragments are held and the whole answer (with a lone `properties`
/// envelope unwrapped) goes out as content when the block stops, and a `tool_use` stop with
/// no other tool call finishes as `stop`.
///
/// DIVERGENCE: Python fake-streams a json_mode Converse call (one non-streaming request
/// replayed as chunks), so its chunk boundaries differ; the content and finish reason match.
/// `redactedContent` is passed through without unwrapping, and the reasoning token estimate
/// Python makes from streamed thinking text is not made.
pub struct ConverseStreamIterator {
@ -211,6 +221,10 @@ pub struct ConverseStreamIterator {
thinking_ran: bool,
thinking_block_index: i64,
thinking_content_block_index: Option<u64>,
/// The block in progress is the `json_tool_call` tool a `response_format` became.
in_response_format_tool: bool,
/// Some `json_tool_call` block has been streamed back as text.
converted_response_format_tool: bool,
}
#[derive(Default)]
@ -233,6 +247,8 @@ impl ConverseStreamIterator {
thinking_ran: false,
thinking_block_index: -1,
thinking_content_block_index: None,
in_response_format_tool: false,
converted_response_format_tool: false,
}
}
@ -295,7 +311,15 @@ impl ConverseStreamIterator {
fn start(&mut self, start: &Value, index: u64) -> ConverseParsed {
let mut parsed = ConverseParsed::default();
self.content_blocks.clear();
self.in_response_format_tool = false;
if let Some(tool) = start.get("toolUse").filter(|tool| !tool.is_null()) {
if self.shape.json_mode
&& tool.get("name").and_then(Value::as_str) == Some(RESPONSE_FORMAT_TOOL_NAME)
{
self.in_response_format_tool = true;
self.converted_response_format_tool = true;
return parsed;
}
self.tool_calls_index = Some(self.tool_calls_index.map_or(0, |index| index + 1));
let name = tool.get("name").and_then(Value::as_str).map(|name| {
self.shape
@ -329,6 +353,9 @@ impl ConverseStreamIterator {
.unwrap_or_default()
.to_owned();
self.content_blocks.push(input.clone());
if self.in_response_format_tool {
return parsed;
}
parsed.tool_use = Some(self.tool_call(None, None, input, index));
} else if let Some(reasoning) = delta.get("reasoningContent") {
parsed
@ -359,6 +386,13 @@ impl ConverseStreamIterator {
self.start(start, index)
} else if let Some(delta) = data.get("delta") {
self.delta(delta, index)
} else if data.get("contentBlockIndex").is_some() && self.in_response_format_tool {
self.in_response_format_tool = false;
let answer = std::mem::take(&mut self.content_blocks).concat();
ConverseParsed {
text: unwrap_bedrock_properties(if answer.is_empty() { "{}" } else { &answer }),
..ConverseParsed::default()
}
} else if data.get("contentBlockIndex").is_some() {
// A block's stop: a tool that streamed only empty fragments closes with `{}`.
let empty =
@ -368,10 +402,15 @@ impl ConverseStreamIterator {
..ConverseParsed::default()
}
} else if let Some(reason) = data.get("stopReason") {
let mut finish = finish_reason_for(reason.as_str().unwrap_or("stop"));
if finish == "tool_calls"
&& self.converted_response_format_tool
&& self.tool_calls_index.is_none()
{
finish = "stop";
}
ConverseParsed {
finish_reason: Some(
finish_reason_for(reason.as_str().unwrap_or("stop")).to_owned(),
),
finish_reason: Some(finish.to_owned()),
..ConverseParsed::default()
}
} else if let Some(usage) = data.get("usage") {
@ -735,6 +774,39 @@ mod tests {
assert_eq!(chunks[7]["choices"][0]["finish_reason"], "tool_calls");
}
#[test]
fn a_response_format_tool_streams_back_as_text() {
let mut iterator = ConverseStreamIterator::new(StreamShape {
json_mode: true,
..StreamShape::default()
});
let chunks: Vec<serde_json::Value> = serde_json::json!([
{"contentBlockIndex": 0, "start": {"toolUse": {"toolUseId": "t1", "name": "json_tool_call"}}},
{"contentBlockIndex": 0, "delta": {"toolUse": {"input": "{\"properties\": {\"a\""}}},
{"contentBlockIndex": 0, "delta": {"toolUse": {"input": ": 1}}"}}},
{"contentBlockIndex": 0},
{"stopReason": "tool_use"}
])
.as_array()
.unwrap()
.iter()
.flat_map(|event| iterator.transform(event.clone()).unwrap())
.map(|chunk| serde_json::to_value(chunk).unwrap())
.collect();
for chunk in &chunks {
assert_eq!(
chunk["choices"][0]["delta"]["tool_calls"],
serde_json::Value::Null
);
}
let content: String = chunks
.iter()
.filter_map(|chunk| chunk["choices"][0]["delta"]["content"].as_str())
.collect();
assert_eq!(content, "{\"a\": 1}");
assert_eq!(chunks[4]["choices"][0]["finish_reason"], "stop");
}
#[test]
fn converse_reasoning_streams_as_thinking_blocks() {
let chunks = converse_chunks(serde_json::json!([

2
litellm-rust/crates/llms/src/bedrock/chat/mod.rs
View file

@ -1,2 +1,4 @@
pub mod converse_messages;
pub mod converse_params;
pub mod converse_transformation;
pub mod invoke_handler;

23
litellm-rust/crates/llms/src/openai_like/chat/transformation.rs
View file

@ -16,9 +16,13 @@ use crate::{
Error,
base_llm::{
auth::AuthScheme,
chat::transformation::{
BaseConfig, Headers, ProviderChatRequestData, ProviderChatResponseData,
ValidatedEnvironment,
chat::{
params::ChatRequestContext,
streaming::StreamShape,
transformation::{
BaseConfig, Headers, ProviderChatRequestData, ProviderChatResponseData,
ValidatedEnvironment,
},
},
},
openai_like::common_utils::{complete_openai_like_url, openai_compatible_provider_info},
@ -90,6 +94,7 @@ impl BaseConfig for OpenAILikeChatConfig {
model: &str,
messages: Vec<ChatMessage>,
optional_params: Map<String, Value>,
_context: &ChatRequestContext<'_>,
) -> Result<ProviderChatRequestData, Error> {
let mut params = Map::from_iter(
optional_params
@ -113,6 +118,7 @@ impl BaseConfig for OpenAILikeChatConfig {
Ok(ProviderChatRequestData {
body: Value::Object(body),
stream_shape: Default::default(),
headers: Vec::new(),
})
}
@ -120,6 +126,7 @@ impl BaseConfig for OpenAILikeChatConfig {
&self,
model: &str,
response: ProviderChatResponseData,
_shape: &StreamShape,
) -> Result<ChatCompletionsResponse, Error> {
let mut body = response.body;
sanitize_usage(&mut body);
@ -266,6 +273,7 @@ fn normalize_choice(position: usize, choice: &Value) -> Result<ChatCompletionsCh
.unwrap_or("assistant")
.to_string(),
content: content.and_then(Value::as_str).map(str::to_string),
..ChatCompletionsChoiceMessage::default()
},
finish_reason: choice
.get("finish_reason")
@ -310,14 +318,17 @@ mod tests {
fn transform(model: &str, msgs: Value, opts: Value) -> Value {
OPENAI_LIKE_CHAT_COMPLETIONS_CONFIG
.transform_request(model, messages(msgs), params(opts))
.transform_request(model, messages(msgs), params(opts), &Default::default())
.expect("request transforms")
.body
}
fn transform_response(body: Value) -> Result<ChatCompletionsResponse, Error> {
OPENAI_LIKE_CHAT_COMPLETIONS_CONFIG
.transform_response("some-model", ProviderChatResponseData { body })
OPENAI_LIKE_CHAT_COMPLETIONS_CONFIG.transform_response(
"some-model",
ProviderChatResponseData { body },
&Default::default(),
)
}
fn reason(msgs: Value, opts: Value) -> Option<Unsupported> {

7
litellm-rust/crates/python-bridge/src/routes/chat_completions/host.rs
View file

@ -1,5 +1,6 @@
use std::convert::Infallible;
use std::{convert::Infallible, sync::Arc};
use super::model_cost::PythonModelCost;
use crate::routes::inference::InferenceHost;
use litellm_host_python::{InvokeError, PythonBinding, PythonHostCalls, PythonOwned};
use litellm_inference_chat::{Error, route::ChatCompletions, types::ChatCompletionsCall};
@ -26,6 +27,10 @@ pub(super) fn project(
custom_llm_provider: call.options.custom_llm_provider,
extra_headers: call.options.extra_headers,
timeout: call.options.timeout,
// Python's own settings decide what a drop would do, so a param that needs one goes
// back to Python. Capabilities come from the cost map Python itself reads.
drop_params: None,
model_info: Some(Arc::new(PythonModelCost::current(py)?)),
})
}

1
litellm-rust/crates/python-bridge/src/routes/chat_completions/mod.rs
View file

@ -1,4 +1,5 @@
mod host;
mod model_cost;
use std::sync::Arc;

88
litellm-rust/crates/python-bridge/src/routes/chat_completions/model_cost.rs Normal file
View file

@ -0,0 +1,88 @@
//! Python's `litellm.model_cost` as the cost map chat translation reads, so a capability
//! lookup sees the map Python itself would, models registered at runtime included.
use std::{borrow::Cow, collections::HashSet, sync::OnceLock};
use litellm_host_python::from_py;
use litellm_llms::base_llm::model_info::ModelInfo;
use pyo3::{prelude::*, types::PyDict};
use serde_json::{Map, Value};
pub(super) struct PythonModelCost {
map: Py<PyDict>,
declared: OnceLock<HashSet<String>>,
}
impl PythonModelCost {
/// The dict `litellm.model_cost` holds now. A map Python replaces later is not followed,
/// as a call already under way would not follow it either.
pub(super) fn current(py: Python<'_>) -> PyResult<Self> {
let map = py
.import("litellm")?
.getattr("model_cost")?
.cast_into::<PyDict>()?
.unbind();
Ok(Self {
map,
declared: OnceLock::new(),
})
}
}
impl ModelInfo for PythonModelCost {
fn entry(&self, model: &str) -> Option<Cow<'_, Map<String, Value>>> {
Python::attach(|py| {
let entry = self.map.bind(py).get_item(model).ok()??;
from_py::<Map<String, Value>>(&entry).ok().map(Cow::Owned)
})
}
fn declares(&self, key: &str) -> bool {
self.declared
.get_or_init(|| {
Python::attach(|py| {
self.map
.bind(py)
.values()
.iter()
.filter_map(|entry| entry.cast_into::<PyDict>().ok())
.flat_map(|entry| {
entry
.keys()
.iter()
.filter_map(|key| key.extract::<String>().ok())
.collect::<Vec<_>>()
})
.collect()
})
})
.contains(key)
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn reads_the_entry_python_holds_and_the_keys_any_entry_sets() {
Python::initialize();
Python::attach(|py| {
let map = PyDict::new(py);
let entry = PyDict::new(py);
entry.set_item("supports_reasoning", true).unwrap();
entry.set_item("max_output_tokens", 64000).unwrap();
map.set_item("claude-x", entry).unwrap();
let cost = PythonModelCost {
map: map.unbind(),
declared: OnceLock::new(),
};
let found = cost.entry("claude-x").unwrap();
assert_eq!(found.get("supports_reasoning"), Some(&Value::Bool(true)));
assert_eq!(found.get("max_output_tokens"), Some(&Value::from(64000)));
assert!(cost.entry("unknown").is_none());
assert!(cost.declares("max_output_tokens"));
assert!(!cost.declares("default_reasoning_effort"));
});
}
}

4
tests/test_litellm_rust/chat_completions/test_callbacks.py
View file

@ -22,13 +22,13 @@ _OBJECT: Final = TypeAdapter(dict[str, object])
async def test_returns_a_model_response_and_logs_success_once(server: RecordingServer, asynchronous: bool) -> None:
recorder: Final = RecordingLogger()
result: Final = await call(server, asynchronous=asynchronous, callbacks=[recorder], temperature=0.25)
result: Final = await call(server, asynchronous=asynchronous, callbacks=[recorder], temperature=1)
assert_served_natively(server)
sent: Final = server.requests[0]
body: Final = _OBJECT.validate_python(sent.body)
assert sent.path == "/v1/messages"
assert body["temperature"] == 0.25
assert body["temperature"] == 1
assert request_body(_OBJECT.validate_python(recorder.wait_for("log_pre_api_call")[0].kwargs)) == body
assert isinstance(result, ModelResponse)
assert result.choices[0].message.content == MESSAGES_TEXT

3
tests/test_litellm_rust/chat_completions/test_requests.py
View file

@ -18,7 +18,8 @@ _OBJECT: Final = TypeAdapter(dict[str, object])
@pytest.mark.asyncio
@pytest.mark.parametrize("asynchronous", (False, True))
async def test_positional_parameters_reach_the_wire(server: RecordingServer, asynchronous: bool) -> None:
temperature: Final = 0.35
# The model takes no sampling params but temperature=1, as Python itself rules.
temperature: Final = 1.0
arguments: Final = (
(MESSAGES_MODEL, list(MESSAGES), None, None, 12.0, temperature)
if asynchronous