diff --git a/litellm-rust/Cargo.lock b/litellm-rust/Cargo.lock index 0c31f12cdc3..163f60bca4a 100644 --- a/litellm-rust/Cargo.lock +++ b/litellm-rust/Cargo.lock @@ -4407,6 +4407,7 @@ dependencies = [ "base64 0.22.1", "criterion", "flate2", + "indexmap 2.14.0", "litellm-http", "litellm-migrate", "litellm-storage-clickhouse", diff --git a/litellm-rust/crates/python-bridge/src/lib.rs b/litellm-rust/crates/python-bridge/src/lib.rs index d269fa4015f..6e091f0594f 100644 --- a/litellm-rust/crates/python-bridge/src/lib.rs +++ b/litellm-rust/crates/python-bridge/src/lib.rs @@ -44,7 +44,10 @@ mod _native { #[pymodule_export] use crate::routes::token_counter::TokenCounter; #[pymodule_export] - use crate::routes::traces::{NativeTraceStorage, trace_decode_otlp, trace_encode_error}; + use crate::routes::traces::{ + NativeTraceStorage, trace_decode_otlp, trace_encode_error, + trace_normalized_field_definitions, + }; #[cfg(feature = "huggingface")] #[pymodule_export] use crate::tokenizer::HuggingFaceEncoding; @@ -112,6 +115,7 @@ mod tests { "NativeTraceStorage", "trace_decode_otlp", "trace_encode_error", + "trace_normalized_field_definitions", "TokenCounter", "Tokenizer", "gil_stats", diff --git a/litellm-rust/crates/python-bridge/src/routes/traces.rs b/litellm-rust/crates/python-bridge/src/routes/traces.rs index ca66e2e46be..06d3f987f27 100644 --- a/litellm-rust/crates/python-bridge/src/routes/traces.rs +++ b/litellm-rust/crates/python-bridge/src/routes/traces.rs @@ -236,6 +236,14 @@ fn spans_to_py<'py>( "events", litellm_host_python::Pythonized(&span.events).into_pyobject(py)?, )?; + row.set_item( + "normalized", + litellm_host_python::Pythonized(&span.normalized).into_pyobject(py)?, + )?; + row.set_item( + "consumed_attributes", + litellm_host_python::Pythonized(&span.consumed_attributes).into_pyobject(py)?, + )?; result.append(row)?; } Ok(result) @@ -288,3 +296,8 @@ mod tests { }); } } + +#[pyfunction] +pub fn trace_normalized_field_definitions<'py>(py: Python<'py>) -> PyResult> { + litellm_host_python::Pythonized(litellm_traces::NORMALIZED_FIELD_DEFINITIONS).into_pyobject(py) +} diff --git a/litellm-rust/crates/traces/Cargo.toml b/litellm-rust/crates/traces/Cargo.toml index 41b6037d015..b21bc886e6a 100644 --- a/litellm-rust/crates/traces/Cargo.toml +++ b/litellm-rust/crates/traces/Cargo.toml @@ -8,6 +8,7 @@ repository.workspace = true [dependencies] base64.workspace = true flate2.workspace = true +indexmap = { version = "2", features = ["serde"] } opentelemetry-proto = { workspace = true, features = ["gen-tonic-messages", "trace", "with-serde"] } prost.workspace = true time = { workspace = true, features = ["formatting"] } diff --git a/litellm-rust/crates/traces/src/error.rs b/litellm-rust/crates/traces/src/error.rs index 18fa4af9b53..9738333b113 100644 --- a/litellm-rust/crates/traces/src/error.rs +++ b/litellm-rust/crates/traces/src/error.rs @@ -4,4 +4,6 @@ pub enum DecodeError { InvalidPayload, #[error("OTLP trace payload exceeds the decoding budget")] TooLarge, + #[error("OTLP token count is outside the storage range")] + TokenCountOutOfRange, } diff --git a/litellm-rust/crates/traces/src/lib.rs b/litellm-rust/crates/traces/src/lib.rs index 1489b44c118..2ad8916731c 100644 --- a/litellm-rust/crates/traces/src/lib.rs +++ b/litellm-rust/crates/traces/src/lib.rs @@ -1,5 +1,6 @@ mod error; mod insert; +mod normalize; mod otlp; mod schema; mod shared; @@ -8,6 +9,9 @@ mod sql; pub use error::DecodeError; pub use insert::{InsertRow, InsertTable, encode_rows, insert_rows, insert_shared_rows}; pub use litellm_storage_clickhouse::{Connection, Error, Parameter, execute_read}; +pub use normalize::{ + NORMALIZED_FIELD_DEFINITIONS, NormalizedFieldDefinition, NormalizedSpan, ObservationType, +}; pub use otlp::{DecodedSpan, decode_otlp}; pub use schema::{ensure_schema, schema_statements}; pub use shared::{Shared, SharedIdentity}; diff --git a/litellm-rust/crates/traces/src/normalize/genai.rs b/litellm-rust/crates/traces/src/normalize/genai.rs new file mode 100644 index 00000000000..15a334ab0fd --- /dev/null +++ b/litellm-rust/crates/traces/src/normalize/genai.rs @@ -0,0 +1,63 @@ +use std::collections::BTreeMap; + +use super::{NormalizedSpan, ObservationType, SpanNormalizer, attr, first, usage_tokens}; +use crate::DecodeError; + +pub(super) struct GenAiNormalizer; + +impl SpanNormalizer for GenAiNormalizer { + fn matches(&self, _scope_name: &str, _attributes: &BTreeMap) -> bool { + true + } + + fn consumed_attributes(&self, attributes: &BTreeMap) -> [&'static str; 2] { + [ + if attr(attributes, "gen_ai.input.messages").is_empty() { + "gen_ai.tool.call.arguments" + } else { + "gen_ai.input.messages" + }, + if attr(attributes, "gen_ai.output.messages").is_empty() { + "gen_ai.tool.call.result" + } else { + "gen_ai.output.messages" + }, + ] + } + + fn normalize( + &self, + _name: &str, + parent_span_id: &str, + attributes: &BTreeMap, + ) -> Result { + let (input_tokens, output_tokens) = usage_tokens(attributes)?; + let observation_type = match attr(attributes, "gen_ai.operation.name") { + "invoke_agent" => ObservationType::Agent, + "chat" | "text_completion" | "generate_content" => ObservationType::Llm, + "execute_tool" => ObservationType::Tool, + _ if parent_span_id.is_empty() => ObservationType::Agent, + _ => ObservationType::Chain, + }; + Ok(NormalizedSpan { + observation_type, + agent_name: attr(attributes, "gen_ai.agent.name").to_owned(), + litellm_request_id: attr(attributes, "gen_ai.response.id").to_owned(), + model: first(attributes, "gen_ai.request.model", "gen_ai.response.model").to_owned(), + input_tokens, + output_tokens, + input: first( + attributes, + "gen_ai.input.messages", + "gen_ai.tool.call.arguments", + ) + .to_owned(), + output: first( + attributes, + "gen_ai.output.messages", + "gen_ai.tool.call.result", + ) + .to_owned(), + }) + } +} diff --git a/litellm-rust/crates/traces/src/normalize/langsmith.rs b/litellm-rust/crates/traces/src/normalize/langsmith.rs new file mode 100644 index 00000000000..bfe7a2c1796 --- /dev/null +++ b/litellm-rust/crates/traces/src/normalize/langsmith.rs @@ -0,0 +1,468 @@ +use std::{collections::BTreeMap, io}; + +use indexmap::IndexMap; +use serde::{Deserialize, Deserializer, Serialize, de::DeserializeOwned}; +use serde_json::{Value, ser::Formatter}; + +use super::{NormalizedSpan, ObservationType, SpanNormalizer, attr, usage_tokens}; +use crate::DecodeError; + +pub(super) struct LangSmithNormalizer; + +#[derive(Deserialize)] +#[serde(untagged)] +enum MessageContent { + Text(String), + Blocks(Vec), + Other(Value), +} + +impl MessageContent { + fn display_text(&self) -> String { + match self { + Self::Text(text) => text.clone(), + Self::Blocks(blocks) => blocks + .iter() + .filter_map(|block| match block { + ContentBlock::Text { text } => Some(text.as_str()), + ContentBlock::Hidden(kind) => match kind { + HiddenBlock::Reasoning + | HiddenBlock::Thinking + | HiddenBlock::RedactedThinking + | HiddenBlock::FunctionCall + | HiddenBlock::ToolUse + | HiddenBlock::ToolCall => None, + }, + }) + .collect::>() + .join("\n\n"), + Self::Other(value) => encode(value), + } + } +} + +#[derive(Deserialize)] +#[serde(untagged)] +enum ContentBlock { + Text { text: String }, + Hidden(HiddenBlock), +} + +#[derive(Deserialize)] +#[serde(tag = "type", rename_all = "snake_case")] +enum HiddenBlock { + Reasoning, + Thinking, + RedactedThinking, + FunctionCall, + ToolUse, + ToolCall, +} + +#[derive(Deserialize, Serialize)] +#[serde(transparent)] +struct RawToolCall(IndexMap); + +#[derive(Deserialize)] +struct ResponseMetadata { + id: Option, +} + +#[derive(Deserialize)] +struct RawMessage { + kwargs: Option>, + #[serde(rename = "type")] + kind: Option, + role: Option, + content: Option, + tool_calls: Option>, + name: Option, + response_metadata: Option, +} + +impl RawMessage { + fn unwrapped(&self) -> &Self { + self.kwargs.as_deref().unwrap_or(self) + } + + fn normalized(&self) -> NormalizedMessage<'_> { + let fields = self.unwrapped(); + let raw_role = fields + .kind + .as_deref() + .filter(|role| !role.is_empty()) + .or_else(|| fields.role.as_deref().filter(|role| !role.is_empty())) + .unwrap_or_default(); + let role = match raw_role { + "human" => "user", + "ai" => "assistant", + other => other, + }; + NormalizedMessage { + role, + content: fields + .content + .as_ref() + .map_or_else(String::new, MessageContent::display_text), + tool_calls: fields + .tool_calls + .as_deref() + .filter(|calls| !calls.is_empty()), + name: (role == "tool") + .then_some(fields.name.as_ref()) + .flatten() + .filter(|name| !name.is_null() && name != &&Value::String(String::new())), + } + } +} + +#[derive(Serialize)] +struct NormalizedMessage<'a> { + role: &'a str, + content: String, + #[serde(skip_serializing_if = "Option::is_none")] + tool_calls: Option<&'a [RawToolCall]>, + #[serde(skip_serializing_if = "Option::is_none")] + name: Option<&'a Value>, +} + +enum MessageBatch { + Flat(Vec), + Nested(Vec>), +} + +impl<'de> Deserialize<'de> for MessageBatch { + fn deserialize>(deserializer: D) -> Result { + let value = Value::deserialize(deserializer)?; + let Value::Array(items) = value else { + return Err(serde::de::Error::custom("messages must be an array")); + }; + let parse = |items: Vec| { + items + .into_iter() + .filter_map(|item| serde_json::from_value(item).ok()) + .collect() + }; + Ok(if items.first().is_some_and(Value::is_array) { + Self::Nested( + items + .into_iter() + .filter_map(|item| item.as_array().cloned()) + .map(parse) + .collect(), + ) + } else { + Self::Flat(parse(items)) + }) + } +} + +fn lenient<'de, D: Deserializer<'de>, T: DeserializeOwned>( + deserializer: D, +) -> Result, D::Error> { + let value = Value::deserialize(deserializer)?; + Ok(serde_json::from_value(value).ok()) +} + +impl MessageBatch { + fn first_batch(&self) -> &[RawMessage] { + match self { + Self::Flat(messages) => messages, + Self::Nested(batches) => batches.first().map(Vec::as_slice).unwrap_or_default(), + } + } + + fn agent_messages(&self) -> &[RawMessage] { + match self { + Self::Flat(messages) => messages, + Self::Nested(_) => &[], + } + } +} + +#[derive(Deserialize)] +struct GenerationMessage { + kwargs: Option, +} + +#[derive(Deserialize)] +struct Generation { + message: Option, +} + +#[derive(Default, Deserialize)] +struct Payload { + #[serde(default, deserialize_with = "lenient")] + messages: Option, + #[serde(default, deserialize_with = "lenient")] + generations: Option>>, +} + +#[derive(Deserialize)] +struct Command { + update: CommandUpdate, +} + +#[derive(Deserialize)] +struct CommandUpdate { + messages: Vec, +} + +#[derive(Deserialize)] +struct ContentValue { + content: Value, +} + +struct SpanIo { + input: String, + output: String, + request_id: String, +} + +struct PythonJsonFormatter; + +impl Formatter for PythonJsonFormatter { + fn begin_array_value( + &mut self, + writer: &mut W, + first: bool, + ) -> io::Result<()> { + if first { + Ok(()) + } else { + writer.write_all(b", ") + } + } + + fn begin_object_key( + &mut self, + writer: &mut W, + first: bool, + ) -> io::Result<()> { + if first { + Ok(()) + } else { + writer.write_all(b", ") + } + } + + fn begin_object_value(&mut self, writer: &mut W) -> io::Result<()> { + writer.write_all(b": ") + } +} + +fn encode(value: &T) -> String { + let mut output = Vec::new(); + let mut serializer = serde_json::Serializer::with_formatter(&mut output, PythonJsonFormatter); + if value.serialize(&mut serializer).is_err() { + return String::new(); + } + String::from_utf8(output).unwrap_or_default() +} + +fn normalized_messages(messages: &[RawMessage]) -> String { + encode( + &messages + .iter() + .map(RawMessage::normalized) + .collect::>(), + ) +} + +fn span_type( + name: &str, + parent_span_id: &str, + attributes: &BTreeMap, +) -> ObservationType { + match attr(attributes, "langsmith.span.kind") { + "llm" => ObservationType::Llm, + "tool" => ObservationType::Tool, + _ if parent_span_id.is_empty() + || name == attr(attributes, "langsmith.metadata.lc_agent_name") => + { + ObservationType::Agent + } + _ if [ + ".wrap_model_call", + ".wrap_tool_call", + ".before_agent", + ".after_agent", + ".before_model", + ".after_model", + ] + .iter() + .any(|suffix| name.ends_with(suffix)) => + { + ObservationType::Framework + } + _ => ObservationType::Chain, + } +} + +fn tool_output(raw_completion: &str) -> String { + let completion = serde_json::from_str::(raw_completion).unwrap_or(Value::Null); + let raw = completion.get("output").cloned().unwrap_or(completion); + let selected = serde_json::from_value::(raw.clone()) + .ok() + .and_then(|command| command.update.messages.into_iter().last()) + .unwrap_or(raw); + let output = serde_json::from_value::(selected.clone()) + .map(|message| message.content) + .unwrap_or(selected); + output + .as_str() + .map(str::to_owned) + .unwrap_or_else(|| encode(&output)) +} + +fn span_io(kind: ObservationType, attributes: &BTreeMap) -> SpanIo { + let raw_prompt = attr(attributes, "gen_ai.prompt"); + let raw_completion = attr(attributes, "gen_ai.completion"); + let prompt = serde_json::from_str::(raw_prompt).unwrap_or_default(); + let completion = serde_json::from_str::(raw_completion).unwrap_or_default(); + if kind == ObservationType::Llm + && serde_json::from_str::(raw_completion).is_ok_and(|value| value.is_object()) + { + let input = prompt.messages.as_ref().map_or_else( + || "[]".to_owned(), + |messages| normalized_messages(messages.first_batch()), + ); + let generation = completion + .generations + .as_ref() + .and_then(|batches| batches.first()) + .and_then(|batch| batch.first()) + .and_then(|generation| generation.message.as_ref()) + .and_then(|message| message.kwargs.as_ref()); + if let Some(generation) = generation { + let id = generation + .response_metadata + .as_ref() + .and_then(|metadata| metadata.id.as_deref()) + .unwrap_or_default() + .to_owned(); + return SpanIo { + input, + output: encode(&generation.normalized()), + request_id: id, + }; + } + return SpanIo { + input, + output: raw_completion.to_owned(), + request_id: String::new(), + }; + } + if kind == ObservationType::Tool { + return SpanIo { + input: raw_prompt.to_owned(), + output: tool_output(raw_completion), + request_id: String::new(), + }; + } + if kind == ObservationType::Agent { + let input = prompt + .messages + .as_ref() + .filter(|messages| !messages.agent_messages().is_empty()) + .map_or_else( + || raw_prompt.to_owned(), + |messages| normalized_messages(messages.agent_messages()), + ); + let output = completion + .messages + .as_ref() + .and_then(|messages| messages.agent_messages().last()) + .map_or_else( + || raw_completion.to_owned(), + |message| encode(&message.normalized()), + ); + return SpanIo { + input, + output, + request_id: String::new(), + }; + } + SpanIo { + input: raw_prompt.to_owned(), + output: raw_completion.to_owned(), + request_id: String::new(), + } +} + +impl SpanNormalizer for LangSmithNormalizer { + fn matches(&self, scope_name: &str, attributes: &BTreeMap) -> bool { + scope_name == "langsmith" || attributes.contains_key("langsmith.span.kind") + } + + fn consumed_attributes(&self, _attributes: &BTreeMap) -> [&'static str; 2] { + ["gen_ai.prompt", "gen_ai.completion"] + } + + fn normalize( + &self, + name: &str, + parent_span_id: &str, + attributes: &BTreeMap, + ) -> Result { + let (input_tokens, output_tokens) = usage_tokens(attributes)?; + let observation_type = span_type(name, parent_span_id, attributes); + let io = span_io(observation_type, attributes); + Ok(NormalizedSpan { + observation_type, + agent_name: attr(attributes, "langsmith.metadata.lc_agent_name").to_owned(), + litellm_request_id: io.request_id, + model: attr(attributes, "gen_ai.request.model").to_owned(), + input_tokens, + output_tokens, + input: io.input, + output: io.output, + }) + } +} + +#[cfg(test)] +mod tests { + use std::collections::BTreeMap; + + use rstest::rstest; + use serde_json::Value; + + use super::{ObservationType, span_io}; + + #[rstest] + fn malformed_messages_preserve_valid_input_and_response_id() { + let attributes = BTreeMap::from([ + ( + "gen_ai.prompt".to_owned(), + r#"{"messages":[[{"kwargs":{"type":"human","content":"hello"}},null]]}"#.to_owned(), + ), + ( + "gen_ai.completion".to_owned(), + r#"{"messages":"unexpected","generations":[[{"message":{"kwargs":{"type":"ai","content":"hi","response_metadata":{"id":"response-1"}}}}]]}"#.to_owned(), + ), + ]); + let io = span_io(ObservationType::Llm, &attributes); + let input: Value = serde_json::from_str(&io.input).expect("normalized input"); + assert_eq!(input.as_array().expect("messages").len(), 1); + assert_eq!(input[0]["content"], "hello"); + assert_eq!(io.request_id, "response-1"); + } + + #[rstest] + fn explicit_null_tool_output_is_preserved() { + let attributes = BTreeMap::from([( + "gen_ai.completion".to_owned(), + r#"{"output":null}"#.to_owned(), + )]); + let io = span_io(ObservationType::Tool, &attributes); + assert_eq!(io.output, "null"); + } + + #[rstest] + fn absent_llm_messages_render_as_an_empty_list() { + let attributes = BTreeMap::from([("gen_ai.completion".to_owned(), "{}".to_owned())]); + let io = span_io(ObservationType::Llm, &attributes); + assert_eq!(io.input, "[]"); + } +} diff --git a/litellm-rust/crates/traces/src/normalize/mod.rs b/litellm-rust/crates/traces/src/normalize/mod.rs new file mode 100644 index 00000000000..b4d5a4a9d06 --- /dev/null +++ b/litellm-rust/crates/traces/src/normalize/mod.rs @@ -0,0 +1,239 @@ +use std::collections::BTreeMap; + +use crate::DecodeError; +use serde::Serialize; + +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)] +#[serde(rename_all = "lowercase")] +pub enum ObservationType { + Agent, + Llm, + Tool, + Chain, + Framework, +} + +#[derive(Debug, Serialize)] +pub struct NormalizedSpan { + pub observation_type: ObservationType, + pub agent_name: String, + pub litellm_request_id: String, + pub model: String, + pub input_tokens: u32, + pub output_tokens: u32, + pub input: String, + pub output: String, +} + +pub(crate) struct Normalization { + pub span: NormalizedSpan, + pub consumed_attributes: [&'static str; 2], +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)] +pub struct NormalizedFieldDefinition { + pub name: &'static str, + pub clickhouse_column: &'static str, + pub clickhouse_type: &'static str, + pub meaning: &'static str, +} + +pub const NORMALIZED_FIELD_DEFINITIONS: [NormalizedFieldDefinition; 8] = [ + NormalizedFieldDefinition { + name: "observation_type", + clickhouse_column: "ObservationType", + clickhouse_type: "LowCardinality(String)", + meaning: "Agent, LLM, tool, chain, or framework span", + }, + NormalizedFieldDefinition { + name: "agent_name", + clickhouse_column: "AgentName", + clickhouse_type: "LowCardinality(String)", + meaning: "Agent associated with this span", + }, + NormalizedFieldDefinition { + name: "litellm_request_id", + clickhouse_column: "LiteLLMRequestId", + clickhouse_type: "String", + meaning: "LiteLLM response ID used to link a span to a spend log", + }, + NormalizedFieldDefinition { + name: "model", + clickhouse_column: "Model", + clickhouse_type: "LowCardinality(String)", + meaning: "Model used by this span", + }, + NormalizedFieldDefinition { + name: "input_tokens", + clickhouse_column: "InputTokens", + clickhouse_type: "UInt32", + meaning: "Input token count", + }, + NormalizedFieldDefinition { + name: "output_tokens", + clickhouse_column: "OutputTokens", + clickhouse_type: "UInt32", + meaning: "Output token count", + }, + NormalizedFieldDefinition { + name: "input", + clickhouse_column: "Input", + clickhouse_type: "String", + meaning: "Normalized input payload", + }, + NormalizedFieldDefinition { + name: "output", + clickhouse_column: "Output", + clickhouse_type: "String", + meaning: "Normalized output payload", + }, +]; + +trait SpanNormalizer { + fn matches(&self, scope_name: &str, attributes: &BTreeMap) -> bool; + fn consumed_attributes(&self, attributes: &BTreeMap) -> [&'static str; 2]; + fn normalize( + &self, + name: &str, + parent_span_id: &str, + attributes: &BTreeMap, + ) -> Result; +} + +mod genai; +mod langsmith; +mod openinference; + +use genai::GenAiNormalizer; +use langsmith::LangSmithNormalizer; +use openinference::OpenInferenceNormalizer; + +fn attr<'a>(attributes: &'a BTreeMap, key: &str) -> &'a str { + attributes.get(key).map(String::as_str).unwrap_or_default() +} + +fn first<'a>(attributes: &'a BTreeMap, left: &str, right: &str) -> &'a str { + let value = attr(attributes, left); + if value.is_empty() { + attr(attributes, right) + } else { + value + } +} + +fn tokens(attributes: &BTreeMap, key: &str) -> Result { + let value = attr(attributes, key).trim(); + if value.is_empty() { + return Ok(0); + } + match value.parse::() { + Ok(number) if (0..=u32::MAX as i128).contains(&number) => Ok(number as u32), + Ok(_) => Err(DecodeError::TokenCountOutOfRange), + Err(_) + if value + .trim_start_matches(['+', '-']) + .bytes() + .all(|byte| byte.is_ascii_digit()) => + { + Err(DecodeError::TokenCountOutOfRange) + } + Err(_) => Ok(0), + } +} + +fn usage_tokens(attributes: &BTreeMap) -> Result<(u32, u32), DecodeError> { + Ok(( + tokens(attributes, "gen_ai.usage.input_tokens")?, + tokens(attributes, "gen_ai.usage.output_tokens")?, + )) +} + +pub fn normalize( + scope_name: &str, + name: &str, + parent_span_id: &str, + attributes: &BTreeMap, +) -> Result { + let normalizers: [&dyn SpanNormalizer; 3] = [ + &LangSmithNormalizer, + &OpenInferenceNormalizer, + &GenAiNormalizer, + ]; + let normalizer = normalizers + .into_iter() + .find(|normalizer| normalizer.matches(scope_name, attributes)) + .expect("GenAI fallback always matches"); + Ok(Normalization { + span: normalizer.normalize(name, parent_span_id, attributes)?, + consumed_attributes: normalizer.consumed_attributes(attributes), + }) +} + +#[cfg(test)] +mod tests { + use std::collections::{BTreeMap, BTreeSet}; + + use rstest::rstest; + + use super::{NORMALIZED_FIELD_DEFINITIONS, ObservationType, normalize}; + + #[rstest] + #[case::langsmith("langsmith", [("langsmith.span.kind", "llm"), ("openinference.span.kind", "TOOL")], ObservationType::Llm)] + #[case::openinference("other", [("openinference.span.kind", "LLM"), ("gen_ai.operation.name", "execute_tool")], ObservationType::Llm)] + #[case::genai("other", [("gen_ai.operation.name", "execute_tool"), ("gen_ai.usage.input_tokens", "7")], ObservationType::Tool)] + fn convention_dispatch_preserves_precedence( + #[case] scope: &str, + #[case] attributes: [(&str, &str); 2], + #[case] expected: ObservationType, + ) { + let attributes = attributes + .into_iter() + .map(|(key, value)| (key.to_owned(), value.to_owned())) + .collect(); + let fields = normalize(scope, "step", "parent", &attributes) + .expect("valid tokens") + .span; + assert_eq!(fields.observation_type, expected); + if expected == ObservationType::Tool { + assert_eq!(fields.input_tokens, 7); + } + } + + #[rstest] + fn field_definitions_match_serialized_normalized_span() { + let fields = normalize("", "root", "", &BTreeMap::new()) + .expect("valid tokens") + .span; + let serialized = serde_json::to_value(fields).expect("serializable fields"); + let keys: BTreeSet<_> = serialized + .as_object() + .expect("field object") + .keys() + .map(String::as_str) + .collect(); + let mapped: BTreeSet<_> = NORMALIZED_FIELD_DEFINITIONS + .iter() + .map(|field| field.name) + .collect(); + assert_eq!(keys, mapped); + } + + #[rstest] + fn token_counts_accept_surrounding_whitespace() { + let attributes = + BTreeMap::from([("gen_ai.usage.input_tokens".to_owned(), " 7 ".to_owned())]); + let fields = normalize("", "root", "", &attributes) + .expect("valid tokens") + .span; + assert_eq!(fields.input_tokens, 7); + } + + #[rstest] + #[case::negative("-1")] + #[case::overflow("4294967296")] + fn token_counts_outside_storage_range_are_rejected(#[case] value: &str) { + let attributes = + BTreeMap::from([("gen_ai.usage.input_tokens".to_owned(), value.to_owned())]); + assert!(normalize("", "root", "", &attributes).is_err()); + } +} diff --git a/litellm-rust/crates/traces/src/normalize/openinference.rs b/litellm-rust/crates/traces/src/normalize/openinference.rs new file mode 100644 index 00000000000..b6156d8f24f --- /dev/null +++ b/litellm-rust/crates/traces/src/normalize/openinference.rs @@ -0,0 +1,53 @@ +use std::collections::BTreeMap; + +use super::{NormalizedSpan, ObservationType, SpanNormalizer, attr, tokens, usage_tokens}; +use crate::DecodeError; + +pub(super) struct OpenInferenceNormalizer; + +impl SpanNormalizer for OpenInferenceNormalizer { + fn matches(&self, _scope_name: &str, attributes: &BTreeMap) -> bool { + attributes.contains_key("openinference.span.kind") + } + + fn consumed_attributes(&self, _attributes: &BTreeMap) -> [&'static str; 2] { + ["input.value", "output.value"] + } + + fn normalize( + &self, + _name: &str, + parent_span_id: &str, + attributes: &BTreeMap, + ) -> Result { + let (usage_input, usage_output) = usage_tokens(attributes)?; + let observation_type = match attr(attributes, "openinference.span.kind") + .to_ascii_uppercase() + .as_str() + { + "AGENT" => ObservationType::Agent, + "LLM" => ObservationType::Llm, + "TOOL" => ObservationType::Tool, + _ if parent_span_id.is_empty() => ObservationType::Agent, + _ => ObservationType::Chain, + }; + Ok(NormalizedSpan { + observation_type, + agent_name: attr(attributes, "agent.name").to_owned(), + litellm_request_id: String::new(), + model: attr(attributes, "llm.model_name").to_owned(), + input_tokens: if attributes.contains_key("llm.token_count.prompt") { + tokens(attributes, "llm.token_count.prompt")? + } else { + usage_input + }, + output_tokens: if attributes.contains_key("llm.token_count.completion") { + tokens(attributes, "llm.token_count.completion")? + } else { + usage_output + }, + input: attr(attributes, "input.value").to_owned(), + output: attr(attributes, "output.value").to_owned(), + }) + } +} diff --git a/litellm-rust/crates/traces/src/otlp/mod.rs b/litellm-rust/crates/traces/src/otlp/mod.rs index fcc42082151..1beef48fe2c 100644 --- a/litellm-rust/crates/traces/src/otlp/mod.rs +++ b/litellm-rust/crates/traces/src/otlp/mod.rs @@ -6,7 +6,7 @@ mod wire; use serde::Serialize; use std::collections::BTreeMap; -use crate::{DecodeError, Shared}; +use crate::{DecodeError, NormalizedSpan, Shared}; #[derive(Serialize)] pub struct DecodedEvent { @@ -31,6 +31,8 @@ pub struct DecodedSpan { pub status_code: String, pub status_message: String, pub events: Vec, + pub normalized: NormalizedSpan, + pub consumed_attributes: [&'static str; 2], } pub fn decode_otlp( diff --git a/litellm-rust/crates/traces/src/otlp/span.rs b/litellm-rust/crates/traces/src/otlp/span.rs index fa993f71e3c..1c1e53e756c 100644 --- a/litellm-rust/crates/traces/src/otlp/span.rs +++ b/litellm-rust/crates/traces/src/otlp/span.rs @@ -10,7 +10,7 @@ use super::{ attributes::attributes, limits::{Budget, MAX_ATTRIBUTES, MAX_DECODED_SPAN_BYTES, MAX_EVENTS, MAX_SPANS}, }; -use crate::{DecodeError, Shared}; +use crate::{DecodeError, Shared, normalize::normalize}; pub(super) fn flatten(request: ExportTraceServiceRequest) -> Result, DecodeError> { let mut budget = Budget::new(MAX_DECODED_SPAN_BYTES); @@ -125,10 +125,26 @@ fn decoded_span( budget: &mut Budget, ) -> Result { let status = span.status.unwrap_or_default(); + let parent_span_id = hex_bytes(&span.parent_span_id); + let span_attributes = attributes(span.attributes, budget)?; + let normalization = normalize( + scope_name.as_ref(), + &span.name, + &parent_span_id, + &span_attributes, + )?; + let normalized = normalization.span; + budget.consume( + normalized.input.len() + + normalized.output.len() + + normalized.agent_name.len() + + normalized.litellm_request_id.len() + + normalized.model.len(), + )?; Ok(DecodedSpan { trace_id: hex_bytes(&span.trace_id), span_id: hex_bytes(&span.span_id), - parent_span_id: hex_bytes(&span.parent_span_id), + parent_span_id, trace_state: span.trace_state, name: span.name, kind: SpanKind::try_from(span.kind) @@ -143,7 +159,7 @@ fn decoded_span( })?, scope_name: budget.clone_shared(scope_name, String::len)?, scope_version: budget.clone_shared(scope_version, String::len)?, - attributes: attributes(span.attributes, budget)?, + attributes: span_attributes, start_ns: span.start_time_unix_nano, end_ns: span.end_time_unix_nano, status_code: StatusCode::try_from(status.code) @@ -162,5 +178,7 @@ fn decoded_span( }) }) .collect::, DecodeError>>()?, + normalized, + consumed_attributes: normalization.consumed_attributes, }) } diff --git a/litellm-rust/crates/traces/tests/migrations.rs b/litellm-rust/crates/traces/tests/migrations.rs index 1d21010b0d8..a7fb3b57db2 100644 --- a/litellm-rust/crates/traces/tests/migrations.rs +++ b/litellm-rust/crates/traces/tests/migrations.rs @@ -2,8 +2,8 @@ use std::{collections::BTreeMap, time::Duration}; use litellm_http::Client; use litellm_traces::{ - Connection, Error, InsertTable, Parameter, ReadQuery, encode_rows, ensure_schema, - execute_named_read, execute_read, schema_statements, + Connection, Error, InsertTable, NORMALIZED_FIELD_DEFINITIONS, Parameter, ReadQuery, + encode_rows, ensure_schema, execute_named_read, execute_read, schema_statements, }; use rstest::{fixture, rstest}; use testcontainers_modules::{ @@ -210,6 +210,43 @@ async fn schema_supports_span_rollups_and_spend_joins( Ok(()) } +#[rstest] +#[tokio::test] +async fn normalized_fields_match_clickhouse_catalog( + #[future(awt)] database: TestResult, +) -> TestResult { + let database = database?; + ensure_schema( + &database.client, + &Connection::writer(&database.url)?, + "trace_test", + 7, + 14, + ) + .await?; + let catalog = read_json(&database, "SELECT name, type FROM system.columns WHERE database = 'trace_test' AND table = 'otel_traces'").await?; + let columns: BTreeMap<&str, &str> = catalog["data"] + .as_array() + .expect("catalog rows") + .iter() + .map(|row| { + ( + row["name"].as_str().expect("column name"), + row["type"].as_str().expect("column type"), + ) + }) + .collect(); + for field in NORMALIZED_FIELD_DEFINITIONS { + assert_eq!( + columns.get(field.clickhouse_column).copied(), + Some(field.clickhouse_type), + "{}", + field.name + ); + } + Ok(()) +} + #[rstest] #[tokio::test] async fn insert_rejects_unknown_columns_even_if_url_requests_skipping_them( diff --git a/litellm-rust/crates/traces/tests/otlp.rs b/litellm-rust/crates/traces/tests/otlp.rs index 8aa2cbedeb3..3baa820d312 100644 --- a/litellm-rust/crates/traces/tests/otlp.rs +++ b/litellm-rust/crates/traces/tests/otlp.rs @@ -1,5 +1,5 @@ -use litellm_traces::Shared; use litellm_traces::decode_otlp; +use litellm_traces::{ObservationType, Shared}; use rstest::rstest; const FIXTURE: &[u8] = include_bytes!( @@ -341,3 +341,48 @@ fn escaped_attribute_expansion_is_bounded_below_four_mib( Err(litellm_traces::DecodeError::TooLarge) )); } + +#[rstest] +fn normalizes_langsmith_fixture() { + let spans = decode_otlp(FIXTURE, Some("application/json")).expect("valid OTLP export"); + let llm = spans + .iter() + .find(|span| span.name == "ChatOpenAI") + .expect("LLM span"); + assert_eq!(llm.normalized.observation_type, ObservationType::Llm); + assert_eq!(llm.normalized.agent_name, "deep_research_agent"); + assert_eq!(llm.normalized.model, "claude-sonnet-4-5"); + assert_eq!( + (llm.normalized.input_tokens, llm.normalized.output_tokens), + (3332, 467) + ); + assert_eq!( + llm.normalized.litellm_request_id, + "chatcmpl-4077bb36-9380-4a3b-9481-245700cef09a" + ); + let input: serde_json::Value = + serde_json::from_str(&llm.normalized.input).expect("message input"); + assert_eq!(input[0]["role"], "system"); + assert_eq!(input[1]["role"], "user"); + let output: serde_json::Value = + serde_json::from_str(&llm.normalized.output).expect("message output"); + assert_eq!(output["role"], "assistant"); + assert!(output["tool_calls"][0]["name"].is_string()); + assert!(output["tool_calls"][0]["id"].is_string()); + assert_eq!(output["tool_calls"][0]["type"], "tool_call"); + let root = spans + .iter() + .find(|span| span.name == "deep_research_agent") + .expect("root span"); + assert_eq!(root.normalized.observation_type, ObservationType::Agent); + assert_eq!( + root.normalized.input, + "[{\"role\": \"user\", \"content\": \"Should we store OTEL agent spans in ClickHouse or Postgres at 50k spans/sec?\"}]" + ); + let tool = spans + .iter() + .find(|span| span.name == "task") + .expect("tool span"); + assert_eq!(tool.normalized.observation_type, ObservationType::Tool); + assert!(tool.normalized.output.starts_with("Based on my research")); +} diff --git a/litellm/proxy/tracing_endpoints.py b/litellm/proxy/tracing_endpoints.py index 46d29c50c1b..b44a3605659 100644 --- a/litellm/proxy/tracing_endpoints.py +++ b/litellm/proxy/tracing_endpoints.py @@ -117,7 +117,7 @@ async def ingest_otlp_traces( return Response(content=body, media_type=media_type) -@router.get("/v1/traces", response_model=None) +@router.get("/v1/traces", response_model=TracePage) async def list_agent_traces( context: Annotated[TraceAccessContext, Depends(provide_trace_access)], start_ms: Annotated[int | None, Query(description="Window start, unix ms. Default: 24h ago")] = None, @@ -137,7 +137,7 @@ async def list_agent_traces( raise HTTPException(status_code=400, detail=str(error)) from error -@router.get("/v1/traces/{trace_id}", response_model=None) +@router.get("/v1/traces/{trace_id}", response_model=Trace) async def get_agent_trace( trace_id: str, context: Annotated[TraceAccessContext, Depends(provide_trace_access)], @@ -150,7 +150,7 @@ async def get_agent_trace( return trace -@router.get("/v1/traces/{trace_id}/spans/{span_id}", response_model=None) +@router.get("/v1/traces/{trace_id}/spans/{span_id}", response_model=SpanDetail) async def get_agent_trace_span( trace_id: str, span_id: str, diff --git a/litellm/rust_bridge/_native.pyi b/litellm/rust_bridge/_native.pyi index a3d8ba0e582..921badafb3b 100644 --- a/litellm/rust_bridge/_native.pyi +++ b/litellm/rust_bridge/_native.pyi @@ -23,6 +23,7 @@ class ProcessReservedForForking(RuntimeError): ... def trace_decode_otlp(body: bytes, content_type: str | None) -> list[DecodedSpan]: ... def trace_encode_error(message: str) -> bytes: ... +def trace_normalized_field_definitions() -> list[dict[str, str]]: ... @final class NativeTraceStorage: @@ -353,6 +354,7 @@ __all__ = [ "responses", "trace_decode_otlp", "trace_encode_error", + "trace_normalized_field_definitions", "transcription", ] diff --git a/litellm/rust_bridge/traces.py b/litellm/rust_bridge/traces.py index 06e006be89c..d779166bade 100644 --- a/litellm/rust_bridge/traces.py +++ b/litellm/rust_bridge/traces.py @@ -2,7 +2,7 @@ from collections.abc import Awaitable, Mapping, Sequence from types import MappingProxyType from typing import Final, Literal, Protocol, TypedDict, cast -from pydantic import BaseModel, ConfigDict, JsonValue, TypeAdapter +from pydantic import BaseModel, ConfigDict, Field, JsonValue, TypeAdapter from typing_extensions import ReadOnly from litellm.rust_bridge.loader import get_native_bridge @@ -13,6 +13,28 @@ class DecodedEvent(TypedDict): attributes: ReadOnly[dict[str, str]] +class NormalizedSpan(BaseModel): + model_config = ConfigDict(frozen=True, extra="forbid", strict=True) + + observation_type: Literal["agent", "llm", "tool", "chain", "framework"] + agent_name: str + litellm_request_id: str + model: str + input_tokens: int = Field(ge=0, le=2**32 - 1) + output_tokens: int = Field(ge=0, le=2**32 - 1) + input: str + output: str + + +class NormalizedFieldDefinition(BaseModel): + model_config = ConfigDict(frozen=True, extra="forbid", strict=True) + + name: str + clickhouse_column: str + clickhouse_type: str + meaning: str + + class DecodedSpan(TypedDict): trace_id: ReadOnly[str] span_id: ReadOnly[str] @@ -29,6 +51,8 @@ class DecodedSpan(TypedDict): status_code: ReadOnly[str] status_message: ReadOnly[str] events: ReadOnly[list[DecodedEvent]] + normalized: ReadOnly[NormalizedSpan] + consumed_attributes: ReadOnly[tuple[str, str]] ReadQueryName = Literal["list_traces", "trace_spans", "span_detail", "span_error", "spend_by_response_ids"] @@ -57,6 +81,8 @@ class NativeTraces(Protocol): def trace_encode_error(self, message: str) -> bytes: ... + def trace_normalized_field_definitions(self) -> list[dict[str, str]]: ... + class QueryResponse(BaseModel): model_config = ConfigDict(frozen=True) @@ -64,6 +90,7 @@ class QueryResponse(BaseModel): QUERY_PARAMETERS: Final = TypeAdapter(dict[str, str | int | list[str]]) +_FIELD_DEFINITIONS_ADAPTER: Final = TypeAdapter(tuple[NormalizedFieldDefinition, ...]) def _native() -> NativeTraces: @@ -74,7 +101,17 @@ def _native() -> NativeTraces: def decode_otlp(body: bytes, content_type: str | None) -> list[DecodedSpan]: - return _native().trace_decode_otlp(body, content_type) + return [ + {**span, "normalized": NormalizedSpan.model_validate(span["normalized"])} + for span in _native().trace_decode_otlp(body, content_type) + ] + + +def normalized_field_definitions() -> tuple[NormalizedFieldDefinition, ...]: + fields: Final = _FIELD_DEFINITIONS_ADAPTER.validate_python(_native().trace_normalized_field_definitions()) + if frozenset(field.name for field in fields) != frozenset(NormalizedSpan.model_fields): + raise ValueError("Rust and Python normalized trace fields disagree") + return fields def encode_error(message: str) -> bytes: diff --git a/litellm/tracing/decode.py b/litellm/tracing/decode.py index d8b5f70de68..e76c9aad97d 100644 --- a/litellm/tracing/decode.py +++ b/litellm/tracing/decode.py @@ -1,50 +1,23 @@ -""" -OTLP/HTTP trace export -> `SpanRow`s. - -Pure functions, no I/O. Two steps: -1. `decode_otlp()` protobuf / JSON / gzip `ExportTraceServiceRequest` -> flat spans -2. `normalize()` framework conventions -> LiteLLM columns (type, agent, input/output, - LiteLLM request id). Supported: LangSmith (LangChain, LangGraph, - Deep Agents), OTEL GenAI semconv, OpenInference. -""" - import gzip import json import zlib from collections.abc import Mapping -from dataclasses import dataclass from io import BytesIO from itertools import accumulate from types import MappingProxyType from typing import Final from pydantic import JsonValue, TypeAdapter, ValidationError -from typing_extensions import NotRequired, ReadOnly, TypedDict +from typing_extensions import ReadOnly, TypedDict from litellm.constants import OTLP_MAX_ATTRIBUTE_VALUE_BYTES, OTLP_MAX_BODY_BYTES from litellm.rust_bridge.traces import DecodedSpan from litellm.rust_bridge.traces import decode_otlp as native_decode_otlp from litellm.rust_bridge.traces import encode_error as native_encode_error -from litellm.tracing.normalizers.messages import content_text -from litellm.tracing.types import SpanRow, SpanType +from litellm.tracing.types import SpanRow -_FRAMEWORK_SUFFIXES: Final = ( - ".wrap_model_call", - ".wrap_tool_call", - ".before_agent", - ".after_agent", - ".before_model", - ".after_model", -) -_LLM_OPERATIONS: Final = frozenset({"chat", "text_completion", "generate_content"}) -_LC_ROLES: Final = MappingProxyType({"human": "user", "ai": "assistant", "system": "system", "tool": "tool"}) -_OPENINFERENCE_TYPES: Final[Mapping[str, SpanType]] = MappingProxyType({"AGENT": "agent", "LLM": "llm", "TOOL": "tool"}) - - -_JSON: Final[TypeAdapter[JsonValue]] = TypeAdapter(JsonValue) _MESSAGE_LIST: Final = TypeAdapter(tuple[dict[str, JsonValue], ...]) _MAX_JSON_ESCAPE_BYTES: Final = 6 -_MAX_TOKENS: Final = (1 << 32) - 1 class InvalidOTLPPayloadError(ValueError): @@ -55,33 +28,10 @@ class OTLPPayloadTooLargeError(OverflowError): pass -class MessageExtras(TypedDict): - tool_calls: ReadOnly[NotRequired[JsonValue]] - name: ReadOnly[NotRequired[str]] - - -class NormalizedMessage(MessageExtras): - role: ReadOnly[str] - content: ReadOnly[str] - - class OTLPError(TypedDict): message: ReadOnly[str] -@dataclass(frozen=True, slots=True) -class NormalizedSpan: - kind: SpanType - agent: str = "" - model: str = "" - request_id: str = "" - input: str = "" - output: str = "" - input_tokens: int = 0 - output_tokens: int = 0 - consumed: frozenset[str] = frozenset() - - def _truncate(value: str) -> str: encoded: Final = value.encode("utf-8") if len(encoded) <= OTLP_MAX_ATTRIBUTE_VALUE_BYTES: @@ -206,7 +156,7 @@ def _exception_message(span: DecodedSpan) -> str: def _span_row(span: DecodedSpan) -> SpanRow: attributes: Final = span["attributes"] - normalized: Final = normalize(span) + normalized: Final = span["normalized"] return SpanRow( Timestamp=span["start_ns"], TraceId=span["trace_id"], @@ -220,17 +170,17 @@ def _span_row(span: DecodedSpan) -> SpanRow: ScopeName=span["scope_name"], ScopeVersion=span["scope_version"], SpanAttributes=MappingProxyType( - {key: _truncate(value) for key, value in attributes.items() if key not in normalized.consumed} + {key: _truncate(value) for key, value in attributes.items() if key not in span["consumed_attributes"]} ), Duration=span["end_ns"] - span["start_ns"], StatusCode=span["status_code"], StatusMessage=span["status_message"] or _exception_message(span), TeamId="", ApiKeyHash="", - ObservationType=normalized.kind, - AgentName=normalized.agent, + ObservationType=normalized.observation_type, + AgentName=normalized.agent_name, Model=normalized.model, - LiteLLMRequestId=attributes.get("gen_ai.response.id") or normalized.request_id, + LiteLLMRequestId=attributes.get("gen_ai.response.id") or normalized.litellm_request_id, InputTokens=normalized.input_tokens, OutputTokens=normalized.output_tokens, Input=_truncate_payload(normalized.input), @@ -238,173 +188,6 @@ def _span_row(span: DecodedSpan) -> SpanRow: ) -def _loads(value: str) -> JsonValue: - if len(value.encode("utf-8")) > OTLP_MAX_BODY_BYTES: - return None - try: - return _JSON.validate_json(value) - except ValidationError: - return None - - -def _text(value: JsonValue) -> str: - return value if isinstance(value, str) else "" - - -def _message(value: JsonValue) -> NormalizedMessage | None: - if not isinstance(value, dict): - return None - kwargs: Final = value.get("kwargs", value) - if not isinstance(kwargs, dict): - return None - kind: Final = _text(kwargs.get("type")) or _text(kwargs.get("role")) - if not kind: - return None - calls: Final = kwargs.get("tool_calls") - if calls is not None and (not isinstance(calls, list) or not all(isinstance(call, dict) for call in calls)): - return None - role: Final = _LC_ROLES.get(kind, kind) - content: Final = kwargs.get("content", "") - name: Final = kwargs.get("name") - tool_calls: Final = MessageExtras(tool_calls=calls) if calls else MessageExtras() - tool_name: Final = MessageExtras(name=name) if role == "tool" and isinstance(name, str) else MessageExtras() - message: Final[NormalizedMessage] = { - "role": role, - "content": content_text(content), - **tool_calls, - **tool_name, - } - return message - - -def _messages(value: JsonValue, raw: str) -> str: - if not isinstance(value, list): - return raw - messages: Final = tuple(_message(item) for item in value) - return json.dumps(messages) if all(message is not None for message in messages) else raw - - -def _langsmith_type(span: DecodedSpan) -> SpanType: - attributes: Final = span["attributes"] - kind: Final = attributes.get("langsmith.span.kind", "chain") - if kind in ("llm", "tool"): - return "llm" if kind == "llm" else "tool" - if not span["parent_span_id"] or span["name"] == attributes.get("langsmith.metadata.lc_agent_name"): - return "agent" - return "framework" if span["name"].endswith(_FRAMEWORK_SUFFIXES) else "chain" - - -def _langsmith_io(kind: SpanType, attributes: Mapping[str, str]) -> tuple[str, str, str]: - raw_prompt: Final = attributes.get("gen_ai.prompt", "") - raw_completion: Final = attributes.get("gen_ai.completion", "") - prompt: Final = _loads(raw_prompt) - completion: Final = _loads(raw_completion) - messages: Final = prompt.get("messages") if isinstance(prompt, dict) else None - if kind == "llm": - batch: Final = ( - messages[0] if isinstance(messages, list) and messages and isinstance(messages[0], list) else messages - ) - generations: Final = completion.get("generations") if isinstance(completion, dict) else None - first: Final = generations[0] if isinstance(generations, list) and generations else None - item: Final = first[0] if isinstance(first, list) and first else first - message: Final = item.get("message") if isinstance(item, dict) else None - parsed: Final = _message(message) - kwargs: Final = message.get("kwargs", message) if isinstance(message, dict) else None - metadata: Final = kwargs.get("response_metadata") if isinstance(kwargs, dict) else None - request_id: Final = _text(metadata.get("id")) if isinstance(metadata, dict) else "" - return _messages(batch, raw_prompt), json.dumps(parsed) if parsed is not None else raw_completion, request_id - if kind == "tool": - output: Final = completion.get("output", completion) if isinstance(completion, dict) else completion - update: Final = output.get("update") if isinstance(output, dict) else None - updates: Final = update.get("messages") if isinstance(update, dict) else None - final: Final = updates[-1] if isinstance(updates, list) and updates else output - content: Final = final.get("content", final) if isinstance(final, dict) else final - return ( - raw_prompt, - (content if isinstance(content, str) else json.dumps(content)) if content is not None else raw_completion, - "", - ) - if kind == "agent": - outputs: Final = completion.get("messages") if isinstance(completion, dict) else None - last: Final = _message(outputs[-1]) if isinstance(outputs, list) and outputs else None - return _messages(messages, raw_prompt), json.dumps(last) if last is not None else raw_completion, "" - return raw_prompt, raw_completion, "" - - -def _to_int(value: str | None) -> int: - try: - number: Final = int(value) if value else 0 - except ValueError: - return 0 - if not 0 <= number <= _MAX_TOKENS: - raise InvalidOTLPPayloadError("OTLP token count is outside the storage range") - return number - - -def normalize(span: DecodedSpan) -> NormalizedSpan: - attributes: Final = span["attributes"] - fallback: Final[SpanType] = "agent" if not span["parent_span_id"] else "chain" - input_tokens: Final = _to_int(attributes.get("gen_ai.usage.input_tokens")) - output_tokens: Final = _to_int(attributes.get("gen_ai.usage.output_tokens")) - if span["scope_name"] == "langsmith" or "langsmith.span.kind" in attributes: - kind: Final = _langsmith_type(span) - prompt, completion, request_id = _langsmith_io(kind, attributes) - return NormalizedSpan( - kind, - attributes.get("langsmith.metadata.lc_agent_name", ""), - attributes.get("gen_ai.request.model", ""), - request_id, - prompt, - completion, - input_tokens, - output_tokens, - frozenset({"gen_ai.prompt", "gen_ai.completion"}), - ) - if "openinference.span.kind" in attributes: - return NormalizedSpan( - _OPENINFERENCE_TYPES.get(attributes["openinference.span.kind"].upper(), fallback), - attributes.get("agent.name", ""), - attributes.get("llm.model_name", ""), - "", - attributes.get("input.value", ""), - attributes.get("output.value", ""), - _to_int(attributes.get("llm.token_count.prompt")) - if "llm.token_count.prompt" in attributes - else input_tokens, - _to_int(attributes.get("llm.token_count.completion")) - if "llm.token_count.completion" in attributes - else output_tokens, - frozenset({"input.value", "output.value"}), - ) - operation: Final = attributes.get("gen_ai.operation.name", "") - genai_kind: Final[SpanType] = ( - "llm" - if operation in _LLM_OPERATIONS - else "tool" - if operation == "execute_tool" - else "agent" - if operation == "invoke_agent" - else fallback - ) - input_key: Final = ( - "gen_ai.input.messages" if attributes.get("gen_ai.input.messages") else "gen_ai.tool.call.arguments" - ) - output_key: Final = ( - "gen_ai.output.messages" if attributes.get("gen_ai.output.messages") else "gen_ai.tool.call.result" - ) - return NormalizedSpan( - genai_kind, - attributes.get("gen_ai.agent.name", ""), - attributes.get("gen_ai.request.model") or attributes.get("gen_ai.response.model", ""), - "", - attributes.get(input_key, ""), - attributes.get(output_key, ""), - input_tokens, - output_tokens, - frozenset({input_key, output_key}), - ) - - def encode_otlp_response(content_type: str | None, error: str | None = None) -> tuple[bytes, str]: media_type: Final = (content_type or "application/x-protobuf").split(";", 1)[0].strip().lower() if media_type == "application/json": diff --git a/litellm/tracing/normalizers/messages.py b/litellm/tracing/messages.py similarity index 61% rename from litellm/tracing/normalizers/messages.py rename to litellm/tracing/messages.py index 8a9aa914dfd..09f57e25308 100644 --- a/litellm/tracing/normalizers/messages.py +++ b/litellm/tracing/messages.py @@ -1,7 +1,7 @@ import json from collections.abc import Mapping from types import MappingProxyType -from typing import Any, Final, Literal, TypeAlias +from typing import Final, Literal, TypeAlias from pydantic import BaseModel, ConfigDict, TypeAdapter, ValidationError @@ -37,20 +37,3 @@ def content_text(content: object) -> str: if not all(block.text is not None or block.type in _NON_TEXT_BLOCKS for block in blocks): return json.dumps(content) return "\n\n".join(block.text for block in blocks if block.text is not None) - - -def lc_message(message: Mapping[str, Any]) -> dict[str, Any]: - """LangChain serialized message (or plain {role, content}) -> {role, content, tool_calls?}.""" - kwargs: Final = message.get("kwargs", message) - role: Final = MESSAGE_ROLES.get( - kwargs.get("type") or kwargs.get("role"), kwargs.get("role") or kwargs.get("type") or "" - ) - out: Final[dict[str, Any]] = { # mutable-ok: the framework message is built for JSON serialization - "role": role, - "content": content_text(kwargs.get("content", "")), - } - if kwargs.get("tool_calls"): - out["tool_calls"] = tuple({"name": t.get("name"), "args": t.get("args")} for t in kwargs["tool_calls"]) - if role == "tool" and kwargs.get("name"): - out["name"] = kwargs["name"] - return out diff --git a/litellm/tracing/normalizers/__init__.py b/litellm/tracing/normalizers/__init__.py deleted file mode 100644 index 2f861330a36..00000000000 --- a/litellm/tracing/normalizers/__init__.py +++ /dev/null @@ -1,32 +0,0 @@ -"""Per-convention span normalizers, tried in order: the first whose `matches()` is true wins.""" - -from collections.abc import Mapping, Sequence -from typing import Final - -from litellm.tracing.normalizers.base import SpanNormalizer -from litellm.tracing.normalizers.genai import GenAISemconvNormalizer -from litellm.tracing.normalizers.langsmith import LangSmithNormalizer -from litellm.tracing.normalizers.openinference import OpenInferenceNormalizer - -NORMALIZERS: Final[tuple[SpanNormalizer, ...]] = ( - LangSmithNormalizer(), - OpenInferenceNormalizer(), - GenAISemconvNormalizer(), -) -_FALLBACK: Final[SpanNormalizer] = GenAISemconvNormalizer() - - -def select_normalizer( - scope_name: str, attributes: Mapping[str, str], registry: Sequence[SpanNormalizer] = NORMALIZERS -) -> SpanNormalizer: - return next((n for n in registry if n.matches(scope_name, attributes)), _FALLBACK) - - -__all__ = ( - "NORMALIZERS", - "GenAISemconvNormalizer", - "LangSmithNormalizer", - "OpenInferenceNormalizer", - "SpanNormalizer", - "select_normalizer", -) diff --git a/litellm/tracing/normalizers/base.py b/litellm/tracing/normalizers/base.py deleted file mode 100644 index 37735113ce2..00000000000 --- a/litellm/tracing/normalizers/base.py +++ /dev/null @@ -1,22 +0,0 @@ -from collections.abc import Mapping -from typing import Protocol - -from litellm.tracing.types import SpanRow - - -class SpanNormalizer(Protocol): - """Maps one tracing convention's span attributes onto the LiteLLM `SpanRow` columns.""" - - @property - def name(self) -> str: ... - - def matches(self, scope_name: str, attributes: Mapping[str, str]) -> bool: ... - - def normalize(self, row: SpanRow, attributes: Mapping[str, str]) -> None: ... - - -def to_int(value: str | None) -> int: - try: - return int(value) if value else 0 - except ValueError: - return 0 diff --git a/litellm/tracing/normalizers/genai.py b/litellm/tracing/normalizers/genai.py deleted file mode 100644 index 16986607396..00000000000 --- a/litellm/tracing/normalizers/genai.py +++ /dev/null @@ -1,31 +0,0 @@ -from collections.abc import Mapping -from dataclasses import dataclass -from typing import Final - -from litellm.tracing.types import SpanRow - -_LLM_OPERATIONS: Final = frozenset({"chat", "text_completion", "generate_content"}) - - -@dataclass(frozen=True, slots=True) -class GenAISemconvNormalizer: - """OTEL `gen_ai.*` semantic conventions. Matches every span, so it belongs last as the fallback.""" - - name: str = "genai" - - def matches(self, scope_name: str, attributes: Mapping[str, str]) -> bool: - return True - - def normalize(self, row: SpanRow, attributes: Mapping[str, str]) -> None: - operation: Final = attributes.get("gen_ai.operation.name", "") - if operation == "invoke_agent" or not row["ParentSpanId"]: - row["ObservationType"] = "agent" - elif operation in _LLM_OPERATIONS: - row["ObservationType"] = "llm" - elif operation == "execute_tool": - row["ObservationType"] = "tool" - row["AgentName"] = attributes.get("gen_ai.agent.name", "") - row["Model"] = attributes.get("gen_ai.request.model") or attributes.get("gen_ai.response.model", "") - row["LiteLLMRequestId"] = attributes.get("gen_ai.response.id", "") - row["Input"] = attributes.get("gen_ai.input.messages") or attributes.get("gen_ai.tool.call.arguments", "") - row["Output"] = attributes.get("gen_ai.output.messages") or attributes.get("gen_ai.tool.call.result", "") diff --git a/litellm/tracing/normalizers/langsmith.py b/litellm/tracing/normalizers/langsmith.py deleted file mode 100644 index daca932d57d..00000000000 --- a/litellm/tracing/normalizers/langsmith.py +++ /dev/null @@ -1,115 +0,0 @@ -"""LangSmith OTEL mode, which LangChain, LangGraph and Deep Agents export through.""" - -import json -from collections.abc import Mapping -from dataclasses import dataclass -from types import MappingProxyType -from typing import Final - -from litellm.tracing.normalizers.messages import lc_message -from litellm.tracing.types import SpanRow, SpanType - -# LangChain / Deep Agents middleware wrappers: real spans, but noise in the UI -_FRAMEWORK_SUFFIXES: Final = ( - ".wrap_model_call", - ".wrap_tool_call", - ".before_agent", - ".after_agent", - ".before_model", - ".after_model", -) - - -def _loads(value: str) -> object: - try: - return json.loads(value) - except (ValueError, TypeError): - return None - - -def _span_type(row: SpanRow, attributes: Mapping[str, str]) -> SpanType: - kind: Final = attributes.get("langsmith.span.kind", "chain") - name: Final = row["SpanName"] - if not row["ParentSpanId"] or name == attributes.get("langsmith.metadata.lc_agent_name"): - return "agent" - if kind in ("llm", "tool"): - return kind - if name.endswith(_FRAMEWORK_SUFFIXES): - return "framework" - return "chain" - - -def _tool_output(completion: object) -> object: - raw: Final = completion.get("output", completion) if isinstance(completion, dict) else completion - update: Final = raw.get("update") if isinstance(raw, dict) else None - update_messages: Final = update.get("messages") or () if isinstance(update, dict) else () - is_command: Final = isinstance(raw, dict) and "update" in raw - # LangGraph Command (e.g. the Deep Agents `task` tool): the result is the last update message - output: Final = update_messages[-1] if is_command and update_messages else raw - return output.get("content", output) if isinstance(output, dict) else output - - -def _set_agent_io(row: SpanRow, attributes: Mapping[str, str], prompt: object, completion: object) -> None: - input_messages: Final = prompt.get("messages") if isinstance(prompt, dict) else None - output_messages: Final = completion.get("messages") if isinstance(completion, dict) else None - # agents built with @traceable take arbitrary args, not a message list: keep the raw payload then - row["Input"] = ( - json.dumps(tuple(lc_message(m) for m in input_messages if isinstance(m, dict))) - if input_messages - else attributes.get("gen_ai.prompt", "") - ) - row["Output"] = ( - json.dumps(lc_message(output_messages[-1])) - if output_messages and isinstance(output_messages[-1], dict) - else attributes.get("gen_ai.completion", "") - ) - - -def _set_io(row: SpanRow, attributes: Mapping[str, str]) -> None: - prompt: Final = _loads(attributes.get("gen_ai.prompt", "")) - completion: Final = _loads(attributes.get("gen_ai.completion", "")) - if row["ObservationType"] == "llm" and isinstance(completion, dict): - prompt_payload: Final = prompt if isinstance(prompt, dict) else MappingProxyType({}) - messages: Final = prompt_payload.get("messages") or ((),) - batch: Final = messages[0] if messages and isinstance(messages[0], list) else messages - row["Input"] = ( - json.dumps(tuple(lc_message(m) for m in batch if isinstance(m, dict))) - if isinstance(batch, (list, tuple)) - else "" - ) - generations: Final = completion.get("generations") - first: Final = generations[0] if isinstance(generations, list) and generations else None - item: Final = first[0] if isinstance(first, list) and first else None - message: Final = item.get("message") if isinstance(item, dict) else None - generation: Final = message.get("kwargs") if isinstance(message, dict) else None - if isinstance(generation, dict): - row["Output"] = json.dumps(lc_message(generation)) - metadata: Final = generation.get("response_metadata") - row["LiteLLMRequestId"] = metadata.get("id", "") if isinstance(metadata, dict) else "" - return - row["Output"] = attributes.get("gen_ai.completion", "") - return - if row["ObservationType"] == "tool": - output: Final = _tool_output(completion) - row["Input"] = attributes.get("gen_ai.prompt", "") - row["Output"] = output if isinstance(output, str) else json.dumps(output) - return - if row["ObservationType"] == "agent": - _set_agent_io(row, attributes, prompt, completion) - return - row["Input"] = attributes.get("gen_ai.prompt", "") - row["Output"] = attributes.get("gen_ai.completion", "") - - -@dataclass(frozen=True, slots=True) -class LangSmithNormalizer: - name: str = "langsmith" - - def matches(self, scope_name: str, attributes: Mapping[str, str]) -> bool: - return scope_name == "langsmith" or "langsmith.span.kind" in attributes - - def normalize(self, row: SpanRow, attributes: Mapping[str, str]) -> None: - row["ObservationType"] = _span_type(row, attributes) - row["AgentName"] = attributes.get("langsmith.metadata.lc_agent_name", "") - row["Model"] = attributes.get("gen_ai.request.model", "") - _set_io(row, attributes) diff --git a/litellm/tracing/normalizers/openinference.py b/litellm/tracing/normalizers/openinference.py deleted file mode 100644 index f9e1295148c..00000000000 --- a/litellm/tracing/normalizers/openinference.py +++ /dev/null @@ -1,27 +0,0 @@ -from collections.abc import Mapping -from dataclasses import dataclass -from types import MappingProxyType -from typing import Final - -from litellm.tracing.normalizers.base import to_int -from litellm.tracing.types import SpanRow, SpanType - -_OPENINFERENCE_TYPES: Final[Mapping[str, SpanType]] = MappingProxyType({"AGENT": "agent", "LLM": "llm", "TOOL": "tool"}) - - -@dataclass(frozen=True, slots=True) -class OpenInferenceNormalizer: - name: str = "openinference" - - def matches(self, scope_name: str, attributes: Mapping[str, str]) -> bool: - return "openinference.span.kind" in attributes - - def normalize(self, row: SpanRow, attributes: Mapping[str, str]) -> None: - kind: Final = attributes.get("openinference.span.kind", "").upper() - row["ObservationType"] = _OPENINFERENCE_TYPES.get(kind, "agent" if not row["ParentSpanId"] else "chain") - row["AgentName"] = attributes.get("agent.name", "") - row["Model"] = attributes.get("llm.model_name", "") - row["Input"] = attributes.get("input.value", "") - row["Output"] = attributes.get("output.value", "") - row["InputTokens"] = to_int(attributes.get("llm.token_count.prompt")) - row["OutputTokens"] = to_int(attributes.get("llm.token_count.completion")) diff --git a/litellm/tracing/ui_format.py b/litellm/tracing/ui_format.py index d7ecf48078f..51ec2876fbb 100644 --- a/litellm/tracing/ui_format.py +++ b/litellm/tracing/ui_format.py @@ -7,7 +7,7 @@ from typing import Final, Literal, TypeAlias from pydantic import BaseModel, ConfigDict, JsonValue, TypeAdapter, ValidationError from typing_extensions import NotRequired, ReadOnly, TypedDict -from litellm.tracing.normalizers.messages import MESSAGE_ROLES, ChatRole, content_text +from litellm.tracing.messages import MESSAGE_ROLES, ChatRole, content_text class UIToolCall(TypedDict): diff --git a/tests/test_litellm/tracing/normalizers/test_registry.py b/tests/test_litellm/tracing/normalizers/test_registry.py deleted file mode 100644 index 4c2fd051d6c..00000000000 --- a/tests/test_litellm/tracing/normalizers/test_registry.py +++ /dev/null @@ -1,68 +0,0 @@ -from collections.abc import Mapping -from dataclasses import dataclass -from types import MappingProxyType -from typing import Final - -from litellm.tracing.normalizers import ( - NORMALIZERS, - GenAISemconvNormalizer, - LangSmithNormalizer, - OpenInferenceNormalizer, - select_normalizer, -) -from litellm.tracing.types import SpanRow - -_NO_ATTRIBUTES: Final[Mapping[str, str]] = MappingProxyType({}) - - -def test_langsmith_scope_selects_langsmith_without_any_attributes(): - assert isinstance(select_normalizer("langsmith", _NO_ATTRIBUTES), LangSmithNormalizer) - - -def test_langsmith_kind_attribute_selects_langsmith_under_any_scope(): - assert isinstance(select_normalizer("other", MappingProxyType({"langsmith.span.kind": "llm"})), LangSmithNormalizer) - - -def test_langsmith_wins_over_openinference_when_both_markers_present(): - attributes: Final = MappingProxyType({"langsmith.span.kind": "llm", "openinference.span.kind": "LLM"}) - assert isinstance(select_normalizer("other", attributes), LangSmithNormalizer) - - -def test_openinference_kind_attribute_selects_openinference(): - assert isinstance( - select_normalizer("other", MappingProxyType({"openinference.span.kind": "LLM"})), OpenInferenceNormalizer - ) - - -def test_unmarked_span_falls_back_to_genai(): - assert isinstance( - select_normalizer("other", MappingProxyType({"gen_ai.operation.name": "chat"})), GenAISemconvNormalizer - ) - - -def test_empty_registry_falls_back_to_genai(): - assert isinstance(select_normalizer("langsmith", _NO_ATTRIBUTES, registry=()), GenAISemconvNormalizer) - - -def test_registry_names_are_unique(): - names: Final = tuple(n.name for n in NORMALIZERS) - assert len(names) == len(frozenset(names)) - - -@dataclass(frozen=True, slots=True) -class _CustomNormalizer: - name: str = "custom" - - def matches(self, scope_name: str, attributes: Mapping[str, str]) -> bool: - return scope_name == "custom-sdk" - - def normalize(self, row: SpanRow, attributes: Mapping[str, str]) -> None: - return None - - -def test_normalizer_inserted_ahead_in_custom_registry_wins_only_where_it_matches(): - registry: Final = (_CustomNormalizer(), *NORMALIZERS) - assert isinstance( - select_normalizer("custom-sdk", MappingProxyType({"langsmith.span.kind": "llm"}), registry), _CustomNormalizer - ) - assert isinstance(select_normalizer("langsmith", _NO_ATTRIBUTES, registry), LangSmithNormalizer) diff --git a/tests/test_litellm_rust/test_traces.py b/tests/test_litellm_rust/test_traces.py index e6492c9bca6..9b4d8c4fac1 100644 --- a/tests/test_litellm_rust/test_traces.py +++ b/tests/test_litellm_rust/test_traces.py @@ -9,7 +9,7 @@ from urllib.parse import parse_qs, urlsplit import pytest from litellm.rust_bridge._native import NativeTraceStorage, trace_decode_otlp -from litellm.rust_bridge.traces import ClickHouseStorage +from litellm.rust_bridge.traces import ClickHouseStorage, NormalizedSpan, normalized_field_definitions from litellm.tracing import Tenant, TraceReceiver, TracingPayloadTooLargeError from litellm.tracing.decode import decode_otlp from litellm.tracing.store import TraceStore @@ -157,6 +157,15 @@ def test_decode_and_tenant_stamping_share_resources_without_crossing_groups() -> assert rows[0]["ResourceAttributes"] == {"shared": "x" * 128, "litellm.team_id": "spoofed"} +def test_normalized_field_contract_matches_decoded_rust_span() -> None: + body: Final = _resource_export(8, 1) + spans: Final = trace_decode_otlp(body, "application/json") + fields: Final = normalized_field_definitions() + assert len(spans) == 1 + assert {field.name for field in fields} == set(spans[0]["normalized"]) == set(NormalizedSpan.model_fields) + assert len({field.clickhouse_column for field in fields}) == len(fields) + + @pytest.mark.asyncio async def test_resource_fanout_reaches_insert_with_identical_values(recording_server: RecordingServer) -> None: body: Final = _resource_export(16 * 1024, 1024) diff --git a/tests/unit/proxy/test_tracing_endpoints.py b/tests/unit/proxy/test_tracing_endpoints.py index aa1403b8db9..287092c9f91 100644 --- a/tests/unit/proxy/test_tracing_endpoints.py +++ b/tests/unit/proxy/test_tracing_endpoints.py @@ -23,6 +23,37 @@ from litellm.tracing.types import TraceScope TEAM_KEY = UserAPIKeyAuth( token="hashed-key", team_id="team-research", org_id="org-1", user_role=LitellmUserRoles.INTERNAL_USER ) +TRACE_RESPONSE: Final = { + "summary": { + "trace_id": "t1", + "name": "trace", + "service": "test", + "input_preview": "", + "start_time": "2026-01-01T00:00:00Z", + "duration_ms": 0, + "status": "ok", + "span_count": 0, + "agent_count": 0, + "agent_invocations": 0, + "llm_calls": 0, + "tool_calls": 0, + "error_count": 0, + "input_tokens": 0, + "output_tokens": 0, + "models": [], + "spend": None, + }, + "agents": [], + "spans": [], +} +SPAN_DETAIL_RESPONSE: Final = { + "span_id": "s1", + "input": "", + "output": "", + "input_ui": {"kind": "text", "text": ""}, + "output_ui": {"kind": "text", "text": ""}, + "attributes": {}, +} @pytest.mark.parametrize( @@ -171,17 +202,16 @@ def test_list_traces_defaults_to_last_24h(client, receiver): def test_get_trace_404_and_200(client, receiver): assert client.get("/v1/traces/missing").status_code == 404 - trace = {"summary": {"trace_id": "t1"}, "agents": [], "spans": []} - receiver.get_trace.return_value = trace + receiver.get_trace.return_value = TRACE_RESPONSE response = client.get("/v1/traces/t1") assert response.status_code == 200 - assert response.json() == trace + assert response.json() == TRACE_RESPONSE receiver.get_trace.assert_awaited_with("t1", {"team_ids": ("team-research",), "api_key_hash": ""}, "") def test_get_span_404_and_200(client, receiver): assert client.get("/v1/traces/t1/spans/s1").status_code == 404 - receiver.get_span.return_value = {"span_id": "s1", "input": "", "output": "", "attributes": {}} + receiver.get_span.return_value = SPAN_DETAIL_RESPONSE response = client.get("/v1/traces/t1/spans/s1") assert response.status_code == 200 assert response.json()["span_id"] == "s1" @@ -207,7 +237,7 @@ def test_get_span_serves_ui_content_from_stored_payloads(client): def test_trace_detail_passes_scoped_reference(client, receiver): - receiver.get_trace.return_value = {"summary": {"trace_id": "t1"}, "agents": [], "spans": []} + receiver.get_trace.return_value = TRACE_RESPONSE assert client.get("/v1/traces/t1?trace_ref=run-one").status_code == 200 receiver.get_trace.assert_awaited_with("t1", {"team_ids": ("team-research",), "api_key_hash": ""}, "run-one") diff --git a/ui/litellm-dashboard/src/components/view_logs/TraceView/DetailPane.integration.test.tsx b/ui/litellm-dashboard/src/components/view_logs/TraceView/DetailPane.integration.test.tsx index be5eefa1546..4af95c9d7b1 100644 --- a/ui/litellm-dashboard/src/components/view_logs/TraceView/DetailPane.integration.test.tsx +++ b/ui/litellm-dashboard/src/components/view_logs/TraceView/DetailPane.integration.test.tsx @@ -32,6 +32,7 @@ const span = (overrides: SpanFields): Span => ({ input_tokens: 0, output_tokens: 0, litellm_request_id: null, + spend: null, ...overrides, }); diff --git a/ui/litellm-dashboard/src/components/view_logs/TraceView/RunDrawer.test.tsx b/ui/litellm-dashboard/src/components/view_logs/TraceView/RunDrawer.test.tsx index 5ec0df55dab..8f205de441a 100644 --- a/ui/litellm-dashboard/src/components/view_logs/TraceView/RunDrawer.test.tsx +++ b/ui/litellm-dashboard/src/components/view_logs/TraceView/RunDrawer.test.tsx @@ -19,12 +19,14 @@ const run = (trace_id: string): TraceSummary => ({ status: "ok", span_count: 1, agent_count: 1, + agent_invocations: 1, llm_calls: 0, tool_calls: 0, error_count: 0, input_tokens: 0, output_tokens: 0, models: [], + spend: null, }); const mockReducedMotion = (reduce: boolean) => diff --git a/ui/litellm-dashboard/src/components/view_logs/TraceView/traceTypes.ts b/ui/litellm-dashboard/src/components/view_logs/TraceView/traceTypes.ts index d3080f8aab4..390a296aa00 100644 --- a/ui/litellm-dashboard/src/components/view_logs/TraceView/traceTypes.ts +++ b/ui/litellm-dashboard/src/components/view_logs/TraceView/traceTypes.ts @@ -1,114 +1,22 @@ -/** - * Agent tracing types. Mirrors `litellm/tracing/types.py` exactly. - * - * A trace is one agent run made of spans (agent / llm / tool / chain / framework). - */ +import type { components, paths } from "@/lib/http/schema"; -export type SpanType = "agent" | "llm" | "tool" | "chain" | "framework"; -export type SpanStatus = "ok" | "error" | "unset"; - -export interface Span { - span_id: string; - parent_span_id: string | null; - name: string; - type: SpanType; - /** The agent this span runs inside, e.g. "researcher". */ - agent: string; - /** Relative to trace start. */ - start_offset_ms: number; - duration_ms: number; - status: SpanStatus; - /** Exception message when status is "error". */ - error?: string | null; - error_truncated?: boolean; - input_preview: string; - model: string | null; - input_tokens: number; - output_tokens: number; - litellm_request_id: string | null; - spend?: number | null; -} - -/** One distinct agent in a trace. 200 invocations of `researcher` = one node. */ -export interface AgentNode { - name: string; - parent_agent: string | null; - invocations: number; - llm_calls: number; - tool_calls: number; - duration_ms: number; - spend?: number | null; -} - -export interface TraceSummary { - trace_id: string; - trace_ref?: string; - name: string; - service: string; - input_preview: string; - /** ISO 8601 */ - start_time: string; - duration_ms: number; - status: SpanStatus; - span_count: number; - agent_count: number; - llm_calls: number; - tool_calls: number; - /** Spans with an error status; > 0 means the run shows as failed. */ - error_count: number; - input_tokens: number; - output_tokens: number; - models: string[]; - spend?: number | null; -} - -export interface Trace { - summary: TraceSummary; - agents: AgentNode[]; - spans: Span[]; -} - -export interface TracePage { - data: TraceSummary[]; - next_cursor: string | null; -} - -/** Mirrors `litellm/tracing/ui_format.py`: span content already reduced to what the UI renders. */ -export interface UIToolCall { - name: string; - /** JSON-encoded arguments. */ - arguments: string; -} - -export interface UIMessage { - role: "system" | "user" | "assistant" | "tool"; - content: string; - name?: string; - tool_calls?: UIToolCall[]; -} - -export interface UIField { - key: string; - value: string; -} - -export type UIContent = - | { kind: "messages"; messages: UIMessage[] } - | { kind: "fields"; fields: UIField[] } - | { kind: "text"; text: string }; - -/** - * `input` / `output` are the raw stored JSON strings. `input_ui` / `output_ui` are the - * standard rendering; optional because older proxies don't send them. - */ -export interface SpanDetail { - span_id: string; - input: string; - output: string; - input_ui?: UIContent; - output_ui?: UIContent; - attributes: Record; -} +export type Trace = paths["/v1/traces/{trace_id}"]["get"]["responses"][200]["content"]["application/json"]; +export type TracePage = paths["/v1/traces"]["get"]["responses"][200]["content"]["application/json"]; +export type SpanErrorPage = + paths["/v1/traces/{trace_id}/spans/{span_id}/error"]["get"]["responses"][200]["content"]["application/json"]; +type ApiSpanDetail = + paths["/v1/traces/{trace_id}/spans/{span_id}"]["get"]["responses"][200]["content"]["application/json"]; +export type Span = Trace["spans"][number]; +export type SpanType = Span["type"]; +export type SpanStatus = Span["status"]; +export type AgentNode = Trace["agents"][number]; +export type TraceSummary = Trace["summary"]; +export type SpanDetail = Omit & + Partial>; +export type UIToolCall = components["schemas"]["UIToolCall"]; +export type UIMessage = components["schemas"]["UIMessage"]; +export type UIField = components["schemas"]["UIField"]; +export type UIContent = ApiSpanDetail["input_ui"]; export interface TraceToolCall { name: string; @@ -121,10 +29,3 @@ export interface TraceMessage { name?: string; tool_calls?: TraceToolCall[]; } - -export interface SpanErrorPage { - span_id: string; - message: string; - total_chars: number; - next_cursor: string | null; -} diff --git a/ui/litellm-dashboard/src/components/view_logs/TraceView/traceUtils.test.ts b/ui/litellm-dashboard/src/components/view_logs/TraceView/traceUtils.test.ts index 878ac58ab2f..9353f2024c8 100644 --- a/ui/litellm-dashboard/src/components/view_logs/TraceView/traceUtils.test.ts +++ b/ui/litellm-dashboard/src/components/view_logs/TraceView/traceUtils.test.ts @@ -50,6 +50,7 @@ const span = (overrides: SpanOverrides): Span => ({ input_tokens: 0, output_tokens: 0, litellm_request_id: null, + spend: null, ...overrides, }); diff --git a/ui/litellm-dashboard/src/lib/http/schema.d.ts b/ui/litellm-dashboard/src/lib/http/schema.d.ts index 3b91f2a17e9..dab0d6c5048 100644 --- a/ui/litellm-dashboard/src/lib/http/schema.d.ts +++ b/ui/litellm-dashboard/src/lib/http/schema.d.ts @@ -25603,6 +25603,26 @@ export interface components { /** Updated By */ updated_by: string; }; + /** + * AgentNode + * @description One distinct agent in a trace. 200 invocations of `researcher` = one node. + */ + AgentNode: { + /** Duration Ms */ + duration_ms: number; + /** Invocations */ + invocations: number; + /** Llm Calls */ + llm_calls: number; + /** Name */ + name: string; + /** Parent Agent */ + parent_agent: string | null; + /** Spend */ + spend: number | null; + /** Tool Calls */ + tool_calls: number; + }; /** AgentObjectPermission */ AgentObjectPermission: { /** Agents */ @@ -44876,6 +44896,64 @@ export interface components { /** Version */ version?: string; }; + /** Span */ + Span: { + /** Agent */ + agent: string; + /** Duration Ms */ + duration_ms: number; + /** Error */ + error: string | null; + /** Error Truncated */ + error_truncated: boolean; + /** Input Preview */ + input_preview: string; + /** Input Tokens */ + input_tokens: number; + /** Litellm Request Id */ + litellm_request_id: string | null; + /** Model */ + model: string | null; + /** Name */ + name: string; + /** Output Tokens */ + output_tokens: number; + /** Parent Span Id */ + parent_span_id: string | null; + /** Span Id */ + span_id: string; + /** Spend */ + spend: number | null; + /** Start Offset Ms */ + start_offset_ms: number; + /** + * Status + * @enum {string} + */ + status: "ok" | "error" | "unset"; + /** + * Type + * @enum {string} + */ + type: "agent" | "llm" | "tool" | "chain" | "framework"; + }; + /** SpanDetail */ + SpanDetail: { + /** Attributes */ + attributes: { + [key: string]: string; + }; + /** Input */ + input: string; + /** Input Ui */ + input_ui: components["schemas"]["UIMessages"] | components["schemas"]["UIFields"] | components["schemas"]["UIText"]; + /** Output */ + output: string; + /** Output Ui */ + output_ui: components["schemas"]["UIMessages"] | components["schemas"]["UIFields"] | components["schemas"]["UIText"]; + /** Span Id */ + span_id: string; + }; /** SpanErrorPage */ SpanErrorPage: { /** Message */ @@ -46688,6 +46766,21 @@ export interface components { } & { [key: string]: unknown; }; + /** Trace */ + Trace: { + /** Agents */ + agents: components["schemas"]["AgentNode"][]; + /** Spans */ + spans: components["schemas"]["Span"][]; + summary: components["schemas"]["TraceSummary"]; + }; + /** TracePage */ + TracePage: { + /** Data */ + data: components["schemas"]["TraceSummary"][]; + /** Next Cursor */ + next_cursor: string | null; + }; /** TracePart */ TracePart: { /** Content */ @@ -46711,6 +46804,48 @@ export interface components { */ truncated: boolean; }; + /** TraceSummary */ + TraceSummary: { + /** Agent Count */ + agent_count: number; + /** Agent Invocations */ + agent_invocations: number; + /** Duration Ms */ + duration_ms: number; + /** Error Count */ + error_count: number; + /** Input Preview */ + input_preview: string; + /** Input Tokens */ + input_tokens: number; + /** Llm Calls */ + llm_calls: number; + /** Models */ + models: string[]; + /** Name */ + name: string; + /** Output Tokens */ + output_tokens: number; + /** Service */ + service: string; + /** Span Count */ + span_count: number; + /** Spend */ + spend: number | null; + /** Start Time */ + start_time: string; + /** + * Status + * @enum {string} + */ + status: "ok" | "error" | "unset"; + /** Tool Calls */ + tool_calls: number; + /** Trace Id */ + trace_id: string; + /** Trace Ref */ + trace_ref?: string; + }; /** TrainedTierArtifact */ TrainedTierArtifact: { /** @@ -46785,6 +46920,47 @@ export interface components { } & { [key: string]: unknown; }; + /** UIField */ + UIField: { + /** Key */ + key: string; + /** Value */ + value: string; + }; + /** UIFields */ + UIFields: { + /** Fields */ + fields: components["schemas"]["UIField"][]; + /** + * Kind + * @constant + */ + kind: "fields"; + }; + /** UIMessage */ + UIMessage: { + /** Content */ + content: string; + /** Name */ + name?: string; + /** + * Role + * @enum {string} + */ + role: "system" | "user" | "assistant" | "tool"; + /** Tool Calls */ + tool_calls?: components["schemas"]["UIToolCall"][]; + }; + /** UIMessages */ + UIMessages: { + /** + * Kind + * @constant + */ + kind: "messages"; + /** Messages */ + messages: components["schemas"]["UIMessage"][]; + }; /** * UISettingsResponse * @description Response model for UI settings @@ -46803,6 +46979,16 @@ export interface components { [key: string]: unknown; }; }; + /** UIText */ + UIText: { + /** + * Kind + * @constant + */ + kind: "text"; + /** Text */ + text: string; + }; /** * UIThemeConfig * @description Configuration for UI theme customization @@ -46838,6 +47024,13 @@ export interface components { [key: string]: unknown; }; }; + /** UIToolCall */ + UIToolCall: { + /** Arguments */ + arguments: string; + /** Name */ + name: string; + }; /** UiDiscoveryEndpoints */ UiDiscoveryEndpoints: { /** Admin Ui Disabled */ @@ -79412,7 +79605,7 @@ export interface operations { [name: string]: unknown; }; content: { - "application/json": unknown; + "application/json": components["schemas"]["TracePage"]; }; }; /** @description Validation Error */ @@ -79465,7 +79658,7 @@ export interface operations { [name: string]: unknown; }; content: { - "application/json": unknown; + "application/json": components["schemas"]["Trace"]; }; }; /** @description Validation Error */ @@ -79499,7 +79692,7 @@ export interface operations { [name: string]: unknown; }; content: { - "application/json": unknown; + "application/json": components["schemas"]["SpanDetail"]; }; }; /** @description Validation Error */