fix(otel v2): map the caller's Langfuse user, session and tags onto the root and generation spans

litellm/integrations/otel/langfuse_logger.py

@@ -6,10 +6,10 @@
 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.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:
@@ -18,14 +18,13 @@
 
 
 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)
 
 

litellm/integrations/otel/logger.py

@@ -555,7 +555,7 @@ def _finish_carrier(
             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:

litellm/integrations/otel/mappers/langfuse.py

@@ -6,7 +6,8 @@
 
 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.
+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
@@ -16,6 +17,7 @@
 from litellm.integrations.otel.mappers.base import AttributeMap, AttrValue, SpanData
 from litellm.integrations.otel.mappers.utils import (
     collect,
+    drop_none_pairs,
     json_if,
     output_messages,
     serialize_messages,
@@ -25,10 +27,14 @@
     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"
 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,7 +43,6 @@ 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,
     }
@@ -77,9 +82,21 @@ def map(self, data: SpanData) -> AttributeMap:
             case _:
                 return {}
 
+    @staticmethod
+    def trace_attributes(trace: TraceControls) -> AttributeMap:
+        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
     def _llm_call(cls, data: LLMCallSpanData) -> AttributeMap:
         return {
             **collect(cls._LLM_CALL_ATTRS, data),
+            **cls.trace_attributes(data.trace),
             **collect(cls._BLOB_ATTRS, data),
         }

litellm/integrations/otel/mappers/utils.py

@@ -6,7 +6,7 @@
 """
 
 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(

litellm/integrations/otel/model/metadata.py

@@ -43,12 +43,12 @@
 
 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_TRACE_NAME_HEADER: Final = "langfuse_trace_name"
 REQUESTER_METADATA_KEY: Final = "requester_metadata"
 REQUESTER_METADATA_PATH: Final = f"{REQUESTER_METADATA_KEY}."
 
@@ -225,7 +225,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:
@@ -242,30 +242,10 @@ def from_dict(cls, kwargs: Mapping[str, Any]) -> 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:
-    request: Final = _as_str_mapping(kwargs.get("litellm_params"))
-    if request is None:
-        return None
-    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,
-    )
-
-
 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
@@ -300,15 +280,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)})
@@ -324,18 +297,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
     )
 
 
@@ -365,14 +338,14 @@ def metadata_from_request_data(data: object) -> Mapping[str, object] | None:
     The proxy stores it under ``metadata`` or ``litellm_metadata`` depending on the route;
     the proxy-owned siblings (``user_api_key_*``, ``requester_ip_address``) are not read.
     """
-    top: Final = _as_str_mapping(data)
+    top: Final = as_str_mapping(data)
     if top is None:
         return None
     snapshots: Final = tuple(
         snapshot
         for name in ("metadata", "litellm_metadata")
-        if (nested := _as_str_mapping(top.get(name))) is not None
-        and (snapshot := _as_str_mapping(nested.get(REQUESTER_METADATA_KEY))) is not None
+        if (nested := as_str_mapping(top.get(name))) is not None
+        and (snapshot := as_str_mapping(nested.get(REQUESTER_METADATA_KEY))) is not None
     )
     return MappingProxyType({REQUESTER_METADATA_KEY: snapshots[0]}) if snapshots else None
 
@@ -382,7 +355,7 @@ def flatten_metadata(raw: Mapping[str, object]) -> Iterator[tuple[str, str]]:
     stack: Final = list(tuple(raw.items())[::-1])  # mutable-ok: iterative worklist keeps the walk off the call stack
     while stack:
         key, value = stack.pop()
-        if (nested := _as_str_mapping(value)) is not None:
+        if (nested := as_str_mapping(value)) is not None:
             stack.extend(tuple((f"{key}.{sub_key}", sub_value) for sub_key, sub_value in nested.items())[::-1])
         elif isinstance(value, (str, bool, int, float)):
             yield key, str(value)

litellm/integrations/otel/model/payloads.py

@@ -10,10 +10,7 @@
 from typing import TYPE_CHECKING, ClassVar, Final, cast
 from urllib.parse import urlsplit
 
-from litellm.integrations.otel.model.metadata import (
-    RequestContext,
-    RequestIdentity,
-)
+from litellm.integrations.otel.model.metadata import RequestContext, RequestIdentity
 from litellm.integrations.otel.model.semconv import (
     GenAIOperation,
     GenAIOutputType,
@@ -22,6 +19,7 @@
     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,
@@ -387,7 +385,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 +394,7 @@ def from_standard_logging_payload(
         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 +436,7 @@ def from_standard_logging_payload(
             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(),
         )
 
 

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_<control>`` 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
+    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 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)
+
+    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)

litellm/integrations/otel/model/utils.py

@@ -8,7 +8,13 @@
 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