From 1e583e8e79d37e48f8989c08efa08531511b5c95 Mon Sep 17 00:00:00 2001 From: yucheng Date: Mon, 14 Sep 2026 22:31:21 +0000 Subject: [PATCH 1/4] fix(otel v2): map the caller's Langfuse user, session and tags onto the root and generation spans `langfuse_otel` (OTel v2) only carried `trace_name` from the caller's metadata, so `metadata.trace_user_id` / `session_id` / `tags` (and the `langfuse_trace_user_id` / `langfuse_session_id` proxy headers) never reached Langfuse's user, session and tags fields. Widen the typed caller boundary to `TraceControls`, map it through one `LangfuseMapper.trace_attributes` table on both the root observation and the generation span, and keep `team_id` / `team_alias` proxy-authoritative Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- litellm/integrations/otel/langfuse_logger.py | 13 ++- litellm/integrations/otel/logger.py | 2 +- litellm/integrations/otel/mappers/langfuse.py | 27 +++++- litellm/integrations/otel/model/metadata.py | 61 ++++++++---- litellm/integrations/otel/model/payloads.py | 7 +- .../integrations/otel/test_langfuse_logger.py | 94 +++++++++++++++++++ .../otel/test_otel_v2_sources_of_truth.py | 61 ++++++++++-- .../otel/test_otel_v2_vendor_mappers.py | 32 ++++++- 8 files changed, 255 insertions(+), 42 deletions(-) diff --git a/litellm/integrations/otel/langfuse_logger.py b/litellm/integrations/otel/langfuse_logger.py index f8fd417392f..177bba0c1e1 100644 --- a/litellm/integrations/otel/langfuse_logger.py +++ b/litellm/integrations/otel/langfuse_logger.py @@ -6,9 +6,9 @@ from litellm.integrations.otel.logger import OpenTelemetryV2 from litellm.integrations.otel.mappers.langfuse import ( LANGFUSE_OBSERVATION_INPUT, LANGFUSE_OBSERVATION_OUTPUT, - LANGFUSE_TRACE_NAME, + LangfuseMapper, ) -from litellm.integrations.otel.model.metadata import caller_trace_name +from litellm.integrations.otel.model.metadata import caller_trace_controls from litellm.integrations.otel.model.request_io import request_input, response_output, stream_output from litellm.integrations.otel.plumbing.context import request_root_span @@ -18,14 +18,13 @@ if TYPE_CHECKING: class LangfuseOpenTelemetryV2(OpenTelemetryV2): - """Names the trace from the request. Langfuse reads ``langfuse.trace.name`` off the root observation, - and the proxy's root span is still recording when the LLM call starts.""" + """Stamps the caller's trace controls (name, user, session, tags) on the request. Langfuse reads them off + the root observation, and the proxy's root span is still recording when the LLM call starts.""" def log_pre_api_call(self, model: str, messages: object, kwargs: Mapping[str, object]) -> None: root: Final = request_root_span() - name: Final = caller_trace_name(kwargs) - if root is not None and root.is_recording() and name is not None: - root.set_attribute(LANGFUSE_TRACE_NAME, name) + if root is not None and root.is_recording(): + root.set_attributes(LangfuseMapper.trace_attributes(caller_trace_controls(kwargs))) super().log_pre_api_call(model, messages, kwargs) diff --git a/litellm/integrations/otel/logger.py b/litellm/integrations/otel/logger.py index 9ac748b231c..cf0eb04add7 100644 --- a/litellm/integrations/otel/logger.py +++ b/litellm/integrations/otel/logger.py @@ -554,7 +554,7 @@ class OpenTelemetryV2(CustomLogger): capture_content=self.config.capture_span_content, time_to_first_chunk_seconds=call.time_to_first_chunk_seconds, request_route=request_root_http_route(), - trace_name=call.trace_name, + trace=call.trace, ) end_time_ns: Final = to_ns(end_time) if carrier is not None and carrier.span is not None: diff --git a/litellm/integrations/otel/mappers/langfuse.py b/litellm/integrations/otel/mappers/langfuse.py index 98ff0f155a1..11a199966b9 100644 --- a/litellm/integrations/otel/mappers/langfuse.py +++ b/litellm/integrations/otel/mappers/langfuse.py @@ -5,12 +5,14 @@ Langfuse ingests OTLP spans and reads from its own vendor namespace ``GenAIMapper`` to send canonical + Langfuse-flavored spans simultaneously. Every attribute is declared as a ``key -> extractor`` table entry (one callable -per mapping operation): ``_LLM_CALL_ATTRS`` for scalars and ``_BLOB_ATTRS`` for -the JSON-serialized payloads. ``_llm_call`` just applies both tables. +per mapping operation): ``_LLM_CALL_ATTRS`` for scalars, ``_TRACE_ATTRS`` for the +caller's trace controls (shared with the root observation), and ``_BLOB_ATTRS`` for +the JSON-serialized payloads. ``_llm_call`` just applies the three tables. """ import json -from collections.abc import Callable +from collections.abc import Callable, Mapping +from types import MappingProxyType from typing import Final from litellm.integrations.otel.mappers.base import AttributeMap, AttrValue, SpanData @@ -20,6 +22,7 @@ from litellm.integrations.otel.mappers.utils import ( output_messages, serialize_messages, ) +from litellm.integrations.otel.model.metadata import TraceControls from litellm.integrations.otel.model.payloads import ( LLMCallSpanData, LLMRequestParams, @@ -29,6 +32,9 @@ from litellm.integrations.otel.model.payloads import ( LANGFUSE_OBSERVATION_INPUT: Final = "langfuse.observation.input" LANGFUSE_OBSERVATION_OUTPUT: Final = "langfuse.observation.output" LANGFUSE_TRACE_NAME: Final = "langfuse.trace.name" +LANGFUSE_TRACE_USER_ID: Final = "user.id" +LANGFUSE_TRACE_SESSION_ID: Final = "session.id" +LANGFUSE_TRACE_TAGS: Final = "langfuse.trace.tags" class LangfuseMapper: @@ -37,11 +43,19 @@ class LangfuseMapper: "langfuse.observation.model.name": lambda d: d.request_model or None, "langfuse.observation.metadata.provider": lambda d: d.provider or None, "langfuse.observation.id": lambda d: d.identity.call_id or None, - LANGFUSE_TRACE_NAME: lambda d: d.trace_name or None, "langfuse.trace.metadata.team_id": lambda d: d.identity.team_id or None, "langfuse.trace.metadata.team_alias": lambda d: d.identity.team_alias or None, } + _TRACE_ATTRS: Mapping[str, Callable[[TraceControls], AttrValue | None]] = MappingProxyType( + { + LANGFUSE_TRACE_NAME: lambda t: t.name or None, + LANGFUSE_TRACE_USER_ID: lambda t: t.user_id or None, + LANGFUSE_TRACE_SESSION_ID: lambda t: t.session_id or None, + LANGFUSE_TRACE_TAGS: lambda t: t.tags or None, + } + ) + # Sub-tables folded into their respective JSON blobs. _MODEL_PARAMS: dict[str, Callable[[LLMRequestParams], AttrValue | None]] = { "temperature": lambda rp: rp.temperature, @@ -77,9 +91,14 @@ class LangfuseMapper: case _: return {} + @classmethod + def trace_attributes(cls, trace: TraceControls) -> AttributeMap: + return collect(cls._TRACE_ATTRS, trace) + @classmethod def _llm_call(cls, data: LLMCallSpanData) -> AttributeMap: return { **collect(cls._LLM_CALL_ATTRS, data), + **cls.trace_attributes(data.trace), **collect(cls._BLOB_ATTRS, data), } diff --git a/litellm/integrations/otel/model/metadata.py b/litellm/integrations/otel/model/metadata.py index cc81b689708..8a5383e981a 100644 --- a/litellm/integrations/otel/model/metadata.py +++ b/litellm/integrations/otel/model/metadata.py @@ -36,7 +36,7 @@ model. They coincide on the SDK path, which is correct. from __future__ import annotations -from collections.abc import Iterator, Mapping +from collections.abc import Iterator, Mapping, Sequence from dataclasses import dataclass, field from types import MappingProxyType from typing import TYPE_CHECKING, Any, Final, cast @@ -48,7 +48,20 @@ from litellm.integrations.otel.model.utils import as_str, to_seconds if TYPE_CHECKING: from litellm.types.utils import StandardLoggingPayload -LANGFUSE_TRACE_NAME_HEADER: Final = "langfuse_trace_name" +LANGFUSE_HEADER_PREFIX: Final = "langfuse_" + + +@dataclass(frozen=True, slots=True) +class TraceControls: + """The caller's trace-level Langfuse controls: ``metadata.trace_name`` / ``trace_user_id`` / ``session_id`` / + ``tags`` on the request (SDK or proxy body), with the proxy's ``langfuse_`` headers winning over the + body for the scalar ones. Mutation controls (``trace_id``, ``existing_trace_id``, ``update_trace_keys``) are + deliberately not carried.""" + + name: str | None = None + user_id: str | None = None + session_id: str | None = None + tags: tuple[str, ...] = () @dataclass(frozen=True) @@ -217,7 +230,7 @@ class LLMCallEvent: # needs to be reasonable for a span that never gets closed (a leak). provisional_span_name: str time_to_first_chunk_seconds: float | None - trace_name: str | None + trace: TraceControls @classmethod def from_dict(cls, kwargs: Mapping[str, Any]) -> LLMCallEvent: @@ -234,29 +247,41 @@ class LLMCallEvent: upstream_started=kwargs.get("api_call_start_time") is not None, provisional_span_name=f"{operation.value} {model}".strip(), time_to_first_chunk_seconds=time_to_first_chunk_seconds(kwargs), - trace_name=caller_trace_name(kwargs), + trace=caller_trace_controls(kwargs), ) -def caller_trace_name(kwargs: Mapping[str, object]) -> str | None: +def caller_trace_controls(kwargs: Mapping[str, object]) -> TraceControls: request: Final = _as_str_mapping(kwargs.get("litellm_params")) if request is None: - return None + return TraceControls() proxy_request: Final = _as_str_mapping(request.get("proxy_server_request")) - headers: Final = _as_str_mapping(proxy_request.get("headers")) if proxy_request is not None else None - from_header: Final = as_str(headers.get(LANGFUSE_TRACE_NAME_HEADER)) if headers is not None else None - if from_header: - return from_header - return next( - ( - name - for key in ("metadata", "litellm_metadata") - if (metadata := _as_str_mapping(request.get(key))) is not None - and (name := as_str(metadata.get("trace_name"))) - ), - None, + headers: Final = (_as_str_mapping(proxy_request.get("headers")) if proxy_request is not None else None) or {} + bodies: Final = tuple( + metadata + for key in ("metadata", "litellm_metadata") + if (metadata := _as_str_mapping(request.get(key))) is not None ) + def scalar(control: str) -> str | None: + from_header: Final = as_str(headers.get(f"{LANGFUSE_HEADER_PREFIX}{control}")) + if from_header: + return from_header + return next((value for body in bodies if (value := as_str(body.get(control)))), None) + + return TraceControls( + name=scalar("trace_name"), + user_id=scalar("trace_user_id"), + session_id=scalar("session_id"), + tags=next((tags for body in bodies if (tags := _str_items(body.get("tags")))), ()), + ) + + +def _str_items(value: object) -> tuple[str, ...]: + if not isinstance(value, (list, tuple)): + return () + return tuple(item for item in cast("Sequence[object]", value) if isinstance(item, str) and item) + def time_to_first_chunk_seconds(kwargs: Mapping[str, Any]) -> float | None: """Seconds from the upstream request being issued (``api_call_start_time``) diff --git a/litellm/integrations/otel/model/payloads.py b/litellm/integrations/otel/model/payloads.py index c11c4a7a27d..068c3a38e21 100644 --- a/litellm/integrations/otel/model/payloads.py +++ b/litellm/integrations/otel/model/payloads.py @@ -13,6 +13,7 @@ from urllib.parse import urlsplit from litellm.integrations.otel.model.metadata import ( RequestContext, RequestIdentity, + TraceControls, ) from litellm.integrations.otel.model.semconv import ( GenAIOperation, @@ -387,7 +388,7 @@ class LLMCallSpanData: output_type: GenAIOutputType | None = None call_type: str | None = None request_route: str | None = None - trace_name: str | None = None + trace: TraceControls = field(default_factory=TraceControls) @classmethod def from_standard_logging_payload( @@ -396,7 +397,7 @@ class LLMCallSpanData: capture_content: bool = False, time_to_first_chunk_seconds: float | None = None, request_route: str | None = None, - trace_name: str | None = None, + trace: TraceControls | None = None, ) -> LLMCallSpanData: params: Final = cast(Mapping[str, object], payload.get("model_parameters") or {}) # The single parse of the request's metadata — the request-vs-provider @@ -438,7 +439,7 @@ class LLMCallSpanData: output_type=resolve_output_type(call_type), call_type=call_type or None, request_route=request_route or context.identity.request_route, - trace_name=trace_name, + trace=trace or TraceControls(), ) diff --git a/tests/test_litellm/integrations/otel/test_langfuse_logger.py b/tests/test_litellm/integrations/otel/test_langfuse_logger.py index 8db84b090a0..aca9dcc8a5e 100644 --- a/tests/test_litellm/integrations/otel/test_langfuse_logger.py +++ b/tests/test_litellm/integrations/otel/test_langfuse_logger.py @@ -42,6 +42,7 @@ from litellm.types.utils import ( # noqa: E402 INPUT_ATTR: Final = "langfuse.observation.input" OUTPUT_ATTR: Final = "langfuse.observation.output" TRACE_NAME_ATTR: Final = "langfuse.trace.name" +TRACE_CONTROL_ATTRS: Final = (TRACE_NAME_ATTR, "user.id", "session.id", "langfuse.trace.tags") CHAT_DATA: Final = {"model": "gpt-5.4-mini", "messages": [{"role": "user", "content": "ping"}]} @@ -374,6 +375,99 @@ def test_unnamed_request_leaves_the_trace_name_off_both_spans(): assert TRACE_NAME_ATTR not in root_attrs and TRACE_NAME_ATTR not in generation_attrs +@pytest.mark.parametrize("capture", ["span_only", "no_content"]) +def test_body_metadata_user_session_and_tags_land_on_the_root_and_the_generation(capture): + logger, exporter = _logger(capture=capture) + + root_attrs, generation_attrs = _run_named_request( + logger, + exporter, + { + "metadata": { + "trace_user_id": "user-42", + "session_id": "session-7", + "tags": ["prod", "eval", "nightly"], + "user_api_key_team_id": "team-from-proxy", + }, + "proxy_server_request": {"headers": {}}, + }, + ) + + for attrs in (root_attrs, generation_attrs): + assert attrs["user.id"] == "user-42" + assert attrs["session.id"] == "session-7" + assert tuple(attrs["langfuse.trace.tags"]) == ("prod", "eval", "nightly") + assert TRACE_NAME_ATTR not in attrs + + +def test_langfuse_user_and_session_headers_beat_body_metadata_on_both_spans(): + logger, exporter = _logger() + + root_attrs, generation_attrs = _run_named_request( + logger, + exporter, + { + "metadata": {"trace_user_id": "from-body", "session_id": "from-body"}, + "proxy_server_request": { + "headers": {"langfuse_trace_user_id": "from-header", "langfuse_session_id": "from-header-s"} + }, + }, + ) + + for attrs in (root_attrs, generation_attrs): + assert attrs["user.id"] == "from-header" + assert attrs["session.id"] == "from-header-s" + + +def test_caller_metadata_cannot_override_the_proxy_team_identity(): + logger, exporter = _logger() + response: Final = ModelResponse(choices=[Choices(message=Message(role="assistant", content="pong"))]) + litellm_params: Final = { + "metadata": {"trace_user_id": "u", "trace_metadata": {"team_id": "spoofed"}, "team_id": "spoofed"} + } + logger.log_pre_api_call( + model="gpt-5.4-mini", messages=[], kwargs={"litellm_call_id": "call_1", "litellm_params": litellm_params} + ) + payload: Final = { + "call_type": "acompletion", + "custom_llm_provider": "openai", + "model": "gpt-5.4-mini", + "messages": CHAT_DATA["messages"], + "response": response.model_dump(), + "status": "success", + "litellm_call_id": "call_1", + "metadata": { + "user_api_key_team_id": "real-team", + "user_api_key_team_alias": "real-alias", + "team_id": "spoofed", + "team_alias": "spoofed", + }, + "hidden_params": {}, + } + asyncio.run( + logger.async_log_success_event( + {"standard_logging_object": payload, "litellm_params": litellm_params}, response, None, None + ) + ) + + attrs: Final = dict(exporter.get_finished_spans()[0].attributes or {}) + assert attrs["user.id"] == "u" + assert attrs["langfuse.trace.metadata.team_id"] == "real-team" + assert attrs["langfuse.trace.metadata.team_alias"] == "real-alias" + assert "langfuse.trace.metadata" not in attrs and "langfuse.trace.id" not in attrs + + +def test_a_request_without_trace_controls_stamps_none_of_them(): + logger, exporter = _logger() + + root_attrs, generation_attrs = _run_named_request( + logger, exporter, {"metadata": {"user_api_key_team_id": "t1", "tags": []}, "proxy_server_request": {"headers": {}}} + ) + + assert set(TRACE_CONTROL_ATTRS).isdisjoint(root_attrs) + assert set(TRACE_CONTROL_ATTRS).isdisjoint(generation_attrs) + + @pytest.mark.parametrize( ("capture", "mappers"), [("no_content", ("genai", "langfuse")), ("span_only", ("genai",))], diff --git a/tests/test_litellm/integrations/otel/test_otel_v2_sources_of_truth.py b/tests/test_litellm/integrations/otel/test_otel_v2_sources_of_truth.py index 8baf9310538..d42077d6e6a 100644 --- a/tests/test_litellm/integrations/otel/test_otel_v2_sources_of_truth.py +++ b/tests/test_litellm/integrations/otel/test_otel_v2_sources_of_truth.py @@ -28,7 +28,7 @@ from litellm.integrations.otel import ( ) from litellm.integrations.otel.mappers.genai import GenAIMapper from litellm.integrations.otel.model import spans as spans_mod -from litellm.integrations.otel.model.metadata import LLMCallEvent, caller_trace_name +from litellm.integrations.otel.model.metadata import LLMCallEvent, TraceControls, caller_trace_controls from litellm.integrations.otel.model.payloads import ( LLMCallSpanData, RequestIdentity, @@ -743,15 +743,62 @@ def test_request_identity_falls_back_to_legacy_team_keys(): ids=["header", "body", "anthropic-body", "header-beats-body", "blank-header-falls-through", "neither", "empty"], ) def test_caller_trace_name_prefers_the_langfuse_header_over_body_metadata(request_data, expected): - assert caller_trace_name({"litellm_params": request_data}) == expected - assert LLMCallEvent.from_dict({"litellm_params": request_data}).trace_name == expected + assert caller_trace_controls({"litellm_params": request_data}).name == expected + assert LLMCallEvent.from_dict({"litellm_params": request_data}).trace.name == expected -def test_llm_span_data_carries_the_caller_trace_name(): - data: Final = LLMCallSpanData.from_standard_logging_payload(_sample_payload(), trace_name="nightly-eval") +@pytest.mark.parametrize( + ("request_data", "expected"), + [ + ( + {"metadata": {"trace_user_id": "u-body", "session_id": "s-body", "tags": ["a", "b", "c"]}}, + TraceControls(user_id="u-body", session_id="s-body", tags=("a", "b", "c")), + ), + ( + { + "proxy_server_request": { + "headers": {"langfuse_trace_user_id": "u-header", "langfuse_session_id": "s-header"} + }, + "metadata": {"trace_user_id": "u-body", "session_id": "s-body"}, + }, + TraceControls(user_id="u-header", session_id="s-header"), + ), + ( + {"litellm_metadata": {"trace_user_id": "u-anthropic", "session_id": "s-anthropic", "tags": ["x"]}}, + TraceControls(user_id="u-anthropic", session_id="s-anthropic", tags=("x",)), + ), + ( + {"metadata": {"tags": ["kept", 7, "", None, "also-kept"]}}, + TraceControls(tags=("kept", "also-kept")), + ), + ({"metadata": {"tags": "not-a-list", "trace_user_id": "", "session_id": 12}}, TraceControls(session_id="12")), + ( + { + "metadata": { + "trace_id": "forced", + "existing_trace_id": "forced", + "update_trace_keys": ["name"], + "trace_metadata": {"team_id": "spoofed"}, + "user_api_key_team_id": "t1", + } + }, + TraceControls(), + ), + ({}, TraceControls()), + ], + ids=["body", "headers-beat-body", "anthropic-body", "non-string-tags-dropped", "scalar-coercion", "mutation-controls-ignored", "empty"], +) +def test_caller_trace_controls_carry_user_session_and_tags(request_data, expected): + assert caller_trace_controls({"litellm_params": request_data}) == expected + assert LLMCallEvent.from_dict({"litellm_params": request_data}).trace == expected - assert data.trace_name == "nightly-eval" - assert LLMCallSpanData.from_standard_logging_payload(_sample_payload()).trace_name is None + +def test_llm_span_data_carries_the_caller_trace_controls(): + controls: Final = TraceControls(name="nightly-eval", user_id="u1", session_id="s1", tags=("a", "b")) + data: Final = LLMCallSpanData.from_standard_logging_payload(_sample_payload(), trace=controls) + + assert data.trace == controls + assert LLMCallSpanData.from_standard_logging_payload(_sample_payload()).trace == TraceControls() def test_llm_span_carries_proxy_request_route(): diff --git a/tests/test_litellm/integrations/otel/test_otel_v2_vendor_mappers.py b/tests/test_litellm/integrations/otel/test_otel_v2_vendor_mappers.py index bcdda93383a..81c9c2cdc9a 100644 --- a/tests/test_litellm/integrations/otel/test_otel_v2_vendor_mappers.py +++ b/tests/test_litellm/integrations/otel/test_otel_v2_vendor_mappers.py @@ -18,6 +18,7 @@ from litellm.integrations.otel.mappers import ( WeaveMapper, resolve_mappers, ) +from litellm.integrations.otel.model.metadata import TraceControls from litellm.integrations.otel.model.payloads import ( LLMCallSpanData, LLMRequestParams, @@ -135,8 +136,35 @@ def test_langfuse_mapper_observation_attrs(): def test_langfuse_mapper_names_the_trace_from_the_caller(): - assert LangfuseMapper().map(_llm_call(trace_name="nightly-eval"))["langfuse.trace.name"] == "nightly-eval" - assert "langfuse.trace.name" not in LangfuseMapper().map(_llm_call(trace_name=None)) + named = LangfuseMapper().map(_llm_call(trace=TraceControls(name="nightly-eval"))) + assert named["langfuse.trace.name"] == "nightly-eval" + assert "langfuse.trace.name" not in LangfuseMapper().map(_llm_call(trace=TraceControls())) + + +def test_langfuse_mapper_carries_the_caller_user_session_and_tags(): + controls = TraceControls(user_id="u-42", session_id="s-7", tags=("prod", "eval", "nightly")) + attrs = LangfuseMapper().map(_llm_call(trace=controls)) + + assert attrs["user.id"] == "u-42" + assert attrs["session.id"] == "s-7" + assert attrs["langfuse.trace.tags"] == ("prod", "eval", "nightly") + assert attrs["langfuse.trace.metadata.team_id"] == "t1" + assert attrs["langfuse.trace.metadata.team_alias"] == "team one" + + +def test_langfuse_mapper_omits_unset_trace_controls(): + attrs = LangfuseMapper().map(_llm_call(trace=TraceControls(user_id="", session_id=None, tags=()))) + + assert {"user.id", "session.id", "langfuse.trace.tags", "langfuse.trace.name"}.isdisjoint(attrs) + + +def test_langfuse_trace_attributes_match_between_root_and_generation(): + controls = TraceControls(name="n", user_id="u", session_id="s", tags=("t",)) + generation = LangfuseMapper().map(_llm_call(trace=controls)) + + root = LangfuseMapper.trace_attributes(controls) + assert root == {"langfuse.trace.name": "n", "user.id": "u", "session.id": "s", "langfuse.trace.tags": ("t",)} + assert all(generation[key] == value for key, value in root.items()) def test_langfuse_mapper_skips_when_no_messages(): From 75e7b402c8df7d23c40cac24fd7076f8800ddf43 Mon Sep 17 00:00:00 2001 From: yucheng Date: Mon, 14 Sep 2026 23:17:02 +0000 Subject: [PATCH 2/4] refactor(otel v2): move TraceControls into its own module to break the metadata <-> payloads import cycle CodeQL flagged that TraceControls could be undefined when metadata is imported before payloads. trace_controls now depends only on utils, and the mapping / sequence narrowing parses via pydantic TypeAdapter instead of cast. Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- litellm/integrations/otel/langfuse_logger.py | 2 +- litellm/integrations/otel/mappers/langfuse.py | 2 +- litellm/integrations/otel/model/metadata.py | 69 +++---------------- litellm/integrations/otel/model/payloads.py | 7 +- .../integrations/otel/model/trace_controls.py | 61 ++++++++++++++++ litellm/integrations/otel/model/utils.py | 13 ++++ .../otel/test_otel_v2_sources_of_truth.py | 3 +- .../otel/test_otel_v2_vendor_mappers.py | 2 +- 8 files changed, 89 insertions(+), 70 deletions(-) create mode 100644 litellm/integrations/otel/model/trace_controls.py diff --git a/litellm/integrations/otel/langfuse_logger.py b/litellm/integrations/otel/langfuse_logger.py index 177bba0c1e1..d029b153c52 100644 --- a/litellm/integrations/otel/langfuse_logger.py +++ b/litellm/integrations/otel/langfuse_logger.py @@ -8,8 +8,8 @@ from litellm.integrations.otel.mappers.langfuse import ( LANGFUSE_OBSERVATION_OUTPUT, LangfuseMapper, ) -from litellm.integrations.otel.model.metadata import caller_trace_controls from litellm.integrations.otel.model.request_io import request_input, response_output, stream_output +from litellm.integrations.otel.model.trace_controls import caller_trace_controls from litellm.integrations.otel.plumbing.context import request_root_span if TYPE_CHECKING: diff --git a/litellm/integrations/otel/mappers/langfuse.py b/litellm/integrations/otel/mappers/langfuse.py index 11a199966b9..8651cd15945 100644 --- a/litellm/integrations/otel/mappers/langfuse.py +++ b/litellm/integrations/otel/mappers/langfuse.py @@ -22,12 +22,12 @@ from litellm.integrations.otel.mappers.utils import ( output_messages, serialize_messages, ) -from litellm.integrations.otel.model.metadata import TraceControls from litellm.integrations.otel.model.payloads import ( LLMCallSpanData, LLMRequestParams, LLMUsage, ) +from litellm.integrations.otel.model.trace_controls import TraceControls LANGFUSE_OBSERVATION_INPUT: Final = "langfuse.observation.input" LANGFUSE_OBSERVATION_OUTPUT: Final = "langfuse.observation.output" diff --git a/litellm/integrations/otel/model/metadata.py b/litellm/integrations/otel/model/metadata.py index 8a5383e981a..2fdbc826803 100644 --- a/litellm/integrations/otel/model/metadata.py +++ b/litellm/integrations/otel/model/metadata.py @@ -36,33 +36,19 @@ model. They coincide on the SDK path, which is correct. from __future__ import annotations -from collections.abc import Iterator, Mapping, Sequence +from collections.abc import Iterator, Mapping from dataclasses import dataclass, field from types import MappingProxyType from typing import TYPE_CHECKING, Any, Final, cast from litellm.constants import LITELLM_LOGGING_NO_UPSTREAM_LLM_CALL from litellm.integrations.otel.model.semconv import resolve_operation -from litellm.integrations.otel.model.utils import as_str, to_seconds +from litellm.integrations.otel.model.trace_controls import TraceControls, caller_trace_controls +from litellm.integrations.otel.model.utils import as_str, as_str_mapping, to_seconds if TYPE_CHECKING: from litellm.types.utils import StandardLoggingPayload -LANGFUSE_HEADER_PREFIX: Final = "langfuse_" - - -@dataclass(frozen=True, slots=True) -class TraceControls: - """The caller's trace-level Langfuse controls: ``metadata.trace_name`` / ``trace_user_id`` / ``session_id`` / - ``tags`` on the request (SDK or proxy body), with the proxy's ``langfuse_`` headers winning over the - body for the scalar ones. Mutation controls (``trace_id``, ``existing_trace_id``, ``update_trace_keys``) are - deliberately not carried.""" - - name: str | None = None - user_id: str | None = None - session_id: str | None = None - tags: tuple[str, ...] = () - @dataclass(frozen=True) class RequestIdentity: @@ -251,38 +237,6 @@ class LLMCallEvent: ) -def caller_trace_controls(kwargs: Mapping[str, object]) -> TraceControls: - request: Final = _as_str_mapping(kwargs.get("litellm_params")) - if request is None: - return TraceControls() - proxy_request: Final = _as_str_mapping(request.get("proxy_server_request")) - headers: Final = (_as_str_mapping(proxy_request.get("headers")) if proxy_request is not None else None) or {} - bodies: Final = tuple( - metadata - for key in ("metadata", "litellm_metadata") - if (metadata := _as_str_mapping(request.get(key))) is not None - ) - - def scalar(control: str) -> str | None: - from_header: Final = as_str(headers.get(f"{LANGFUSE_HEADER_PREFIX}{control}")) - if from_header: - return from_header - return next((value for body in bodies if (value := as_str(body.get(control)))), None) - - return TraceControls( - name=scalar("trace_name"), - user_id=scalar("trace_user_id"), - session_id=scalar("session_id"), - tags=next((tags for body in bodies if (tags := _str_items(body.get("tags")))), ()), - ) - - -def _str_items(value: object) -> tuple[str, ...]: - if not isinstance(value, (list, tuple)): - return () - return tuple(item for item in cast("Sequence[object]", value) if isinstance(item, str) and item) - - def time_to_first_chunk_seconds(kwargs: Mapping[str, Any]) -> float | None: """Seconds from the upstream request being issued (``api_call_start_time``) to the first streamed chunk (``completion_start_time``); ``None`` for @@ -317,15 +271,8 @@ def auth_metadata(payload: StandardLoggingPayload | None, kwargs: Mapping[str, o ) -def _as_str_mapping(value: object) -> Mapping[str, object] | None: - """A read-only view of ``value`` when it is a mapping, else ``None``.""" - if not isinstance(value, Mapping): - return None - return cast("Mapping[str, object]", value) # cast-ok: isinstance-guarded, JSON metadata has str keys - - def _string_entries(value: object) -> Mapping[str, str] | None: - entries: Final = _as_str_mapping(value) + entries: Final = as_str_mapping(value) if entries is None: return None typed: Final = MappingProxyType({key: item for key, item in entries.items() if isinstance(item, str)}) @@ -341,18 +288,18 @@ def _metadata_dicts( litellm copies it onto ``metadata``, but both are yielded so a route that populates only one is still covered. """ - payload_view: Final = _as_str_mapping(payload) + payload_view: Final = as_str_mapping(payload) if payload_view is not None: - payload_metadata: Final = _as_str_mapping(payload_view.get("metadata")) + payload_metadata: Final = as_str_mapping(payload_view.get("metadata")) if payload_metadata is not None: yield payload_metadata - params: Final = _as_str_mapping(kwargs.get("litellm_params")) + params: Final = as_str_mapping(kwargs.get("litellm_params")) if params is None: return yield from ( metadata for key in ("metadata", "litellm_metadata") - if (metadata := _as_str_mapping(params.get(key))) is not None + if (metadata := as_str_mapping(params.get(key))) is not None ) diff --git a/litellm/integrations/otel/model/payloads.py b/litellm/integrations/otel/model/payloads.py index 068c3a38e21..33da1549fd5 100644 --- a/litellm/integrations/otel/model/payloads.py +++ b/litellm/integrations/otel/model/payloads.py @@ -10,11 +10,7 @@ from types import MappingProxyType from typing import TYPE_CHECKING, ClassVar, Final, cast from urllib.parse import urlsplit -from litellm.integrations.otel.model.metadata import ( - RequestContext, - RequestIdentity, - TraceControls, -) +from litellm.integrations.otel.model.metadata import RequestContext, RequestIdentity from litellm.integrations.otel.model.semconv import ( GenAIOperation, GenAIOutputType, @@ -23,6 +19,7 @@ from litellm.integrations.otel.model.semconv import ( resolve_output_type, resolve_provider, ) +from litellm.integrations.otel.model.trace_controls import TraceControls from litellm.integrations.otel.model.utils import ( as_bool, as_float, diff --git a/litellm/integrations/otel/model/trace_controls.py b/litellm/integrations/otel/model/trace_controls.py new file mode 100644 index 00000000000..884c51a420b --- /dev/null +++ b/litellm/integrations/otel/model/trace_controls.py @@ -0,0 +1,61 @@ +"""The caller's Langfuse trace controls, parsed from the live callback kwargs.""" + +from __future__ import annotations + +from collections.abc import Mapping +from dataclasses import dataclass +from typing import Final + +from pydantic import TypeAdapter, ValidationError + +from litellm.integrations.otel.model.utils import as_str, as_str_mapping + +LANGFUSE_HEADER_PREFIX: Final = "langfuse_" +_ITEMS: Final = TypeAdapter(tuple[object, ...]) + + +@dataclass(frozen=True, slots=True) +class TraceControls: + """The caller's trace-level Langfuse controls: ``metadata.trace_name`` / ``trace_user_id`` / ``session_id`` / + ``tags`` on the request (SDK or proxy body), with the proxy's ``langfuse_`` headers winning over the + body for the scalar ones. Mutation controls (``trace_id``, ``existing_trace_id``, ``update_trace_keys``) are + deliberately not carried.""" + + name: str | None = None + user_id: str | None = None + session_id: str | None = None + tags: tuple[str, ...] = () + + +def caller_trace_controls(kwargs: Mapping[str, object]) -> TraceControls: + request: Final = as_str_mapping(kwargs.get("litellm_params")) + if request is None: + return TraceControls() + proxy_request: Final = as_str_mapping(request.get("proxy_server_request")) + headers: Final = (as_str_mapping(proxy_request.get("headers")) if proxy_request is not None else None) or {} + bodies: Final = tuple( + metadata + for key in ("metadata", "litellm_metadata") + if (metadata := as_str_mapping(request.get(key))) is not None + ) + + def scalar(control: str) -> str | None: + from_header: Final = as_str(headers.get(f"{LANGFUSE_HEADER_PREFIX}{control}")) + if from_header: + return from_header + return next((value for body in bodies if (value := as_str(body.get(control)))), None) + + return TraceControls( + name=scalar("trace_name"), + user_id=scalar("trace_user_id"), + session_id=scalar("session_id"), + tags=next((tags for body in bodies if (tags := _str_items(body.get("tags")))), ()), + ) + + +def _str_items(value: object) -> tuple[str, ...]: + try: + items: Final = _ITEMS.validate_python(value) + except ValidationError: + return () + return tuple(item for item in items if isinstance(item, str) and item) diff --git a/litellm/integrations/otel/model/utils.py b/litellm/integrations/otel/model/utils.py index fb35e9abf51..a3276f30078 100644 --- a/litellm/integrations/otel/model/utils.py +++ b/litellm/integrations/otel/model/utils.py @@ -8,7 +8,13 @@ parsing lives in :mod:`litellm.integrations.otel.plumbing.providers` instead, because it delegates to the OTel SDK's own W3C Baggage parser. """ +from collections.abc import Mapping from datetime import datetime +from typing import Final + +from pydantic import TypeAdapter, ValidationError + +_STR_MAPPING: Final = TypeAdapter(Mapping[str, object]) def as_str(value: object) -> str | None: @@ -55,6 +61,13 @@ def as_bool(value: object) -> bool | None: return bool(value) +def as_str_mapping(value: object) -> Mapping[str, object] | None: + try: + return _STR_MAPPING.validate_python(value) + except ValidationError: + return None + + def as_str_tuple(value: object) -> tuple[str, ...] | None: if value is None: return None diff --git a/tests/test_litellm/integrations/otel/test_otel_v2_sources_of_truth.py b/tests/test_litellm/integrations/otel/test_otel_v2_sources_of_truth.py index d42077d6e6a..c5c77a12a62 100644 --- a/tests/test_litellm/integrations/otel/test_otel_v2_sources_of_truth.py +++ b/tests/test_litellm/integrations/otel/test_otel_v2_sources_of_truth.py @@ -28,7 +28,8 @@ from litellm.integrations.otel import ( ) from litellm.integrations.otel.mappers.genai import GenAIMapper from litellm.integrations.otel.model import spans as spans_mod -from litellm.integrations.otel.model.metadata import LLMCallEvent, TraceControls, caller_trace_controls +from litellm.integrations.otel.model.metadata import LLMCallEvent +from litellm.integrations.otel.model.trace_controls import TraceControls, caller_trace_controls from litellm.integrations.otel.model.payloads import ( LLMCallSpanData, RequestIdentity, diff --git a/tests/test_litellm/integrations/otel/test_otel_v2_vendor_mappers.py b/tests/test_litellm/integrations/otel/test_otel_v2_vendor_mappers.py index 81c9c2cdc9a..bd83357305e 100644 --- a/tests/test_litellm/integrations/otel/test_otel_v2_vendor_mappers.py +++ b/tests/test_litellm/integrations/otel/test_otel_v2_vendor_mappers.py @@ -18,7 +18,7 @@ from litellm.integrations.otel.mappers import ( WeaveMapper, resolve_mappers, ) -from litellm.integrations.otel.model.metadata import TraceControls +from litellm.integrations.otel.model.trace_controls import TraceControls from litellm.integrations.otel.model.payloads import ( LLMCallSpanData, LLMRequestParams, From 923853016603c5867548d52450659ac01ff08d32 Mon Sep 17 00:00:00 2001 From: yucheng Date: Mon, 14 Sep 2026 23:53:27 +0000 Subject: [PATCH 3/4] refactor(otel v2): map Langfuse trace controls in a plain function instead of a MappingProxyType table CodeQL resolved the stdlib types import in mappers/langfuse.py to litellm.proxy.management_endpoints.types and reported a new import cycle through the OTel package Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- litellm/integrations/otel/mappers/langfuse.py | 32 +++++++++---------- 1 file changed, 15 insertions(+), 17 deletions(-) diff --git a/litellm/integrations/otel/mappers/langfuse.py b/litellm/integrations/otel/mappers/langfuse.py index 8651cd15945..ae8c26721d8 100644 --- a/litellm/integrations/otel/mappers/langfuse.py +++ b/litellm/integrations/otel/mappers/langfuse.py @@ -5,19 +5,19 @@ Langfuse ingests OTLP spans and reads from its own vendor namespace ``GenAIMapper`` to send canonical + Langfuse-flavored spans simultaneously. Every attribute is declared as a ``key -> extractor`` table entry (one callable -per mapping operation): ``_LLM_CALL_ATTRS`` for scalars, ``_TRACE_ATTRS`` for the -caller's trace controls (shared with the root observation), and ``_BLOB_ATTRS`` for -the JSON-serialized payloads. ``_llm_call`` just applies the three tables. +per mapping operation): ``_LLM_CALL_ATTRS`` for scalars and ``_BLOB_ATTRS`` for +the JSON-serialized payloads. ``trace_attributes`` maps the caller's trace controls +(shared with the root observation); ``_llm_call`` applies both tables plus it. """ import json -from collections.abc import Callable, Mapping -from types import MappingProxyType +from collections.abc import Callable from typing import Final from litellm.integrations.otel.mappers.base import AttributeMap, AttrValue, SpanData from litellm.integrations.otel.mappers.utils import ( collect, + drop_none, json_if, output_messages, serialize_messages, @@ -47,15 +47,6 @@ class LangfuseMapper: "langfuse.trace.metadata.team_alias": lambda d: d.identity.team_alias or None, } - _TRACE_ATTRS: Mapping[str, Callable[[TraceControls], AttrValue | None]] = MappingProxyType( - { - LANGFUSE_TRACE_NAME: lambda t: t.name or None, - LANGFUSE_TRACE_USER_ID: lambda t: t.user_id or None, - LANGFUSE_TRACE_SESSION_ID: lambda t: t.session_id or None, - LANGFUSE_TRACE_TAGS: lambda t: t.tags or None, - } - ) - # Sub-tables folded into their respective JSON blobs. _MODEL_PARAMS: dict[str, Callable[[LLMRequestParams], AttrValue | None]] = { "temperature": lambda rp: rp.temperature, @@ -91,9 +82,16 @@ class LangfuseMapper: case _: return {} - @classmethod - def trace_attributes(cls, trace: TraceControls) -> AttributeMap: - return collect(cls._TRACE_ATTRS, trace) + @staticmethod + def trace_attributes(trace: TraceControls) -> AttributeMap: + return drop_none( + { + LANGFUSE_TRACE_NAME: trace.name or None, + LANGFUSE_TRACE_USER_ID: trace.user_id or None, + LANGFUSE_TRACE_SESSION_ID: trace.session_id or None, + LANGFUSE_TRACE_TAGS: trace.tags or None, + } + ) @classmethod def _llm_call(cls, data: LLMCallSpanData) -> AttributeMap: From 79aae7f06248965f2f4cfb9fd8248b6dbdf58f2a Mon Sep 17 00:00:00 2001 From: yucheng Date: Thu, 17 Sep 2026 00:49:12 +0000 Subject: [PATCH 4/4] refactor(otel v2): build Langfuse trace attributes from pairs to satisfy the type discipline gate Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- litellm/integrations/otel/mappers/langfuse.py | 16 ++++++++-------- litellm/integrations/otel/mappers/utils.py | 9 +++++++-- .../integrations/otel/model/trace_controls.py | 4 ++-- 3 files changed, 17 insertions(+), 12 deletions(-) diff --git a/litellm/integrations/otel/mappers/langfuse.py b/litellm/integrations/otel/mappers/langfuse.py index ae8c26721d8..e76cffde881 100644 --- a/litellm/integrations/otel/mappers/langfuse.py +++ b/litellm/integrations/otel/mappers/langfuse.py @@ -17,7 +17,7 @@ from typing import Final from litellm.integrations.otel.mappers.base import AttributeMap, AttrValue, SpanData from litellm.integrations.otel.mappers.utils import ( collect, - drop_none, + drop_none_pairs, json_if, output_messages, serialize_messages, @@ -84,13 +84,13 @@ class LangfuseMapper: @staticmethod def trace_attributes(trace: TraceControls) -> AttributeMap: - return drop_none( - { - LANGFUSE_TRACE_NAME: trace.name or None, - LANGFUSE_TRACE_USER_ID: trace.user_id or None, - LANGFUSE_TRACE_SESSION_ID: trace.session_id or None, - LANGFUSE_TRACE_TAGS: trace.tags or None, - } + return drop_none_pairs( + ( + (LANGFUSE_TRACE_NAME, trace.name or None), + (LANGFUSE_TRACE_USER_ID, trace.user_id or None), + (LANGFUSE_TRACE_SESSION_ID, trace.session_id or None), + (LANGFUSE_TRACE_TAGS, trace.tags or None), + ) ) @classmethod diff --git a/litellm/integrations/otel/mappers/utils.py b/litellm/integrations/otel/mappers/utils.py index c023621d2ef..f8644765f59 100644 --- a/litellm/integrations/otel/mappers/utils.py +++ b/litellm/integrations/otel/mappers/utils.py @@ -6,7 +6,7 @@ they live in one place. """ import json -from collections.abc import Callable, Mapping, Sequence +from collections.abc import Callable, Iterable, Mapping, Sequence from typing import Final from litellm.integrations.otel.mappers.base import AttributeMap, AttrValue @@ -47,7 +47,12 @@ def tool_attr_budget(vocabularies: int) -> int: def drop_none(values: Mapping[str, AttrValue | None]) -> AttributeMap: """Return ``values`` with ``None``-valued entries removed.""" - return {k: v for k, v in values.items() if v is not None} + return drop_none_pairs(values.items()) + + +def drop_none_pairs(pairs: Iterable[tuple[str, AttrValue | None]]) -> AttributeMap: + """Return ``pairs`` as a map with ``None``-valued entries removed.""" + return {k: v for k, v in pairs if v is not None} def tool_definition_attrs( diff --git a/litellm/integrations/otel/model/trace_controls.py b/litellm/integrations/otel/model/trace_controls.py index 884c51a420b..eac7b5c897b 100644 --- a/litellm/integrations/otel/model/trace_controls.py +++ b/litellm/integrations/otel/model/trace_controls.py @@ -32,7 +32,7 @@ def caller_trace_controls(kwargs: Mapping[str, object]) -> TraceControls: if request is None: return TraceControls() proxy_request: Final = as_str_mapping(request.get("proxy_server_request")) - headers: Final = (as_str_mapping(proxy_request.get("headers")) if proxy_request is not None else None) or {} + headers: Final = as_str_mapping(proxy_request.get("headers")) if proxy_request is not None else None bodies: Final = tuple( metadata for key in ("metadata", "litellm_metadata") @@ -40,7 +40,7 @@ def caller_trace_controls(kwargs: Mapping[str, object]) -> TraceControls: ) def scalar(control: str) -> str | None: - from_header: Final = as_str(headers.get(f"{LANGFUSE_HEADER_PREFIX}{control}")) + from_header: Final = as_str(headers.get(f"{LANGFUSE_HEADER_PREFIX}{control}")) if headers is not None else None if from_header: return from_header return next((value for body in bodies if (value := as_str(body.get(control)))), None)