diff --git a/litellm/integrations/otel/mappers/langfuse.py b/litellm/integrations/otel/mappers/langfuse.py index b55d9746f63..5aed173254c 100644 --- a/litellm/integrations/otel/mappers/langfuse.py +++ b/litellm/integrations/otel/mappers/langfuse.py @@ -103,6 +103,18 @@ class LangfuseMapper: ), } + #: Every attribute this mapper can write on a model call. + LLM_CALL_ATTRIBUTES: Final[frozenset[str]] = frozenset( + { + *_LLM_CALL_ATTRS, + *_BLOB_ATTRS, + LANGFUSE_TRACE_NAME, + LANGFUSE_TRACE_USER_ID, + LANGFUSE_TRACE_SESSION_ID, + LANGFUSE_TRACE_TAGS, + } + ) + def map(self, data: SpanData) -> AttributeMap: match data: case LLMCallSpanData(): diff --git a/litellm/integrations/otel/plumbing/providers.py b/litellm/integrations/otel/plumbing/providers.py index a65a75d2128..bcbeb5bd2f3 100644 --- a/litellm/integrations/otel/plumbing/providers.py +++ b/litellm/integrations/otel/plumbing/providers.py @@ -46,6 +46,7 @@ from litellm.integrations.otel.mappers.langfuse import ( LANGFUSE_OBSERVATION_OUTPUT, LANGFUSE_REQUEST_METADATA_ATTRIBUTES, LANGFUSE_TRACE_NAME, + LangfuseMapper, ) from litellm.integrations.otel.model.config import ( CaptureMessageContent, @@ -488,19 +489,25 @@ def _is_litellm_span(span: ReadableSpan) -> bool: return scope is not None and scope.name == LITELLM_TRACER_NAME -def without_langfuse_request_metadata(span: ReadableSpan) -> ReadableSpan: - """``span`` without the request metadata the Langfuse mapper writes for a Langfuse destination. +def without_destination_langfuse_attributes(span: ReadableSpan, langfuse_mapped: bool) -> ReadableSpan: + """``span`` without the Langfuse attributes only a Langfuse destination asked for. - Only LiteLLM's own spans lose it: any other span keeps every key its own code set. + That is the request metadata the Langfuse mapper writes and, on a model call written by + a logger whose own mappers leave the Langfuse vocabulary out (``langfuse_mapped`` off), + every Langfuse attribute, since that logger wrote them for a key's or team's Langfuse + destination alone. Only LiteLLM's own spans lose them: any other span keeps every key + its own code set. """ attributes: Final = span.attributes or _NO_ATTRIBUTES - if not (_is_litellm_span(span) and any(key in LANGFUSE_REQUEST_METADATA_ATTRIBUTES for key in attributes)): + dropped: Final = ( + LANGFUSE_REQUEST_METADATA_ATTRIBUTES + if langfuse_mapped or not is_llm_call_span(span) + else LangfuseMapper.LLM_CALL_ATTRIBUTES + ) + if not (_is_litellm_span(span) and any(key in dropped for key in attributes)): return span return _CountedSpanView( - span, - MappingProxyType( - {key: value for key, value in attributes.items() if key not in LANGFUSE_REQUEST_METADATA_ATTRIBUTES} - ), + span, MappingProxyType({key: value for key, value in attributes.items() if key not in dropped}) ) @@ -656,6 +663,7 @@ def _for_destination( span: ReadableSpan, destination: "OtelDestination", default_capture: str, + langfuse_mapped: bool, ) -> ReadableSpan: """The view of ``span`` a tenant destination receives. @@ -670,13 +678,16 @@ def _for_destination( and the query string of the request URL, which can hold the same key. A destination whose effective setting disables span content loses the prompt and response content too, and a destination for any backend but Langfuse loses the - request metadata the Langfuse mapper writes on LiteLLM's own spans. The span itself - stays, so the tenant still gets the whole trace tree. + Langfuse attributes only a Langfuse destination asked for (see + :func:`without_destination_langfuse_attributes`). The span itself stays, so the + tenant still gets the whole trace tree. """ extra: Final = destination.resource_attributes defaults: Final = destination.resource_defaults source: Final = ( - span if destination.callback_name == ExporterOwner.LANGFUSE_OTEL else without_langfuse_request_metadata(span) + span + if destination.callback_name == ExporterOwner.LANGFUSE_OTEL + else without_destination_langfuse_attributes(span, langfuse_mapped) ) attributes: Final = source.attributes or _NO_ATTRIBUTES database: Final = _is_database_span(attributes) @@ -726,8 +737,10 @@ class TenantFanOutSpanProcessor(SpanProcessor): default_capture_by_backend: Mapping[str, str] = MappingProxyType({}), pending_drains: int = _MAX_PENDING_DRAINS, drain_pool: _DrainPool | None = None, + langfuse_mapped: bool = True, ) -> None: self._operator_sinks: Final = operator_sinks + self._langfuse_mapped: Final = langfuse_mapped self._excluded_db_systems: Final = excluded_db_systems self._default_capture: Final = default_capture self._default_capture_by_backend: Final = default_capture_by_backend @@ -820,7 +833,8 @@ class TenantFanOutSpanProcessor(SpanProcessor): try: processor.on_end( _scoped( - _for_destination(span, destination, self._default_capture_for(destination)), destination.span_scope + _for_destination(span, destination, self._default_capture_for(destination), self._langfuse_mapped), + destination.span_scope, ) ) except Exception as exc: # noqa: BLE001 # one destination's failure must not cost the others their span @@ -1196,17 +1210,19 @@ class _OverriddenBackendFilter(SpanProcessor): return self._inner.force_flush(timeout_millis) -class _LangfuseRequestMetadataFilter(SpanProcessor): - """Strip the Langfuse request metadata from LiteLLM's own spans before ``inner`` exports them.""" +class _DestinationLangfuseAttributeFilter(SpanProcessor): + """Strip the Langfuse attributes only a Langfuse destination asked for from LiteLLM's own spans before ``inner`` + exports them.""" - def __init__(self, inner: SpanProcessor) -> None: + def __init__(self, inner: SpanProcessor, langfuse_mapped: bool) -> None: self._inner: Final = inner + self._langfuse_mapped: Final = langfuse_mapped def on_start(self, span: SDKSpan, parent_context: Context | None = None) -> None: self._inner.on_start(span, parent_context) def on_end(self, span: ReadableSpan) -> None: - self._inner.on_end(without_langfuse_request_metadata(span)) + self._inner.on_end(without_destination_langfuse_attributes(span, self._langfuse_mapped)) def shutdown(self) -> None: self._inner.shutdown() @@ -1489,13 +1505,18 @@ def _spec_processor( spec: ExporterSpec, use_simple_processor: bool | None, strip_message_content: bool, - strip_langfuse_request_metadata: bool, + strip_destination_langfuse_attributes: bool, + langfuse_mapped: bool, ) -> SpanProcessor: exporting: Final = _processor_for( _exporter_from_spec(spec), (spec.use_simple_processor if spec.use_simple_processor is not None else use_simple_processor), ) - scoped: Final = _LangfuseRequestMetadataFilter(exporting) if strip_langfuse_request_metadata else exporting + scoped: Final = ( + _DestinationLangfuseAttributeFilter(exporting, langfuse_mapped) + if strip_destination_langfuse_attributes + else exporting + ) return _MessageContentFilter(scoped) if strip_message_content else scoped @@ -1524,9 +1545,9 @@ def build_tracer_provider( ``config.langfuse_span_scope`` narrows the exporter owned by ``langfuse_otel`` alone; a collector or any other backend in the same config keeps the full tree. - Every exporter but ``langfuse_otel``'s goes without the request metadata the Langfuse - mapper writes on LiteLLM's own spans, which any logger writes for a request with a - Langfuse destination. + Every exporter but ``langfuse_otel``'s goes without the Langfuse attributes only a + Langfuse destination asked for (see :func:`without_destination_langfuse_attributes`), + which any logger writes for a request with a Langfuse destination. ``strip_message_content`` removes prompt and response content before every exporter, for a routed provider whose spans can carry content another destination asked for. @@ -1540,6 +1561,7 @@ def build_tracer_provider( provider.add_span_processor(_processor_for(exporter, use_simple_processor)) return provider + langfuse_mapped: Final = _maps_langfuse(config) # ``config._normalize`` guarantees at least one spec (it folds the top-level # ``exporter``/``endpoint``/``headers`` fields in when ``exporters`` is empty). for spec in config.exporters: @@ -1549,7 +1571,8 @@ def build_tracer_provider( spec, use_simple_processor, strip_message_content, - strip_langfuse_request_metadata=spec.owner is not ExporterOwner.LANGFUSE_OTEL, + strip_destination_langfuse_attributes=spec.owner is not ExporterOwner.LANGFUSE_OTEL, + langfuse_mapped=langfuse_mapped, ) owner = spec.owner.value if tenant_overrides and spec.owner is not None else None scope = _operator_scope(config, spec) @@ -1586,8 +1609,9 @@ def attach_tenant_fan_out( so exactly one fan-out lands. ``configs`` name the operator's own exporters, one config per v2 logger since each keeps its own provider and still writes its account, so an additive destination pointing at any of them is delivered once - rather than twice. ``excluded_db_systems`` only filters what the fan-out - delivers, never the operator's own exporters. + rather than twice. The first is the config of the logger that owns ``provider``, + whose model calls the fan-out carries. ``excluded_db_systems`` only filters what + the fan-out delivers, never the operator's own exporters. """ with _FAN_OUT_ATTACH_LOCK: if any(isinstance(processor, TenantFanOutSpanProcessor) for processor in _attached_processors(provider)): @@ -1598,10 +1622,16 @@ def attach_tenant_fan_out( excluded_db_systems=excluded_db_systems, default_capture=_operator_capture_mode(configs), default_capture_by_backend=MappingProxyType(dict(_owned_capture_modes(configs))), + langfuse_mapped=any(_maps_langfuse(config) for config in configs[:1]), ) ) +def _maps_langfuse(config: OpenTelemetryV2Config) -> bool: + """Whether a logger built from ``config`` writes the Langfuse vocabulary on every model call.""" + return "langfuse" in config.mapper_names + + def _operator_capture_mode(configs: Sequence[OpenTelemetryV2Config]) -> str: """The mode of the operator's generic ``otel`` callback, so ``callback_settings.otel`` counts as the global default""" return next( diff --git a/tests/integration/observability/test_langfuse_otel_tenant_only.py b/tests/integration/observability/test_langfuse_otel_tenant_only.py index 88398e367b2..3ab1b30562d 100644 --- a/tests/integration/observability/test_langfuse_otel_tenant_only.py +++ b/tests/integration/observability/test_langfuse_otel_tenant_only.py @@ -136,7 +136,9 @@ def test_a_tenant_langfuse_project_gets_user_session_and_tags_without_operator_k ] == [{"user.id": "user-42", "session.id": session, "langfuse.trace.tags": TAGS}], [s["name"] for s in tenant] assert "langfuse.observation.metadata" in tenant[0]["attributes"], sorted(tenant[0]["attributes"]) assert len(operator) == 1, [span["name"] for span in operator] - assert "langfuse.observation.metadata" not in operator[0]["attributes"], sorted(operator[0]["attributes"]) + assert [ + name for name in operator[0]["attributes"] if name in ("user.id", "session.id") or name.startswith("langfuse.") + ] == [], sorted(operator[0]["attributes"]) def test_a_tenant_langfuse_project_gets_the_keys_langfuse_environment( diff --git a/tests/unit/integrations/otel/test_otel_v2_destinations.py b/tests/unit/integrations/otel/test_otel_v2_destinations.py index d26d238b0b7..a35580e0e8f 100644 --- a/tests/unit/integrations/otel/test_otel_v2_destinations.py +++ b/tests/unit/integrations/otel/test_otel_v2_destinations.py @@ -35,7 +35,7 @@ from litellm.integrations.otel.logger import ( publish_global_otel_v2_provider, ) from litellm.integrations.otel.mappers import resolve_mappers -from litellm.integrations.otel.mappers.langfuse import LANGFUSE_REQUEST_METADATA_ATTRIBUTES +from litellm.integrations.otel.mappers.langfuse import LANGFUSE_REQUEST_METADATA_ATTRIBUTES, LangfuseMapper from litellm.integrations.otel.model.config import ( CaptureMessageContent, ExporterOwner, @@ -3481,6 +3481,62 @@ class TestLangfuseDestinationFields: assert "langfuse.observation.metadata" in to_langfuse assert "langfuse.observation.metadata" not in received[None] + @staticmethod + def _received_by_account( + monkeypatch: pytest.MonkeyPatch, config: OpenTelemetryV2Config, destinations: tuple[OtelDestination, ...] + ) -> Mapping[str | None, Mapping[str, object]]: + """The attributes of the model call each account received, keyed by the headers it was sent with.""" + credential_less_proxy(monkeypatch) + capture: Final = _ExporterCapture() + for kind in ("otlp_grpc", "otlp_http", "console"): + monkeypatch.setitem(otel_providers._EXPORTER_FACTORIES, kind, capture.build) # pyright: ignore[reportPrivateUsage] # the factory table is the seam for capturing each exporter + provider: Final = build_tracer_provider(config, tenant_overrides=True) + attach_tenant_fan_out(provider, config) + emitter: Final = SpanEmitter(get_tracer(provider, "litellm"), config) + call: Final = replace( + _MODEL_CALL_WITHOUT_CONTENT, + trace=TraceControls(user_id="user-42", session_id="sess-1", tags=("agent:orders",)), + request_metadata=MappingProxyType({"user_api_key_alias": "prod-key"}), + ) + + def with_destinations() -> None: + set_request_destinations(deliverable_destinations(destinations, provider)) + emitter.emit(SpanRole.LLM_CALL, call) + + in_fresh_context(with_destinations) + provider.force_flush() + return MappingProxyType( + {headers: dict(exporter.get_finished_spans()[0].attributes or {}) for headers, exporter in capture.built} + ) + + def test_only_the_langfuse_destination_gets_the_langfuse_fields_written_for_it( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """The operator's own exporter and the key's Arize destination get the model call as they would with no + Langfuse destination on the request.""" + received: Final = self._received_by_account(monkeypatch, OpenTelemetryV2Config(), (LANGFUSE_DEST, ARIZE_DEST)) + + assert received[LANGFUSE_DEST.header_string()]["user.id"] == "user-42" + assert { + headers: sorted(LangfuseMapper.LLM_CALL_ATTRIBUTES & attributes.keys()) + for headers, attributes in received.items() + if headers != LANGFUSE_DEST.header_string() + } == {None: [], ARIZE_DEST.header_string(): []} + + def test_an_exporter_whose_config_maps_langfuse_keeps_the_langfuse_fields_but_the_request_metadata( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + received: Final = self._received_by_account( + monkeypatch, OpenTelemetryV2Config(mapper_names=["genai", "langfuse"]), (LANGFUSE_DEST,) + ) + + assert {key: received[None].get(key) for key in ("user.id", "session.id", "langfuse.trace.tags")} == { + "user.id": "user-42", + "session.id": "sess-1", + "langfuse.trace.tags": ("agent:orders",), + } + assert not LANGFUSE_REQUEST_METADATA_ATTRIBUTES & received[None].keys() + @pytest.mark.parametrize( ("key_environment", "tracing_environment", "expected"), [