diff --git a/litellm-rust/Cargo.lock b/litellm-rust/Cargo.lock index 9654113e2b9..25584f4599a 100644 --- a/litellm-rust/Cargo.lock +++ b/litellm-rust/Cargo.lock @@ -1978,6 +1978,7 @@ dependencies = [ "strum", "subtle", "thiserror 2.0.19", + "time", "tokio", "tokio-tungstenite", "url", diff --git a/litellm-rust/Cargo.toml b/litellm-rust/Cargo.toml index 6033e6957f3..fa2457d3c2a 100644 --- a/litellm-rust/Cargo.toml +++ b/litellm-rust/Cargo.toml @@ -40,6 +40,7 @@ base64 = "0.22" moka = { version = "0.12.16", features = ["future"] } strum = { version = "0.28.0", features = ["derive"] } url = "2.5.8" +time = { version = "0.3.53", features = ["parsing"] } criterion = "0.8.2" veil = "0.3.0" diff --git a/litellm-rust/crates/core/Cargo.toml b/litellm-rust/crates/core/Cargo.toml index eccd753b3a0..da1cd92868f 100644 --- a/litellm-rust/crates/core/Cargo.toml +++ b/litellm-rust/crates/core/Cargo.toml @@ -30,6 +30,7 @@ subtle.workspace = true tokio = { workspace = true, features = ["sync"] } tokio-tungstenite.workspace = true thiserror.workspace = true +time.workspace = true sha2.workspace = true url.workspace = true veil.workspace = true diff --git a/litellm-rust/crates/core/src/providers/anthropic/messages/batches.rs b/litellm-rust/crates/core/src/providers/anthropic/messages/batches.rs new file mode 100644 index 00000000000..cf9bb0964be --- /dev/null +++ b/litellm-rust/crates/core/src/providers/anthropic/messages/batches.rs @@ -0,0 +1,344 @@ +use serde::{Deserialize, Serialize}; +use serde_json::Value; +use time::OffsetDateTime; +use url::Url; + +use crate::messages::Error; +use crate::messages::types::AnthropicMessagesResponse; +use crate::providers::anthropic::messages::transformation::resolve_anthropic_api_base; + +const BATCHES_PATH_SUFFIX: &str = "/v1/messages/batches"; + +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct AnthropicBatchRequestCounts { + #[serde(default)] + pub processing: u64, + #[serde(default)] + pub succeeded: u64, + #[serde(default)] + pub errored: u64, + #[serde(default)] + pub canceled: u64, + #[serde(default)] + pub expired: u64, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct AnthropicMessageBatch { + #[serde(default)] + pub id: String, + #[serde(default = "default_processing_status")] + pub processing_status: String, + pub created_at: Option, + pub ended_at: Option, + pub expires_at: Option, + pub cancel_initiated_at: Option, + pub archived_at: Option, + #[serde(default)] + pub request_counts: AnthropicBatchRequestCounts, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum BatchStatus { + InProgress, + Cancelling, + Completed, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct BatchRequestCounts { + pub total: u64, + pub completed: u64, + pub failed: u64, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct LiteLlmMessageBatch { + pub id: String, + pub object: String, + pub endpoint: String, + pub input_file_id: String, + pub completion_window: String, + pub status: BatchStatus, + pub output_file_id: String, + pub created_at: i64, + pub in_progress_at: Option, + pub expires_at: Option, + pub completed_at: Option, + pub expired_at: Option, + pub cancelling_at: Option, + pub cancelled_at: Option, + pub request_counts: BatchRequestCounts, +} + +pub trait AnthropicBatchesConfig { + fn create_batch_url( + &self, + api_base: Option<&str>, + env_lookup: &dyn Fn(&str) -> Option, + ) -> Result; + + fn transform_create_batch_request(&self) -> Result; + + fn transform_create_batch_response( + &self, + response: AnthropicMessageBatch, + now: i64, + ) -> Result; + + fn retrieve_batch_url( + &self, + api_base: Option<&str>, + batch_id: &str, + env_lookup: &dyn Fn(&str) -> Option, + ) -> Result; + + fn transform_retrieve_batch_request(&self) -> Value; + + fn transform_retrieve_batch_response( + &self, + response: AnthropicMessageBatch, + now: i64, + ) -> LiteLlmMessageBatch; + + fn transform_batch_results(&self, body: &str) -> Result, Error>; +} + +pub struct AnthropicBatchesTransformation; + +pub const ANTHROPIC_BATCHES_TRANSFORMATION: AnthropicBatchesTransformation = + AnthropicBatchesTransformation; + +fn default_processing_status() -> String { + "in_progress".into() +} + +fn timestamp(value: Option<&str>) -> Option { + value + .and_then(|value| { + OffsetDateTime::parse(value, &time::format_description::well_known::Rfc3339).ok() + }) + .map(OffsetDateTime::unix_timestamp) +} + +fn batches_base_url( + api_base: Option<&str>, + env_lookup: &dyn Fn(&str) -> Option, +) -> Result { + let api_base = resolve_anthropic_api_base(api_base, env_lookup); + let api_base = api_base.trim_end_matches('/'); + let complete_url = if api_base.ends_with(BATCHES_PATH_SUFFIX) { + api_base.to_string() + } else if let Some(base) = api_base.strip_suffix("/v1/messages") { + format!("{base}{BATCHES_PATH_SUFFIX}") + } else { + format!("{api_base}{BATCHES_PATH_SUFFIX}") + }; + Url::parse(&complete_url) + .map_err(|error| Error::InvalidRequest(format!("invalid Anthropic API base: {error}"))) +} + +impl AnthropicBatchesConfig for AnthropicBatchesTransformation { + fn create_batch_url( + &self, + api_base: Option<&str>, + env_lookup: &dyn Fn(&str) -> Option, + ) -> Result { + Ok(batches_base_url(api_base, env_lookup)?.into()) + } + + fn transform_create_batch_request(&self) -> Result { + Err(Error::InvalidRequest( + "Batch creation not yet implemented for Anthropic".into(), + )) + } + + fn transform_create_batch_response( + &self, + _response: AnthropicMessageBatch, + _now: i64, + ) -> Result { + Err(Error::InvalidResponse( + "Batch creation not yet implemented for Anthropic".into(), + )) + } + + fn retrieve_batch_url( + &self, + api_base: Option<&str>, + batch_id: &str, + env_lookup: &dyn Fn(&str) -> Option, + ) -> Result { + if batch_id.is_empty() { + return Err(Error::InvalidRequest("batch_id is required".into())); + } + let mut url = batches_base_url(api_base, env_lookup)?; + url.path_segments_mut() + .map_err(|_| Error::InvalidRequest("Anthropic API base cannot be a base URL".into()))? + .push(batch_id); + Ok(url.into()) + } + + fn transform_retrieve_batch_request(&self) -> Value { + Value::Object(Default::default()) + } + + fn transform_retrieve_batch_response( + &self, + response: AnthropicMessageBatch, + now: i64, + ) -> LiteLlmMessageBatch { + let created_at = timestamp(response.created_at.as_deref()); + let ended_at = timestamp(response.ended_at.as_deref()); + let expires_at = timestamp(response.expires_at.as_deref()); + let cancel_initiated_at = timestamp(response.cancel_initiated_at.as_deref()); + let archived_at = timestamp(response.archived_at.as_deref()); + let status = match response.processing_status.as_str() { + "canceling" => BatchStatus::Cancelling, + "ended" => BatchStatus::Completed, + _ => BatchStatus::InProgress, + }; + let request_counts = BatchRequestCounts { + total: response.request_counts.processing + + response.request_counts.succeeded + + response.request_counts.errored + + response.request_counts.canceled + + response.request_counts.expired, + completed: response.request_counts.succeeded, + failed: response.request_counts.errored, + }; + + LiteLlmMessageBatch { + id: response.id.clone(), + object: "batch".into(), + endpoint: "/v1/messages".into(), + input_file_id: "None".into(), + completion_window: "24h".into(), + status, + output_file_id: response.id, + created_at: created_at.unwrap_or(now), + in_progress_at: (response.processing_status == "in_progress") + .then_some(created_at) + .flatten(), + expires_at, + completed_at: (response.processing_status == "ended") + .then_some(ended_at) + .flatten(), + expired_at: archived_at, + cancelling_at: (response.processing_status == "canceling") + .then_some(cancel_initiated_at) + .flatten(), + cancelled_at: (response.processing_status == "canceling") + .then_some(ended_at) + .flatten(), + request_counts, + } + } + + fn transform_batch_results(&self, body: &str) -> Result, Error> { + body.lines() + .filter(|line| !line.trim().is_empty()) + .filter_map(|line| serde_json::from_str::(line.trim()).ok()) + .map(|record| { + serde_json::from_value(record["result"]["message"].clone()).map_err(|error| { + Error::InvalidResponse(format!("invalid Anthropic batch result: {error}")) + }) + }) + .collect() + } +} + +#[cfg(test)] +mod tests { + use serde_json::json; + + use super::*; + + #[test] + fn builds_and_encodes_message_batch_urls() { + assert_eq!( + ANTHROPIC_BATCHES_TRANSFORMATION + .create_batch_url(None, &|_| None) + .unwrap(), + "https://api.anthropic.com/v1/messages/batches" + ); + assert_eq!( + ANTHROPIC_BATCHES_TRANSFORMATION + .create_batch_url(Some("https://proxy.test/v1/messages/batches"), &|_| None) + .unwrap(), + "https://proxy.test/v1/messages/batches" + ); + assert_eq!( + ANTHROPIC_BATCHES_TRANSFORMATION + .retrieve_batch_url(Some("https://proxy.test"), "batch/id ?", &|_| None) + .unwrap(), + "https://proxy.test/v1/messages/batches/batch%2Fid%20%3F" + ); + assert_eq!( + ANTHROPIC_BATCHES_TRANSFORMATION.transform_retrieve_batch_request(), + json!({}) + ); + } + + #[test] + fn maps_retrieved_batch_status_counts_and_timestamps_like_python() { + let response: AnthropicMessageBatch = serde_json::from_value(json!({ + "id": "msgbatch_1", + "processing_status": "ended", + "created_at": "2025-01-01T00:00:00Z", + "ended_at": "2025-01-01T00:01:00Z", + "expires_at": "not-a-timestamp", + "request_counts": { + "processing": 1, + "succeeded": 2, + "errored": 3, + "canceled": 4, + "expired": 5 + } + })) + .unwrap(); + + let batch = ANTHROPIC_BATCHES_TRANSFORMATION.transform_retrieve_batch_response(response, 7); + assert_eq!(batch.status, BatchStatus::Completed); + assert_eq!(batch.created_at, 1_735_689_600); + assert_eq!(batch.completed_at, Some(1_735_689_660)); + assert_eq!(batch.expires_at, None); + assert_eq!( + batch.request_counts, + BatchRequestCounts { + total: 15, + completed: 2, + failed: 3 + } + ); + } + + #[test] + fn extracts_message_responses_from_ndjson_and_skips_non_json_lines() { + let body = r#"not-json +{"result":{"message":{"id":"msg_1","type":"message","role":"assistant","model":"claude-test","content":[],"stop_reason":"end_turn","stop_sequence":null}}} +"#; + let messages = ANTHROPIC_BATCHES_TRANSFORMATION + .transform_batch_results(body) + .unwrap(); + + assert_eq!(messages.len(), 1); + assert_eq!(messages[0].id, "msg_1"); + } + + #[test] + fn preserves_python_placeholder_for_batch_creation() { + assert!(matches!( + ANTHROPIC_BATCHES_TRANSFORMATION.transform_create_batch_request(), + Err(Error::InvalidRequest(message)) + if message == "Batch creation not yet implemented for Anthropic" + )); + let response: AnthropicMessageBatch = serde_json::from_value(json!({})).unwrap(); + assert!(matches!( + ANTHROPIC_BATCHES_TRANSFORMATION.transform_create_batch_response(response, 0), + Err(Error::InvalidResponse(message)) + if message == "Batch creation not yet implemented for Anthropic" + )); + } +} diff --git a/litellm-rust/crates/core/src/providers/anthropic/messages/count_tokens.rs b/litellm-rust/crates/core/src/providers/anthropic/messages/count_tokens.rs new file mode 100644 index 00000000000..34c3dfdde56 --- /dev/null +++ b/litellm-rust/crates/core/src/providers/anthropic/messages/count_tokens.rs @@ -0,0 +1,170 @@ +use serde::{Deserialize, Serialize}; +use serde_json::Value; + +use crate::constants::ANTHROPIC_OAUTH_TOKEN_PREFIX; +use crate::messages::Error; +use crate::messages::types::{AnthropicMessage, SystemPrompt}; + +const COUNT_TOKENS_ENDPOINT: &str = "https://api.anthropic.com/v1/messages/count_tokens"; +const TOKEN_COUNTING_BETA: &str = "token-counting-2024-11-01"; + +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub struct AnthropicCountTokensRequest { + pub model: String, + pub messages: Vec, + #[serde(skip_serializing_if = "Option::is_none")] + pub tools: Option>, + #[serde(skip_serializing_if = "Option::is_none")] + pub system: Option, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct AnthropicCountTokensResponse { + pub input_tokens: u64, +} + +pub trait AnthropicCountTokensConfig { + fn endpoint(&self) -> &'static str; + + fn validate_request(&self, model: &str, messages: &[AnthropicMessage]) -> Result<(), Error>; + + fn transform_request( + &self, + model: &str, + messages: Vec, + tools: Option>, + system: Option, + ) -> Result; + + fn required_headers(&self, api_key: &str) -> Vec<(&'static str, String)>; +} + +pub struct AnthropicCountTokensTransformation; + +pub const ANTHROPIC_COUNT_TOKENS_TRANSFORMATION: AnthropicCountTokensTransformation = + AnthropicCountTokensTransformation; + +impl AnthropicCountTokensConfig for AnthropicCountTokensTransformation { + fn endpoint(&self) -> &'static str { + COUNT_TOKENS_ENDPOINT + } + + fn transform_request( + &self, + model: &str, + messages: Vec, + tools: Option>, + system: Option, + ) -> Result { + self.validate_request(model, &messages)?; + + Ok(AnthropicCountTokensRequest { + model: model.to_string(), + messages, + tools, + system, + }) + } + + fn validate_request(&self, model: &str, messages: &[AnthropicMessage]) -> Result<(), Error> { + if model.is_empty() { + return Err(Error::InvalidRequest("model parameter is required".into())); + } + if messages.is_empty() { + return Err(Error::InvalidRequest( + "messages parameter is required".into(), + )); + } + Ok(()) + } + + fn required_headers(&self, api_key: &str) -> Vec<(&'static str, String)> { + let auth = if api_key.starts_with(ANTHROPIC_OAUTH_TOKEN_PREFIX) { + ("authorization", format!("Bearer {api_key}")) + } else { + ("x-api-key", api_key.to_string()) + }; + vec![ + ("content-type", "application/json".to_string()), + auth, + ("anthropic-version", "2023-06-01".to_string()), + ("anthropic-beta", TOKEN_COUNTING_BETA.to_string()), + ] + } +} + +#[cfg(test)] +mod tests { + use serde_json::{Map, json}; + + use super::*; + use crate::messages::types::MessageContent; + + fn message() -> AnthropicMessage { + AnthropicMessage { + role: "user".into(), + content: MessageContent::Text("hello".into()), + extra: Map::new(), + } + } + + #[test] + fn maps_the_python_count_tokens_contract() { + let request = ANTHROPIC_COUNT_TOKENS_TRANSFORMATION + .transform_request( + "claude-test", + vec![message()], + Some(vec![json!({"name": "lookup"})]), + Some(SystemPrompt::Text("system".into())), + ) + .unwrap(); + + assert_eq!( + serde_json::to_value(request).unwrap(), + json!({ + "model": "claude-test", + "messages": [{"role": "user", "content": "hello"}], + "tools": [{"name": "lookup"}], + "system": "system" + }) + ); + assert_eq!( + ANTHROPIC_COUNT_TOKENS_TRANSFORMATION.endpoint(), + COUNT_TOKENS_ENDPOINT + ); + } + + #[test] + fn rejects_the_invalid_requests_python_rejects() { + assert!(matches!( + ANTHROPIC_COUNT_TOKENS_TRANSFORMATION.transform_request( + "", + vec![message()], + None, + None + ), + Err(Error::InvalidRequest(message)) if message == "model parameter is required" + )); + assert!(matches!( + ANTHROPIC_COUNT_TOKENS_TRANSFORMATION.transform_request( + "claude-test", + vec![], + None, + None + ), + Err(Error::InvalidRequest(message)) if message == "messages parameter is required" + )); + } + + #[test] + fn uses_api_key_or_oauth_headers_without_combining_credentials() { + let api_key = ANTHROPIC_COUNT_TOKENS_TRANSFORMATION.required_headers("sk-ant-api"); + assert!(api_key.contains(&("x-api-key", "sk-ant-api".into()))); + assert!(!api_key.iter().any(|(name, _)| *name == "authorization")); + + let oauth = ANTHROPIC_COUNT_TOKENS_TRANSFORMATION.required_headers("sk-ant-oat-test"); + assert!(oauth.contains(&("authorization", "Bearer sk-ant-oat-test".into()))); + assert!(!oauth.iter().any(|(name, _)| *name == "x-api-key")); + assert!(oauth.contains(&("anthropic-beta", TOKEN_COUNTING_BETA.into()))); + } +} diff --git a/litellm-rust/crates/core/src/providers/anthropic/messages/mod.rs b/litellm-rust/crates/core/src/providers/anthropic/messages/mod.rs index fa7df180f50..3b1da7dc069 100644 --- a/litellm-rust/crates/core/src/providers/anthropic/messages/mod.rs +++ b/litellm-rust/crates/core/src/providers/anthropic/messages/mod.rs @@ -1,2 +1,4 @@ +pub mod batches; +pub mod count_tokens; pub mod streaming; pub mod transformation; diff --git a/litellm-rust/crates/core/src/providers/anthropic/messages/transformation.rs b/litellm-rust/crates/core/src/providers/anthropic/messages/transformation.rs index 080f11c8cac..0f1294a412c 100644 --- a/litellm-rust/crates/core/src/providers/anthropic/messages/transformation.rs +++ b/litellm-rust/crates/core/src/providers/anthropic/messages/transformation.rs @@ -31,10 +31,7 @@ pub fn complete_anthropic_url( api_base: Option<&str>, env_lookup: &dyn Fn(&str) -> Option, ) -> String { - let api_base = non_empty(api_base) - .map(str::to_string) - .or_else(|| env_lookup(ANTHROPIC_API_BASE_ENV).filter(|value| !value.trim().is_empty())) - .unwrap_or_else(|| DEFAULT_ANTHROPIC_API_BASE.to_string()); + let api_base = resolve_anthropic_api_base(api_base, env_lookup); let api_base = api_base.trim_end_matches('/'); if api_base.ends_with(MESSAGES_PATH_SUFFIX) { @@ -43,6 +40,16 @@ pub fn complete_anthropic_url( format!("{api_base}{MESSAGES_PATH_SUFFIX}") } +pub fn resolve_anthropic_api_base( + api_base: Option<&str>, + env_lookup: &dyn Fn(&str) -> Option, +) -> String { + non_empty(api_base) + .map(str::to_string) + .or_else(|| env_lookup(ANTHROPIC_API_BASE_ENV).filter(|value| !value.trim().is_empty())) + .unwrap_or_else(|| DEFAULT_ANTHROPIC_API_BASE.to_string()) +} + impl AnthropicMessagesProviderConfig for AnthropicMessagesConfig { fn complete_url( &self,