mirror of
https://github.com/BerriAI/litellm.git
synced 2026-11-07 07:58:09 +00:00
feat(router-core): derive serialize on router events (#50416)
RouterEvent and the types it carries (FailureKind, Blocked, MockFailure, RetryCause, PickError) now serialize, so a host can log or forward events as JSON instead of parsing their Debug text. Events are tagged by "event", durations are seconds, failure kinds use litellm's exception class names
This commit is contained in:
parent
bc1309f18e
commit
4fe1cfe097
4 changed files with 174 additions and 6 deletions
|
|
@ -284,6 +284,7 @@ impl Error {
|
|||
strum::EnumString,
|
||||
strum::IntoStaticStr,
|
||||
strum::VariantArray,
|
||||
serde_with::SerializeDisplay,
|
||||
)]
|
||||
pub enum FailureKind {
|
||||
#[strum(serialize = "RateLimitError")]
|
||||
|
|
@ -471,6 +472,17 @@ mod tests {
|
|||
assert_eq!(<&'static str>::from(kind), wire);
|
||||
}
|
||||
|
||||
#[rstest]
|
||||
#[case::rate_limit(FailureKind::RateLimit, "RateLimitError")]
|
||||
#[case::timeout(FailureKind::Timeout, "Timeout")]
|
||||
#[case::api_connection(FailureKind::ApiConnection, "APIConnectionError")]
|
||||
fn failure_kind_serializes_as_its_exception_name(
|
||||
#[case] kind: FailureKind,
|
||||
#[case] wire: &str,
|
||||
) {
|
||||
assert_eq!(serde_json::to_value(kind).unwrap(), serde_json::json!(wire));
|
||||
}
|
||||
|
||||
#[rstest]
|
||||
#[case::empty("")]
|
||||
#[case::lowercase("ratelimiterror")]
|
||||
|
|
|
|||
|
|
@ -342,7 +342,10 @@ pub struct Output {
|
|||
}
|
||||
|
||||
/// How a provider blocked a response the call otherwise served.
|
||||
#[derive(Clone, Copy, Debug, PartialEq, Eq, strum::IntoStaticStr, strum::EnumString)]
|
||||
#[derive(
|
||||
Clone, Copy, Debug, PartialEq, Eq, serde::Serialize, strum::IntoStaticStr, strum::EnumString,
|
||||
)]
|
||||
#[serde(rename_all = "snake_case")]
|
||||
#[strum(serialize_all = "snake_case")]
|
||||
pub enum Blocked {
|
||||
/// A chat response whose first choice was content-filtered (or that has no choices).
|
||||
|
|
@ -402,8 +405,10 @@ pub enum SelectStage {
|
|||
OverrideCheck(RoutingStrategy),
|
||||
}
|
||||
|
||||
/// Routing events, delivered through [`Host::observe`].
|
||||
#[derive(Clone, Debug, PartialEq)]
|
||||
/// Routing events, delivered through [`Host::observe`]. They serialize as JSON objects tagged by
|
||||
/// `event` (the variant name), durations in seconds.
|
||||
#[derive(Clone, Debug, PartialEq, serde::Serialize)]
|
||||
#[serde(tag = "event")]
|
||||
pub enum RouterEvent {
|
||||
/// A pick for `model` (the level's group) begins: its route phase opens.
|
||||
PickStarted {
|
||||
|
|
@ -413,6 +418,7 @@ pub enum RouterEvent {
|
|||
model_group: String,
|
||||
deployment_id: String,
|
||||
/// How long the pick took (`_acompletion`'s `async_get_available_deployment` service event).
|
||||
#[serde(serialize_with = "seconds")]
|
||||
pick: Duration,
|
||||
},
|
||||
Succeeded {
|
||||
|
|
@ -425,11 +431,13 @@ pub enum RouterEvent {
|
|||
},
|
||||
CooldownSet {
|
||||
deployment_id: String,
|
||||
#[serde(serialize_with = "seconds")]
|
||||
cooldown: Duration,
|
||||
},
|
||||
Retry {
|
||||
model_group: String,
|
||||
retry: u32,
|
||||
#[serde(serialize_with = "seconds")]
|
||||
delay: Duration,
|
||||
},
|
||||
Fallback {
|
||||
|
|
@ -584,6 +592,13 @@ impl PickError {
|
|||
}
|
||||
}
|
||||
|
||||
/// A pick's error serializes as its message.
|
||||
impl serde::Serialize for PickError {
|
||||
fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
|
||||
serializer.collect_str(&self.0)
|
||||
}
|
||||
}
|
||||
|
||||
impl PartialEq for PickError {
|
||||
fn eq(&self, other: &Self) -> bool {
|
||||
format!("{:?}", self.0) == format!("{:?}", other.0)
|
||||
|
|
@ -591,7 +606,8 @@ impl PartialEq for PickError {
|
|||
}
|
||||
|
||||
/// Which error a [`RouterEvent::RetryLogged`] records.
|
||||
#[derive(Clone, Debug, PartialEq)]
|
||||
#[derive(Clone, Debug, PartialEq, serde::Serialize)]
|
||||
#[serde(tag = "cause", content = "error", rename_all = "snake_case")]
|
||||
pub enum RetryCause {
|
||||
/// The last attempt's failure.
|
||||
Attempt,
|
||||
|
|
@ -745,3 +761,83 @@ pub trait Host: Send + Sync {
|
|||
Ok(Value::Null)
|
||||
}
|
||||
}
|
||||
|
||||
fn seconds<S: serde::Serializer>(duration: &Duration, serializer: S) -> Result<S::Ok, S::Error> {
|
||||
serializer.serialize_f64(duration.as_secs_f64())
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use rstest::rstest;
|
||||
use serde_json::{Value, json};
|
||||
|
||||
use super::*;
|
||||
|
||||
#[rstest]
|
||||
#[case::selected(
|
||||
RouterEvent::Selected {
|
||||
model_group: "gpt-4o".to_owned(),
|
||||
deployment_id: "azure-eastus".to_owned(),
|
||||
pick: Duration::from_millis(1500),
|
||||
},
|
||||
json!({"event": "Selected", "model_group": "gpt-4o", "deployment_id": "azure-eastus", "pick": 1.5})
|
||||
)]
|
||||
#[case::attempt_failed(
|
||||
RouterEvent::AttemptFailed {
|
||||
deployment_id: "openai".to_owned(),
|
||||
kind: FailureKind::RateLimit,
|
||||
status: Some(429),
|
||||
},
|
||||
json!({"event": "AttemptFailed", "deployment_id": "openai", "kind": "RateLimitError", "status": 429})
|
||||
)]
|
||||
#[case::cooldown_set(
|
||||
RouterEvent::CooldownSet { deployment_id: "openai".to_owned(), cooldown: Duration::from_secs(30) },
|
||||
json!({"event": "CooldownSet", "deployment_id": "openai", "cooldown": 30.0})
|
||||
)]
|
||||
#[case::retry(
|
||||
RouterEvent::Retry { model_group: "gpt-4o".to_owned(), retry: 1, delay: Duration::ZERO },
|
||||
json!({"event": "Retry", "model_group": "gpt-4o", "retry": 1, "delay": 0.0})
|
||||
)]
|
||||
#[case::blocked(
|
||||
RouterEvent::Blocked { level: 0, model: "claude".to_owned(), blocked: Blocked::ContentFilter },
|
||||
json!({"event": "Blocked", "level": 0, "model": "claude", "blocked": "content_filter"})
|
||||
)]
|
||||
#[case::retry_logged_after_an_attempt(
|
||||
RouterEvent::RetryLogged {
|
||||
scope: 0,
|
||||
level: 0,
|
||||
model: "gpt-4o".to_owned(),
|
||||
attempted_retries: None,
|
||||
cause: RetryCause::Attempt,
|
||||
},
|
||||
json!({
|
||||
"event": "RetryLogged", "scope": 0, "level": 0, "model": "gpt-4o",
|
||||
"attempted_retries": null, "cause": {"cause": "attempt"},
|
||||
})
|
||||
)]
|
||||
#[case::retry_logged_after_a_pick(
|
||||
RouterEvent::RetryLogged {
|
||||
scope: 1,
|
||||
level: 1,
|
||||
model: "backup".to_owned(),
|
||||
attempted_retries: Some(2),
|
||||
cause: RetryCause::Pick(PickError(Arc::new(Error::UnknownModel { model: "backup".to_owned() }))),
|
||||
},
|
||||
json!({
|
||||
"event": "RetryLogged", "scope": 1, "level": 1, "model": "backup", "attempted_retries": 2,
|
||||
"cause": {"cause": "pick", "error": "no deployment serves model `backup`"},
|
||||
})
|
||||
)]
|
||||
#[case::mock_failed(
|
||||
RouterEvent::MockFailed {
|
||||
mock: MockFailure::RateLimit,
|
||||
level: 0,
|
||||
model: "gpt-4o".to_owned(),
|
||||
num_retries: None,
|
||||
},
|
||||
json!({"event": "MockFailed", "mock": "rate_limit", "level": 0, "model": "gpt-4o", "num_retries": null})
|
||||
)]
|
||||
fn an_event_serializes_tagged_by_its_name(#[case] event: RouterEvent, #[case] expected: Value) {
|
||||
assert_eq!(serde_json::to_value(&event).unwrap(), expected);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -7,7 +7,15 @@ use crate::error::{AttemptError, FailureKind};
|
|||
|
||||
/// Which mock failure a request asks for.
|
||||
#[derive(
|
||||
Clone, Copy, Debug, PartialEq, Eq, Deserialize, strum::IntoStaticStr, strum::EnumString,
|
||||
Clone,
|
||||
Copy,
|
||||
Debug,
|
||||
PartialEq,
|
||||
Eq,
|
||||
Deserialize,
|
||||
serde::Serialize,
|
||||
strum::IntoStaticStr,
|
||||
strum::EnumString,
|
||||
)]
|
||||
#[serde(rename_all = "snake_case")]
|
||||
#[strum(serialize_all = "snake_case")]
|
||||
|
|
|
|||
|
|
@ -17,7 +17,7 @@ use litellm_router_core::{
|
|||
};
|
||||
use parking_lot::Mutex;
|
||||
use rstest::{fixture, rstest};
|
||||
use serde_json::json;
|
||||
use serde_json::{Value, json};
|
||||
|
||||
#[derive(Default)]
|
||||
struct ScriptedHost {
|
||||
|
|
@ -1809,3 +1809,55 @@ fn request_for(model: &str) -> RouterRequest {
|
|||
..RouterRequest::default()
|
||||
}
|
||||
}
|
||||
|
||||
#[rstest]
|
||||
#[tokio::test]
|
||||
async fn every_event_of_a_retried_and_fallen_back_call_serializes_tagged_by_its_name(
|
||||
config: RouterConfig,
|
||||
) {
|
||||
let host = Arc::new(ScriptedHost::failing(
|
||||
"primary",
|
||||
vec![failure(FailureKind::RateLimit, 429); 3],
|
||||
));
|
||||
let router = router(
|
||||
config,
|
||||
json!({"num_retries": 1, "fallbacks": [{"primary": ["backup"]}]}),
|
||||
host.clone(),
|
||||
);
|
||||
router
|
||||
.call(Op::from("acompletion"), request())
|
||||
.await
|
||||
.unwrap();
|
||||
let tags: Value = host
|
||||
.events
|
||||
.lock()
|
||||
.iter()
|
||||
.map(|event| serde_json::to_value(event).unwrap()["event"].clone())
|
||||
.collect();
|
||||
assert_eq!(
|
||||
tags,
|
||||
json!([
|
||||
"RetriesOpened",
|
||||
"PickStarted",
|
||||
"Selected",
|
||||
"AttemptFailed",
|
||||
"CooldownSet",
|
||||
"RetryBudget",
|
||||
"RetryLogged",
|
||||
"Retry",
|
||||
"PickStarted",
|
||||
"Selected",
|
||||
"AttemptFailed",
|
||||
"CooldownSet",
|
||||
"RetryLogged",
|
||||
"FallingBack",
|
||||
"RetryLogged",
|
||||
"Fallback",
|
||||
"RetriesOpened",
|
||||
"PickStarted",
|
||||
"Selected",
|
||||
"Succeeded",
|
||||
"FallbackDone",
|
||||
])
|
||||
);
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue