Skip to content
Merged
Show file tree
Hide file tree
Changes from 5 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
234 changes: 75 additions & 159 deletions litellm/tracing/decode.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,14 +9,22 @@
"""

import json
from collections.abc import Callable, Mapping
from collections.abc import Mapping
from itertools import accumulate
from types import MappingProxyType
from typing import Any, Final
from typing import Final

from pydantic import JsonValue, TypeAdapter, ValidationError
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.tracing.types import SpanRow, SpanType
from litellm.tracing.normalizers import select_normalizer
from litellm.tracing.normalizers.base import to_int
from litellm.tracing.types import SpanRow

_MESSAGE_LIST: Final = TypeAdapter(tuple[dict[str, JsonValue], ...])

# attributes whose content we lift into Input/Output and drop from SpanAttributes
_HEAVY_ATTRIBUTES: Final = frozenset(
Expand All @@ -30,18 +38,6 @@
"output.value",
}
)
# 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",
)
_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"})


class InvalidOTLPPayloadError(ValueError):
Expand All @@ -63,6 +59,66 @@ def _truncate(value: str) -> str:
return f"{kept}…[truncated {size - OTLP_MAX_ATTRIBUTE_VALUE_BYTES} bytes]"


def _size(value: str) -> int:
return len(value.encode("utf-8"))


class _ElisionMarker(TypedDict):
role: ReadOnly[str]
content: ReadOnly[str]


def _elided(count: int) -> str:
marker: Final[_ElisionMarker] = {"role": "system", "content": f"…[{count} earlier messages truncated]"}
return json.dumps(marker)


def _with_content(message: Mapping[str, JsonValue], content: str) -> str:
return json.dumps(MappingProxyType({**message, "content": content}), default=lambda proxy: proxy.copy())


def _shrunk_message(message: Mapping[str, JsonValue], budget: int) -> str:
"""One message cut to `budget` bytes by shortening its text content, so the array stays valid JSON."""
content: Final = message.get("content")
text: Final = content if isinstance(content, str) else json.dumps(content)
overhead: Final = _size(_with_content(message, ""))
kept: Final = text.encode("utf-8")[: max(0, budget - overhead - 64)].decode("utf-8", "ignore")
return _with_content(message, f"{kept}…[truncated {_size(text) - _size(kept)} bytes]")
Comment thread
This conversation was marked as resolved.
Resolved


def _newest_that_fit(encoded: tuple[str, ...], budget: int) -> int:
"""How many trailing messages fit in `budget` bytes (comma separators included), scanning newest first."""
sizes: Final = tuple(_size(m) + 1 for m in reversed(encoded))
totals: Final = tuple(accumulate(sizes))
return next((count for count, total in enumerate(totals) if total > budget), len(totals))


def _truncate_payload(value: str) -> str:
"""Message arrays keep the first message, an elision marker and the newest messages that fit.

The result is always valid JSON: if even those don't fit, the first and last messages are shortened.
Anything that isn't a message array is byte-truncated as before.
"""
if _size(value) <= OTLP_MAX_ATTRIBUTE_VALUE_BYTES or not value.startswith("["):
return _truncate(value)
try:
messages: Final = _MESSAGE_LIST.validate_json(value)
except ValidationError:
return _truncate(value)
if len(messages) < 2:
return _truncate(value)
encoded: Final = tuple(json.dumps(m) for m in messages)
marker_budget: Final = _size(_elided(len(messages))) + 1
budget: Final = OTLP_MAX_ATTRIBUTE_VALUE_BYTES - 2 - _size(encoded[0]) - 1 - marker_budget
kept: Final = min(_newest_that_fit(encoded[1:], budget), len(messages) - 2)
if kept > 0:
tail: Final = encoded[len(encoded) - kept :]
return "[" + ", ".join((encoded[0], _elided(len(messages) - 1 - kept), *tail)) + "]"
half: Final = (OTLP_MAX_ATTRIBUTE_VALUE_BYTES - marker_budget - 4) // 2
middle: Final = (_elided(len(messages) - 2),) if len(messages) > 2 else ()
return "[" + ", ".join((_shrunk_message(messages[0], half), *middle, _shrunk_message(messages[-1], half))) + "]"


def decode_otlp(
body: bytes, content_type: str | None = None, content_encoding: str | None = None
) -> tuple[SpanRow, ...]:
Expand Down Expand Up @@ -119,160 +175,20 @@ def _span_row(span: DecodedSpan) -> SpanRow:
row["SpanAttributes"] = { # mutable-ok: the Rust JSON bridge requires a plain dict for span attributes
k: _truncate(v) for k, v in attributes.items() if k not in _HEAVY_ATTRIBUTES
}
row["Input"], row["Output"] = _truncate(row["Input"]), _truncate(row["Output"])
row["Input"], row["Output"] = _truncate_payload(row["Input"]), _truncate(row["Output"])
return row


# ---------------------------------------------------------------- normalize


def _loads(value: str) -> object:
try:
return json.loads(value)
except (ValueError, TypeError):
return None


def _lc_message(message: Mapping[str, Any]) -> dict[str, Any]:
"""LangChain serialized message (or plain {role, content}) -> {role, content, tool_calls?}."""
kwargs = message.get("kwargs", message)
role = _LC_ROLES.get(kwargs.get("type") or kwargs.get("role"), kwargs.get("role") or kwargs.get("type") or "")
content = kwargs.get("content", "")
out: dict[str, Any] = { # mutable-ok: the framework message is built for JSON serialization
"role": role,
"content": content if isinstance(content, str) else json.dumps(content),
}
if kwargs.get("tool_calls"):
out["tool_calls"] = tuple(
{"name": t.get("name"), "args": t.get("args")} # mutable-ok: JSON tool calls need object payloads
for t in kwargs["tool_calls"]
)
if role == "tool" and kwargs.get("name"):
out["name"] = kwargs["name"]
return out


def _langsmith_type(row: SpanRow, attributes: Mapping[str, str]) -> SpanType:
kind = attributes.get("langsmith.span.kind", "chain")
name = 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 _langsmith_io(row: SpanRow, attributes: Mapping[str, str]) -> None:
prompt = _loads(attributes.get("gen_ai.prompt", ""))
completion = _loads(attributes.get("gen_ai.completion", ""))
prompt_payload = prompt if isinstance(prompt, dict) else MappingProxyType({})
if row["ObservationType"] == "llm" and isinstance(completion, dict):
messages = prompt_payload.get("messages") or ((),)
batch = 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 ""
else:
row["Output"] = attributes.get("gen_ai.completion", "")
return
if row["ObservationType"] == "tool":
output = completion.get("output", completion) if isinstance(completion, dict) else completion
if isinstance(output, dict) and "update" in output: # LangGraph Command, e.g. Deep Agents `task`
update: Final = output.get("update")
update_messages = update.get("messages") or () if isinstance(update, dict) else ()
output = update_messages[-1] if update_messages else output
if isinstance(output, dict):
output = output.get("content", output)
row["Input"] = attributes.get("gen_ai.prompt", "")
row["Output"] = output if isinstance(output, str) else json.dumps(output)
return
if row["ObservationType"] == "agent":
input_messages = prompt.get("messages") if isinstance(prompt, dict) else None
output_messages = 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", "")
)
return
row["Input"] = attributes.get("gen_ai.prompt", "")
row["Output"] = attributes.get("gen_ai.completion", "")


def normalize_langsmith(row: SpanRow, attributes: Mapping[str, str]) -> None:
row["ObservationType"] = _langsmith_type(row, attributes)
row["AgentName"] = attributes.get("langsmith.metadata.lc_agent_name", "")
row["Model"] = attributes.get("gen_ai.request.model", "")
_langsmith_io(row, attributes)


def normalize_genai(row: SpanRow, attributes: Mapping[str, str]) -> None:
operation = 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", "")


def normalize_openinference(row: SpanRow, attributes: Mapping[str, str]) -> None:
kind = 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"))


def _set_tokens(row: SpanRow, attributes: Mapping[str, str]) -> None:
row["InputTokens"] = _to_int(attributes.get("gen_ai.usage.input_tokens"))
row["OutputTokens"] = _to_int(attributes.get("gen_ai.usage.output_tokens"))


def _to_int(value: str | None) -> int:
try:
return int(value) if value else 0
except ValueError:
return 0


def select_normalizer(scope_name: str, attributes: Mapping[str, str]) -> Callable[[SpanRow, Mapping[str, str]], None]:
if scope_name == "langsmith" or "langsmith.span.kind" in attributes:
return normalize_langsmith
if "openinference.span.kind" in attributes:
return normalize_openinference
return normalize_genai
row["InputTokens"] = to_int(attributes.get("gen_ai.usage.input_tokens"))
row["OutputTokens"] = to_int(attributes.get("gen_ai.usage.output_tokens"))


def normalize(row: SpanRow, attributes: Mapping[str, str]) -> None:
select_normalizer(row["ScopeName"], attributes)(row, attributes)
select_normalizer(row["ScopeName"], attributes).normalize(row, attributes)
if not row["InputTokens"] and not row["OutputTokens"]:
_set_tokens(row, attributes)

Expand Down
32 changes: 32 additions & 0 deletions litellm/tracing/normalizers/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,32 @@
"""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",
)
22 changes: 22 additions & 0 deletions litellm/tracing/normalizers/base.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,22 @@
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
31 changes: 31 additions & 0 deletions litellm/tracing/normalizers/genai.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,31 @@
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", "")
Comment on lines +25 to +28

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Normalizer mutates span rows These assignments violate the repository's immutability instruction for new code. Return normalized values or justify unavoidable mutation before merging

Context Used: AGENTS.md (source)

Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!

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", "")
Loading
Loading