diff --git a/litellm/integrations/langfuse/langfuse_otel.py b/litellm/integrations/langfuse/langfuse_otel.py index 3558f6c6426..b0456817af5 100644 --- a/litellm/integrations/langfuse/langfuse_otel.py +++ b/litellm/integrations/langfuse/langfuse_otel.py @@ -6,6 +6,8 @@ from datetime import datetime from itertools import chain from typing import TYPE_CHECKING, Any, Final, Optional, Protocol, runtime_checkable +from opentelemetry import trace + import litellm from litellm._logging import verbose_logger from litellm.constants import REDACTED_BY_LITELLM @@ -13,7 +15,7 @@ from litellm.integrations.arize import _utils from litellm.integrations.langfuse.langfuse_otel_attributes import ( LangfuseLLMObsOTELAttributes, ) -from litellm.integrations.opentelemetry import OpenTelemetry, OpenTelemetryConfig +from litellm.integrations.opentelemetry import LITELLM_PROXY_REQUEST_SPAN_NAME, OpenTelemetry, OpenTelemetryConfig from litellm.litellm_core_utils.safe_json_dumps import safe_dumps from litellm.litellm_core_utils.safe_json_loads import safe_json_loads from litellm.secret_managers.main import str_to_bool @@ -73,6 +75,48 @@ def _caller_tag_list(value: object) -> tuple[str, ...]: return () +def _parent_of(span: object) -> trace.SpanContext | None: + parent: Final = getattr(span, "parent", None) + return parent if isinstance(parent, trace.SpanContext) else None + + +def _litellm_proxy_span(kwargs: Mapping[str, object]) -> trace.Span | None: + """The proxy request span another OTel callback opened for this request, when the request carries one.""" + litellm_params: Final = _as_gettable(kwargs.get("litellm_params")) or {} + carried: Final = ( + (_as_gettable(litellm_params.get(key)) or {}).get("litellm_parent_otel_span") + for key in ("metadata", "litellm_metadata") + ) + return next( + ( + span + for span in carried + if isinstance(span, trace.Span) and getattr(span, "name", None) == LITELLM_PROXY_REQUEST_SPAN_NAME + ), + None, + ) + + +def _starts_its_own_trace(span: object, kwargs: Mapping[str, object]) -> bool: + """Whether the trace `span` belongs to was started by litellm rather than by the app calling it. + + litellm writes onto the app's own span when one is active, and otherwise parents its span to an inbound + traceparent, a caller-provided span or its proxy request span. The trace is litellm's only when its span has no + parent, or hangs off a proxy request span that has none. + """ + if span is trace.get_current_span(): + return False + parent: Final = _parent_of(span) + if parent is None: + return True + proxy_span: Final = _litellm_proxy_span(kwargs) + return ( + proxy_span is not None + and _parent_of(proxy_span) is None + and proxy_span.get_span_context().span_id == parent.span_id + ) + + class LangfuseOtelLogger(OpenTelemetry): def __init__(self, config=None, *args, **kwargs): # Prevent LangfuseOtelLogger from modifying global environment variables by constructing config manually @@ -235,7 +279,9 @@ class LangfuseOtelLogger(OpenTelemetry): metadata: Final = LangfuseOtelLogger._extract_langfuse_metadata(kwargs) LangfuseOtelLogger._set_metadata_attributes(span=span, metadata=metadata) - joins_existing_trace: Final = metadata.get("existing_trace_id") is not None + joins_existing_trace: Final = metadata.get("existing_trace_id") is not None or not _starts_its_own_trace( + span, kwargs + ) if metadata.get("trace_name") is None and not joins_existing_trace: safe_set_attribute( span, diff --git a/litellm/integrations/otel/langfuse_logger.py b/litellm/integrations/otel/langfuse_logger.py index 87796f2b7bc..ce3329d55dc 100644 --- a/litellm/integrations/otel/langfuse_logger.py +++ b/litellm/integrations/otel/langfuse_logger.py @@ -26,7 +26,10 @@ class LangfuseOpenTelemetryV2(OpenTelemetryV2): root: Final = request_root_span() if root is not None and root.is_recording(): root.set_attributes( - LangfuseMapper.trace_attributes(caller_trace_controls(kwargs), as_str(kwargs.get("call_type"))) + LangfuseMapper.trace_attributes( + caller_trace_controls(kwargs), + as_str(kwargs.get("call_type")) if getattr(root, "parent", None) is None else None, + ) ) super().log_pre_api_call(model, messages, kwargs) diff --git a/litellm/integrations/otel/mappers/langfuse.py b/litellm/integrations/otel/mappers/langfuse.py index f34b6500acd..5dc5f180c22 100644 --- a/litellm/integrations/otel/mappers/langfuse.py +++ b/litellm/integrations/otel/mappers/langfuse.py @@ -103,6 +103,6 @@ class LangfuseMapper: def _llm_call(cls, data: LLMCallSpanData) -> AttributeMap: return { **collect(cls._LLM_CALL_ATTRS, data), - **cls.trace_attributes(data.trace, data.call_type), + **cls.trace_attributes(data.trace), **collect(cls._BLOB_ATTRS, data), } diff --git a/tests/integration/observability/test_langfuse_otel_trace_fields.py b/tests/integration/observability/test_langfuse_otel_trace_fields.py index 365798850b2..d415b1caaea 100644 --- a/tests/integration/observability/test_langfuse_otel_trace_fields.py +++ b/tests/integration/observability/test_langfuse_otel_trace_fields.py @@ -1,6 +1,7 @@ import json import threading import uuid +from itertools import product from collections.abc import Iterator, Mapping from dataclasses import dataclass from pathlib import Path @@ -14,8 +15,10 @@ from integration._support.process import owned_proxy from integration._support.wire import Reply, Request, wire_server from opentelemetry.proto.collector.trace.v1.trace_service_pb2 import ExportTraceServiceRequest from opentelemetry.proto.common.v1.common_pb2 import AnyValue +from opentelemetry.proto.trace.v1.trace_pb2 import Span REDACTED: Final = "redacted-by-litellm" +TRACE_FIELDS: Final = ("langfuse.trace.name", "langfuse.trace.input", "langfuse.trace.output", "langfuse.trace.tags") SURFACE_CALL_TYPES: Final = { "chat": "acompletion", "chat_stream": "acompletion", @@ -29,20 +32,21 @@ ANTHROPIC_MODEL: Final = "anthropic/claude-sonnet-4-5-20250929" @dataclass(frozen=True, slots=True) class SpanRecord: + name: str trace_id: str parent_span_id: str attributes: Mapping[str, object] - def tags(self) -> list[object]: + def tags(self) -> tuple[object, ...]: """The trace tags Langfuse ingestion reads from the raw attribute.""" raw: Final = self.attributes.get("langfuse.trace.tags") if isinstance(raw, tuple): - return list(raw) + return raw if not isinstance(raw, str) or not raw: - return [] + return () if raw.strip().startswith("["): - return json.loads(raw) - return [tag.strip() for tag in raw.split(",")] if "," in raw else [raw] + return tuple(json.loads(raw)) + return tuple(tag.strip() for tag in raw.split(",")) if "," in raw else (raw,) def _attribute_value(value: AnyValue) -> object: @@ -52,17 +56,21 @@ def _attribute_value(value: AnyValue) -> object: return getattr(value, kind) if kind is not None else None +def _exported_spans(body: bytes) -> Iterator[Span]: + for resource in ExportTraceServiceRequest.FromString(body).resource_spans: + for scope in resource.scope_spans: + yield from scope.spans + + def _span_records(body: bytes) -> tuple[SpanRecord, ...]: - export: Final = ExportTraceServiceRequest.FromString(body) return tuple( SpanRecord( + name=span.name, trace_id=span.trace_id.hex(), parent_span_id=span.parent_span_id.hex(), attributes={pair.key: _attribute_value(pair.value) for pair in span.attributes}, ) - for resource in export.resource_spans - for scope in resource.scope_spans - for span in scope.spans + for span in _exported_spans(body) ) @@ -75,6 +83,7 @@ class LangfuseSink: self.outage: Final = threading.Event() self.accepted: Final[list[SpanRecord]] = [] # mutable-ok: spans accumulate as the exporter delivers them self.roots: Final[list[SpanRecord]] = [] # mutable-ok: root spans accumulate as the exporter delivers them + self.spans: Final[list[SpanRecord]] = [] # mutable-ok: every span accumulates as the exporter delivers it self.refusals: Final[list[int]] = [] # mutable-ok: one entry per batch refused during the outage def respond(self, request: Request) -> Reply: @@ -91,6 +100,7 @@ class LangfuseSink: record for record in records if record.attributes.get("langfuse.observation.type") == "generation" ) self.roots.extend(record for record in records if not record.parent_span_id) + self.spans.extend(records) return Reply(body=b"") def matching(self, marker: str) -> tuple[SpanRecord, ...]: @@ -101,6 +111,10 @@ class LangfuseSink: def one(self, marker: str) -> SpanRecord: return eventually(lambda: self.matching(marker), lambda found: len(found) == 1, seconds=70)[0] + def in_trace(self, trace_id: str) -> tuple[SpanRecord, ...]: + with self.lock: + return tuple(record for record in self.spans if record.trace_id == trace_id) + def root_of(self, span: SpanRecord) -> SpanRecord: def roots() -> tuple[SpanRecord, ...]: with self.lock: @@ -241,6 +255,12 @@ def _chat( return response.text +def _trace_fields(spans: tuple[SpanRecord, ...]) -> Iterator[str]: + """Every trace-level key any of the spans carries.""" + for record in spans: + yield from (key for key in record.attributes if key in TRACE_FIELDS) + + def _assert_derived(span: SpanRecord, call_type: str) -> None: attributes: Final = span.attributes assert attributes.get("langfuse.trace.name") == f"litellm-{call_type}", attributes @@ -262,14 +282,32 @@ def test_every_surface_derives_trace_name_io_and_tags(v1: Rig, surface: str) -> assert any(str(tag).startswith("cache_hit:") for tag in span.tags()), span.tags() -def test_inbound_traceparent_keeps_the_caller_trace_and_derives_trace_fields(v1: Rig) -> None: +def test_inbound_traceparent_joins_the_caller_trace_without_derived_trace_fields(v1: Rig) -> None: marker: Final = "lit8281-parent-" + uuid.uuid4().hex trace_id: Final = uuid.uuid4().hex parent_id: Final = uuid.uuid4().hex[:16] _chat(v1, marker, headers={"traceparent": f"00-{trace_id}-{parent_id}-01"}) span: Final = v1.sink.one(marker) assert (span.trace_id, span.parent_span_id) == (trace_id, parent_id), span - _assert_derived(span, "acompletion") + assert not set(TRACE_FIELDS) & set(span.attributes), span.attributes + assert marker in str(span.attributes.get("langfuse.observation.input")), span.attributes + assert span.attributes.get("langfuse.observation.output") is not None, span.attributes + + +def test_inbound_traceparent_still_carries_the_caller_trace_name_and_tags(v1: Rig) -> None: + marker: Final = "lit8281-parent-caller-" + uuid.uuid4().hex + trace_id: Final = uuid.uuid4().hex + _chat( + v1, + marker, + metadata={"trace_name": "caller-trace", "tags": ["caller-tag"]}, + headers={"traceparent": f"00-{trace_id}-{uuid.uuid4().hex[:16]}-01"}, + ) + span: Final = v1.sink.one(marker) + assert span.trace_id == trace_id, span + assert span.attributes.get("langfuse.trace.name") == "caller-trace", span.attributes + assert span.tags() == ("caller-tag",), span.attributes + assert not {"langfuse.trace.input", "langfuse.trace.output"} & set(span.attributes), span.attributes def test_caller_trace_name_and_tags_win_over_derived_values(v1: Rig) -> None: @@ -277,20 +315,20 @@ def test_caller_trace_name_and_tags_win_over_derived_values(v1: Rig) -> None: _chat(v1, marker, metadata={"trace_name": "caller-trace", "tags": ["caller-tag", "second-tag"]}) span: Final = v1.sink.one(marker) assert span.attributes["langfuse.trace.name"] == "caller-trace", span.attributes - assert span.tags()[:2] == ["caller-tag", "second-tag"], span.tags() + assert span.tags()[:2] == ("caller-tag", "second-tag"), span.tags() assert span.attributes.get("langfuse.trace.input") == span.attributes.get("langfuse.observation.input") @pytest.mark.parametrize( ("metadata", "headers", "expected"), [ - pytest.param({"tags": "c1, c2"}, None, ["c1", "c2"], id="body_comma_string"), - pytest.param({"tags": '["j1", "j2"]'}, None, ["j1", "j2"], id="body_json_array_string"), - pytest.param(None, {"langfuse_tags": "h1,h2"}, ["h1", "h2"], id="header_comma_string"), + pytest.param({"tags": "c1, c2"}, None, ("c1", "c2"), id="body_comma_string"), + pytest.param({"tags": '["j1", "j2"]'}, None, ("j1", "j2"), id="body_json_array_string"), + pytest.param(None, {"langfuse_tags": "h1,h2"}, ("h1", "h2"), id="header_comma_string"), ], ) def test_string_caller_tags_keep_the_tags_langfuse_split_them_into( - v1: Rig, metadata: Mapping[str, object] | None, headers: Mapping[str, str] | None, expected: list[str] + v1: Rig, metadata: Mapping[str, object] | None, headers: Mapping[str, str] | None, expected: tuple[str, ...] ) -> None: marker: Final = "lit8281-strtags-" + uuid.uuid4().hex _chat(v1, marker, metadata=metadata, headers=headers) @@ -312,7 +350,7 @@ def test_joined_trace_keeps_caller_tags_and_gets_no_derived_trace_fields(v1: Rig span: Final = v1.sink.one(marker) attributes: Final = span.attributes assert attributes.get("langfuse.trace.existing_id") == existing, attributes - assert span.tags() == ["join-tag"], attributes + assert span.tags() == ("join-tag",), attributes assert "langfuse.trace.name" not in attributes, attributes assert "langfuse.trace.input" not in attributes, attributes assert "langfuse.trace.output" not in attributes, attributes @@ -410,29 +448,42 @@ def test_identical_requests_each_log_exactly_one_span_with_derived_fields(v1: Ri _assert_derived(span, "acompletion") -def test_otel_v2_derives_the_trace_name_and_leaves_trace_io_to_the_root_observation(v2: Rig) -> None: +def test_otel_v2_names_the_request_root_and_leaves_trace_io_to_the_root_observation(v2: Rig) -> None: derived: Final = "lit8281-v2-" + uuid.uuid4().hex caller: Final = "lit8281-v2-caller-" + uuid.uuid4().hex _chat(v2, derived) _chat(v2, caller, metadata={"trace_name": "caller-trace", "tags": ["caller-tag"]}) derived_span: Final = v2.sink.one(derived) - assert derived_span.attributes.get("langfuse.trace.name") == "litellm-acompletion", derived_span.attributes - assert not {"langfuse.trace.input", "langfuse.trace.output"} & set(derived_span.attributes), derived_span.attributes + assert not set(_trace_fields((derived_span,))), derived_span.attributes root: Final = v2.sink.root_of(derived_span) assert root.attributes.get("langfuse.trace.name") == "litellm-acompletion", root.attributes assert derived in str(root.attributes.get("langfuse.observation.input")), root.attributes assert root.attributes.get("langfuse.observation.output") is not None, root.attributes caller_span: Final = v2.sink.one(caller) assert caller_span.attributes.get("langfuse.trace.name") == "caller-trace", caller_span.attributes - assert caller_span.tags() == ["caller-tag"], caller_span.attributes + assert caller_span.tags() == ("caller-tag",), caller_span.attributes -def test_otel_v2_llm_only_root_generation_is_named_after_the_call_type(v2_llm_only: Rig) -> None: +def test_otel_v2_inbound_traceparent_leaves_the_trace_name_to_the_caller(v2: Rig) -> None: + marker: Final = "lit8281-v2-parent-" + uuid.uuid4().hex + trace_id: Final = uuid.uuid4().hex + _chat(v2, marker, headers={"traceparent": f"00-{trace_id}-{uuid.uuid4().hex[:16]}-01"}) + span: Final = v2.sink.one(marker) + assert span.trace_id == trace_id, span + spans: Final = eventually( + lambda: v2.sink.in_trace(trace_id), + lambda found: any(record.parent_span_id != span.parent_span_id for record in found), + seconds=70, + ) + assert not set(_trace_fields(spans)), spans + + +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) span: Final = v2_llm_only.sink.one(marker) assert span.parent_span_id == "", span - assert span.attributes.get("langfuse.trace.name") == "litellm-acompletion", span.attributes + assert span.attributes.get("langfuse.trace.name") == span.name, span assert marker in str(span.attributes.get("langfuse.observation.input")), span.attributes @@ -445,9 +496,9 @@ def test_destination_outage_mid_burst_lands_every_span_once_with_derived_fields( finally: v1.sink.outage.clear() assert len(answered) == 4 * len(SURFACES) - expected: Final = {f"{marker}-{surface}-{index}": surface for surface in SURFACES for index in range(4)} + expected: Final = {f"{marker}-{surface}-{index}": surface for surface, index in product(SURFACES, range(4))} - def landed() -> dict[str, tuple[SpanRecord, ...]]: + def landed() -> Mapping[str, tuple[SpanRecord, ...]]: return {identity: v1.sink.matching(identity) for identity in expected} spans: Final = eventually(landed, lambda found: all(len(records) >= 1 for records in found.values()), seconds=120) diff --git a/tests/integration/sdk/test_langfuse_otel_trace_fields_sdk.py b/tests/integration/sdk/test_langfuse_otel_trace_fields_sdk.py index 0a07e7b9bd5..ddb790ef6f5 100644 --- a/tests/integration/sdk/test_langfuse_otel_trace_fields_sdk.py +++ b/tests/integration/sdk/test_langfuse_otel_trace_fields_sdk.py @@ -4,12 +4,15 @@ import subprocess import sys import threading import uuid -from collections.abc import Mapping +from collections.abc import Iterator, Mapping +from dataclasses import dataclass from typing import Final +import pytest from integration._support.client import eventually from integration._support.wire import Reply, Request, wire_server from opentelemetry.proto.collector.trace.v1.trace_service_pb2 import ExportTraceServiceRequest +from opentelemetry.proto.trace.v1.trace_pb2 import Span CALLS: Final = { "sync": "completion", @@ -52,6 +55,165 @@ sys.stdin.read() """ +APP_TRACE_CALLS: Final = """ +import asyncio +import json +import sys +import threading +import time + +import litellm +from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.trace.export import SimpleSpanProcessor + +litellm.callbacks = ["langfuse_otel"] +api_base, marker, sink, writes_onto_app_span = sys.argv[1], sys.argv[2], sys.argv[3], sys.argv[4] == "1" +app = TracerProvider() +app.add_span_processor(SimpleSpanProcessor(OTLPSpanExporter(endpoint=sink + "/api/public/otel/v1/traces"))) +tracer = app.get_tracer("checkout-app") + + +def request(call, **extra): + return { + "model": "openai/gpt-4o-mini", + "api_base": api_base, + "api_key": "synthetic-provider-key", + "messages": [{"role": "user", "content": f"{marker}-{call}"}], + **extra, + } + + +def name_trace(root, call): + root.set_attribute("app.call", f"{marker}-{call}") + root.set_attribute("langfuse.trace.name", f"checkout-{marker}-{call}") + root.set_attribute("langfuse.trace.input", json.dumps({"cart": f"{marker}-{call}"})) + + +def close_trace(root, call): + root.set_attribute("langfuse.trace.output", json.dumps({"order": f"{marker}-{call}"})) + root.end() + + +def litellm_wrote_onto(root): + return not writes_onto_app_span or "langfuse.observation.type" in root.attributes + + +def ambient_sync(): + with tracer.start_as_current_span("checkout", end_on_exit=False) as root: + name_trace(root, "ambient_sync") + litellm.completion(**request("ambient_sync")) + deadline = time.monotonic() + 30 + while not litellm_wrote_onto(root) and time.monotonic() < deadline: + threading.Event().wait(0.05) + close_trace(root, "ambient_sync") + + +async def ambient_async(): + with tracer.start_as_current_span("checkout", end_on_exit=False) as root: + name_trace(root, "ambient_async") + await litellm.acompletion(**request("ambient_async")) + deadline = time.monotonic() + 30 + while not litellm_wrote_onto(root) and time.monotonic() < deadline: + await asyncio.sleep(0.05) + close_trace(root, "ambient_async") + + +def caller_span_sync(): + root = tracer.start_span("checkout") + name_trace(root, "caller_span_sync") + litellm.completion(**request("caller_span_sync", metadata={"litellm_parent_otel_span": root})) + close_trace(root, "caller_span_sync") + + +async def caller_span_async(): + root = tracer.start_span("checkout") + name_trace(root, "caller_span_async") + await litellm.acompletion(**request("caller_span_async", metadata={"litellm_parent_otel_span": root})) + close_trace(root, "caller_span_async") + + +async def async_calls(): + await ambient_async() + await caller_span_async() + + +ambient_sync() +caller_span_sync() +asyncio.run(async_calls()) +print("calls done", flush=True) +sys.stdin.read() +""" +APP_CALLS: Final = ("ambient_sync", "ambient_async", "caller_span_sync", "caller_span_async") + + +def _exported_spans(body: bytes) -> Iterator[Span]: + for resource in ExportTraceServiceRequest.FromString(body).resource_spans: + for scope in resource.scope_spans: + yield from scope.spans + + +def _attributes(span: Span) -> Mapping[str, str]: + return {pair.key: pair.value.string_value for pair in span.attributes} + + +@dataclass(frozen=True, slots=True) +class TraceSpan: + trace_id: str + parent_span_id: str + attributes: Mapping[str, str] + + +class TraceSink: + """The Langfuse OTLP endpoint for both the app's own exporter and litellm's: keeps every span it receives.""" + + def __init__(self) -> None: + self.lock: Final = threading.Lock() + self.spans: Final[list[TraceSpan]] = [] # mutable-ok: spans accumulate as both exporters deliver + + def respond(self, request: Request) -> Reply: + if not request.target.endswith("/v1/traces"): + return Reply(body=b"") + spans: Final = tuple( + TraceSpan(span.trace_id.hex(), span.parent_span_id.hex(), _attributes(span)) + for span in _exported_spans(request.body) + ) + with self.lock: + self.spans.extend(spans) + return Reply(body=b"") + + def traces(self, marker: str, calls: tuple[str, ...]) -> Mapping[str, tuple[TraceSpan, ...]]: + """Every span of each app trace, keyed by call, once the app root and a litellm generation both landed.""" + with self.lock: + snapshot: Final = tuple(self.spans) + roots: Final = { + call: next( + ( + span + for span in snapshot + if not span.parent_span_id and span.attributes.get("app.call") == f"{marker}-{call}" + ), + None, + ) + for call in calls + } + return { + call: tuple(span for span in snapshot if span.trace_id == root.trace_id) + for call, root in roots.items() + if root is not None + } + + +def _trace_values(spans: tuple[TraceSpan, ...], keys: tuple[str, ...]) -> Iterator[tuple[str, str]]: + """Every (key, value) pair of the given trace-level keys that any span of the trace carries.""" + for span in spans: + yield from ((key, span.attributes[key]) for key in keys if key in span.attributes) + + +def _logged(spans: tuple[TraceSpan, ...]) -> bool: + return any(span.attributes.get("langfuse.observation.type") == "generation" for span in spans) + + def _chat_reply(request: Request) -> Reply: body: Final = json.loads(request.body) content: Final = body["messages"][0]["content"] @@ -87,17 +249,12 @@ class GenerationSink: def respond(self, request: Request) -> Reply: if not request.target.endswith("/v1/traces"): return Reply(body=b"") - spans: Final = tuple( - {pair.key: pair.value.string_value for pair in span.attributes} - for resource in ExportTraceServiceRequest.FromString(request.body).resource_spans - for scope in resource.scope_spans - for span in scope.spans - ) + spans: Final = tuple(_attributes(span) for span in _exported_spans(request.body)) with self.lock: self.generations.extend(span for span in spans if span.get("langfuse.observation.type") == "generation") return Reply(body=b"") - def by_call(self, marker: str) -> dict[str, tuple[Mapping[str, str], ...]]: + def by_call(self, marker: str) -> Mapping[str, tuple[Mapping[str, str], ...]]: with self.lock: snapshot: Final = tuple(self.generations) return { @@ -138,3 +295,68 @@ def test_sdk_sync_and_async_calls_derive_trace_name_and_io_from_their_call_type( assert attributes["langfuse.trace.input"] == attributes["langfuse.observation.input"], (call, attributes) assert attributes["langfuse.trace.output"] == attributes["langfuse.observation.output"], (call, attributes) assert json.loads(attributes["langfuse.trace.output"]) == {"role": "assistant", "content": "ok"}, call + + +@pytest.mark.parametrize( + ("environment", "writes_onto_app_span", "calls"), + [ + pytest.param({}, True, APP_CALLS, id="litellm_writes_onto_the_app_span"), + pytest.param({"USE_OTEL_LITELLM_REQUEST_SPAN": "true"}, False, APP_CALLS, id="litellm_request_child_span"), + pytest.param( + {"LITELLM_OTEL_V2": "1", "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT": "span_only"}, + False, + ("ambient_async",), + id="otel_v2_llm_call_child_span", + ), + ], +) +def test_sdk_call_inside_an_app_trace_leaves_the_app_trace_name_and_io_alone( + environment: Mapping[str, str], writes_onto_app_span: bool, calls: tuple[str, ...] +) -> None: + marker: Final = "lit8281-app-" + uuid.uuid4().hex + sink: Final = TraceSink() + with wire_server(sink.respond) as destination, wire_server(_chat_reply) as provider: + child: Final = subprocess.Popen( + [ + sys.executable, + "-P", + "-c", + APP_TRACE_CALLS, + provider.url + "/v1", + marker, + destination.url, + "1" if writes_onto_app_span else "0", + ], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + env={ + **os.environ, + "LANGFUSE_PUBLIC_KEY": "pk-lf-sdk", + "LANGFUSE_SECRET_KEY": "sk-lf-sdk", + "LANGFUSE_HOST": destination.url, + "OTEL_BSP_SCHEDULE_DELAY": "200", + **environment, + }, + ) + try: + traces: Final = eventually( + lambda: sink.traces(marker, calls), + lambda found: ( + (len(found) == len(calls) and all(map(_logged, found.values()))) or child.poll() is not None + ), + seconds=70, + ) + finally: + output: Final = child.communicate(input=b"", timeout=60)[0].decode() + assert sorted(call for call, spans in traces.items() if _logged(spans)) == sorted(calls), output + for call, spans in traces.items(): + app: Final = { + "langfuse.trace.name": f"checkout-{marker}-{call}", + "langfuse.trace.input": json.dumps({"cart": f"{marker}-{call}"}), + "langfuse.trace.output": json.dumps({"order": f"{marker}-{call}"}), + } + root: Final = next(span for span in spans if not span.parent_span_id) + assert {key: root.attributes.get(key) for key in app} == app, (call, root.attributes) + written: Final = set(_trace_values(spans, (*app, "langfuse.trace.tags"))) + assert written == set(app.items()), (call, written) diff --git a/tests/unit/integrations/otel/test_langfuse_logger.py b/tests/unit/integrations/otel/test_langfuse_logger.py index b84f4512676..e4c4f5e6406 100644 --- a/tests/unit/integrations/otel/test_langfuse_logger.py +++ b/tests/unit/integrations/otel/test_langfuse_logger.py @@ -10,6 +10,8 @@ import pytest pytest.importorskip("opentelemetry") +from opentelemetry import trace # noqa: E402 # needs the importorskip above +from opentelemetry.context import Context # noqa: E402 # needs the importorskip above from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter # noqa: E402 import litellm # noqa: E402 @@ -61,8 +63,8 @@ def _logger(*, capture: str = "span_only", mappers: Sequence[str] = ("genai", "l return build_otel_v2_logger(config=cfg, tracer_provider=tracer_provider), exporter -def _start_root(logger): - root = logger._emitter.start_span(SpanRole.PROXY_REQUEST, LITELLM_PROXY_REQUEST_SPAN_NAME) +def _start_root(logger, parent_context: Context | None = None): + root = logger._emitter.start_span(SpanRole.PROXY_REQUEST, LITELLM_PROXY_REQUEST_SPAN_NAME, parent_context) set_request_root_span(root) return root @@ -328,11 +330,24 @@ def test_unrenderable_output_never_raises_into_the_request(): assert INPUT_ATTR not in attrs and OUTPUT_ATTR not in attrs +def _inbound_traceparent() -> Context: + caller: Final = trace.SpanContext( + trace_id=0x4BF92F3577B34DA6A3CE929D0E0E4736, + span_id=0x00F067AA0BA902B7, + is_remote=True, + trace_flags=trace.TraceFlags(trace.TraceFlags.SAMPLED), + ) + return trace.set_span_in_context(trace.NonRecordingSpan(caller)) + + def _run_named_request( - logger: OpenTelemetryV2, exporter: InMemorySpanExporter, litellm_params: Mapping[str, object] + logger: OpenTelemetryV2, + exporter: InMemorySpanExporter, + litellm_params: Mapping[str, object], + parent_context: Context | None = None, ) -> tuple[Mapping[str, object], Mapping[str, object]]: response: Final = ModelResponse(choices=[Choices(message=Message(role="assistant", content="pong"))]) - root: Final = _start_root(logger) + root: Final = _start_root(logger, parent_context) logger.log_pre_api_call( model="gpt-5.4-mini", messages=[], @@ -389,13 +404,37 @@ def test_body_metadata_trace_name_names_the_root_and_the_generation(): assert generation_attrs[TRACE_NAME_ATTR] == "from-body" -def test_unnamed_request_derives_the_trace_name_on_both_spans(): +def test_unnamed_request_derives_the_trace_name_on_the_root_only(): logger, exporter = _logger() root_attrs, generation_attrs = _run_named_request(logger, exporter, {"proxy_server_request": {"headers": {}}}) assert root_attrs[TRACE_NAME_ATTR] == "litellm-acompletion" - assert generation_attrs[TRACE_NAME_ATTR] == "litellm-acompletion" + assert TRACE_NAME_ATTR not in generation_attrs + + +def test_unnamed_request_under_an_inbound_traceparent_leaves_the_trace_name_to_the_caller(): + logger, exporter = _logger() + + root_attrs, generation_attrs = _run_named_request( + logger, exporter, {"proxy_server_request": {"headers": {}}}, _inbound_traceparent() + ) + + assert TRACE_NAME_ATTR not in root_attrs and TRACE_NAME_ATTR not in generation_attrs + + +def test_named_request_under_an_inbound_traceparent_keeps_the_caller_trace_name(): + logger, exporter = _logger() + + root_attrs, generation_attrs = _run_named_request( + logger, + exporter, + {"metadata": {"trace_name": "from-body"}, "proxy_server_request": {"headers": {}}}, + _inbound_traceparent(), + ) + + assert root_attrs[TRACE_NAME_ATTR] == "from-body" + assert generation_attrs[TRACE_NAME_ATTR] == "from-body" @pytest.mark.parametrize("capture", ["span_only", "no_content"]) @@ -420,7 +459,8 @@ def test_body_metadata_user_session_and_tags_land_on_the_root_and_the_generation assert attrs["user.id"] == "user-42" assert attrs["session.id"] == "session-7" assert tuple(attrs["langfuse.trace.tags"]) == ("prod", "eval", "nightly") - assert attrs[TRACE_NAME_ATTR] == "litellm-acompletion" + assert root_attrs[TRACE_NAME_ATTR] == "litellm-acompletion" + assert TRACE_NAME_ATTR not in generation_attrs def test_langfuse_user_and_session_headers_beat_body_metadata_on_both_spans(): @@ -491,9 +531,8 @@ def test_a_request_without_trace_controls_stamps_none_of_them(): caller_only: Final = tuple(attr for attr in TRACE_CONTROL_ATTRS if attr != TRACE_NAME_ATTR) assert set(caller_only).isdisjoint(root_attrs) - assert set(caller_only).isdisjoint(generation_attrs) assert root_attrs[TRACE_NAME_ATTR] == "litellm-acompletion" - assert generation_attrs[TRACE_NAME_ATTR] == "litellm-acompletion" + assert set(TRACE_CONTROL_ATTRS).isdisjoint(generation_attrs) @pytest.mark.parametrize( diff --git a/tests/unit/integrations/otel/test_otel_v2_vendor_mappers.py b/tests/unit/integrations/otel/test_otel_v2_vendor_mappers.py index f53c9149e02..54f4fc0b66d 100644 --- a/tests/unit/integrations/otel/test_otel_v2_vendor_mappers.py +++ b/tests/unit/integrations/otel/test_otel_v2_vendor_mappers.py @@ -777,28 +777,27 @@ def test_resolve_mappers_rejects_unknown_name(): resolve_mappers(["genai", "nope"]) -def test_langfuse_mapper_derives_the_trace_name_from_the_call_type(): - attrs = LangfuseMapper().map(_llm_call(call_type="acompletion")) - - assert attrs["langfuse.trace.name"] == "litellm-acompletion" - - -def test_langfuse_mapper_derived_name_yields_to_the_caller_name(): - attrs = LangfuseMapper().map(_llm_call(call_type="acompletion", trace=TraceControls(name="caller-trace"))) - - assert attrs["langfuse.trace.name"] == "caller-trace" - - -def test_langfuse_mapper_without_call_type_or_name_emits_no_trace_name(): - attrs = LangfuseMapper().map(_llm_call(call_type=None, trace=TraceControls())) +def test_langfuse_mapper_leaves_the_trace_name_off_an_unnamed_generation() -> None: + attrs: Final = LangfuseMapper().map(_llm_call(call_type="acompletion", trace=TraceControls())) assert "langfuse.trace.name" not in attrs -def test_langfuse_root_trace_attributes_match_the_generation_pairs(): - data = _llm_call(call_type="aresponses", trace=TraceControls()) - generation = LangfuseMapper().map(data) +def test_langfuse_mapper_carries_the_caller_trace_name_on_the_generation() -> None: + attrs: Final = LangfuseMapper().map(_llm_call(call_type="acompletion", trace=TraceControls(name="caller-trace"))) - root = LangfuseMapper.trace_attributes(TraceControls(), "aresponses") - assert root == {"langfuse.trace.name": "litellm-aresponses"} - assert generation["langfuse.trace.name"] == root["langfuse.trace.name"] + assert attrs["langfuse.trace.name"] == "caller-trace" + + +@pytest.mark.parametrize( + ("trace", "call_type", "expected"), + [ + (TraceControls(), "aresponses", {"langfuse.trace.name": "litellm-aresponses"}), + (TraceControls(name="caller-trace"), "aresponses", {"langfuse.trace.name": "caller-trace"}), + (TraceControls(), None, {}), + ], +) +def test_langfuse_root_trace_attributes_derive_the_name_only_when_the_caller_gave_none( + trace: TraceControls, call_type: str | None, expected: Mapping[str, object] +) -> None: + assert LangfuseMapper.trace_attributes(trace, call_type) == expected diff --git a/tests/unit/integrations/test_langfuse_otel.py b/tests/unit/integrations/test_langfuse_otel.py index 598ab8e03f1..6474f5e8ff1 100644 --- a/tests/unit/integrations/test_langfuse_otel.py +++ b/tests/unit/integrations/test_langfuse_otel.py @@ -1,19 +1,22 @@ import base64 import json import os -from collections.abc import Mapping +from collections.abc import Mapping, Sequence from datetime import datetime, timezone from typing import Final from unittest.mock import MagicMock, patch import pytest from openai.types.responses import ResponseFunctionToolCall +from opentelemetry import trace +from opentelemetry.context import Context from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.trace import Span import litellm from litellm.constants import REDACTED_BY_LITELLM from litellm.integrations.langfuse.langfuse_otel import LangfuseOtelLogger -from litellm.integrations.opentelemetry import OpenTelemetry, OpenTelemetryConfig +from litellm.integrations.opentelemetry import LITELLM_PROXY_REQUEST_SPAN_NAME, OpenTelemetry, OpenTelemetryConfig from litellm.types.llms.openai import ResponsesAPIResponse from litellm.types.utils import Choices, ModelResponse, StandardCallbackDynamicParams @@ -913,7 +916,7 @@ class TestLangfuseOtelResponsesAPI: == "The weather in San Francisco is sunny, 20°C." ) - trace_output_calls = [ + trace_output_calls: Final = [ call for call in mock_safe_set_attribute.call_args_list if call.args[1] == LangfuseSpanAttributes.TRACE_OUTPUT.value @@ -1030,135 +1033,133 @@ class TestLangfuseOtelResponsesAPI: assert output_data[0]["arguments"] == {} -class _RecordingSpan: - def __init__(self) -> None: - self.attributes: dict[str, object] = {} +_TRACER: Final = TracerProvider().get_tracer("langfuse-otel-tests") +TRACE_FIELD_KEYS: Final = ( + "langfuse.trace.name", + "langfuse.trace.input", + "langfuse.trace.output", + "langfuse.trace.tags", +) +CALLER: Final = trace.SpanContext( + trace_id=0x4BF92F3577B34DA6A3CE929D0E0E4736, + span_id=0x00F067AA0BA902B7, + is_remote=True, + trace_flags=trace.TraceFlags(trace.TraceFlags.SAMPLED), +) +CHAT_KWARGS: Final = { + "call_type": "acompletion", + "messages": [{"role": "user", "content": "weather?"}], +} - def set_attribute(self, key: str, value: object) -> None: - self.attributes[key] = value + +def _chat_response(content: str) -> ModelResponse: + return ModelResponse( + id="chatcmpl-derived", + model="gpt-4o", + choices=[Choices(finish_reason="stop", message={"role": "assistant", "content": content})], + ) + + +def _stamped(span: Span, kwargs: Mapping[str, object], response_obj: object = None) -> Mapping[str, object]: + LangfuseOtelLogger._set_langfuse_specific_attributes(span, kwargs, response_obj) + span.end() + return dict(getattr(span, "attributes", None) or {}) def _emitted(kwargs: Mapping[str, object], response_obj: object = None) -> Mapping[str, object]: - span: Final = _RecordingSpan() - LangfuseOtelLogger._set_langfuse_specific_attributes(span, kwargs, response_obj) - return span.attributes + return _stamped(_TRACER.start_span("litellm_request"), kwargs, response_obj) + + +def _trace_fields(attributes: Mapping[str, object]) -> Mapping[str, object]: + return {key: attributes[key] for key in TRACE_FIELD_KEYS if key in attributes} + + +class _PresetCache: + @staticmethod + def get_preset_cache_key_from_kwargs(**kwargs: object) -> str: # kwargs-ok: mirrors Cache's signature + return "preset-abc" class TestDerivedTraceFields: - def test_chat_call_derives_trace_name_input_and_output(self): - response_obj = ModelResponse( - id="chatcmpl-derived", - model="gpt-4o", - choices=[ - Choices( - finish_reason="stop", - message={"role": "assistant", "content": "Sunny."}, - ) - ], - ) - attributes = _emitted( - { - "call_type": "acompletion", - "messages": [{"role": "user", "content": "weather?"}], - "litellm_params": {"metadata": {}}, - }, - response_obj, - ) + def test_chat_call_derives_trace_name_input_and_output(self) -> None: + attributes: Final = _emitted({**CHAT_KWARGS, "litellm_params": {"metadata": {}}}, _chat_response("Sunny.")) assert attributes["langfuse.trace.name"] == "litellm-acompletion" assert attributes["langfuse.trace.input"] == attributes["langfuse.observation.input"] assert attributes["langfuse.trace.output"] == attributes["langfuse.observation.output"] - assert json.loads(attributes["langfuse.trace.input"]) == [{"role": "user", "content": "weather?"}] - assert json.loads(attributes["langfuse.trace.output"]) == {"role": "assistant", "content": "Sunny."} + assert json.loads(str(attributes["langfuse.trace.input"])) == [{"role": "user", "content": "weather?"}] + assert json.loads(str(attributes["langfuse.trace.output"])) == {"role": "assistant", "content": "Sunny."} - def test_missing_call_type_defaults_to_completion(self): - attributes = _emitted( - { - "messages": [{"role": "user", "content": "hi"}], - "litellm_params": {"metadata": {}}, - } + def test_missing_call_type_defaults_to_completion(self) -> None: + attributes: Final = _emitted( + {"messages": [{"role": "user", "content": "hi"}], "litellm_params": {"metadata": {}}} ) + assert attributes["langfuse.trace.name"] == "litellm-completion" - def test_caller_trace_name_wins_and_is_the_only_name(self): - attributes = _emitted( - { - "call_type": "acompletion", - "litellm_params": {"metadata": {"trace_name": "caller-trace"}}, - } + def test_caller_trace_name_wins_and_is_the_only_name(self) -> None: + attributes: Final = _emitted( + {"call_type": "acompletion", "litellm_params": {"metadata": {"trace_name": "caller-trace"}}} ) + assert attributes["langfuse.trace.name"] == "caller-trace" - def test_existing_trace_id_keeps_caller_tags_and_leaves_derived_trace_fields_off(self, monkeypatch): + def test_existing_trace_id_keeps_caller_tags_and_leaves_derived_trace_fields_off( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: monkeypatch.setattr(litellm, "langfuse_default_tags", ["cache_hit"]) - response_obj = ModelResponse( - id="chatcmpl-joined", - model="gpt-4o", - choices=[ - Choices( - finish_reason="stop", - message={"role": "assistant", "content": "hi"}, - ) - ], - ) - attributes = _emitted( + attributes: Final = _emitted( { - "call_type": "acompletion", - "messages": [{"role": "user", "content": "hi"}], + **CHAT_KWARGS, "cache_hit": True, - "litellm_params": { - "metadata": {"existing_trace_id": "abc123", "tags": ["experiment"]}, - }, + "litellm_params": {"metadata": {"existing_trace_id": "abc123", "tags": ["experiment"]}}, "standard_logging_object": {"request_tags": ["injected"]}, }, - response_obj, + _chat_response("hi"), ) - assert "langfuse.trace.name" not in attributes + + assert _trace_fields(attributes) == {"langfuse.trace.tags": json.dumps(["experiment"])} assert attributes["langfuse.trace.existing_id"] == "abc123" - assert "langfuse.trace.input" not in attributes - assert "langfuse.trace.output" not in attributes - assert json.loads(attributes["langfuse.trace.tags"]) == ["experiment"] assert "langfuse.observation.input" in attributes assert "langfuse.observation.output" in attributes @pytest.mark.parametrize( ("mask_flags", "masked_trace_keys"), [ - ({"mask_input": True}, {"langfuse.trace.input"}), - ({"mask_output": True}, {"langfuse.trace.output"}), - ({"mask_input": True, "mask_output": True}, {"langfuse.trace.input", "langfuse.trace.output"}), - ({"mask_input": "true", "mask_output": "TRUE"}, {"langfuse.trace.input", "langfuse.trace.output"}), - ({"mask_input": "false", "mask_output": "False"}, set()), - ({"mask_input": False, "mask_output": None}, set()), + ({"mask_input": True}, frozenset({"langfuse.trace.input"})), + ({"mask_output": True}, frozenset({"langfuse.trace.output"})), + ({"mask_input": True, "mask_output": True}, frozenset({"langfuse.trace.input", "langfuse.trace.output"})), + ( + {"mask_input": "true", "mask_output": "TRUE"}, + frozenset({"langfuse.trace.input", "langfuse.trace.output"}), + ), + ({"mask_input": "false", "mask_output": "False"}, frozenset()), + ({"mask_input": False, "mask_output": None}, frozenset()), ], ) - def test_mask_flags_redact_the_derived_trace_io(self, mask_flags, masked_trace_keys): - response_obj = ModelResponse( - id="chatcmpl-masked", - model="gpt-4o", - choices=[Choices(finish_reason="stop", message={"role": "assistant", "content": "secret answer"})], - ) - attributes = _emitted( + def test_mask_flags_redact_the_derived_trace_io( + self, mask_flags: Mapping[str, object], masked_trace_keys: frozenset[str] + ) -> None: + attributes: Final = _emitted( { "call_type": "acompletion", "messages": [{"role": "user", "content": "secret question"}], "litellm_params": {"metadata": mask_flags}, }, - response_obj, + _chat_response("secret answer"), ) - trace_io = { - key: attributes[key] for key in ("langfuse.trace.input", "langfuse.trace.output") if key in attributes - } - observation_of = { + observation_of: Final = { "langfuse.trace.input": attributes["langfuse.observation.input"], "langfuse.trace.output": attributes["langfuse.observation.output"], } - assert trace_io == { - key: REDACTED_BY_LITELLM if key in masked_trace_keys else observation_of[key] for key in observation_of + + assert {key: attributes.get(key) for key in observation_of} == { + key: REDACTED_BY_LITELLM if key in masked_trace_keys else observation + for key, observation in observation_of.items() } - def test_responses_api_output_mirrors_observation_output_items(self): - response_obj = ResponsesAPIResponse( + def test_responses_api_output_mirrors_observation_output_items(self) -> None: + response_obj: Final = ResponsesAPIResponse( id="response-derived", created_at=1625247700, output=[ @@ -1172,16 +1173,11 @@ class TestDerivedTraceFields: ) ], ) - attributes = _emitted( - { - "call_type": "aresponses", - "litellm_params": {"metadata": {}}, - }, - response_obj, - ) + attributes: Final = _emitted({"call_type": "aresponses", "litellm_params": {"metadata": {}}}, response_obj) + assert attributes["langfuse.trace.name"] == "litellm-aresponses" assert attributes["langfuse.trace.output"] == attributes["langfuse.observation.output"] - assert json.loads(attributes["langfuse.trace.output"]) == [ + assert json.loads(str(attributes["langfuse.trace.output"])) == [ { "id": "fc-1", "name": "get_weather", @@ -1191,19 +1187,18 @@ class TestDerivedTraceFields: } ] - def test_tags_merge_caller_request_and_default_tags_in_order(self, monkeypatch): + def test_tags_merge_caller_request_and_default_tags_in_order(self, monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(litellm, "langfuse_default_tags", ["cache_hit", "user_api_key_alias"]) - attributes = _emitted( + attributes: Final = _emitted( { "call_type": "acompletion", "cache_hit": True, - "litellm_params": { - "metadata": {"tags": ["a", "b"], "user_api_key_alias": "k1"}, - }, + "litellm_params": {"metadata": {"tags": ["a", "b"], "user_api_key_alias": "k1"}}, "standard_logging_object": {"request_tags": ["b", "c"]}, } ) - assert json.loads(attributes["langfuse.trace.tags"]) == [ + + assert json.loads(str(attributes["langfuse.trace.tags"])) == [ "a", "b", "c", @@ -1211,37 +1206,22 @@ class TestDerivedTraceFields: "user_api_key_alias:k1", ] - def test_cache_key_default_tag_falls_back_to_preset_cache_key(self, monkeypatch): - class _PresetCache: - @staticmethod - def get_preset_cache_key_from_kwargs(**kwargs) -> str: - return "preset-abc" - + @pytest.mark.parametrize( + ("metadata", "expected"), + [ + ({}, ["cache_key:preset-abc"]), + ({"hidden_params": {"cache_key": "explicit-key"}}, ["cache_key:explicit-key"]), + ], + ids=["falls_back_to_the_preset_cache_key", "prefers_hidden_params_over_the_preset"], + ) + def test_cache_key_default_tag( + self, monkeypatch: pytest.MonkeyPatch, metadata: Mapping[str, object], expected: Sequence[str] + ) -> None: monkeypatch.setattr(litellm, "langfuse_default_tags", ["cache_key"]) monkeypatch.setattr(litellm, "cache", _PresetCache()) - attributes = _emitted( - { - "call_type": "acompletion", - "litellm_params": {"metadata": {}}, - } - ) - assert json.loads(attributes["langfuse.trace.tags"]) == ["cache_key:preset-abc"] + attributes: Final = _emitted({"call_type": "acompletion", "litellm_params": {"metadata": metadata}}) - def test_cache_key_default_tag_prefers_hidden_params_over_preset(self, monkeypatch): - class _PresetCache: - @staticmethod - def get_preset_cache_key_from_kwargs(**kwargs) -> str: - return "preset-abc" - - monkeypatch.setattr(litellm, "langfuse_default_tags", ["cache_key"]) - monkeypatch.setattr(litellm, "cache", _PresetCache()) - attributes = _emitted( - { - "call_type": "acompletion", - "litellm_params": {"metadata": {"hidden_params": {"cache_key": "explicit-key"}}}, - } - ) - assert json.loads(attributes["langfuse.trace.tags"]) == ["cache_key:explicit-key"] + assert json.loads(str(attributes["langfuse.trace.tags"])) == expected @pytest.mark.parametrize( ("caller_tags", "expected"), @@ -1253,62 +1233,133 @@ class TestDerivedTraceFields: ("", ["req"]), ], ) - def test_caller_tags_are_read_the_way_langfuse_reads_a_raw_tags_value(self, caller_tags, expected): - attributes = _emitted( + def test_caller_tags_are_read_the_way_langfuse_reads_a_raw_tags_value( + self, caller_tags: object, expected: Sequence[str] + ) -> None: + attributes: Final = _emitted( { "call_type": "acompletion", "litellm_params": {"metadata": {"tags": caller_tags}}, "standard_logging_object": {"request_tags": ["req"]}, } ) - assert json.loads(attributes["langfuse.trace.tags"]) == expected - def test_no_tags_anywhere_leaves_trace_tags_unset(self): - attributes = _emitted( - { - "call_type": "acompletion", - "litellm_params": {"metadata": {}}, - } - ) + assert json.loads(str(attributes["langfuse.trace.tags"])) == expected + + def test_no_tags_anywhere_leaves_trace_tags_unset(self) -> None: + attributes: Final = _emitted({"call_type": "acompletion", "litellm_params": {"metadata": {}}}) + assert "langfuse.trace.tags" not in attributes - def test_empty_string_role_is_omitted_from_output(self): - response_obj = { - "id": "chatcmpl-norole", - "choices": [{"message": {"role": "", "content": "hi"}}], - } - attributes = _emitted( - { - "call_type": "acompletion", - "litellm_params": {"metadata": {}}, - }, - response_obj, - ) - assert json.loads(attributes["langfuse.observation.output"]) == {"content": "hi"} + def test_empty_string_role_is_omitted_from_output(self) -> None: + response_obj: Final = {"id": "chatcmpl-norole", "choices": [{"message": {"role": "", "content": "hi"}}]} + attributes: Final = _emitted({"call_type": "acompletion", "litellm_params": {"metadata": {}}}, response_obj) - def test_no_messages_leaves_trace_input_unset_but_emits_output(self): - response_obj = ModelResponse( - id="chatcmpl-nomsg", - model="gpt-4o", - choices=[ - Choices( - finish_reason="stop", - message={"role": "assistant", "content": "hi"}, - ) - ], - ) - attributes = _emitted( - { - "call_type": "acompletion", - "litellm_params": {"metadata": {}}, - }, - response_obj, + assert json.loads(str(attributes["langfuse.observation.output"])) == {"content": "hi"} + + def test_no_messages_leaves_trace_input_unset_but_emits_output(self) -> None: + attributes: Final = _emitted( + {"call_type": "acompletion", "litellm_params": {"metadata": {}}}, _chat_response("hi") ) + assert "langfuse.trace.input" not in attributes assert "langfuse.observation.input" not in attributes assert attributes["langfuse.trace.output"] == attributes["langfuse.observation.output"] +def _proxy_span(parent: Context | None = None) -> Span: + return _TRACER.start_span(LITELLM_PROXY_REQUEST_SPAN_NAME, context=parent) + + +def _child_of(parent: Span) -> Span: + return _TRACER.start_span("litellm_request", context=trace.set_span_in_context(parent)) + + +def _carrying(parent: Span, key: str = "metadata") -> Mapping[str, object]: + return {**CHAT_KWARGS, "litellm_params": {key: {"litellm_parent_otel_span": parent}}} + + +class TestDerivedTraceFieldsOnlyOnTracesLitellmStarts: + """litellm derives a trace's name, io and tags only when its own span opened the trace.""" + + def test_a_span_under_an_inbound_traceparent_joins_the_caller_trace(self) -> None: + span: Final = _TRACER.start_span( + "litellm_request", context=trace.set_span_in_context(trace.NonRecordingSpan(CALLER)) + ) + + attributes: Final = _stamped(span, {**CHAT_KWARGS, "litellm_params": {"metadata": {}}}, _chat_response("ok")) + + assert _trace_fields(attributes) == {} + assert "langfuse.observation.input" in attributes and "langfuse.observation.output" in attributes + + def test_the_active_app_span_litellm_writes_onto_keeps_its_trace_fields(self) -> None: + with _TRACER.start_as_current_span("checkout") as app_span: + app_span.set_attribute("langfuse.trace.name", "checkout") + + attributes: Final = _stamped( + app_span, {**CHAT_KWARGS, "litellm_params": {"metadata": {}}}, _chat_response("ok") + ) + + assert _trace_fields(attributes) == {"langfuse.trace.name": "checkout"} + + def test_a_litellm_root_span_derives_while_an_app_span_is_current(self) -> None: + with _TRACER.start_as_current_span("checkout"): + span: Final = _TRACER.start_span("litellm_request", context=Context()) + + attributes: Final = _stamped( + span, {**CHAT_KWARGS, "litellm_params": {"metadata": {}}}, _chat_response("ok") + ) + + assert attributes["langfuse.trace.name"] == "litellm-acompletion" + + def test_a_span_under_a_caller_provided_parent_span_joins_the_caller_trace(self) -> None: + caller_span: Final = _TRACER.start_span("caller-root") + + attributes: Final = _stamped(_child_of(caller_span), _carrying(caller_span), _chat_response("ok")) + + assert _trace_fields(attributes) == {} + + @pytest.mark.parametrize("key", ["metadata", "litellm_metadata"]) + def test_a_span_under_litellms_own_unparented_proxy_span_derives(self, key: str) -> None: + proxy_span: Final = _proxy_span() + + attributes: Final = _stamped(_child_of(proxy_span), _carrying(proxy_span, key), _chat_response("ok")) + + assert attributes["langfuse.trace.name"] == "litellm-acompletion" + assert attributes["langfuse.trace.input"] == attributes["langfuse.observation.input"] + assert attributes["langfuse.trace.output"] == attributes["langfuse.observation.output"] + + def test_a_span_under_a_proxy_span_with_an_inbound_traceparent_joins_the_caller_trace(self) -> None: + proxy_span: Final = _proxy_span(trace.set_span_in_context(trace.NonRecordingSpan(CALLER))) + + attributes: Final = _stamped(_child_of(proxy_span), _carrying(proxy_span), _chat_response("ok")) + + assert _trace_fields(attributes) == {} + + def test_a_span_whose_parent_is_not_the_carried_proxy_span_joins_that_parent_trace(self) -> None: + other_parent: Final = _TRACER.start_span("caller-root") + + attributes: Final = _stamped(_child_of(other_parent), _carrying(_proxy_span()), _chat_response("ok")) + + assert _trace_fields(attributes) == {} + + def test_an_inbound_traceparent_still_carries_the_caller_trace_name_and_tags(self) -> None: + span: Final = _TRACER.start_span( + "litellm_request", context=trace.set_span_in_context(trace.NonRecordingSpan(CALLER)) + ) + + attributes: Final = _stamped( + span, + {**CHAT_KWARGS, "litellm_params": {"metadata": {"trace_name": "caller-trace", "tags": ["caller"]}}}, + _chat_response("ok"), + ) + + assert _trace_fields(attributes) == { + "langfuse.trace.name": "caller-trace", + "langfuse.trace.tags": json.dumps(["caller"]), + } + + LANGFUSE_ENV_ONLY: Final = { key: value for key, value in os.environ.items()