From 84f2f28f9e09de22500193bb4c2a84ba0e66ebb8 Mon Sep 17 00:00:00 2001 From: yucheng-berri <293926326+yucheng-berri@users.noreply.github.com> Date: Sat, 17 Oct 2026 01:52:29 -0700 Subject: [PATCH] fix(langfuse_otel): name the OTel v2 request root once, from the request's own call The v2 Langfuse logger re-derived langfuse.trace.name on the request root for every LLM call in the request. A post_call llm_as_a_judge guardrail's judge call replaced the caller's trace_name on the root with litellm-acompletion, and with another v2 callback (otel, arize_phoenix) listed before langfuse_otel that vendor's root span gained langfuse.trace.name on every request. The derived name now fills only an unnamed root this logger's provider started, and never comes from a call litellm makes for its own features. A caller trace_name still names the root on every call, as on main. --- litellm/integrations/otel/langfuse_logger.py | 20 ++- .../test_langfuse_otel_trace_fields.py | 116 +++++++++++++++++- .../integrations/otel/test_langfuse_logger.py | 72 +++++++++++ 3 files changed, 203 insertions(+), 5 deletions(-) diff --git a/litellm/integrations/otel/langfuse_logger.py b/litellm/integrations/otel/langfuse_logger.py index ce3329d55dc..5ceca672590 100644 --- a/litellm/integrations/otel/langfuse_logger.py +++ b/litellm/integrations/otel/langfuse_logger.py @@ -1,19 +1,25 @@ from collections.abc import AsyncGenerator, AsyncIterator, Callable, Mapping from typing import TYPE_CHECKING, Final +from opentelemetry.sdk.trace import ReadableSpan + from litellm._logging import verbose_logger 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 internal_call_origin 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.model.utils import as_str from litellm.integrations.otel.plumbing.context import request_root_span if TYPE_CHECKING: + from opentelemetry.trace import Span + from litellm.types.proxy.auth.user_api_key_auth import UserAPIKeyAuth from litellm.types.utils import ModelResponseStream @@ -28,11 +34,23 @@ class LangfuseOpenTelemetryV2(OpenTelemetryV2): root.set_attributes( LangfuseMapper.trace_attributes( caller_trace_controls(kwargs), - as_str(kwargs.get("call_type")) if getattr(root, "parent", None) is None else None, + as_str(kwargs.get("call_type")) if self._derives_trace_name(root, kwargs) else None, ) ) super().log_pre_api_call(model, messages, kwargs) + def _derives_trace_name(self, root: "Span", kwargs: Mapping[str, object]) -> bool: + """Whether this call may name the request root after its call type: the call is not one litellm made for + its own features, this logger's provider started the root with no parent, and neither the caller nor an + earlier call in the request has named it.""" + return ( + internal_call_origin(None, kwargs) is None + and isinstance(root, ReadableSpan) + and root.parent is None + and root.resource is self.tracer_provider.resource + and LANGFUSE_TRACE_NAME not in (root.attributes or {}) + ) + class LangfuseContentOpenTelemetryV2(LangfuseOpenTelemetryV2): """Stamps the request's input and output on the root observation while it is still recording. diff --git a/tests/integration/observability/test_langfuse_otel_trace_fields.py b/tests/integration/observability/test_langfuse_otel_trace_fields.py index d415b1caaea..e2c58164c7d 100644 --- a/tests/integration/observability/test_langfuse_otel_trace_fields.py +++ b/tests/integration/observability/test_langfuse_otel_trace_fields.py @@ -2,7 +2,7 @@ import json import threading import uuid from itertools import product -from collections.abc import Iterator, Mapping +from collections.abc import Iterator, Mapping, Sequence from dataclasses import dataclass from pathlib import Path from typing import Final @@ -28,6 +28,8 @@ SURFACE_CALL_TYPES: Final = { "responses_stream": "aresponses", } ANTHROPIC_MODEL: Final = "anthropic/claude-sonnet-4-5-20250929" +JUDGE_MODEL: Final = "lit8281-judge-model" +JUDGE_GUARDRAIL: Final = "lit8281-judge" @dataclass(frozen=True, slots=True) @@ -123,6 +125,32 @@ class LangfuseSink: return eventually(roots, lambda found: len(found) == 1, seconds=70)[0] +class OtlpSink: + """A generic OTLP endpoint: keeps every span it receives.""" + + def __init__(self) -> None: + self.lock: Final = threading.Lock() + self.spans: Final[list[SpanRecord]] = [] # mutable-ok: spans accumulate as the exporter delivers them + + def respond(self, request: Request) -> Reply: + if request.target.endswith("/v1/traces"): + records: Final = _span_records(request.body) + with self.lock: + self.spans.extend(records) + return Reply(body=b"") + + def root_of_request(self, marker: str) -> SpanRecord: + """The root span of the one trace whose spans carry the marker.""" + + def roots() -> tuple[SpanRecord, ...]: + with self.lock: + snapshot: Final = tuple(self.spans) + traces: Final = {record.trace_id for record in snapshot if marker in str(tuple(record.attributes.values()))} + return tuple(record for record in snapshot if record.trace_id in traces and not record.parent_span_id) + + return eventually(roots, lambda found: len(found) == 1, seconds=70)[0] + + def _chat_completion(marker: str, text: str = "langfuse-otel scripted reply") -> bytes: return json.dumps( { @@ -152,6 +180,8 @@ def _provider(request: Request) -> Reply: } ).encode() ) + if body.get("response_format") == {"type": "json_object"}: + return Reply(body=_chat_completion("lit8281-judge-verdict", json.dumps({"overall_score": 100}))) messages: Final = body.get("messages") first: Final = messages[0]["content"] if isinstance(messages, list) and messages else None if isinstance(first, str) and first.startswith("fail-"): @@ -159,10 +189,33 @@ def _provider(request: Request) -> Reply: return surface_reply(request) -def _langfuse_config(directory: Path, name: str, settings: Mapping[str, object]) -> Path: +def _langfuse_config(directory: Path, name: str, settings: Mapping[str, object], provider_url: str) -> Path: + """The integration config with `langfuse_otel`, plus an opt-in post_call `llm_as_a_judge` guardrail whose judge + model is served by the scripted provider.""" base: Final = yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text()) + judge: Final = { + "model_name": JUDGE_MODEL, + "litellm_params": { + "model": "openai/gpt-4o-mini", + "api_base": provider_url + "/v1", + "api_key": "synthetic-provider-key", + }, + } config: Final = { **base, + "model_list": [*base["model_list"], judge], + "guardrails": [ + { + "guardrail_name": JUDGE_GUARDRAIL, + "litellm_params": { + "guardrail": "llm_as_a_judge", + "mode": "post_call", + "judge_model": JUDGE_MODEL, + "on_failure": "log", + "criteria": [{"name": "helpful", "description": "The reply answers the user", "weight": 100}], + }, + } + ], "litellm_settings": {**base["litellm_settings"], "callbacks": ["langfuse_otel"], **settings}, } path: Final = directory / f"{name}.yaml" @@ -198,7 +251,7 @@ def _rig( "LANGFUSE_HOST": destination.url, **environment, }, - config=_langfuse_config(directory, name, settings), + config=_langfuse_config(directory, name, settings, provider.url), workers=2, ) as proxy, proxy.scenario() as scenario, @@ -237,8 +290,35 @@ def v2_llm_only(tmp_path_factory: pytest.TempPathFactory) -> Iterator[Rig]: ) +@dataclass(frozen=True, slots=True) +class SharedRig: + rig: Rig + otel: OtlpSink + + +@pytest.fixture(scope="module") +def v2_after_otel(tmp_path_factory: pytest.TempPathFactory) -> Iterator[SharedRig]: + """OTel v2 with the generic `otel` callback listed before `langfuse_otel`, so the request root belongs to the + generic exporter.""" + otel: Final = OtlpSink() + with wire_server(otel.respond) as destination: + environment: Final = { + **V2_ENVIRONMENT, + "OTEL_EXPORTER": "otlp_http", + "OTEL_ENDPOINT": destination.url + "/v1/traces", + } + for rig in _rig( + tmp_path_factory, "langfuse-v2-after-otel", environment, {"callbacks": ["otel", "langfuse_otel"]} + ): + yield SharedRig(rig, otel) + + def _chat( - rig: Rig, content: str, metadata: Mapping[str, object] | None = None, headers: Mapping[str, str] | None = None + rig: Rig, + content: str, + metadata: Mapping[str, object] | None = None, + headers: Mapping[str, str] | None = None, + guardrails: Sequence[str] = (), ) -> str: response: Final = rig.proxy.request( "POST", @@ -247,6 +327,7 @@ def _chat( "model": rig.openai_model, "messages": [{"role": "user", "content": content}], **({"metadata": dict(metadata)} if metadata is not None else {}), + **({"guardrails": list(guardrails)} if guardrails else {}), }, key=rig.key, headers=headers, @@ -478,6 +559,33 @@ def test_otel_v2_inbound_traceparent_leaves_the_trace_name_to_the_caller(v2: Rig assert not set(_trace_fields(spans)), spans +@pytest.mark.parametrize( + ("metadata", "headers"), + [({"trace_name": "caller-trace"}, None), (None, {"langfuse_trace_name": "caller-trace"})], + ids=["body", "header"], +) +def test_otel_v2_post_call_judge_keeps_the_caller_trace_name_on_the_root( + v2: Rig, metadata: Mapping[str, object] | None, headers: Mapping[str, str] | None +) -> None: + marker: Final = "lit8281-v2-judge-" + uuid.uuid4().hex + _chat(v2, marker, metadata=metadata, headers=headers, guardrails=[JUDGE_GUARDRAIL]) + generations: Final = eventually(lambda: v2.sink.matching(marker), lambda found: len(found) == 2, seconds=70) + assert len({generation.trace_id for generation in generations}) == 1, generations + root: Final = v2.sink.root_of(generations[0]) + assert root.attributes.get("langfuse.trace.name") == "caller-trace", root.attributes + + +def test_otel_v2_generic_otel_root_gets_no_derived_langfuse_trace_name(v2_after_otel: SharedRig) -> None: + derived: Final = "lit8281-v2-after-otel-" + uuid.uuid4().hex + caller: Final = "lit8281-v2-after-otel-caller-" + uuid.uuid4().hex + _chat(v2_after_otel.rig, derived) + _chat(v2_after_otel.rig, caller, metadata={"trace_name": "caller-trace"}) + root: Final = v2_after_otel.otel.root_of_request(derived) + assert "langfuse.trace.name" not in root.attributes, root.attributes + caller_root: Final = v2_after_otel.otel.root_of_request(caller) + assert caller_root.attributes.get("langfuse.trace.name") == "caller-trace", caller_root.attributes + + def test_otel_v2_llm_only_root_generation_keeps_its_span_name_as_the_trace_name(v2_llm_only: Rig) -> None: marker: Final = "lit8281-v2-llm-only-" + uuid.uuid4().hex _chat(v2_llm_only, marker) diff --git a/tests/unit/integrations/otel/test_langfuse_logger.py b/tests/unit/integrations/otel/test_langfuse_logger.py index e4c4f5e6406..ac51dfdbe93 100644 --- a/tests/unit/integrations/otel/test_langfuse_logger.py +++ b/tests/unit/integrations/otel/test_langfuse_logger.py @@ -437,6 +437,78 @@ def test_named_request_under_an_inbound_traceparent_keeps_the_caller_trace_name( assert generation_attrs[TRACE_NAME_ATTR] == "from-body" +UNNAMED: Final = {"proxy_server_request": {"headers": {}}} +JUDGE_PARAMS: Final = {"metadata": {"internal_call_origin": "llm_as_a_judge_guardrail"}} +CALLER_NAMES: Final = ( + {"metadata": {"trace_name": "from-caller"}, "proxy_server_request": {"headers": {}}}, + {"proxy_server_request": {"headers": {"langfuse_trace_name": "from-caller"}}}, +) + + +def _root_trace_name_after( + logger: OpenTelemetryV2, + calls: Sequence[tuple[str, Mapping[str, object]]], + root_logger: OpenTelemetryV2 | None = None, +) -> object: + root: Final = _start_root(root_logger or logger) + for index, (call_type, litellm_params) in enumerate(calls): + logger.log_pre_api_call( + model="gpt-5.4-mini", + messages=[], + kwargs={"litellm_call_id": f"call_{index}", "litellm_params": litellm_params, "call_type": call_type}, + ) + root.end() + return dict(root.attributes or {}).get(TRACE_NAME_ATTR) + + +@pytest.mark.parametrize("later", [JUDGE_PARAMS, UNNAMED], ids=["litellm-sub-call", "untagged-call"]) +@pytest.mark.parametrize("caller", CALLER_NAMES, ids=["body", "header"]) +def test_a_later_call_in_the_request_keeps_the_caller_trace_name_on_the_root(caller, later): + logger, _ = _logger() + + assert _root_trace_name_after(logger, (("acompletion", caller), ("acompletion", later))) == "from-caller" + + +@pytest.mark.parametrize("caller", CALLER_NAMES, ids=["body", "header"]) +def test_a_caller_trace_name_replaces_the_name_an_earlier_call_derived(caller): + logger, _ = _logger() + + assert _root_trace_name_after(logger, (("acompletion", JUDGE_PARAMS), ("acompletion", caller))) == "from-caller" + + +def test_a_later_call_keeps_the_trace_name_the_first_call_derived(): + logger, _ = _logger() + + assert ( + _root_trace_name_after(logger, (("anthropic_messages", UNNAMED), ("acompletion", UNNAMED))) + == "litellm-anthropic_messages" + ) + + +def test_a_litellm_sub_call_before_the_request_call_leaves_the_trace_name_to_the_request_call(): + logger, _ = _logger() + + assert ( + _root_trace_name_after(logger, (("acompletion", JUDGE_PARAMS), ("anthropic_messages", UNNAMED))) + == "litellm-anthropic_messages" + ) + + +def test_a_root_another_vendor_started_gets_no_derived_trace_name(): + logger, _ = _logger() + other_vendor, _ = _logger(mappers=("genai",)) + + assert _root_trace_name_after(logger, (("acompletion", UNNAMED),), root_logger=other_vendor) is None + + +@pytest.mark.parametrize("caller", CALLER_NAMES, ids=["body", "header"]) +def test_a_root_another_vendor_started_still_gets_the_caller_trace_name(caller): + logger, _ = _logger() + other_vendor, _ = _logger(mappers=("genai",)) + + assert _root_trace_name_after(logger, (("acompletion", caller),), root_logger=other_vendor) == "from-caller" + + @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)