diff --git a/litellm/integrations/langfuse/langfuse_otel.py b/litellm/integrations/langfuse/langfuse_otel.py index a8651bc2555..3efa474b6db 100644 --- a/litellm/integrations/langfuse/langfuse_otel.py +++ b/litellm/integrations/langfuse/langfuse_otel.py @@ -51,7 +51,14 @@ class LangfuseOtelLogger(OpenTelemetry): Sets OpenTelemetry span attributes for Langfuse observability. Uses the same attribute setting logic as Arize Phoenix for consistency. """ + from opentelemetry.sdk.trace import ReadableSpan + # The caller identity goes first, so a span that reaches its attribute limit evicts it before any + # other attribute. A span that already carries attributes (a caller's active span) does not get it. + if not (isinstance(span, ReadableSpan) and span.attributes): + LangfuseOtelLogger._set_request_metadata_attributes( + span=span, kwargs=_MAPPING_ADAPTER.validate_python(kwargs) + ) _utils.set_attributes(span, kwargs, response_obj, LangfuseLLMObsOTELAttributes) span.set_attribute("langfuse.observation.type", "generation") @@ -278,7 +285,6 @@ class LangfuseOtelLogger(OpenTelemetry): metadata: Final = LangfuseOtelLogger._extract_langfuse_metadata(kwargs) LangfuseOtelLogger._set_metadata_attributes(span=span, metadata=metadata) - LangfuseOtelLogger._set_request_metadata_attributes(span=span, kwargs=_MAPPING_ADAPTER.validate_python(kwargs)) messages: Final = kwargs.get("messages") if messages: diff --git a/litellm/integrations/otel/emitter.py b/litellm/integrations/otel/emitter.py index a4cb0a84541..225ea87f078 100644 --- a/litellm/integrations/otel/emitter.py +++ b/litellm/integrations/otel/emitter.py @@ -13,6 +13,7 @@ from opentelemetry.trace.status import Status, StatusCode from litellm.integrations.otel.mappers import resolve_mappers from litellm.integrations.otel.mappers.base import AttributeMapper, AttrValue, SpanData +from litellm.integrations.otel.mappers.langfuse import LANGFUSE_REQUEST_METADATA_ATTRIBUTES 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 ( @@ -303,7 +304,9 @@ class SpanEmitter: ) 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, frozenset(mapped))).items(): + # The Langfuse request metadata sheds first, so a span at its attribute limit loses nothing else to it. + budget: Final = attribute_budget(span, reserved, frozenset(mapped)) + for key, value in fit_indexed_messages(mapped, budget, LANGFUSE_REQUEST_METADATA_ATTRIBUTES).items(): span.set_attribute(key, value) if error: stamped: Final = stamp_error(span, error) diff --git a/litellm/integrations/otel/mappers/openinference.py b/litellm/integrations/otel/mappers/openinference.py index 8d7b4d7cf4a..0c465bbe5cb 100644 --- a/litellm/integrations/otel/mappers/openinference.py +++ b/litellm/integrations/otel/mappers/openinference.py @@ -153,7 +153,9 @@ def _shed_order(groups: Mapping[tuple[str, int, int], tuple[str, ...]]) -> tuple _METADATA_KEY: Final = "metadata" -def fit_indexed_messages(attrs: Mapping[str, AttrValue], budget: int | None) -> Mapping[str, AttrValue]: +def fit_indexed_messages( + attrs: Mapping[str, AttrValue], budget: int | None, shed_first: frozenset[str] = frozenset() +) -> Mapping[str, AttrValue]: """``attrs`` with indexed message attributes 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 @@ -161,12 +163,14 @@ def fit_indexed_messages(attrs: Mapping[str, AttrValue], budget: int | None) -> after every indexed message attribute: message attributes are the indexed, queryable view (the blobs carry no per-index keys), so a count-squeezed span keeps them and the single metadata key absorbs only the residual shortfall. Shedding happens here, before ``span.set_attribute``, so the fit is exact and - the SDK's dropped-attributes counter never silently masks the choice. + the SDK's dropped-attributes counter never silently masks the choice. ``shed_first`` keys shed one at a + time before anything else, so they never cost the span any other attribute. """ if budget is None or len(attrs) <= budget: return attrs groups: Final = _message_key_groups(attrs) - sheddable: Final = (*(groups[group] for group in _shed_order(groups)),) + leading: Final = tuple((key,) for key in attrs if key in shed_first) + sheddable: Final = (*leading, *(groups[group] for group in _shed_order(groups))) flex: Final = (_METADATA_KEY,) if _METADATA_KEY in attrs else () candidates: Final = (*sheddable, flex) if flex else sheddable running: Final = tuple(accumulate(len(candidate) for candidate in candidates)) diff --git a/tests/integration/observability/test_langfuse_otel_metadata.py b/tests/integration/observability/test_langfuse_otel_metadata.py index be867d07fe6..d43ff4779bb 100644 --- a/tests/integration/observability/test_langfuse_otel_metadata.py +++ b/tests/integration/observability/test_langfuse_otel_metadata.py @@ -1396,3 +1396,157 @@ def test_langfuse_otel_metadata_omits_prompt_quoting_routing_fields( assert keyword not in json.dumps(langfuse_metadata, default=str), langfuse_metadata if otel_v2: assert keyword not in json.dumps(attrs, default=str), attrs + + +def _span_records(body: bytes) -> tuple[tuple[dict[str, object], int], ...]: + request: Final = ExportTraceServiceRequest.FromString(body) + return tuple( + ( + { + attribute.key: getattr(attribute.value, attribute.value.WhichOneof("value")) + for attribute in span.attributes + }, + span.dropped_attributes_count, + ) + for resource in request.resource_spans + for scope in resource.scope_spans + for span in scope.spans + ) + + +def _identity_attributes(attrs: Mapping[str, object]) -> frozenset[str]: + names: Final = {"langfuse.observation.metadata", *(f"langfuse.trace.metadata.{field}" for field in _IDENTITIES)} + return frozenset(key for key in attrs if key in names) + + +def _limit_sweep( + rig: _Rig, + key: str, + marker: str, + bodies: Mapping[int, Mapping[str, object]], +) -> dict[int, tuple[dict[str, object], int]]: + """The exported span and dropped-attribute count per size; each body's prompt carries ``::``.""" + drained: list[ + tuple[dict[str, object], int] + ] = [] # mutable-ok: drain() consumes the wire queue, so observed spans must accumulate across eventually polls + for body in bodies.values(): + response = rig.candidate.request("POST", "/v1/chat/completions", dict(body), key=key) + assert response.status_code == 200, response.text + + def by_size() -> dict[int, tuple[dict[str, object], int] | None]: + drained.extend(record for batch in rig.collector.drain() for record in _span_records(batch.body)) + return { + size: next( + ( + (attrs, dropped) + for attrs, dropped in drained + if f"{marker}:{size}:" in str(attrs.get("langfuse.observation.input", "")) + ), + None, + ) + for size in bodies + } + + observed: Final = eventually(by_size, lambda spans: None not in spans.values(), seconds=30) + return {size: record for size, record in observed.items() if record is not None} + + +def test_langfuse_otel_identity_is_evicted_before_any_other_attribute_of_a_full_span( + gateway: Gateway, tmp_path: Path +) -> None: + marker: Final = "lf-limit-" + uuid.uuid4().hex + with ( + _langfuse_rig(gateway, tmp_path, {}, marker) as rig, + rig.candidate.scenario() as scenario, + ): + key, identity = _scenario_identity(scenario, rig.provider) + + def body(tool_count: int) -> dict[str, object]: + return { + "model": identity["model"], + "messages": [{"role": "user", "content": f"{marker}:{tool_count}:"}], + "user": identity["end_user"], + "tools": [ + { + "type": "function", + "function": { + "name": f"tool_{n}", + "description": f"looks up thing {n}", + "parameters": {"type": "object", "properties": {"q": {"type": "string"}}}, + }, + } + for n in range(tool_count) + ], + "cache": {"no-cache": True}, + } + + spans: Final = _limit_sweep( + rig, key, marker, {tool_count: body(tool_count) for tool_count in (0, *range(30, 41))} + ) + + reference, reference_dropped = spans[0] + identity_count: Final = len(_identity_attributes(reference)) + assert reference_dropped == 0, reference_dropped + assert identity_count == len(_chat_identity_fields(identity)) + 1, sorted(reference) + for tool_count, (attrs, dropped) in spans.items(): + kept = _identity_attributes(attrs) + assert len(kept) == identity_count - min(dropped, identity_count), (tool_count, dropped, sorted(kept)) + if dropped <= identity_count: + expected = {"openinference.span.kind", "metadata", "llm.model_name", "llm.request.type"} + assert expected <= set(attrs), (tool_count, dropped, sorted(expected - set(attrs))) + tools = {f"llm.tools.{n}.name" for n in range(tool_count)} + assert tools <= set(attrs), (tool_count, dropped, sorted(tools - set(attrs))) + assert [tool_count for tool_count, (_, dropped) in spans.items() if 0 < dropped <= identity_count], { + tool_count: dropped for tool_count, (_, dropped) in spans.items() + } + + +def test_otel_v2_langfuse_identity_sheds_before_any_message_attribute(gateway: Gateway, tmp_path: Path) -> None: + marker: Final = "lf-v2-limit-" + uuid.uuid4().hex + with ( + _langfuse_rig( + gateway, + tmp_path, + {}, + marker, + overrides={ + "LITELLM_OTEL_V2": "1", + "MAPPER_NAMES": "genai,openinference", + "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT": "span_only", + }, + ) as rig, + rig.candidate.scenario() as scenario, + ): + key, identity = _scenario_identity(scenario, rig.provider) + + def body(message_count: int) -> dict[str, object]: + return { + "model": identity["model"], + "messages": [ + {"role": ("user", "assistant")[n % 2], "content": f"{marker} turn {n}"} + for n in range(message_count - 1) + ] + + [{"role": "user", "content": f"{marker}:{message_count}: turn {message_count - 1}"}], + "user": identity["end_user"], + "cache": {"no-cache": True}, + } + + spans: Final = _limit_sweep( + rig, key, marker, {message_count: body(message_count) for message_count in (1, *range(20, 61))} + ) + + reference, _ = spans[1] + identity_count: Final = len(_identity_attributes(reference)) + assert identity_count == len(_chat_identity_fields(identity)) + 1, sorted(reference) + shed_by_size: Final[dict[int, tuple[int, int]]] = {} # mutable-ok: filled once per swept span below + for message_count, (attrs, _) in spans.items(): + kept = {n for n in range(message_count) if f"llm.input_messages.{n}.message.content" in attrs} + shed_messages = message_count - len(kept) + shed_identity = identity_count - len(_identity_attributes(attrs)) + shed_by_size[message_count] = (shed_identity, shed_messages) + if shed_messages: + assert shed_identity == identity_count, (message_count, shed_identity, shed_messages) + assert [ + size for size, (shed_identity, shed_messages) in shed_by_size.items() if shed_identity and not shed_messages + ], shed_by_size + assert [size for size, (_, shed_messages) in shed_by_size.items() if shed_messages], shed_by_size diff --git a/tests/unit/integrations/otel/test_otel_v2_emitter.py b/tests/unit/integrations/otel/test_otel_v2_emitter.py index 949eb97bccd..044ecd0c09e 100644 --- a/tests/unit/integrations/otel/test_otel_v2_emitter.py +++ b/tests/unit/integrations/otel/test_otel_v2_emitter.py @@ -795,6 +795,37 @@ def test_indexed_messages_follow_the_span_limits_of_a_per_request_tracer_overrid assert span.attributes["llm.output_messages.0.message.content"] == "reply 0" +def _team_conversation_span(monkeypatch, redact, max_span_attributes): + monkeypatch.setattr("litellm.redact_user_api_key_info", redact) + return _conversation_span( + ["genai", "openinference", "langfuse"], + _conversation_payload(60), + span_limits=SpanLimits(max_span_attributes=max_span_attributes), + ) + + +def test_langfuse_identity_sheds_before_any_attribute_the_span_carries_without_it(monkeypatch): + """At its attribute limit a span drops the Langfuse identity keys first, then exactly what it drops without them.""" + plain = _team_conversation_span(monkeypatch, True, SpanLimits.UNSET).attributes + full = _team_conversation_span(monkeypatch, False, SpanLimits.UNSET).attributes + identity_keys = set(full) - set(plain) + assert identity_keys == { + "langfuse.observation.metadata", + "langfuse.trace.metadata.user_api_key_team_id", + "langfuse.trace.metadata.user_api_key_team_alias", + } + + squeezed_limit = len(plain) - 10 + squeezed_plain = _team_conversation_span(monkeypatch, True, squeezed_limit) + squeezed = _team_conversation_span(monkeypatch, False, squeezed_limit) + assert dict(squeezed.attributes) == dict(squeezed_plain.attributes) + assert squeezed.dropped_attributes == squeezed_plain.dropped_attributes == 0 + + roomy = _team_conversation_span(monkeypatch, False, len(plain) + 2).attributes + assert {key: roomy.get(key) for key in plain} == dict(plain) + assert len(set(roomy) & identity_keys) == 2 + + 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 diff --git a/tests/unit/integrations/test_langfuse_otel.py b/tests/unit/integrations/test_langfuse_otel.py index 8b5332db190..eb7630781e4 100644 --- a/tests/unit/integrations/test_langfuse_otel.py +++ b/tests/unit/integrations/test_langfuse_otel.py @@ -6,7 +6,9 @@ from typing import Final from unittest.mock import MagicMock, patch import pytest -from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.trace import ReadableSpan, SpanLimits, TracerProvider +from opentelemetry.sdk.trace.export import SimpleSpanProcessor +from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter import litellm from litellm.integrations.langfuse.langfuse_otel import LangfuseOtelLogger @@ -506,8 +508,11 @@ class TestLangfuseOtelIntegration: "standard_logging_object": {"metadata": request_metadata}, } - with patch("litellm.integrations.arize._utils.safe_set_attribute") as mock_safe_set_attribute: - LangfuseOtelLogger._set_langfuse_specific_attributes(MagicMock(), kwargs, None) + with ( + patch("litellm.integrations.arize._utils.safe_set_attribute") as mock_safe_set_attribute, + patch("litellm.integrations.arize._utils.set_attributes"), + ): + LangfuseOtelLogger.set_langfuse_otel_attributes(MagicMock(), kwargs, None) actual: Final = {call.args[1]: call.args[2] for call in mock_safe_set_attribute.call_args_list} @@ -545,8 +550,11 @@ class TestLangfuseOtelIntegration: "standard_logging_object": {"metadata": request_metadata}, } - with patch("litellm.integrations.arize._utils.safe_set_attribute") as mock_safe_set_attribute: - LangfuseOtelLogger._set_langfuse_specific_attributes(MagicMock(), kwargs, None) + with ( + patch("litellm.integrations.arize._utils.safe_set_attribute") as mock_safe_set_attribute, + patch("litellm.integrations.arize._utils.set_attributes"), + ): + LangfuseOtelLogger.set_langfuse_otel_attributes(MagicMock(), kwargs, None) actual: Final = {call.args[1]: call.args[2] for call in mock_safe_set_attribute.call_args_list} @@ -574,8 +582,11 @@ class TestLangfuseOtelIntegration: "standard_logging_object": {"metadata": request_metadata}, } - with patch("litellm.integrations.arize._utils.safe_set_attribute") as mock_safe_set_attribute: - LangfuseOtelLogger._set_langfuse_specific_attributes(MagicMock(), kwargs, None) + with ( + patch("litellm.integrations.arize._utils.safe_set_attribute") as mock_safe_set_attribute, + patch("litellm.integrations.arize._utils.set_attributes"), + ): + LangfuseOtelLogger.set_langfuse_otel_attributes(MagicMock(), kwargs, None) actual: Final = {call.args[1]: call.args[2] for call in mock_safe_set_attribute.call_args_list} langfuse_metadata: Final = { @@ -605,8 +616,11 @@ class TestLangfuseOtelIntegration: "standard_logging_object": {"metadata": request_metadata}, } - with patch("litellm.integrations.arize._utils.safe_set_attribute") as mock_safe_set_attribute: - LangfuseOtelLogger._set_langfuse_specific_attributes(MagicMock(), kwargs, None) + with ( + patch("litellm.integrations.arize._utils.safe_set_attribute") as mock_safe_set_attribute, + patch("litellm.integrations.arize._utils.set_attributes"), + ): + LangfuseOtelLogger.set_langfuse_otel_attributes(MagicMock(), kwargs, None) actual: Final = {call.args[1]: call.args[2] for call in mock_safe_set_attribute.call_args_list} @@ -627,8 +641,11 @@ class TestLangfuseOtelIntegration: }, } - with patch("litellm.integrations.arize._utils.safe_set_attribute") as mock_safe_set_attribute: - LangfuseOtelLogger._set_langfuse_specific_attributes(MagicMock(), kwargs, None) + with ( + patch("litellm.integrations.arize._utils.safe_set_attribute") as mock_safe_set_attribute, + patch("litellm.integrations.arize._utils.set_attributes"), + ): + LangfuseOtelLogger.set_langfuse_otel_attributes(MagicMock(), kwargs, None) actual: Final = {call.args[1]: call.args[2] for call in mock_safe_set_attribute.call_args_list} @@ -638,8 +655,11 @@ class TestLangfuseOtelIntegration: def test_request_metadata_keys_are_absent_without_standard_logging_object(self) -> None: kwargs: Final = {"litellm_params": {"metadata": {"trace_metadata": {"k": "v"}}}} - with patch("litellm.integrations.arize._utils.safe_set_attribute") as mock_safe_set_attribute: - LangfuseOtelLogger._set_langfuse_specific_attributes(MagicMock(), kwargs, None) + with ( + patch("litellm.integrations.arize._utils.safe_set_attribute") as mock_safe_set_attribute, + patch("litellm.integrations.arize._utils.set_attributes"), + ): + LangfuseOtelLogger.set_langfuse_otel_attributes(MagicMock(), kwargs, None) actual: Final = {call.args[1]: call.args[2] for call in mock_safe_set_attribute.call_args_list} @@ -669,14 +689,89 @@ class TestLangfuseOtelIntegration: } response_obj: Final = {"choices": [{"message": {"role": "assistant", "content": "hello"}}]} - with patch("litellm.integrations.arize._utils.safe_set_attribute") as mock_safe_set_attribute: - LangfuseOtelLogger._set_langfuse_specific_attributes(MagicMock(), kwargs, response_obj) + with ( + patch("litellm.integrations.arize._utils.safe_set_attribute") as mock_safe_set_attribute, + patch("litellm.integrations.arize._utils.set_attributes"), + ): + LangfuseOtelLogger.set_langfuse_otel_attributes(MagicMock(), kwargs, response_obj) actual: Final = {call.args[1]: call.args[2] for call in mock_safe_set_attribute.call_args_list} assert json.loads(actual["langfuse.observation.metadata"]) == {"user_api_key_alias": "prod-key"} assert json.loads(actual["langfuse.observation.output"]) == {"role": "assistant", "content": "hello"} + @staticmethod + def _exported_span( + kwargs: dict[str, object], max_span_attributes: int | None, carried: dict[str, str] | None = None + ) -> ReadableSpan: + exporter: Final = InMemorySpanExporter() + provider: Final = TracerProvider(span_limits=SpanLimits(max_span_attributes=max_span_attributes)) + provider.add_span_processor(SimpleSpanProcessor(exporter)) + span: Final = provider.get_tracer("langfuse-otel-test").start_span("litellm_request", attributes=carried) + response_obj: Final = {"id": "chatcmpl-1", "choices": [{"message": {"role": "assistant", "content": "hello"}}]} + LangfuseOtelLogger.set_langfuse_otel_attributes(span, kwargs, response_obj) + span.end() + (finished,) = exporter.get_finished_spans() + return finished + + @staticmethod + def _tool_call_kwargs() -> dict[str, object]: + tools: Final = [ + { + "type": "function", + "function": {"name": f"tool_{n}", "description": f"does thing {n}", "parameters": {"type": "object"}}, + } + for n in range(3) + ] + messages: Final = [{"role": "user", "content": "hello"}] + return { + "model": "gpt-4o", + "messages": messages, + "optional_params": {"tools": tools}, + "litellm_params": {"metadata": {}}, + "standard_logging_object": { + "model": "gpt-4o", + "custom_llm_provider": "openai", + "call_type": "completion", + "messages": messages, + "model_parameters": {"tools": tools}, + "metadata": {"user_api_key_alias": "prod-key", "user_api_key_team_id": "team-1"}, + }, + } + + def test_request_metadata_never_evicts_another_attribute_at_the_span_limit( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + kwargs: Final = self._tool_call_kwargs() + monkeypatch.setattr(litellm, "redact_user_api_key_info", True) + plain: Final = dict(self._exported_span(kwargs, SpanLimits.UNSET).attributes or {}) + monkeypatch.setattr(litellm, "redact_user_api_key_info", False) + full: Final = dict(self._exported_span(kwargs, SpanLimits.UNSET).attributes or {}) + identity_keys: Final = set(full) - set(plain) + assert identity_keys == { + "langfuse.observation.metadata", + "langfuse.trace.metadata.user_api_key_alias", + "langfuse.trace.metadata.user_api_key_team_id", + } + + at_limit: Final = self._exported_span(kwargs, len(plain)) + assert dict(at_limit.attributes or {}) == plain + + one_spare: Final = dict(self._exported_span(kwargs, len(plain) + 1).attributes or {}) + assert {key: one_spare.get(key) for key in plain} == plain + assert len(set(one_spare) & identity_keys) == 1 + + def test_request_metadata_skips_a_span_that_already_carries_attributes(self) -> None: + attributes: Final = dict( + self._exported_span(self._tool_call_kwargs(), SpanLimits.UNSET, {"caller.attribute": "kept"}).attributes + or {} + ) + + assert attributes["caller.attribute"] == "kept" + assert attributes["langfuse.observation.type"] == "generation" + assert "langfuse.observation.metadata" not in attributes + assert not [key for key in attributes if key.startswith("langfuse.trace.metadata.")] + class TestLangfuseOtelKeyDynamicConfig: """Key/team-scoped Langfuse credentials must define the full export target