mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-23 05:34:41 +00:00
fix(langfuse_otel): derive trace fields only on traces litellm starts
A litellm span that joins a trace an app or caller started (inbound traceparent, the app's active span, a caller-provided parent span) no longer gets a derived trace name, input, output or tags, like a request joining an existing_trace_id. Langfuse applies the later-starting span's trace update, so the derived values overwrote the app's own trace name and io. OTel v2 derives the name on the request root only when it has no parent, and the generation span keeps main's caller-only trace name.
This commit is contained in:
yucheng
parent
4564d8debe
commit
f21309773d
8 changed files with 647 additions and 236 deletions
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
||||
|
|
|
|||
|
|
@ -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),
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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(
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue