mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-20 00:11:50 +00:00
Merge pull request #41498 from BerriAI/litellm_otel_indexed_messages_span_headroom
This commit is contained in:
yucheng-berri
GitHub
commit
375cd4a668
4 changed files with 316 additions and 132 deletions
|
|
@ -1,15 +1,19 @@
|
|||
"""The span engine: dedup, start, run the mapper chain, set status, end."""
|
||||
|
||||
from collections import OrderedDict
|
||||
from collections.abc import Callable, Sequence
|
||||
from collections.abc import Callable, Mapping, Sequence
|
||||
from types import MappingProxyType
|
||||
from typing import Final
|
||||
|
||||
from opentelemetry.context import Context
|
||||
from opentelemetry.sdk.trace import ReadableSpan, SpanLimits
|
||||
from opentelemetry.sdk.trace import Span as SdkSpan
|
||||
from opentelemetry.trace import Link, Span, Tracer
|
||||
from opentelemetry.trace.status import Status, StatusCode
|
||||
|
||||
from litellm.integrations.otel.mappers import resolve_mappers
|
||||
from litellm.integrations.otel.mappers.base import AttributeMapper, SpanData
|
||||
from litellm.integrations.otel.mappers.base import AttributeMapper, AttrValue, SpanData
|
||||
from litellm.integrations.otel.mappers.openinference import fit_indexed_messages
|
||||
from litellm.integrations.otel.model.config import OpenTelemetryV2Config
|
||||
from litellm.integrations.otel.model.payloads import (
|
||||
GuardrailSpanData,
|
||||
|
|
@ -52,25 +56,48 @@ _NAME_BUILDERS: Final[dict[SpanRole, Callable[..., str]]] = {
|
|||
_DEDUP_CACHE_MAX: Final = 10_000
|
||||
|
||||
|
||||
def _stamp_otel_error_attributes(span: Span, error_type: str, resolved_message: str) -> None:
|
||||
"""Stamp the OTel-semconv error attributes (``error.type`` + ``error.message``).
|
||||
``error_type`` and ``resolved_message`` are ``finish_span``'s already-computed
|
||||
fallback chains, so the pair on the status, event, and attributes stays in
|
||||
lockstep."""
|
||||
span.set_attribute(Error.TYPE, error_type)
|
||||
span.set_attribute(Error.MESSAGE, resolved_message)
|
||||
def _resolve_error(error: SpanError) -> tuple[str, str] | None:
|
||||
"""The ``(error_type, message)`` fallback chain shared by the status, the event and the attributes, or
|
||||
``None`` when ``error`` carries neither a type nor a message."""
|
||||
if not (error.error_type or error.message):
|
||||
return None
|
||||
return error.error_type or "error", error.message or error.error_type or "error"
|
||||
|
||||
|
||||
def _stamp_litellm_error_attributes(span: Span, error: SpanError) -> None:
|
||||
"""Stamp litellm-specific error detail attributes. Emitted only when the
|
||||
corresponding field is populated so guardrail-shape errors carrying only a
|
||||
message aren't polluted with empty detail keys."""
|
||||
if error.code:
|
||||
span.set_attribute(LiteLLMError.CODE, error.code)
|
||||
if error.stack_trace:
|
||||
span.set_attribute(LiteLLMError.STACK_TRACE, error.stack_trace)
|
||||
if error.llm_provider:
|
||||
span.set_attribute(LiteLLMError.LLM_PROVIDER, error.llm_provider)
|
||||
_NO_ATTRIBUTES: Final[Mapping[str, AttrValue]] = MappingProxyType({})
|
||||
|
||||
|
||||
def error_attributes(error: SpanError) -> Mapping[str, AttrValue]:
|
||||
"""The v2 error attribute set: the OTel-semconv ``error.*`` pair plus the litellm detail keys that are
|
||||
populated, so guardrail-shape errors carrying only a message aren't polluted with empty detail keys."""
|
||||
resolved: Final = _resolve_error(error)
|
||||
if resolved is None:
|
||||
return _NO_ATTRIBUTES
|
||||
error_type, message = resolved
|
||||
pairs: Final = (
|
||||
(Error.TYPE, error_type),
|
||||
(Error.MESSAGE, message),
|
||||
(LiteLLMError.CODE, error.code),
|
||||
(LiteLLMError.STACK_TRACE, error.stack_trace),
|
||||
(LiteLLMError.LLM_PROVIDER, error.llm_provider),
|
||||
)
|
||||
return MappingProxyType({key: value for key, value in pairs if value})
|
||||
|
||||
|
||||
def span_attribute_limit(span: Span) -> int | None:
|
||||
"""The attribute count limit ``span`` was built with, ``None`` when unbounded."""
|
||||
if not isinstance(span, SdkSpan):
|
||||
return SpanLimits().max_span_attributes
|
||||
return span._limits.max_span_attributes # pyright: ignore[reportPrivateUsage] # SDK has no public getter
|
||||
|
||||
|
||||
def attribute_budget(span: Span, reserved: int) -> int | None:
|
||||
"""How many mapped attributes fit on ``span`` next to what it already carries and ``reserved`` more."""
|
||||
limit: Final = span_attribute_limit(span)
|
||||
if limit is None:
|
||||
return None
|
||||
on_span: Final = len(span.attributes or ()) if isinstance(span, ReadableSpan) else 0
|
||||
return limit - on_span - reserved
|
||||
|
||||
|
||||
def stamp_error(
|
||||
|
|
@ -93,12 +120,12 @@ def stamp_error(
|
|||
``set_status`` are opt-outs for callers whose span lifecycle (``use_span``) or
|
||||
owner (the FastAPI instrumentor) already records the event or the status.
|
||||
"""
|
||||
if not (error.error_type or error.message):
|
||||
resolved: Final = _resolve_error(error)
|
||||
if resolved is None:
|
||||
return None
|
||||
error_type: Final = error.error_type or "error"
|
||||
message: Final = error.message or error.error_type or "error"
|
||||
_stamp_otel_error_attributes(span, error_type, message)
|
||||
_stamp_litellm_error_attributes(span, error)
|
||||
error_type, message = resolved
|
||||
for key, value in error_attributes(error).items():
|
||||
span.set_attribute(key, value)
|
||||
if set_status:
|
||||
span.set_status(Status(StatusCode.ERROR, message))
|
||||
if record_event:
|
||||
|
|
@ -238,9 +265,6 @@ class SpanEmitter:
|
|||
data, since the boundary opener only has a provisional name.
|
||||
"""
|
||||
span.update_name(_NAME_BUILDERS[role](data))
|
||||
for mapper in self._mappers:
|
||||
for key, value in mapper.map(data).items():
|
||||
span.set_attribute(key, value)
|
||||
error: Final = (
|
||||
data.error
|
||||
if isinstance(
|
||||
|
|
@ -255,6 +279,13 @@ class SpanEmitter:
|
|||
)
|
||||
else None
|
||||
)
|
||||
mapped: Final = MappingProxyType(
|
||||
{key: value for mapper in self._mappers for key, value in mapper.map(data).items()}
|
||||
)
|
||||
stamped_later: Final = error_attributes(error) if error else _NO_ATTRIBUTES
|
||||
reserved: Final = len(stamped_later.keys() - mapped.keys())
|
||||
for key, value in fit_indexed_messages(mapped, attribute_budget(span, reserved)).items():
|
||||
span.set_attribute(key, value)
|
||||
if error:
|
||||
stamped: Final = stamp_error(span, error)
|
||||
if stamped is not None and self._event_recorder is not None and role is SpanRole.LLM_CALL:
|
||||
|
|
|
|||
|
|
@ -7,12 +7,13 @@ Phoenix + any other OpenInference-aware backend simultaneously.
|
|||
"""
|
||||
|
||||
import json
|
||||
from collections.abc import Callable, Sequence
|
||||
from collections.abc import Callable, Mapping, Sequence
|
||||
from itertools import accumulate, chain, groupby
|
||||
from types import MappingProxyType
|
||||
from typing import Final
|
||||
|
||||
from litellm.integrations.otel.mappers.base import AttributeMap, AttrValue, SpanData
|
||||
from litellm.integrations.otel.mappers.utils import (
|
||||
MAX_MESSAGE_ATTRS_PER_SPAN,
|
||||
MAX_TOOL_DEFINITION_ATTRS_PER_SPAN,
|
||||
collect,
|
||||
drop_none,
|
||||
|
|
@ -27,7 +28,53 @@ from litellm.integrations.otel.model.payloads import (
|
|||
ToolDefinition,
|
||||
)
|
||||
|
||||
_MAX_INDEXED_MESSAGES: Final = MAX_MESSAGE_ATTRS_PER_SPAN // 2
|
||||
_INPUT_MESSAGES: Final = "llm.input_messages"
|
||||
_OUTPUT_MESSAGES: Final = "llm.output_messages"
|
||||
_MESSAGE_FAMILIES: Final = (_INPUT_MESSAGES, _OUTPUT_MESSAGES)
|
||||
|
||||
|
||||
def _message_key_groups(attrs: Mapping[str, AttrValue]) -> Mapping[tuple[str, int], tuple[str, ...]]:
|
||||
"""Per-index message keys in ``attrs`` grouped by ``(family, index)``."""
|
||||
tagged: Final = sorted(
|
||||
(family, int(key.split(".")[2]), key)
|
||||
for key in attrs
|
||||
for family in _MESSAGE_FAMILIES
|
||||
if key.startswith(f"{family}.")
|
||||
)
|
||||
return MappingProxyType(
|
||||
{group: tuple(key for _, _, key in keys) for group, keys in groupby(tagged, key=lambda tag: tag[:2])}
|
||||
)
|
||||
|
||||
|
||||
def _shed_order(groups: Mapping[tuple[str, int], tuple[str, ...]]) -> tuple[tuple[str, int], ...]:
|
||||
"""Message groups least valuable first: middle prompt turns, extra choices, then the opener, the newest turn
|
||||
and the first choice."""
|
||||
inputs: Final = sorted(idx for family, idx in groups if family == _INPUT_MESSAGES)
|
||||
outputs: Final = sorted(idx for family, idx in groups if family == _OUTPUT_MESSAGES)
|
||||
pinned_inputs: Final = tuple(dict.fromkeys((*inputs[:1], *inputs[-1:])))
|
||||
return (
|
||||
*((_INPUT_MESSAGES, idx) for idx in inputs[1:-1]),
|
||||
*((_OUTPUT_MESSAGES, idx) for idx in reversed(outputs[1:])),
|
||||
*((_INPUT_MESSAGES, idx) for idx in pinned_inputs),
|
||||
*((_OUTPUT_MESSAGES, idx) for idx in outputs[:1]),
|
||||
)
|
||||
|
||||
|
||||
def fit_indexed_messages(attrs: Mapping[str, AttrValue], budget: int | None) -> Mapping[str, AttrValue]:
|
||||
"""``attrs`` with whole per-index messages shed, least valuable first, until at most ``budget`` keys remain.
|
||||
|
||||
``None`` means the span has no attribute count limit. Every message still rides the ``input.value`` and
|
||||
``output.value`` blobs, so shedding a per-index pair loses no content.
|
||||
"""
|
||||
if budget is None or len(attrs) <= budget:
|
||||
return attrs
|
||||
groups: Final = _message_key_groups(attrs)
|
||||
order: Final = _shed_order(groups)
|
||||
running: Final = tuple(accumulate(len(groups[group]) for group in order))
|
||||
excess: Final = len(attrs) - budget
|
||||
shed_count: Final = next((n + 1 for n, total in enumerate(running) if total >= excess), len(order))
|
||||
shed: Final = frozenset(chain.from_iterable(groups[group] for group in order[:shed_count]))
|
||||
return MappingProxyType({key: value for key, value in attrs.items() if key not in shed})
|
||||
|
||||
|
||||
class OpenInferenceMapper:
|
||||
|
|
@ -87,42 +134,22 @@ class OpenInferenceMapper:
|
|||
return {}
|
||||
|
||||
def _llm_call(self, data: LLMCallSpanData) -> AttributeMap:
|
||||
outputs: Final = output_messages(data)
|
||||
indexed_in, indexed_out = self._indexed_split(len(data.messages_in), len(outputs))
|
||||
return {
|
||||
**collect(self._LLM_CALL_ATTRS, data),
|
||||
**collect(self._BLOB_ATTRS, data),
|
||||
**self._messages(
|
||||
"llm.input_messages",
|
||||
"input.value",
|
||||
data.messages_in,
|
||||
self._prompt_positions(len(data.messages_in), indexed_in),
|
||||
),
|
||||
**self._messages("llm.output_messages", "output.value", outputs, range(indexed_out)),
|
||||
**self._messages(_INPUT_MESSAGES, "input.value", data.messages_in),
|
||||
**self._messages(_OUTPUT_MESSAGES, "output.value", output_messages(data)),
|
||||
**self._tools(data),
|
||||
}
|
||||
|
||||
@staticmethod
|
||||
def _indexed_split(inputs: int, outputs: int) -> tuple[int, int]:
|
||||
"""Prompt and response share one allowance; the response is reserved at least half of it."""
|
||||
indexed_out: Final = min(outputs, max(_MAX_INDEXED_MESSAGES // 2, _MAX_INDEXED_MESSAGES - inputs))
|
||||
return _MAX_INDEXED_MESSAGES - indexed_out, indexed_out
|
||||
|
||||
@staticmethod
|
||||
def _prompt_positions(total: int, indexed: int) -> tuple[int, ...]:
|
||||
"""Prompt messages that get per-index attributes: message 0 and the most recent turns."""
|
||||
if total <= indexed:
|
||||
return tuple(range(total))
|
||||
return (0, *range(total - indexed + 1, total))
|
||||
|
||||
@staticmethod
|
||||
def _messages(prefix: str, value_key: str, messages: Sequence[object], positions: Sequence[int]) -> AttributeMap:
|
||||
"""``{prefix}.{idx}.message.*`` keys for the messages at ``positions`` + the ``value_key`` blob of all."""
|
||||
def _messages(prefix: str, value_key: str, messages: Sequence[object]) -> AttributeMap:
|
||||
"""``{prefix}.{idx}.message.*`` keys for every message + the ``value_key`` blob of all of them."""
|
||||
parsed: Final = [(m.get("role") if isinstance(m, dict) else None, message_content(m)) for m in messages]
|
||||
attrs: Final = drop_none(
|
||||
{
|
||||
key: value
|
||||
for idx, (role, content) in ((idx, parsed[idx]) for idx in positions)
|
||||
for idx, (role, content) in enumerate(parsed)
|
||||
for key, value in (
|
||||
(f"{prefix}.{idx}.message.role", role if isinstance(role, str) else None),
|
||||
(f"{prefix}.{idx}.message.content", content),
|
||||
|
|
|
|||
|
|
@ -32,14 +32,6 @@ core telemetry no matter how many vocabularies are configured.
|
|||
"""
|
||||
|
||||
|
||||
MAX_MESSAGE_ATTRS_PER_SPAN: Final = DEFAULT_SPAN_ATTRIBUTE_LIMIT // 8
|
||||
"""Span-wide ceiling on per-index chat message attributes, prompt and response together.
|
||||
|
||||
An eighth is the largest share that still fits beside the tool ceiling and the core
|
||||
of every vocabulary at once. The complete conversation still rides the JSON blobs.
|
||||
"""
|
||||
|
||||
|
||||
def tool_attr_budget(vocabularies: int) -> int:
|
||||
"""Split the span-wide tool-definition ceiling across active vocabularies."""
|
||||
return MAX_TOOL_DEFINITION_ATTRS_PER_SPAN // max(vocabularies, 1)
|
||||
|
|
|
|||
|
|
@ -7,7 +7,10 @@ import pytest
|
|||
|
||||
pytest.importorskip("opentelemetry")
|
||||
|
||||
from opentelemetry.trace import SpanKind # noqa: E402
|
||||
from opentelemetry.sdk.trace import SpanLimits, TracerProvider # noqa: E402
|
||||
from opentelemetry.sdk.trace.export import SimpleSpanProcessor # noqa: E402
|
||||
from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter # noqa: E402
|
||||
from opentelemetry.trace import INVALID_SPAN, SpanKind # noqa: E402
|
||||
from opentelemetry.trace.status import StatusCode # noqa: E402
|
||||
|
||||
from litellm.integrations.otel import ( # noqa: E402
|
||||
|
|
@ -17,12 +20,9 @@ from litellm.integrations.otel import ( # noqa: E402
|
|||
)
|
||||
from litellm.integrations.otel.plumbing import context as ctx_mod # noqa: E402
|
||||
from litellm.integrations.otel.plumbing import providers # noqa: E402
|
||||
from litellm.integrations.otel.emitter import SpanEmitter # noqa: E402
|
||||
from litellm.integrations.otel.emitter import SpanEmitter, span_attribute_limit # noqa: E402
|
||||
from litellm.integrations.otel.emitter import stamp_error # noqa: E402
|
||||
from litellm.integrations.otel.mappers.utils import ( # noqa: E402
|
||||
MAX_MESSAGE_ATTRS_PER_SPAN,
|
||||
MAX_TOOL_DEFINITION_ATTRS_PER_SPAN,
|
||||
)
|
||||
from litellm.integrations.otel.mappers.utils import MAX_TOOL_DEFINITION_ATTRS_PER_SPAN # noqa: E402
|
||||
from litellm.integrations.otel.model.payloads import ( # noqa: E402
|
||||
GuardrailSpanData,
|
||||
LLMCallSpanData,
|
||||
|
|
@ -127,9 +127,7 @@ def test_llm_call_span_golden():
|
|||
|
||||
def test_legacy_dual_emit_on():
|
||||
engine, exporter = _engine(legacy_compat=True)
|
||||
engine.emit(
|
||||
SpanRole.LLM_CALL, LLMCallSpanData.from_standard_logging_payload(_payload())
|
||||
)
|
||||
engine.emit(SpanRole.LLM_CALL, LLMCallSpanData.from_standard_logging_payload(_payload()))
|
||||
(span,) = exporter.get_finished_spans()
|
||||
# canonical AND legacy keys are both present
|
||||
assert span.attributes[GenAI.USAGE_OUTPUT_TOKENS] == 5
|
||||
|
|
@ -139,9 +137,7 @@ def test_legacy_dual_emit_on():
|
|||
|
||||
def test_legacy_dual_emit_off():
|
||||
engine, exporter = _engine(legacy_compat=False)
|
||||
engine.emit(
|
||||
SpanRole.LLM_CALL, LLMCallSpanData.from_standard_logging_payload(_payload())
|
||||
)
|
||||
engine.emit(SpanRole.LLM_CALL, LLMCallSpanData.from_standard_logging_payload(_payload()))
|
||||
(span,) = exporter.get_finished_spans()
|
||||
# canonical present, legacy absent
|
||||
assert span.attributes[GenAI.USAGE_OUTPUT_TOKENS] == 5
|
||||
|
|
@ -155,9 +151,7 @@ def test_error_span_sets_status_and_error_type():
|
|||
status="failure",
|
||||
error_information={"error_class": "RateLimitError", "error_message": "429"},
|
||||
)
|
||||
engine.emit(
|
||||
SpanRole.LLM_CALL, LLMCallSpanData.from_standard_logging_payload(payload)
|
||||
)
|
||||
engine.emit(SpanRole.LLM_CALL, LLMCallSpanData.from_standard_logging_payload(payload))
|
||||
(span,) = exporter.get_finished_spans()
|
||||
assert span.status.status_code is StatusCode.ERROR
|
||||
assert span.attributes["error.type"] == "RateLimitError"
|
||||
|
|
@ -209,15 +203,11 @@ def test_hierarchy_and_kinds_match_registry():
|
|||
root = engine.start_span(SpanRole.PROXY_REQUEST, "POST /chat/completions")
|
||||
root_ctx = ctx_mod.context_from_span(root)
|
||||
engine.emit(SpanRole.LLM_CALL, data, parent_context=root_ctx)
|
||||
engine.emit(
|
||||
SpanRole.GUARDRAIL, GuardrailSpanData("presidio", status="success"), root_ctx
|
||||
)
|
||||
engine.emit(SpanRole.GUARDRAIL, GuardrailSpanData("presidio", status="success"), root_ctx)
|
||||
# An outbound datastore call (DB_CALL) and an internal service call differ in
|
||||
# span kind; both are named "{service} {call_type}".
|
||||
engine.emit(SpanRole.DB_CALL, ServiceSpanData("redis", call_type="set"), root_ctx)
|
||||
engine.emit(
|
||||
SpanRole.SERVICE, ServiceSpanData("router", call_type="acompletion"), root_ctx
|
||||
)
|
||||
engine.emit(SpanRole.SERVICE, ServiceSpanData("router", call_type="acompletion"), root_ctx)
|
||||
root.end()
|
||||
|
||||
by_name = {s.name: s for s in exporter.get_finished_spans()}
|
||||
|
|
@ -255,9 +245,7 @@ def test_dedup_cache_is_bounded(monkeypatch):
|
|||
for i in range(10):
|
||||
engine.emit(
|
||||
SpanRole.LLM_CALL,
|
||||
LLMCallSpanData.from_standard_logging_payload(
|
||||
_payload(litellm_call_id=f"call_{i}")
|
||||
),
|
||||
LLMCallSpanData.from_standard_logging_payload(_payload(litellm_call_id=f"call_{i}")),
|
||||
)
|
||||
assert len(engine._emitted) <= 3
|
||||
|
||||
|
|
@ -268,9 +256,7 @@ def test_service_error_span():
|
|||
engine, exporter = _engine()
|
||||
engine.emit(
|
||||
SpanRole.SERVICE,
|
||||
ServiceSpanData(
|
||||
"postgres", call_type="query", error=SpanError("DBError", "boom")
|
||||
),
|
||||
ServiceSpanData("postgres", call_type="query", error=SpanError("DBError", "boom")),
|
||||
)
|
||||
(span,) = exporter.get_finished_spans()
|
||||
assert span.status.status_code is StatusCode.ERROR
|
||||
|
|
@ -305,9 +291,7 @@ def test_guardrail_success_span_is_unset():
|
|||
engine, exporter = _engine()
|
||||
engine.emit(
|
||||
SpanRole.GUARDRAIL,
|
||||
GuardrailSpanData.from_logging_entry(
|
||||
{"guardrail_name": "g", "guardrail_status": "success"}
|
||||
),
|
||||
GuardrailSpanData.from_logging_entry({"guardrail_name": "g", "guardrail_status": "success"}),
|
||||
)
|
||||
(span,) = exporter.get_finished_spans()
|
||||
assert span.status.status_code is StatusCode.UNSET
|
||||
|
|
@ -396,11 +380,7 @@ def _tool_span(mapper_names, tool_count):
|
|||
|
||||
|
||||
def _tool_definition_keys(attributes):
|
||||
return [
|
||||
key
|
||||
for key in attributes
|
||||
if key.startswith(("gen_ai.tool.", "llm.request.functions.", "llm.tools."))
|
||||
]
|
||||
return [key for key in attributes if key.startswith(("gen_ai.tool.", "llm.request.functions.", "llm.tools."))]
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
|
|
@ -461,15 +441,19 @@ def _conversation_payload(turns, choices=1, **overrides):
|
|||
)
|
||||
|
||||
|
||||
def _conversation_span(mapper_names, payload, legacy_compat=False):
|
||||
"""The exported LLM-call span for ``payload`` with content capture on."""
|
||||
def _conversation_span(mapper_names, payload, legacy_compat=False, span_limits=None):
|
||||
"""The exported LLM-call span for ``payload`` with content capture on.
|
||||
|
||||
``span_limits`` builds the provider with programmatic limits instead of the environment's."""
|
||||
cfg = OpenTelemetryV2Config(
|
||||
exporter="in_memory",
|
||||
legacy_compat=legacy_compat,
|
||||
mapper_names=list(mapper_names),
|
||||
capture_message_content="span_only",
|
||||
)
|
||||
provider, exporter = providers.in_memory_provider(cfg)
|
||||
provider, exporter = (
|
||||
providers.in_memory_provider(cfg) if span_limits is None else _provider_with_limits(span_limits)
|
||||
)
|
||||
engine = SpanEmitter(providers.get_tracer(provider, "litellm-test"), cfg)
|
||||
engine.emit(
|
||||
SpanRole.LLM_CALL,
|
||||
|
|
@ -479,37 +463,56 @@ def _conversation_span(mapper_names, payload, legacy_compat=False):
|
|||
return span
|
||||
|
||||
|
||||
def _indexed_message_count(attributes, prefix):
|
||||
return len({key.split(".")[2] for key in attributes if key.startswith(f"{prefix}.")})
|
||||
def _provider_with_limits(span_limits):
|
||||
provider = TracerProvider(span_limits=span_limits)
|
||||
exporter = InMemorySpanExporter()
|
||||
provider.add_span_processor(SimpleSpanProcessor(exporter))
|
||||
return provider, exporter
|
||||
|
||||
|
||||
@pytest.mark.parametrize("turns", [60, 200])
|
||||
def test_long_conversation_does_not_evict_core_attributes(turns):
|
||||
"""Per-message OpenInference attributes must never crowd core telemetry off the span."""
|
||||
span = _conversation_span(["genai", "openinference"], _conversation_payload(turns))
|
||||
def _indexed_messages(attributes, prefix):
|
||||
return sorted({int(key.split(".")[2]) for key in attributes if key.startswith(f"{prefix}.")})
|
||||
|
||||
|
||||
def _assert_core_intact(span):
|
||||
a = span.attributes
|
||||
|
||||
assert span.dropped_attributes == 0
|
||||
assert a[GenAI.REQUEST_MODEL] == "gpt-4o"
|
||||
assert a[GenAI.PROVIDER_NAME] == "openai"
|
||||
assert a[GenAI.USAGE_INPUT_TOKENS] == 10
|
||||
assert a[GenAI.USAGE_OUTPUT_TOKENS] == 5
|
||||
assert a[GenAI.RESPONSE_FINISH_REASONS] == ("stop",)
|
||||
assert set(a[GenAI.RESPONSE_FINISH_REASONS]) == {"stop"}
|
||||
assert a[f"{LiteLLM.COST_PREFIX}total"] == 0.002
|
||||
|
||||
assert a["llm.input_messages.0.message.content"] == "turn 0"
|
||||
assert a["llm.output_messages.0.message.content"] == "reply 0"
|
||||
|
||||
@pytest.mark.parametrize("turns", [60, 200])
|
||||
def test_long_conversation_does_not_evict_core_attributes(turns):
|
||||
"""Per-message OpenInference attributes fill the span's headroom and never crowd core telemetry off it."""
|
||||
span = _conversation_span(["genai", "openinference"], _conversation_payload(turns))
|
||||
_assert_core_intact(span)
|
||||
a = span.attributes
|
||||
limit = SpanLimits().max_span_attributes
|
||||
|
||||
assert limit - 1 <= len(a) <= limit
|
||||
indexed = _indexed_messages(a, "llm.input_messages")
|
||||
assert 1 < len(indexed) < turns
|
||||
assert indexed[0] == 0
|
||||
assert indexed[1:] == list(range(indexed[1], turns))
|
||||
assert a[f"llm.input_messages.{turns - 1}.message.content"] == f"turn {turns - 1}"
|
||||
assert f"llm.input_messages.{turns // 2}.message.role" not in a
|
||||
assert a["llm.output_messages.0.message.content"] == "reply 0"
|
||||
assert len(json.loads(a["input.value"])) == turns
|
||||
assert len(json.loads(a["output.value"])) == 1
|
||||
assert len(json.loads(a[GenAI.INPUT_MESSAGES])) == turns
|
||||
|
||||
|
||||
def test_short_conversation_keeps_every_message_indexed():
|
||||
"""Below the cap nothing is truncated in either direction."""
|
||||
a = _conversation_span(["genai", "openinference"], _conversation_payload(4, choices=2)).attributes
|
||||
for idx in range(4):
|
||||
@pytest.mark.parametrize("turns", [4, 8, 40])
|
||||
def test_conversation_that_fits_the_span_keeps_every_message_indexed(turns):
|
||||
"""No per-index message is shed while the span has room for all of them."""
|
||||
span = _conversation_span(["genai", "openinference"], _conversation_payload(turns, choices=2))
|
||||
_assert_core_intact(span)
|
||||
a = span.attributes
|
||||
for idx in range(turns):
|
||||
assert a[f"llm.input_messages.{idx}.message.role"] == ("user", "assistant")[idx % 2]
|
||||
assert a[f"llm.input_messages.{idx}.message.content"] == f"turn {idx}"
|
||||
for idx in range(2):
|
||||
assert a[f"llm.output_messages.{idx}.message.content"] == f"reply {idx}"
|
||||
|
|
@ -535,28 +538,159 @@ def test_indexed_prompt_keeps_opener_and_latest_turns_under_a_value_length_limit
|
|||
assert a["llm.input_messages.59.message.role"] == "user"
|
||||
assert a["llm.input_messages.59.message.content"] == "LATEST-TURN"
|
||||
assert a["llm.output_messages.0.message.content"] == "reply 0"
|
||||
assert [int(key.split(".")[2]) for key in a if key.endswith("message.content") and key.startswith("llm.input_")] == [
|
||||
0,
|
||||
*range(54, 60),
|
||||
]
|
||||
indexed = _indexed_messages(a, "llm.input_messages")
|
||||
assert indexed[0] == 0 and indexed[-1] == 59 and len(indexed) < 60
|
||||
assert indexed[1:] == list(range(indexed[1], 60))
|
||||
|
||||
|
||||
def test_message_cap_is_shared_across_input_and_output():
|
||||
"""One span-wide allowance covers both directions, and the response always keeps a share."""
|
||||
def test_prompt_turns_are_shed_before_response_choices():
|
||||
"""Under pressure the middle of the prompt goes first; every response choice keeps its keys."""
|
||||
long_prompt = _conversation_span(["genai", "openinference"], _conversation_payload(60, choices=1)).attributes
|
||||
many_choices = _conversation_span(["genai", "openinference"], _conversation_payload(60, choices=20)).attributes
|
||||
many_choices = _conversation_span(["genai", "openinference"], _conversation_payload(60, choices=20))
|
||||
_assert_core_intact(many_choices)
|
||||
|
||||
single_reply_indexed = _indexed_message_count(long_prompt, "llm.output_messages")
|
||||
assert single_reply_indexed == 1
|
||||
assert _indexed_message_count(long_prompt, "llm.input_messages") + single_reply_indexed == (
|
||||
MAX_MESSAGE_ATTRS_PER_SPAN // 2
|
||||
assert _indexed_messages(long_prompt, "llm.output_messages") == [0]
|
||||
assert _indexed_messages(many_choices.attributes, "llm.output_messages") == list(range(20))
|
||||
assert (
|
||||
1
|
||||
< len(_indexed_messages(many_choices.attributes, "llm.input_messages"))
|
||||
< len(_indexed_messages(long_prompt, "llm.input_messages"))
|
||||
)
|
||||
|
||||
assert _indexed_message_count(many_choices, "llm.input_messages") > 0
|
||||
assert _indexed_message_count(many_choices, "llm.output_messages") > single_reply_indexed
|
||||
assert _indexed_message_count(many_choices, "llm.input_messages") + _indexed_message_count(
|
||||
many_choices, "llm.output_messages"
|
||||
) == (MAX_MESSAGE_ATTRS_PER_SPAN // 2)
|
||||
|
||||
def test_indexed_messages_respect_a_lower_span_attribute_count_limit(monkeypatch):
|
||||
"""The budget follows the SDK's configured limit, not a hardcoded default."""
|
||||
monkeypatch.setenv("OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT", "48")
|
||||
span = _conversation_span(["genai", "openinference"], _conversation_payload(60))
|
||||
_assert_core_intact(span)
|
||||
a = span.attributes
|
||||
assert 47 <= len(a) <= 48
|
||||
assert a["llm.input_messages.0.message.content"] == "turn 0"
|
||||
assert a["llm.input_messages.59.message.content"] == "turn 59"
|
||||
assert a["llm.output_messages.0.message.content"] == "reply 0"
|
||||
|
||||
|
||||
def test_a_tight_span_keeps_the_reply_and_newest_turn_before_the_opener(monkeypatch):
|
||||
monkeypatch.setenv("OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT", "1000")
|
||||
full = _conversation_span(["genai", "openinference"], _conversation_payload(6)).attributes
|
||||
unindexed = [key for key in full if not key.startswith(("llm.input_messages.", "llm.output_messages."))]
|
||||
|
||||
monkeypatch.setenv("OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT", str(len(unindexed) + 4))
|
||||
a = _conversation_span(["genai", "openinference"], _conversation_payload(6)).attributes
|
||||
assert _indexed_messages(a, "llm.output_messages") == [0]
|
||||
assert _indexed_messages(a, "llm.input_messages") == [5]
|
||||
|
||||
monkeypatch.setenv("OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT", str(len(unindexed) + 2))
|
||||
a = _conversation_span(["genai", "openinference"], _conversation_payload(6)).attributes
|
||||
assert _indexed_messages(a, "llm.output_messages") == [0]
|
||||
assert _indexed_messages(a, "llm.input_messages") == []
|
||||
|
||||
|
||||
def test_shedding_stops_exactly_at_the_limit(monkeypatch):
|
||||
"""A span that fits exactly sheds nothing, and shedding never takes one pair more than the excess needs."""
|
||||
monkeypatch.setenv("OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT", "1000")
|
||||
full = dict(_conversation_span(["genai", "openinference"], _conversation_payload(30)).attributes)
|
||||
assert _indexed_messages(full, "llm.input_messages") == list(range(30))
|
||||
|
||||
monkeypatch.setenv("OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT", str(len(full)))
|
||||
exact = _conversation_span(["genai", "openinference"], _conversation_payload(30))
|
||||
assert exact.dropped_attributes == 0
|
||||
assert dict(exact.attributes) == full
|
||||
|
||||
monkeypatch.setenv("OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT", str(len(full) - 2))
|
||||
tight = _conversation_span(["genai", "openinference"], _conversation_payload(30))
|
||||
assert tight.dropped_attributes == 0
|
||||
assert len(tight.attributes) == len(full) - 2
|
||||
assert _indexed_messages(tight.attributes, "llm.input_messages") == [0, *range(2, 30)]
|
||||
|
||||
|
||||
def test_error_and_pre_stamped_attributes_keep_their_room_on_a_long_conversation():
|
||||
"""Attributes already on the span and the error set stamped after mapping both count against the budget."""
|
||||
cfg = OpenTelemetryV2Config(exporter="in_memory", mapper_names=["genai", "openinference"])
|
||||
provider, exporter = providers.in_memory_provider(cfg)
|
||||
engine = SpanEmitter(providers.get_tracer(provider, "litellm-test"), cfg)
|
||||
span = engine.start_span(SpanRole.LLM_CALL, "chat")
|
||||
for idx in range(10):
|
||||
span.set_attribute(f"litellm.metadata.baggage_{idx}", f"value {idx}")
|
||||
payload = _conversation_payload(
|
||||
60,
|
||||
status="failure",
|
||||
error_information={
|
||||
"error_class": "RateLimitError",
|
||||
"error_message": "429",
|
||||
"error_code": "429",
|
||||
"llm_provider": "openai",
|
||||
"traceback": "tb",
|
||||
},
|
||||
)
|
||||
engine.finish_span(
|
||||
SpanRole.LLM_CALL, span, LLMCallSpanData.from_standard_logging_payload(payload, capture_content=True)
|
||||
)
|
||||
(s,) = exporter.get_finished_spans()
|
||||
a = s.attributes
|
||||
|
||||
assert s.dropped_attributes == 0
|
||||
assert SpanLimits().max_span_attributes - 1 <= len(a) <= SpanLimits().max_span_attributes
|
||||
assert a[GenAI.REQUEST_MODEL] == "gpt-4o"
|
||||
assert a["litellm.metadata.baggage_0"] == "value 0"
|
||||
assert a["error.type"] == "RateLimitError"
|
||||
assert a["litellm.provider.error.stack_trace"] == "tb"
|
||||
assert a["llm.input_messages.0.message.content"] == "turn 0"
|
||||
assert a["llm.input_messages.59.message.content"] == "turn 59"
|
||||
|
||||
|
||||
def test_indexed_messages_follow_the_providers_own_span_limits(monkeypatch):
|
||||
"""A provider built with programmatic ``SpanLimits`` sets the budget, whatever the environment says."""
|
||||
monkeypatch.setenv("OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT", "1000")
|
||||
span = _conversation_span(
|
||||
["genai", "openinference"], _conversation_payload(60), span_limits=SpanLimits(max_span_attributes=40)
|
||||
)
|
||||
_assert_core_intact(span)
|
||||
a = span.attributes
|
||||
assert 39 <= len(a) <= 40
|
||||
assert a["llm.input_messages.0.message.content"] == "turn 0"
|
||||
assert a["llm.input_messages.59.message.content"] == "turn 59"
|
||||
assert a["llm.output_messages.0.message.content"] == "reply 0"
|
||||
|
||||
monkeypatch.setenv("OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT", "48")
|
||||
unbounded = _conversation_span(
|
||||
["genai", "openinference"],
|
||||
_conversation_payload(60),
|
||||
span_limits=SpanLimits(max_span_attributes=SpanLimits.UNSET),
|
||||
)
|
||||
_assert_core_intact(unbounded)
|
||||
assert _indexed_messages(unbounded.attributes, "llm.input_messages") == list(range(60))
|
||||
|
||||
|
||||
@pytest.mark.parametrize("opened_at_boundary", [False, True], ids=["emit", "start_span+finish_span"])
|
||||
def test_indexed_messages_follow_the_span_limits_of_a_per_request_tracer_override(monkeypatch, opened_at_boundary):
|
||||
"""A routed ``tracer`` builds the span, so its provider's limits set the budget, not the bound tracer's.
|
||||
|
||||
Holds whether the span is emitted in one shot or opened at the pre_call boundary and finished later.
|
||||
"""
|
||||
monkeypatch.setenv("OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT", "1000")
|
||||
cfg = OpenTelemetryV2Config(
|
||||
exporter="in_memory", mapper_names=["genai", "openinference"], capture_message_content="span_only"
|
||||
)
|
||||
bound_provider, _ = _provider_with_limits(SpanLimits(max_span_attributes=1000))
|
||||
routed_provider, routed_exporter = _provider_with_limits(SpanLimits(max_span_attributes=40))
|
||||
engine = SpanEmitter(providers.get_tracer(bound_provider, "litellm-test"), cfg)
|
||||
routed_tracer = providers.get_tracer(routed_provider, "litellm-routed")
|
||||
data = LLMCallSpanData.from_standard_logging_payload(_conversation_payload(60), capture_content=True)
|
||||
if opened_at_boundary:
|
||||
opened = engine.start_span(SpanRole.LLM_CALL, "chat", tracer=routed_tracer)
|
||||
engine.finish_span(SpanRole.LLM_CALL, opened, data)
|
||||
else:
|
||||
engine.emit(SpanRole.LLM_CALL, data, tracer=routed_tracer)
|
||||
(span,) = routed_exporter.get_finished_spans()
|
||||
_assert_core_intact(span)
|
||||
assert 39 <= len(span.attributes) <= 40
|
||||
assert span.attributes["llm.output_messages.0.message.content"] == "reply 0"
|
||||
|
||||
|
||||
def test_span_attribute_limit_falls_back_to_the_environment_for_spans_outside_the_sdk(monkeypatch):
|
||||
monkeypatch.setenv("OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT", "48")
|
||||
assert span_attribute_limit(INVALID_SPAN) == 48
|
||||
|
||||
|
||||
def test_fully_populated_span_with_every_vocabulary_stays_within_the_attribute_limit():
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue