From 88e44717a4e8ae8b70b3d9fd128aa92cbb538541 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Thu, 27 Aug 2026 06:49:32 +0000 Subject: [PATCH] fix(otel): populate Langfuse root observation IO Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- litellm/integrations/otel/logger.py | 17 +++++- litellm/integrations/otel/mappers/langfuse.py | 8 +++ .../integrations/otel/test_otel_v2_logger.py | 54 ++++++++++++++++--- 3 files changed, 70 insertions(+), 9 deletions(-) diff --git a/litellm/integrations/otel/logger.py b/litellm/integrations/otel/logger.py index 4359b222d06..1e078fe43f7 100644 --- a/litellm/integrations/otel/logger.py +++ b/litellm/integrations/otel/logger.py @@ -24,6 +24,7 @@ from litellm._logging import verbose_logger from litellm.integrations.custom_logger import CustomLogger from litellm.integrations.otel.emitter import SpanEmitter, stamp_error from litellm.integrations.otel.mappers import resolve_mappers +from litellm.integrations.otel.mappers.langfuse import LangfuseMapper from litellm.integrations.otel.model.baggage import promoted_baggage from litellm.integrations.otel.model.config import OpenTelemetryV2Config from litellm.integrations.otel.model.metadata import ( @@ -550,6 +551,7 @@ class OpenTelemetryV2(CustomLogger): # logging object) cannot re-emit through the deferred branch. self._emitter.mark_emitted(call_id, SpanRole.LLM_CALL) self._emitter.finish_span(SpanRole.LLM_CALL, carrier.span, data, end_time_ns=end_time_ns) + self._stamp_langfuse_root_observation(data, carrier.span) return carrier.span # Deferred: ``pre_call`` saw no recordable parent, so create the span now. # The worker copied the request task's context, which carries the anchored @@ -561,7 +563,7 @@ class OpenTelemetryV2(CustomLogger): parent_ctx: Final = self._seed_identity_baggage( data.identity, data.request_model, resolve_request_span_context() ) - return self._emitter.emit( + span: Final = self._emitter.emit( SpanRole.LLM_CALL, data, parent_context=(set_span_in_context(INVALID_SPAN, parent_ctx) if route.detached else parent_ctx), @@ -570,9 +572,22 @@ class OpenTelemetryV2(CustomLogger): tracer=route.tracer, links=_request_trace_links(parent_ctx) if route.detached else None, ) + self._stamp_langfuse_root_observation(data, span) + return span finally: self._tenant_tracers.release(route.provider) + def _stamp_langfuse_root_observation(self, data: LLMCallSpanData, llm_span: Span | None) -> None: + if "langfuse" not in self.config.mapper_names or llm_span is None: + return + root: Final = request_root_span() + if root is None or not root.is_recording(): + return + if root.get_span_context().trace_id != llm_span.get_span_context().trace_id: + return + for key, value in LangfuseMapper.root_observation_io(data).items(): + root.set_attribute(key, value) + # ====================================================================== # # Service hooks # ====================================================================== # diff --git a/litellm/integrations/otel/mappers/langfuse.py b/litellm/integrations/otel/mappers/langfuse.py index 6d4f1b4fd0a..69fd10bbc22 100644 --- a/litellm/integrations/otel/mappers/langfuse.py +++ b/litellm/integrations/otel/mappers/langfuse.py @@ -77,3 +77,11 @@ class LangfuseMapper: **collect(cls._LLM_CALL_ATTRS, data), **collect(cls._BLOB_ATTRS, data), } + + @classmethod + def root_observation_io(cls, data: LLMCallSpanData) -> AttributeMap: + return { + key: value + for key, value in collect(cls._BLOB_ATTRS, data).items() + if key in ("langfuse.observation.input", "langfuse.observation.output") + } diff --git a/tests/test_litellm/integrations/otel/test_otel_v2_logger.py b/tests/test_litellm/integrations/otel/test_otel_v2_logger.py index 455d84c764f..e5e7f00587b 100644 --- a/tests/test_litellm/integrations/otel/test_otel_v2_logger.py +++ b/tests/test_litellm/integrations/otel/test_otel_v2_logger.py @@ -9,8 +9,8 @@ hooks, proxy SERVER span lifecycle (start + setters), parent-context resolution import asyncio import contextlib import os -from unittest.mock import patch from datetime import datetime, timedelta, timezone +from unittest.mock import patch import pytest @@ -28,6 +28,16 @@ from litellm.integrations.otel import ( # noqa: E402 LiteLLM, OpenTelemetryV2Config, ) +from litellm.integrations.otel.logger import OpenTelemetryV2 # noqa: E402 +from litellm.integrations.otel.model.config import ( # noqa: E402 + CaptureMessageContent, + ExporterSpec, +) +from litellm.integrations.otel.model.spans import ( # noqa: E402 + LITELLM_PROXY_REQUEST_SPAN_NAME, + SpanRole, +) +from litellm.integrations.otel.model.utils import to_ns, to_seconds # noqa: E402 from litellm.integrations.otel.plumbing import providers # noqa: E402 from litellm.integrations.otel.plumbing.context import ( # noqa: E402 reset_mcp_message_trace_carrier, @@ -36,13 +46,6 @@ from litellm.integrations.otel.plumbing.context import ( # noqa: E402 set_mcp_message_transport_span, set_request_root_span, ) -from litellm.integrations.otel.logger import OpenTelemetryV2 # noqa: E402 -from litellm.integrations.otel.model.config import ExporterSpec # noqa: E402 -from litellm.integrations.otel.model.spans import ( # noqa: E402 - LITELLM_PROXY_REQUEST_SPAN_NAME, - SpanRole, -) -from litellm.integrations.otel.model.utils import to_ns, to_seconds # noqa: E402 # --------------------------------------------------------------------------- # # Fixtures @@ -959,6 +962,41 @@ def test_llm_span_parents_to_ambient_server_span(): assert llm_span.parent.span_id == server.get_span_context().span_id +def test_langfuse_mapper_stamps_input_output_on_root_observation(): + cfg = OpenTelemetryV2Config( + exporter="in_memory", + legacy_compat=False, + mapper_names=["langfuse"], + capture_message_content=CaptureMessageContent.SPAN_ONLY, + ) + exporter = InMemorySpanExporter() + tracer_provider = providers.build_tracer_provider(cfg, exporter=exporter) + logger = OpenTelemetryV2(config=cfg, tracer_provider=tracer_provider) + server = logger._emitter.start_span(SpanRole.PROXY_REQUEST, LITELLM_PROXY_REQUEST_SPAN_NAME) + set_request_root_span(server) + payload = _payload( + messages=[{"role": "user", "content": "Say hi"}], + response={ + "id": "resp_1", + "model": "gpt-4o", + "choices": [ + { + "finish_reason": "stop", + "message": {"role": "assistant", "content": "Hi"}, + } + ], + }, + ) + + _emit_llm(logger, _kwargs(payload=payload)) + server.end() + + by_name = {span.name: span for span in exporter.get_finished_spans()} + root_attrs = by_name[LITELLM_PROXY_REQUEST_SPAN_NAME].attributes + assert root_attrs["langfuse.observation.input"] == ('[{"role": "user", "content": "Say hi"}]') + assert root_attrs["langfuse.observation.output"] == ('[{"role": "assistant", "content": "Hi"}]') + + def test_llm_span_is_root_without_ambient_server_span(): """No server span at ``pre_call`` → creation is deferred and the span is a root of its own trace (the SDK / no-proxy path)."""