From 45396abb6b7b1e77800879a2e944607d46106712 Mon Sep 17 00:00:00 2001 From: Aastha Joshi Date: Tue, 11 Aug 2026 00:24:44 -0700 Subject: [PATCH] refactor(rust): move audio transcription execution to core --- .../src/audio_transcription/handler.rs | 89 ----------- .../src/audio_transcription/hooks.rs | 127 +++++++-------- .../ai-gateway/src/audio_transcription/mod.rs | 18 ++- .../src/audio_transcription/tests.rs | 148 +++++++++++++++++- .../src/audio_transcription/types.rs | 11 -- .../core/src/audio_transcription/client.rs | 15 ++ .../src/audio_transcription/common_utils.rs | 14 +- .../core/src/audio_transcription/handler.rs | 145 +++++++++++++++++ .../core/src/audio_transcription/mod.rs | 50 ++++++ .../core/src/audio_transcription/prepare.rs | 62 ++++++++ .../core/src/audio_transcription/tests.rs | 140 +++++++++++++++++ .../core/src/audio_transcription/types.rs | 65 +++++++- litellm-rust/crates/python-bridge/src/lib.rs | 18 +-- 13 files changed, 709 insertions(+), 193 deletions(-) delete mode 100644 litellm-rust/crates/ai-gateway/src/audio_transcription/handler.rs create mode 100644 litellm-rust/crates/core/src/audio_transcription/client.rs rename litellm-rust/crates/{ai-gateway => core}/src/audio_transcription/common_utils.rs (82%) create mode 100644 litellm-rust/crates/core/src/audio_transcription/handler.rs create mode 100644 litellm-rust/crates/core/src/audio_transcription/prepare.rs create mode 100644 litellm-rust/crates/core/src/audio_transcription/tests.rs diff --git a/litellm-rust/crates/ai-gateway/src/audio_transcription/handler.rs b/litellm-rust/crates/ai-gateway/src/audio_transcription/handler.rs deleted file mode 100644 index 33c13550f58..00000000000 --- a/litellm-rust/crates/ai-gateway/src/audio_transcription/handler.rs +++ /dev/null @@ -1,89 +0,0 @@ -use std::time::SystemTime; - -use litellm_core::CoreResult; -use litellm_core::audio_transcription::transformation::AudioTranscriptionAuth; -use litellm_core::error::CoreError; -use litellm_core::providers::bedrock::audio_transcription::aws_auth_config; -use litellm_core::providers::bedrock::aws_base::{resolve_credentials, sign_bedrock_post}; -use serde_json::Value; - -use super::common_utils::truncate_error_body; -use super::types::ProviderAudioTranscriptionRequest; -use crate::client::http_client; - -pub(crate) async fn execute_audio_transcription_provider_call( - request: ProviderAudioTranscriptionRequest, -) -> CoreResult { - let body = serde_json::to_vec(&request.body).map_err(|error| { - CoreError::InvalidRequest(format!("invalid audio request body: {error}")) - })?; - let mut request_builder = http_client().post(&request.url).body(body.clone()); - for (key, value) in &request.upstream_headers { - request_builder = request_builder.header(key, value); - } - if let Some(duration) = request.timeout { - request_builder = request_builder.timeout(duration); - } - let response = request_builder - .send() - .await - .map_err(|error| CoreError::Network(error.to_string()))?; - let status = response.status(); - let text = response - .text() - .await - .map_err(|error| CoreError::Network(error.to_string()))?; - if !status.is_success() { - return Err(CoreError::Http { - status: status.as_u16(), - body: truncate_error_body(&text), - }); - } - let response_json: Value = serde_json::from_str(&text).map_err(|error| { - CoreError::InvalidResponse(format!("invalid audio response JSON: {error}")) - })?; - Ok(request - .config - .transform_transcription_response(&request.model, response_json)? - .into_json()) -} - -pub(crate) async fn sign_request( - request: &ProviderAudioTranscriptionRequest, - optional_params: &serde_json::Map, -) -> CoreResult { - let env_lookup = environment_lookup; - let auth = request - .config - .auth_strategy(&request.model, optional_params, &env_lookup)?; - let body = serde_json::to_vec(&request.body).map_err(|error| { - CoreError::InvalidRequest(format!("invalid audio request body: {error}")) - })?; - let mut headers = super::common_utils::string_headers(None)?; - headers.insert("Content-Type".to_string(), "application/json".to_string()); - headers.extend(request.upstream_headers.iter().cloned()); - match auth { - AudioTranscriptionAuth::Bearer => {} - AudioTranscriptionAuth::AwsSigV4 { region, .. } => { - let credentials = - resolve_credentials(aws_auth_config(optional_params, &env_lookup), &env_lookup) - .await?; - headers.extend(sign_bedrock_post( - &request.url, - &body, - &headers, - ®ion, - &credentials, - SystemTime::now(), - )?); - } - } - Ok(ProviderAudioTranscriptionRequest { - upstream_headers: headers.into_iter().collect(), - ..request.clone() - }) -} - -pub(super) fn environment_lookup(key: &str) -> Option { - std::env::var(key).ok() -} diff --git a/litellm-rust/crates/ai-gateway/src/audio_transcription/hooks.rs b/litellm-rust/crates/ai-gateway/src/audio_transcription/hooks.rs index 8b6896f3846..36da54555e5 100644 --- a/litellm-rust/crates/ai-gateway/src/audio_transcription/hooks.rs +++ b/litellm-rust/crates/ai-gateway/src/audio_transcription/hooks.rs @@ -2,14 +2,16 @@ use std::future::Future; use std::pin::Pin; use litellm_core::CoreResult; -use litellm_core::audio_transcription::transformation::AudioTranscriptionAuth; +use litellm_core::audio_transcription::{ + AudioTranscriptionRequest as CoreAudioTranscriptionRequest, + PreparedAudioTranscriptionRequest as CorePreparedAudioTranscriptionRequest, + prepare_audio_transcription_request, +}; use litellm_core::call_lifecycle::{CallLifecycleContext, CallLifecycleHooks, CallLifecycleTiming}; use litellm_core::error::CoreError; use serde_json::{Map, Value, json}; -use super::common_utils::{audio_transcription_provider_config, has_header, string_headers}; -use super::handler::sign_request; -use super::types::{PreparedAudioTranscriptionRequest, ProviderAudioTranscriptionRequest}; +use super::types::PreparedAudioTranscriptionRequest; use crate::integrations::custom_guardrail::{ CustomGuardrailRunner, GuardrailContext, GuardrailError, GuardrailRequest, }; @@ -27,6 +29,7 @@ pub(crate) struct AudioTranscriptionLifecycleHooks { } type AudioFuture<'a, T> = Pin> + Send + 'a>>; + type AudioLogFuture<'a> = Pin + Send + 'a>>; impl AudioTranscriptionLifecycleHooks { @@ -49,27 +52,33 @@ impl AudioTranscriptionLifecycleHooks { if self.guardrail_runner.is_empty() { return Ok(request); } + let (guardrail_request, _) = self .guardrail_runner .run_pre_call( &guardrail_context(&self.request_metadata), GuardrailRequest::new(json!({ "model": request.model, - "custom_llm_provider": request.custom_llm_provider, + "custom_llm_provider": + request.custom_llm_provider, "audio": request.audio, - "optional_params": request.optional_params, + "optional_params": + request.optional_params, })), ) .await .map_err(guardrail_error_to_core_error)?; + let Value::Object(mut data) = guardrail_request.data else { return Err(CoreError::InvalidRequest( "audio transcription pre_call guardrail must return an object".to_string(), )); }; + let audio = data.remove("audio").ok_or_else(|| { CoreError::InvalidRequest("audio transcription guardrail removed audio".to_string()) })?; + let optional_params = match data.remove("optional_params") { Some(Value::Object(value)) => value, Some(_) => { @@ -79,6 +88,7 @@ impl AudioTranscriptionLifecycleHooks { } None => Map::new(), }; + Ok(PreparedAudioTranscriptionRequest { audio, optional_params, @@ -89,78 +99,55 @@ impl AudioTranscriptionLifecycleHooks { async fn prepare_provider_request( &self, request: PreparedAudioTranscriptionRequest, - ) -> CoreResult { - let config = audio_transcription_provider_config(&request.custom_llm_provider) - .ok_or_else(|| CoreError::InvalidProvider(request.custom_llm_provider.clone()))?; - let env_lookup = super::handler::environment_lookup; - let headers = string_headers(request.extra_headers)?; - let url = config.complete_url( - request.api_base.as_deref(), - &request.model, - &request.optional_params, - &env_lookup, - )?; - let filtered_params = config.map_transcription_params(&request.optional_params); - let body = config.transform_transcription_request( - &request.model, - request.audio, - filtered_params, - )?; - let auth = config.auth_strategy(&request.model, &request.optional_params, &env_lookup)?; - let mut upstream_headers = headers.into_iter().collect::>(); - if matches!(auth, AudioTranscriptionAuth::Bearer) - && !has_header( - &upstream_headers - .iter() - .cloned() - .collect::>(), - "authorization", - ) - && let Some(api_key) = request.api_key.as_deref() - { - upstream_headers.push(("Authorization".to_string(), format!("Bearer {api_key}"))); - } - let provider_request = ProviderAudioTranscriptionRequest { - model: request.model, - config, - url, - body: body.body, - upstream_headers, - timeout: request.timeout, - }; - let provider_request = self.run_during_call_guardrails(provider_request).await?; - sign_request(&provider_request, &request.optional_params).await - } + ) -> CoreResult { + let mut provider_request = + prepare_audio_transcription_request(CoreAudioTranscriptionRequest { + model: &request.model, + audio: request.audio, + api_key: request.api_key.as_deref(), + api_base: request.api_base.as_deref(), + custom_llm_provider: Some(&request.custom_llm_provider), + extra_headers: request.extra_headers, + optional_params: request.optional_params, + timeout: request.timeout, + })?; - async fn run_during_call_guardrails( - &self, - request: ProviderAudioTranscriptionRequest, - ) -> CoreResult { if self.guardrail_runner.is_empty() { - return Ok(request); + return Ok(provider_request); } + let (guardrail_request, _) = self .guardrail_runner .run_during_call( &guardrail_context(&self.request_metadata), GuardrailRequest::new(json!({ - "model": request.model, - "custom_llm_provider": "bedrock", - "url": request.url, - "body": request.body, + "model": + provider_request.model(), + "custom_llm_provider": + provider_request + .custom_llm_provider(), + "url": + provider_request.url(), + "body": + provider_request.body(), })), ) .await .map_err(guardrail_error_to_core_error)?; + let Value::Object(mut data) = guardrail_request.data else { return Err(CoreError::InvalidRequest( "audio transcription during_call guardrail must return an object".to_string(), )); }; + let body = data.remove("body").ok_or_else(|| { CoreError::InvalidRequest("audio transcription guardrail removed body".to_string()) })?; - Ok(ProviderAudioTranscriptionRequest { body, ..request }) + + provider_request.replace_body(body); + + Ok(provider_request) } fn logging_payload( @@ -192,11 +179,17 @@ impl AudioTranscriptionLifecycleHooks { } } -impl CallLifecycleHooks - for AudioTranscriptionLifecycleHooks +impl + CallLifecycleHooks< + PreparedAudioTranscriptionRequest, + CorePreparedAudioTranscriptionRequest, + Value, + > for AudioTranscriptionLifecycleHooks { type PreCallFuture<'a> = AudioFuture<'a, PreparedAudioTranscriptionRequest>; - type DuringCallFuture<'a> = AudioFuture<'a, ProviderAudioTranscriptionRequest>; + + type DuringCallFuture<'a> = AudioFuture<'a, CorePreparedAudioTranscriptionRequest>; + type SuccessFuture<'a> = AudioLogFuture<'a>; type FailureFuture<'a> = AudioLogFuture<'a>; @@ -226,6 +219,7 @@ impl CallLifecycleHooks GuardrailContext { } fn guardrail_error_to_core_error(error: GuardrailError) -> CoreError { - CoreError::InvalidRequest(format!("{}: {}", error.kind, error.message)) + CoreError::InvalidRequest(format!("{}: {}", error.kind, error.message,)) } fn core_error_kind(error: &CoreError) -> &'static str { diff --git a/litellm-rust/crates/ai-gateway/src/audio_transcription/mod.rs b/litellm-rust/crates/ai-gateway/src/audio_transcription/mod.rs index 5d33d912c40..c50cdd28f2e 100644 --- a/litellm-rust/crates/ai-gateway/src/audio_transcription/mod.rs +++ b/litellm-rust/crates/ai-gateway/src/audio_transcription/mod.rs @@ -1,23 +1,33 @@ use litellm_core::CoreResult; +use litellm_core::audio_transcription::{ + PreparedAudioTranscriptionRequest as CorePreparedAudioTranscriptionRequest, + execute_prepared_audio_transcription, +}; use litellm_core::call_lifecycle::CallLifecycle; use serde_json::Value; -mod common_utils; -mod handler; mod hooks; mod prepare; mod types; pub use types::AudioTranscriptionRequest; -use handler::execute_audio_transcription_provider_call; use prepare::{PreparedAudioTranscriptionCall, prepare_audio_transcription_call}; +async fn execute_audio_transcription( + request: CorePreparedAudioTranscriptionRequest, +) -> CoreResult { + execute_prepared_audio_transcription(request) + .await + .map(|response| response.into_json()) +} + pub async fn audio_transcription(request: AudioTranscriptionRequest<'_>) -> CoreResult { let PreparedAudioTranscriptionCall { request, hooks } = prepare_audio_transcription_call(request); + CallLifecycle::default() - .run_request(request, &hooks, execute_audio_transcription_provider_call) + .run_request(request, &hooks, execute_audio_transcription) .await } diff --git a/litellm-rust/crates/ai-gateway/src/audio_transcription/tests.rs b/litellm-rust/crates/ai-gateway/src/audio_transcription/tests.rs index 5df04708b7d..f8c9a49bb8d 100644 --- a/litellm-rust/crates/ai-gateway/src/audio_transcription/tests.rs +++ b/litellm-rust/crates/ai-gateway/src/audio_transcription/tests.rs @@ -1,45 +1,97 @@ use std::io::{Read, Write}; use std::net::TcpListener; +use std::sync::Arc; use std::thread; use serde_json::{Map, json}; use super::{AudioTranscriptionRequest, audio_transcription}; +use crate::integrations::custom_guardrail::{ + CustomGuardrail, GuardrailContext, GuardrailDecision, GuardrailEventHook, GuardrailFuture, + GuardrailRequest, +}; + +struct DuringCallAudioGuardrail; + +impl CustomGuardrail for DuringCallAudioGuardrail { + fn guardrail_name(&self) -> &str { + "audio-during-call-test" + } + + fn supported_event_hooks(&self) -> &[GuardrailEventHook] { + static HOOKS: [GuardrailEventHook; 1] = [GuardrailEventHook::DuringCall]; + + &HOOKS + } + + fn async_moderation_hook<'a>( + &'a self, + _context: &'a GuardrailContext, + mut request: GuardrailRequest, + ) -> GuardrailFuture<'a> { + Box::pin(async move { + request.data["body"]["guarded_during"] = json!("added-after-provider-transform"); + + Ok(GuardrailDecision::Mask(request)) + }) + } +} + +fn bedrock_optional_params() -> Map { + Map::from_iter([ + ("aws_access_key_id".to_string(), json!("access-key")), + ("aws_secret_access_key".to_string(), json!("secret-key")), + ("aws_region_name".to_string(), json!("us-east-1")), + ]) +} #[tokio::test] async fn bedrock_request_is_signed_and_contains_audio() { let listener = TcpListener::bind("127.0.0.1:0").expect("listener"); let address = listener.local_addr().expect("address"); + let server = thread::spawn(move || { let (mut stream, _) = listener.accept().expect("connection"); + let mut request = Vec::new(); let mut buffer = [0_u8; 16_384]; + let count = stream.read(&mut buffer).expect("request"); + request.extend_from_slice(&buffer[..count]); + let request = String::from_utf8_lossy(&request); + assert!(request.contains("POST /model/mistral.voxtral-mini-3b-2507/converse")); assert!(request.contains("authorization: AWS4-HMAC-SHA256")); assert!(request.contains("x-amz-date:")); assert!(request.contains("\"bytes\":\"AQI=\"")); assert!(request.contains("Transcribe the audio. Respond with only the transcript.")); - let response = b"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: 53\r\nConnection: close\r\n\r\n{\"output\":{\"message\":{\"content\":[{\"text\":\"hello\"}]}}}"; + + let response = b"HTTP/1.1 200 OK\r\n\ +Content-Type: application/json\r\n\ +Content-Length: 53\r\n\ +Connection: close\r\n\ +\r\n\ +{\"output\":{\"message\":{\"content\":[{\"text\":\"hello\"}]}}}"; + stream.write_all(response).expect("response"); }); - let optional_params = Map::from_iter([ - ("aws_access_key_id".to_string(), json!("access-key")), - ("aws_secret_access_key".to_string(), json!("secret-key")), - ("aws_region_name".to_string(), json!("us-east-1")), - ]); let api_base = format!("http://{address}"); + let response = audio_transcription(AudioTranscriptionRequest { model: "mistral.voxtral-mini-3b-2507", - audio: json!({"data": "AQI=", "format": "wav", "filename": "audio.wav"}), + audio: json!({ + "data": "AQI=", + "format": "wav", + "filename": "audio.wav" + }), api_key: None, api_base: Some(&api_base), custom_llm_provider: Some("bedrock"), extra_headers: None, - optional_params, + optional_params: bedrock_optional_params(), timeout: None, callbacks: Vec::new(), guardrails: Vec::new(), @@ -48,6 +100,86 @@ async fn bedrock_request_is_signed_and_contains_audio() { }) .await .expect("transcription"); + assert_eq!(response, json!({"text": "hello"})); + + server.join().expect("server"); +} + +#[tokio::test] +async fn during_call_guardrail_mutation_is_sent_after_provider_transform() { + let listener = TcpListener::bind("127.0.0.1:0").expect("listener"); + let address = listener.local_addr().expect("address"); + + let server = thread::spawn(move || { + let (mut stream, _) = listener.accept().expect("connection"); + + let mut request = Vec::new(); + let mut buffer = [0_u8; 16_384]; + + let count = stream.read(&mut buffer).expect("request"); + + request.extend_from_slice(&buffer[..count]); + + let request = String::from_utf8_lossy(&request); + + // Provider transformation happened before the during-call hook. + assert!( + request.contains("\"bytes\":\"AQI=\""), + "provider-transformed audio should remain in final request:\n{request}" + ); + + // The during-call guardrail changed the transformed provider body. + assert!( + request.contains("\"guarded_during\":\"added-after-provider-transform\""), + "guardrail mutation should be present in final provider body:\n{request}" + ); + + // Core still finalized provider authentication after the mutation. + assert!( + request.contains("authorization: AWS4-HMAC-SHA256"), + "final request should contain SigV4 authorization:\n{request}" + ); + + assert!( + request.contains("x-amz-date:"), + "final request should contain SigV4 date:\n{request}" + ); + + let response = b"HTTP/1.1 200 OK\r\n\ +Content-Type: application/json\r\n\ +Content-Length: 53\r\n\ +Connection: close\r\n\ +\r\n\ +{\"output\":{\"message\":{\"content\":[{\"text\":\"hello\"}]}}}"; + + stream.write_all(response).expect("response"); + }); + + let api_base = format!("http://{address}"); + + let response = audio_transcription(AudioTranscriptionRequest { + model: "mistral.voxtral-mini-3b-2507", + audio: json!({ + "data": "AQI=", + "format": "wav", + "filename": "audio.wav" + }), + api_key: None, + api_base: Some(&api_base), + custom_llm_provider: Some("bedrock"), + extra_headers: None, + optional_params: bedrock_optional_params(), + timeout: None, + callbacks: Vec::new(), + guardrails: vec![Arc::new(DuringCallAudioGuardrail)], + request_metadata: Default::default(), + litellm_call_id: Some("audio-guardrail-signing-regression"), + }) + .await + .expect("transcription with during-call guardrail"); + + assert_eq!(response, json!({"text": "hello"})); + server.join().expect("server"); } diff --git a/litellm-rust/crates/ai-gateway/src/audio_transcription/types.rs b/litellm-rust/crates/ai-gateway/src/audio_transcription/types.rs index 9697aa98b0a..b470638264e 100644 --- a/litellm-rust/crates/ai-gateway/src/audio_transcription/types.rs +++ b/litellm-rust/crates/ai-gateway/src/audio_transcription/types.rs @@ -1,7 +1,6 @@ use std::sync::Arc; use std::time::Duration; -use litellm_core::audio_transcription::transformation::AudioTranscriptionProviderConfig; use litellm_core::call_lifecycle::{CallLifecycleContext, CallLifecycleRequest}; use serde_json::{Map, Value}; @@ -46,13 +45,3 @@ impl CallLifecycleRequest for PreparedAudioTranscriptionRequest { ) } } - -#[derive(Clone)] -pub(crate) struct ProviderAudioTranscriptionRequest { - pub(crate) model: String, - pub(crate) config: &'static dyn AudioTranscriptionProviderConfig, - pub(crate) url: String, - pub(crate) body: Value, - pub(crate) upstream_headers: Vec<(String, String)>, - pub(crate) timeout: Option, -} diff --git a/litellm-rust/crates/core/src/audio_transcription/client.rs b/litellm-rust/crates/core/src/audio_transcription/client.rs new file mode 100644 index 00000000000..d54b93e866b --- /dev/null +++ b/litellm-rust/crates/core/src/audio_transcription/client.rs @@ -0,0 +1,15 @@ +use std::sync::OnceLock; +use std::time::Duration; + +const AUDIO_TRANSCRIPTION_TIMEOUT_SECS: u64 = 600; + +pub(super) fn http_client() -> &'static reqwest::Client { + static CLIENT: OnceLock = OnceLock::new(); + + CLIENT.get_or_init(|| { + reqwest::Client::builder() + .timeout(Duration::from_secs(AUDIO_TRANSCRIPTION_TIMEOUT_SECS)) + .build() + .expect("failed to build audio transcription reqwest client") + }) +} diff --git a/litellm-rust/crates/ai-gateway/src/audio_transcription/common_utils.rs b/litellm-rust/crates/core/src/audio_transcription/common_utils.rs similarity index 82% rename from litellm-rust/crates/ai-gateway/src/audio_transcription/common_utils.rs rename to litellm-rust/crates/core/src/audio_transcription/common_utils.rs index 270d5c2d97a..9d633e07cbe 100644 --- a/litellm-rust/crates/ai-gateway/src/audio_transcription/common_utils.rs +++ b/litellm-rust/crates/core/src/audio_transcription/common_utils.rs @@ -1,15 +1,20 @@ use std::collections::BTreeMap; -use litellm_core::CoreResult; -use litellm_core::audio_transcription::transformation::AudioTranscriptionProviderConfig; -use litellm_core::error::CoreError; -use litellm_core::providers::bedrock::audio_transcription::BEDROCK_AUDIO_TRANSCRIPTION_CONFIG; use serde_json::{Map, Value}; +use crate::CoreResult; +use crate::error::CoreError; + +use super::transformation::AudioTranscriptionProviderConfig; + +#[cfg(feature = "bedrock-auth")] +use crate::providers::bedrock::audio_transcription::BEDROCK_AUDIO_TRANSCRIPTION_CONFIG; + pub(super) fn audio_transcription_provider_config( provider: &str, ) -> Option<&'static dyn AudioTranscriptionProviderConfig> { match provider { + #[cfg(feature = "bedrock-auth")] "bedrock" => Some(&BEDROCK_AUDIO_TRANSCRIPTION_CONFIG), _ => None, } @@ -40,6 +45,7 @@ pub(super) fn has_header(headers: &BTreeMap, name: &str) -> bool pub(super) fn truncate_error_body(body: &str) -> String { let truncated: String = body.chars().take(256).collect(); + if truncated.chars().count() == body.chars().count() { truncated } else { diff --git a/litellm-rust/crates/core/src/audio_transcription/handler.rs b/litellm-rust/crates/core/src/audio_transcription/handler.rs new file mode 100644 index 00000000000..23ed5c5586c --- /dev/null +++ b/litellm-rust/crates/core/src/audio_transcription/handler.rs @@ -0,0 +1,145 @@ +use crate::CoreResult; +use crate::error::CoreError; + +#[cfg(feature = "bedrock-auth")] +use std::time::SystemTime; + +#[cfg(feature = "bedrock-auth")] +use crate::providers::bedrock::audio_transcription::aws_auth_config; + +#[cfg(feature = "bedrock-auth")] +use crate::providers::bedrock::aws_base::{resolve_credentials, sign_bedrock_post}; + +use super::client::http_client; +use super::common_utils::truncate_error_body; +use super::prepare::environment_lookup; +use super::transformation::AudioTranscriptionAuth; +use super::types::{AudioTranscriptionResponseData, PreparedAudioTranscriptionRequest}; + +pub(super) async fn execute_audio_transcription_provider_call( + mut request: PreparedAudioTranscriptionRequest, +) -> CoreResult { + sign_request(&mut request).await?; + + let body = serde_json::to_vec(&request.body).map_err(|error| { + CoreError::InvalidRequest(format!("invalid audio request body: {error}")) + })?; + + let mut request_builder = http_client().post(&request.url).body(body); + + for (key, value) in &request.upstream_headers { + request_builder = request_builder.header(key, value); + } + + if let Some(timeout) = request.timeout { + request_builder = request_builder.timeout(timeout); + } + + let response = request_builder + .send() + .await + .map_err(|error| CoreError::Network(error.to_string()))?; + + let status = response.status(); + + let response_text = response + .text() + .await + .map_err(|error| CoreError::Network(error.to_string()))?; + + if !status.is_success() { + return Err(CoreError::Http { + status: status.as_u16(), + body: truncate_error_body(&response_text), + }); + } + + let response_json = serde_json::from_str(&response_text).map_err(|error| { + CoreError::InvalidResponse(format!( + "invalid audio transcription response JSON: {error}" + )) + })?; + + request + .config + .transform_transcription_response(&request.model, response_json) +} + +async fn sign_request(request: &mut PreparedAudioTranscriptionRequest) -> CoreResult<()> { + let env_lookup = environment_lookup; + + let auth = + request + .config + .auth_strategy(&request.model, &request.optional_params, &env_lookup)?; + + match auth { + AudioTranscriptionAuth::Bearer => { + ensure_content_type(request); + Ok(()) + } + + AudioTranscriptionAuth::AwsSigV4 { region, .. } => { + sign_aws_sigv4_request(request, ®ion).await + } + } +} + +fn ensure_content_type(request: &mut PreparedAudioTranscriptionRequest) { + if !request + .upstream_headers + .iter() + .any(|(key, _)| key.eq_ignore_ascii_case("content-type")) + { + request + .upstream_headers + .push(("Content-Type".to_string(), "application/json".to_string())); + } +} + +#[cfg(feature = "bedrock-auth")] +async fn sign_aws_sigv4_request( + request: &mut PreparedAudioTranscriptionRequest, + region: &str, +) -> CoreResult<()> { + let env_lookup = environment_lookup; + + let body = serde_json::to_vec(&request.body).map_err(|error| { + CoreError::InvalidRequest(format!("invalid audio request body: {error}")) + })?; + + let mut headers = std::collections::BTreeMap::new(); + + headers.insert("Content-Type".to_string(), "application/json".to_string()); + + headers.extend(request.upstream_headers.iter().cloned()); + + let credentials = resolve_credentials( + aws_auth_config(&request.optional_params, &env_lookup), + &env_lookup, + ) + .await?; + + headers.extend(sign_bedrock_post( + &request.url, + &body, + &headers, + region, + &credentials, + SystemTime::now(), + )?); + + request.upstream_headers = headers.into_iter().collect(); + + Ok(()) +} + +#[cfg(not(feature = "bedrock-auth"))] +async fn sign_aws_sigv4_request( + _request: &mut PreparedAudioTranscriptionRequest, + _region: &str, +) -> CoreResult<()> { + Err(CoreError::InvalidProvider( + "AWS SigV4 authentication requires the `bedrock-auth` feature".to_string(), + )) +} diff --git a/litellm-rust/crates/core/src/audio_transcription/mod.rs b/litellm-rust/crates/core/src/audio_transcription/mod.rs index ec2fbb969a6..20062d8300b 100644 --- a/litellm-rust/crates/core/src/audio_transcription/mod.rs +++ b/litellm-rust/crates/core/src/audio_transcription/mod.rs @@ -1,2 +1,52 @@ +//! Audio transcription SDK entrypoint. +//! +//! Provider resolution, request transformation, authentication, transport, and +//! response transformation live in core so every host uses the same execution +//! path. + +mod client; +mod common_utils; +mod handler; +mod prepare; + pub mod transformation; pub mod types; + +use crate::CoreResult; + +use handler::execute_audio_transcription_provider_call; +use prepare::prepare_audio_transcription_call; + +pub use types::{ + AudioTranscriptionRequest, AudioTranscriptionResponseData, PreparedAudioTranscriptionRequest, +}; + +/// Transform a host request into a provider-specific request. +/// +/// The returned value deliberately exposes only the information required by +/// host lifecycle integrations. Authentication is finalized during execution +/// so body mutations made by guardrails are included in provider signatures. +pub fn prepare_audio_transcription_request( + request: AudioTranscriptionRequest<'_>, +) -> CoreResult { + prepare_audio_transcription_call(request) +} + +/// Execute a previously prepared audio transcription request. +/// +/// Authentication and signing happen here, immediately before transport. +pub async fn execute_prepared_audio_transcription( + request: PreparedAudioTranscriptionRequest, +) -> CoreResult { + execute_audio_transcription_provider_call(request).await +} + +/// Execute an audio transcription request through the resolved provider. +pub async fn audio_transcription( + request: AudioTranscriptionRequest<'_>, +) -> CoreResult { + execute_prepared_audio_transcription(prepare_audio_transcription_request(request)?).await +} + +#[cfg(all(test, feature = "bedrock-auth"))] +mod tests; diff --git a/litellm-rust/crates/core/src/audio_transcription/prepare.rs b/litellm-rust/crates/core/src/audio_transcription/prepare.rs new file mode 100644 index 00000000000..842057e8b8f --- /dev/null +++ b/litellm-rust/crates/core/src/audio_transcription/prepare.rs @@ -0,0 +1,62 @@ +use crate::error::{CoreError, CoreResult}; +use crate::routing_utils::provider::{CustomLlmProvider, get_custom_llm_provider}; + +use super::common_utils::{audio_transcription_provider_config, has_header, string_headers}; +use super::transformation::AudioTranscriptionAuth; +use super::types::{AudioTranscriptionRequest, PreparedAudioTranscriptionRequest}; + +pub(super) fn prepare_audio_transcription_call( + request: AudioTranscriptionRequest<'_>, +) -> CoreResult { + let provider_info = get_custom_llm_provider(request.model, request.custom_llm_provider) + .unwrap_or(CustomLlmProvider { + model: request.model, + custom_llm_provider: "bedrock", + }); + + let model = provider_info.model.to_string(); + let custom_llm_provider = provider_info.custom_llm_provider.to_string(); + + let config = audio_transcription_provider_config(&custom_llm_provider) + .ok_or_else(|| CoreError::InvalidProvider(custom_llm_provider.clone()))?; + + let env_lookup = environment_lookup; + + let mut headers = string_headers(request.extra_headers)?; + + let url = config.complete_url( + request.api_base, + &model, + &request.optional_params, + &env_lookup, + )?; + + let filtered_params = config.map_transcription_params(&request.optional_params); + + let transformed = + config.transform_transcription_request(&model, request.audio, filtered_params)?; + + let auth = config.auth_strategy(&model, &request.optional_params, &env_lookup)?; + + if matches!(auth, AudioTranscriptionAuth::Bearer) + && !has_header(&headers, "authorization") + && let Some(api_key) = request.api_key + { + headers.insert("Authorization".to_string(), format!("Bearer {api_key}")); + } + + Ok(PreparedAudioTranscriptionRequest { + model, + custom_llm_provider, + config, + url, + body: transformed.body, + upstream_headers: headers.into_iter().collect(), + optional_params: request.optional_params, + timeout: request.timeout, + }) +} + +pub(super) fn environment_lookup(key: &str) -> Option { + std::env::var(key).ok() +} diff --git a/litellm-rust/crates/core/src/audio_transcription/tests.rs b/litellm-rust/crates/core/src/audio_transcription/tests.rs new file mode 100644 index 00000000000..0401ea778a0 --- /dev/null +++ b/litellm-rust/crates/core/src/audio_transcription/tests.rs @@ -0,0 +1,140 @@ +use serde_json::{Map, Value, json}; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::TcpListener; + +use super::{AudioTranscriptionRequest, audio_transcription}; + +fn optional_params() -> Map { + let mut params = Map::new(); + + params.insert("aws_access_key_id".to_string(), json!("test-access-key")); + params.insert( + "aws_secret_access_key".to_string(), + json!("test-secret-key"), + ); + params.insert("aws_region_name".to_string(), json!("us-east-1")); + + params +} + +#[tokio::test] +async fn bedrock_request_is_signed_and_contains_audio() { + let listener = TcpListener::bind("127.0.0.1:0") + .await + .expect("bind test listener"); + + let address = listener.local_addr().expect("read listener address"); + + let server = tokio::spawn(async move { + let (mut socket, _) = listener.accept().await.expect("accept request"); + + let mut request = Vec::new(); + let mut buffer = [0_u8; 4096]; + + loop { + let read = socket.read(&mut buffer).await.expect("read request"); + + if read == 0 { + break; + } + + request.extend_from_slice(&buffer[..read]); + + if request.windows(4).any(|window| window == b"\r\n\r\n") { + let request_text = String::from_utf8_lossy(&request); + + let content_length = request_text + .lines() + .find_map(|line| { + let (name, value) = line.split_once(':')?; + + if name.eq_ignore_ascii_case("content-length") { + value.trim().parse::().ok() + } else { + None + } + }) + .unwrap_or(0); + + let header_end = request + .windows(4) + .position(|window| window == b"\r\n\r\n") + .expect("header terminator") + + 4; + + if request.len() >= header_end + content_length { + break; + } + } + } + + let request_text = String::from_utf8_lossy(&request); + + assert!( + request_text.contains("Authorization: AWS4-HMAC-SHA256") + || request_text.contains("authorization: AWS4-HMAC-SHA256"), + "request should contain AWS SigV4 authorization header:\n{request_text}" + ); + + assert!( + request_text.contains("test-audio"), + "request should contain encoded audio payload:\n{request_text}" + ); + + assert!( + request_text.contains("transcribe this"), + "request should preserve instruction:\n{request_text}" + ); + + let response_body = json!({ + "output": { + "message": { + "content": [ + { + "text": "hello" + } + ] + } + } + }) + .to_string(); + + let response = format!( + "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}", + response_body.len(), + response_body, + ); + + socket + .write_all(response.as_bytes()) + .await + .expect("write response"); + }); + + let mut params = optional_params(); + + params.insert("prompt".to_string(), json!("transcribe this")); + + let api_base = format!("http://{address}"); + + let response = audio_transcription(AudioTranscriptionRequest { + model: "amazon.nova-2-sonic-v1:0", + audio: json!({ + "data": "test-audio", + "format": "wav", + "filename": "sample.wav" + }), + api_key: None, + api_base: Some(&api_base), + custom_llm_provider: Some("bedrock"), + extra_headers: None, + optional_params: params, + timeout: None, + }) + .await + .expect("audio transcription succeeds"); + + server.await.expect("server completes"); + + assert_eq!(response.text, "hello"); +} diff --git a/litellm-rust/crates/core/src/audio_transcription/types.rs b/litellm-rust/crates/core/src/audio_transcription/types.rs index 3a9e1ecd88c..d0a888f1dd5 100644 --- a/litellm-rust/crates/core/src/audio_transcription/types.rs +++ b/litellm-rust/crates/core/src/audio_transcription/types.rs @@ -1,5 +1,68 @@ +use std::time::Duration; + use serde::{Deserialize, Serialize}; -use serde_json::Value; +use serde_json::{Map, Value}; + +use super::transformation::AudioTranscriptionProviderConfig; + +/// Host-agnostic input for `audio_transcription`. +/// +/// Logging, guardrails, proxy metadata, and other host concerns intentionally +/// do not belong in this type. +pub struct AudioTranscriptionRequest<'a> { + pub model: &'a str, + pub audio: Value, + pub api_key: Option<&'a str>, + pub api_base: Option<&'a str>, + pub custom_llm_provider: Option<&'a str>, + pub extra_headers: Option>, + pub optional_params: Map, + pub timeout: Option, +} + +/// Provider-transformed audio request owned by core. +/// +/// Hosts may inspect request metadata and replace the transformed body for +/// lifecycle integrations such as guardrails, but provider configuration, +/// authentication state, and transport details remain private to core. +#[derive(Clone)] +pub struct PreparedAudioTranscriptionRequest { + pub(crate) model: String, + pub(crate) custom_llm_provider: String, + pub(crate) config: &'static dyn AudioTranscriptionProviderConfig, + pub(crate) url: String, + pub(crate) body: Value, + pub(crate) upstream_headers: Vec<(String, String)>, + pub(crate) optional_params: Map, + pub(crate) timeout: Option, +} + +impl PreparedAudioTranscriptionRequest { + pub fn model(&self) -> &str { + &self.model + } + + pub fn custom_llm_provider(&self) -> &str { + &self.custom_llm_provider + } + + pub fn url(&self) -> &str { + &self.url + } + + pub fn body(&self) -> &Value { + &self.body + } + + /// Replace the provider-transformed request body. + /// + /// This exists for host lifecycle integrations such as during-call + /// guardrails. Authentication is intentionally performed after this step + /// so signatures always cover the final body sent upstream. + pub fn replace_body(&mut self, body: Value) { + self.body = body; + } +} #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] pub struct AudioTranscriptionRequestData { diff --git a/litellm-rust/crates/python-bridge/src/lib.rs b/litellm-rust/crates/python-bridge/src/lib.rs index f0cc26a0cca..f252f8f5223 100644 --- a/litellm-rust/crates/python-bridge/src/lib.rs +++ b/litellm-rust/crates/python-bridge/src/lib.rs @@ -1,11 +1,11 @@ use std::collections::HashMap; use std::time::Duration; -use litellm_ai_gateway::io::audio_transcription::{ - AudioTranscriptionRequest, audio_transcription as run_audio_transcription, -}; use litellm_ai_gateway::io::ocr::{OcrRequest, ocr as run_ocr}; use litellm_ai_gateway::io::responses_ws::ResponsesWebSocketConnection as RustResponsesWebSocketConnection; +use litellm_core::audio_transcription::{ + AudioTranscriptionRequest, audio_transcription as run_audio_transcription, +}; use litellm_core::error::CoreError; use litellm_core::messages::messages as run_messages; use litellm_core::messages::types::{AnthropicMessagesResponse, MessagesRequest}; @@ -289,15 +289,11 @@ fn transcription( extra_headers, optional_params, timeout, - callbacks: Vec::new(), - guardrails: Vec::new(), - request_metadata: Default::default(), - litellm_call_id: None, }, )) }); match result { - Ok(value) => json_to_py(py, value), + Ok(value) => json_to_py(py, value.into_json()), Err(err) => Err(core_error_to_pyerr(err)), } } @@ -333,14 +329,10 @@ fn atranscription( extra_headers, optional_params, timeout, - callbacks: Vec::new(), - guardrails: Vec::new(), - request_metadata: Default::default(), - litellm_call_id: None, }) .await .map_err(core_error_to_pyerr)?; - Python::attach(|py| json_to_py(py, value)) + Python::attach(|py| json_to_py(py, value.into_json())) }) }