diff --git a/litellm-rust/crates/inference-messages/src/handler.rs b/litellm-rust/crates/inference-messages/src/handler.rs index 443275bf208..6a8ed7fff8e 100644 --- a/litellm-rust/crates/inference-messages/src/handler.rs +++ b/litellm-rust/crates/inference-messages/src/handler.rs @@ -26,10 +26,10 @@ use litellm_llms_types::formats::messages::{ }; use litellm_secrets::source::Secrets; use litellm_tracing::ByteChunk; -use serde_json::Value; +use serde_json::{Map, Value}; use super::{ - Error, MessagesCall, MessagesCallResponse, MessagesRoute, + Error, MessagesBody, MessagesCall, MessagesCallResponse, MessagesRoute, common_utils::{MessagesProvider, merge_messages_header_sources, string_headers}, constants::MESSAGES_TIMEOUT_SECS, route::Messages, @@ -85,7 +85,10 @@ pub(super) async fn execute( ), }; let reported = match &output { - MessagesCallResponse::Complete(response) => provider.config().reported_cost(response), + MessagesCallResponse::Complete(response) => match response { + MessagesBody::Message(message) => provider.config().reported_cost(message), + MessagesBody::Unchecked(_) => None, + }, MessagesCallResponse::Stream { .. } => None, }; context @@ -347,8 +350,7 @@ async fn call_provider( // body it cannot parse, and the public mapping classifies it by that status. return Err(litellm_http::Error::upstream(status, text).into()); } - decode_response(config, &identity.model, &text) - .map(|message| MessagesCallResponse::Complete(Box::new(message))) + decode_response(config, &identity.model, &text).map(MessagesCallResponse::Complete) } fn serialize_failure(err: serde_json::Error) -> Error { @@ -419,15 +421,23 @@ fn decode_response( config: &dyn BaseMessagesConfig, model: &str, text: &str, -) -> Result { - let response = serde_json::from_str(text).map_err(|err| { - Error::invalid_response(litellm_llms::ErrorDetail::invalid( - "messages response JSON", - err, - )) - })?; +) -> Result { + let response = match serde_json::from_str::(text) { + Ok(response) => response, + Err(err) => { + let Ok(mut body) = serde_json::from_str::>(text) else { + return Err(Error::invalid_response(litellm_llms::ErrorDetail::invalid( + "messages response JSON", + err, + ))); + }; + body.entry("stop_sequence").or_insert(Value::Null); + return Ok(MessagesBody::Unchecked(body)); + } + }; config .transform_anthropic_messages_response(model, response) + .map(|message| MessagesBody::Message(Box::new(message))) .map_err(Error::from) .map_err(Error::after_send) } diff --git a/litellm-rust/crates/inference-messages/src/lib.rs b/litellm-rust/crates/inference-messages/src/lib.rs index 745442d0314..25ebeaacb49 100644 --- a/litellm-rust/crates/inference-messages/src/lib.rs +++ b/litellm-rust/crates/inference-messages/src/lib.rs @@ -17,8 +17,8 @@ use crate::common_utils::messages_provider; pub use common_utils::merge_messages_header_sources; pub use litellm_inference::Error; pub use types::{ - MessagesCall, MessagesCallResponse, MessagesSettings, MessagesShaping, litellm_params, - messages_body, + MessagesBody, MessagesCall, MessagesCallResponse, MessagesSettings, MessagesShaping, + litellm_params, messages_body, }; #[derive(Clone)] diff --git a/litellm-rust/crates/inference-messages/src/route.rs b/litellm-rust/crates/inference-messages/src/route.rs index 7ff886689da..5f06d16d49f 100644 --- a/litellm-rust/crates/inference-messages/src/route.rs +++ b/litellm-rust/crates/inference-messages/src/route.rs @@ -5,11 +5,10 @@ use litellm_host::{ call::{HostedCompletion, HostedMachine, hosted_call}, protocol::Protocol, }; -use litellm_llms_types::formats::messages::MessagesResponse; -use super::{Error, MessagesCall}; +use super::{Error, MessagesBody, MessagesCall}; -pub type MessagesOutput = HostedCompletion>; +pub type MessagesOutput = HostedCompletion; /// The upstream response as the caller sees it at stream hand-off, before any chunk. pub struct MessagesStreamHead { @@ -19,7 +18,7 @@ pub struct MessagesStreamHead { pub struct Messages; impl Protocol for Messages { - type Response = Box; + type Response = MessagesBody; type Error = Error; type Request = MessagesCall; type HostCall = Infallible; diff --git a/litellm-rust/crates/inference-messages/src/types.rs b/litellm-rust/crates/inference-messages/src/types.rs index ae346259559..7c183f2bcc7 100644 --- a/litellm-rust/crates/inference-messages/src/types.rs +++ b/litellm-rust/crates/inference-messages/src/types.rs @@ -42,7 +42,16 @@ pub(super) fn invalid_request(err: serde_json::Error) -> Error { } pub type MessagesCallResponse = - CallOutput, super::route::MessagesStreamHead, Bytes, Error>; + CallOutput; + +// Python's `transform_anthropic_messages_response` returns any JSON object the provider sends +// with a 200, message or not, so one that is not a message passes through as sent. +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +#[serde(untagged)] +pub enum MessagesBody { + Message(Box), + Unchecked(Map), +} #[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)] pub struct MessagesShaping { diff --git a/litellm-rust/crates/inference-messages/tests/bedrock_native/main.rs b/litellm-rust/crates/inference-messages/tests/bedrock_native/main.rs index f91d5c79f7c..5207f4e9d58 100644 --- a/litellm-rust/crates/inference-messages/tests/bedrock_native/main.rs +++ b/litellm-rust/crates/inference-messages/tests/bedrock_native/main.rs @@ -1,7 +1,9 @@ use std::{sync::Arc, time::Duration}; use futures_util::TryStreamExt; -use litellm_inference_messages::{Error, MessagesCall, MessagesCallResponse, MessagesShaping}; +use litellm_inference_messages::{ + Error, MessagesBody, MessagesCall, MessagesCallResponse, MessagesShaping, +}; use litellm_inference_testing::{RecordingSecrets, no_secrets}; use litellm_llms_types::formats::messages::{MessagesRequest, MessagesResponse}; use rstest::{fixture, rstest}; @@ -74,7 +76,10 @@ async fn run(call: MessagesCall) -> Result { async fn run_message(call: MessagesCall) -> MessagesResponse { match run(call).await.unwrap() { - MessagesCallResponse::Complete(message) => *message, + MessagesCallResponse::Complete(message) => match message { + MessagesBody::Message(message) => *message, + MessagesBody::Unchecked(_) => panic!("expected a message"), + }, MessagesCallResponse::Stream { .. } => panic!("expected a completed message"), } } diff --git a/litellm-rust/crates/inference-messages/tests/caching.rs b/litellm-rust/crates/inference-messages/tests/caching.rs index 663c87453a9..787501a8fed 100644 --- a/litellm-rust/crates/inference-messages/tests/caching.rs +++ b/litellm-rust/crates/inference-messages/tests/caching.rs @@ -50,42 +50,47 @@ async fn messages_cache_identity_includes_provider_native_parameters( #[case] original: Value, #[case] changed: Value, ) { - use litellm_inference_messages::route::Messages; + use litellm_inference_messages::{MessagesBody, route::Messages}; use litellm_llms_types::formats::messages::MessagesResponse; let calls = AtomicUsize::new(0); for (value, expected_call) in [(original.clone(), 0), (changed, 1), (original, 0)] { - let response = - execute_unary::( - CacheRequest::from_wire( - ProviderIdentity { - model: "test".into(), - provider: "anthropic".into(), - }, - Some(&WireRequest { - url: "https://example.test/v1/messages".into(), - headers: vec![], - body: json!({ - "model":"test", "messages":[{"role":"user","content":"hello"}], - "max_tokens":32, (field):value - }), - }), - ), - Some(cache.clone()), - Some(CacheOptions::new(CacheScope::Shared)), - &(), - None, - || async { - let call = calls.fetch_add(1, Ordering::SeqCst); - Ok(Box::new(serde_json::from_value::(json!({ - "id":call.to_string(), "type":"message", "role":"assistant", "model":"test", - "content":[{"type":"text","text":format!("answer {call}")}], - "stop_reason":"end_turn", "stop_sequence":null - })).unwrap())) + let response = execute_unary::( + CacheRequest::from_wire( + ProviderIdentity { + model: "test".into(), + provider: "anthropic".into(), }, - ) - .await - .unwrap(); + Some(&WireRequest { + url: "https://example.test/v1/messages".into(), + headers: vec![], + body: json!({ + "model":"test", "messages":[{"role":"user","content":"hello"}], + "max_tokens":32, (field):value + }), + }), + ), + Some(cache.clone()), + Some(CacheOptions::new(CacheScope::Shared)), + &(), + None, + || async { + let call = calls.fetch_add(1, Ordering::SeqCst); + Ok(MessagesBody::Message(Box::new( + serde_json::from_value::(json!({ + "id":call.to_string(), "type":"message", "role":"assistant", "model":"test", + "content":[{"type":"text","text":format!("answer {call}")}], + "stop_reason":"end_turn", "stop_sequence":null + })) + .unwrap(), + ))) + }, + ) + .await + .unwrap(); + let MessagesBody::Message(response) = response else { + panic!("expected a message"); + }; assert_eq!(response.id, expected_call.to_string()); assert_eq!( response.content[0]["text"], @@ -253,7 +258,7 @@ async fn provider_reported_cost_is_delivered_once_and_never_charged_for_a_cache_ #[case] cost: Value, #[case] expected: Option, ) { - use litellm_inference_messages::{MessagesCall, MessagesCallResponse}; + use litellm_inference_messages::{MessagesBody, MessagesCall, MessagesCallResponse}; let response = json!({"id": "msg_cost", "type": "message", "role": "assistant", "model": "native-test", "content": [{"type": "text", "text": "answer"}], "stop_reason": "end_turn", "stop_sequence": null, "usage": {"input_tokens": 3, "output_tokens": 2}, "cost": cost}); @@ -277,6 +282,9 @@ async fn provider_reported_cost_is_delivered_once_and_never_charged_for_a_cache_ panic!("expected Messages responses"); }; assert_eq!(first, second); + let MessagesBody::Message(first) = first else { + panic!("expected a message"); + }; assert_eq!(first.extra.get("cost"), Some(&cost)); assert_eq!(support::received(&upstream).await.len(), 1); let facts = hooks.facts.lock().unwrap(); diff --git a/litellm-rust/crates/inference-messages/tests/live/support.rs b/litellm-rust/crates/inference-messages/tests/live/support.rs index 81e295e34ad..52684928f1f 100644 --- a/litellm-rust/crates/inference-messages/tests/live/support.rs +++ b/litellm-rust/crates/inference-messages/tests/live/support.rs @@ -4,7 +4,9 @@ use futures_util::{FutureExt, TryStreamExt}; use litellm_host::interceptors::{ ExecutionFacts, Interceptors, RawResponse, RequestContext, ResultSource, WireRequest, }; -use litellm_inference_messages::{Error, MessagesCall, MessagesCallResponse, MessagesRoute}; +use litellm_inference_messages::{ + Error, MessagesBody, MessagesCall, MessagesCallResponse, MessagesRoute, +}; use litellm_inference_testing::live::{LiveResources, model, within_deadline}; use litellm_llms_types::formats::messages::MessagesResponse; use serde_json::{Value, json}; @@ -111,7 +113,10 @@ pub(super) fn call(provider: &str, payload: Value) -> MessagesCall { pub(super) fn complete(response: MessagesCallResponse) -> Box { match response { - MessagesCallResponse::Complete(message) => message, + MessagesCallResponse::Complete(message) => match message { + MessagesBody::Message(message) => message, + MessagesBody::Unchecked(_) => panic!("a complete request returned no message"), + }, MessagesCallResponse::Stream { .. } => panic!("a complete request returned a stream"), } } diff --git a/litellm-rust/crates/inference-messages/tests/messages/host.rs b/litellm-rust/crates/inference-messages/tests/messages/host.rs index e8df20e7dcb..20f6a97aa37 100644 --- a/litellm-rust/crates/inference-messages/tests/messages/host.rs +++ b/litellm-rust/crates/inference-messages/tests/messages/host.rs @@ -6,7 +6,7 @@ use litellm_host::{ interceptors::{ExecutionFacts, RequestContext, ResultSource, WireRequest}, lifecycle::CallEvent, }; -use litellm_inference_messages::{MessagesCallResponse, route::Messages}; +use litellm_inference_messages::{MessagesBody, MessagesCallResponse, route::Messages}; use litellm_inference_testing::{RecordingSecrets, no_secrets}; use litellm_llms::base_llm::messages::context::{ MessagesModelCapabilities as AnthropicModelCapabilities, SupportedEffortTiers, @@ -340,7 +340,10 @@ async fn response_mode_follows_the_intercepted_request( } MessagesCallResponse::Complete(message) => { assert_eq!(expected_stream, Some(false)); - assert_eq!(*message, serde_json::from_value(message_body()).unwrap()); + assert_eq!( + message, + MessagesBody::Message(serde_json::from_value(message_body()).unwrap()) + ); } } Ok::<_, Error>(()) diff --git a/litellm-rust/crates/inference-messages/tests/messages/main.rs b/litellm-rust/crates/inference-messages/tests/messages/main.rs index a01e571ec75..fe203287766 100644 --- a/litellm-rust/crates/inference-messages/tests/messages/main.rs +++ b/litellm-rust/crates/inference-messages/tests/messages/main.rs @@ -5,7 +5,7 @@ use std::{ use litellm_http::{HttpSettings, Resolution}; use litellm_inference_messages::{ - Error, MessagesCall, MessagesSettings, MessagesShaping, + Error, MessagesBody, MessagesCall, MessagesSettings, MessagesShaping, route::{Messages, MessagesMachine, MessagesOutput}, }; use litellm_inference_testing::RecordingSecrets; @@ -119,7 +119,10 @@ async fn run(call: MessagesCall) -> Result { async fn run_message(call: MessagesCall) -> MessagesResponse { match run(call).await.expect("messages call succeeds") { - MessagesOutput::Complete(message) => *message, + MessagesOutput::Complete(message) => match message { + MessagesBody::Message(message) => *message, + MessagesBody::Unchecked(_) => panic!("a call returned no message"), + }, MessagesOutput::StreamEnded | MessagesOutput::Detached => { panic!("a non-streaming call returned a stream") } diff --git a/litellm-rust/crates/inference-messages/tests/messages/response.rs b/litellm-rust/crates/inference-messages/tests/messages/response.rs index 76813fe0a34..f978da5e995 100644 --- a/litellm-rust/crates/inference-messages/tests/messages/response.rs +++ b/litellm-rust/crates/inference-messages/tests/messages/response.rs @@ -2,7 +2,7 @@ use litellm_host::{ interceptors::{ExecutionFacts, ResultSource}, lifecycle::ExecutionEvent, }; -use litellm_inference_messages::{MessagesCallResponse, messages_body}; +use litellm_inference_messages::{MessagesBody, MessagesCallResponse, messages_body}; use litellm_inference_testing::{ RecordingSecrets, http_config, no_secrets, provider_http, resources, }; @@ -48,6 +48,9 @@ async fn calls_defer_execution_until_polled( let MessagesCallResponse::Complete(response) = future.await.unwrap() else { panic!("expected a completed message"); }; + let MessagesBody::Message(response) = response else { + panic!("expected a message"); + }; assert_eq!( response.content, message_body()["content"].as_array().unwrap().as_slice() @@ -240,10 +243,42 @@ async fn an_unparseable_success_body_is_reported_as_the_upstream_answer(call: Me assert_eq!(error.upstream_response(), Some((200, "not json"))); } +#[rstest] +#[case::error_object( + json!({"type": "error", "error": {"type": "overloaded_error", "message": "Overloaded"}}), + json!({"type": "error", "error": {"type": "overloaded_error", "message": "Overloaded"}, + "stop_sequence": null}), +)] +#[case::message_missing_id( + json!({"type": "message", "content": [], "stop_sequence": "###"}), + json!({"type": "message", "content": [], "stop_sequence": "###"}), +)] +#[tokio::test] +async fn a_success_body_that_is_not_a_message_passes_through( + call: MessagesCall, + #[case] body: Value, + #[case] expected: Value, +) { + let upstream = upstream([json_response(body)]).await; + + let response = run(MessagesCall { + api_key: Some("sk".into()), + api_base: Some(upstream.uri()), + ..call + }) + .await + .expect("any JSON object is returned"); + + let MessagesOutput::Complete(response) = response else { + panic!("expected a completed response"); + }; + assert_eq!(serde_json::to_value(response).unwrap(), expected); +} + #[rstest] #[tokio::test] async fn an_unreadable_success_body_is_an_invalid_response(call: MessagesCall) { - let upstream = upstream([json_response(json!({"unexpected": true}))]).await; + let upstream = upstream([json_response(json!(["unexpected"]))]).await; let error = run(MessagesCall { api_key: Some("sk".into()), @@ -320,6 +355,9 @@ async fn the_facade_sends_through_the_injected_http_pool_configuration(call: Mes let MessagesCallResponse::Complete(message) = response else { panic!("a non-streaming request returns a message"); }; + let MessagesBody::Message(message) = message else { + panic!("expected a message"); + }; assert_eq!(message.id, "msg_1"); let sent = only_request(&upstream).await; assert_eq!(sent.header("x-api-key"), Some("sk-ant")); @@ -412,6 +450,9 @@ async fn route_uses_injected_dependencies_and_optional_cache( else { panic!("expected a completed message"); }; + let MessagesBody::Message(response) = response else { + panic!("expected a message"); + }; assert_eq!( response.content, message_body()["content"].as_array().unwrap().as_slice() @@ -484,6 +525,9 @@ async fn cache_overrides_preserve_the_routes_isolated_scope(call: MessagesCall) else { panic!("expected a completed message"); }; + let MessagesBody::Message(response) = response else { + panic!("expected a message"); + }; assert_eq!(response.id, expected["id"].as_str().unwrap()); } assert_eq!(received(&upstream).await.len(), 2); diff --git a/litellm-rust/crates/python-bridge/src/routes/messages/host.rs b/litellm-rust/crates/python-bridge/src/routes/messages/host.rs index ea27dc69d92..a24126c4f85 100644 --- a/litellm-rust/crates/python-bridge/src/routes/messages/host.rs +++ b/litellm-rust/crates/python-bridge/src/routes/messages/host.rs @@ -239,9 +239,9 @@ impl PythonBinding for MessagesPythonHost { fn encode_response( &mut self, py: Python<'_>, - response: Box, + response: litellm_inference_messages::MessagesBody, ) -> PyResult> { - let response = public_response(py, ROUTE_HOST_MODULE, response.as_ref())?; + let response = public_response(py, ROUTE_HOST_MODULE, &response)?; self.cache.store_response(py, response.bind(py))?; Ok(response) }