mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-06 02:48:13 +00:00
Merge pull request #41140 from BerriAI/litellm_otel_v2_langfuse_user_session_tags
This commit is contained in:
commit
821bcf5d78
11 changed files with 304 additions and 68 deletions
|
|
@ -6,10 +6,10 @@ from litellm.integrations.otel.logger import OpenTelemetryV2
|
|||
from litellm.integrations.otel.mappers.langfuse import (
|
||||
LANGFUSE_OBSERVATION_INPUT,
|
||||
LANGFUSE_OBSERVATION_OUTPUT,
|
||||
LANGFUSE_TRACE_NAME,
|
||||
LangfuseMapper,
|
||||
)
|
||||
from litellm.integrations.otel.model.metadata import caller_trace_name
|
||||
from litellm.integrations.otel.model.request_io import request_input, response_output, stream_output
|
||||
from litellm.integrations.otel.model.trace_controls import caller_trace_controls
|
||||
from litellm.integrations.otel.plumbing.context import request_root_span
|
||||
|
||||
if TYPE_CHECKING:
|
||||
|
|
@ -18,14 +18,13 @@ if TYPE_CHECKING:
|
|||
|
||||
|
||||
class LangfuseOpenTelemetryV2(OpenTelemetryV2):
|
||||
"""Names the trace from the request. Langfuse reads ``langfuse.trace.name`` off the root observation,
|
||||
and the proxy's root span is still recording when the LLM call starts."""
|
||||
"""Stamps the caller's trace controls (name, user, session, tags) on the request. Langfuse reads them off
|
||||
the root observation, and the proxy's root span is still recording when the LLM call starts."""
|
||||
|
||||
def log_pre_api_call(self, model: str, messages: object, kwargs: Mapping[str, object]) -> None:
|
||||
root: Final = request_root_span()
|
||||
name: Final = caller_trace_name(kwargs)
|
||||
if root is not None and root.is_recording() and name is not None:
|
||||
root.set_attribute(LANGFUSE_TRACE_NAME, name)
|
||||
if root is not None and root.is_recording():
|
||||
root.set_attributes(LangfuseMapper.trace_attributes(caller_trace_controls(kwargs)))
|
||||
super().log_pre_api_call(model, messages, kwargs)
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -555,7 +555,7 @@ class OpenTelemetryV2(CustomLogger):
|
|||
capture_content=self.config.capture_span_content,
|
||||
time_to_first_chunk_seconds=call.time_to_first_chunk_seconds,
|
||||
request_route=request_root_http_route(),
|
||||
trace_name=call.trace_name,
|
||||
trace=call.trace,
|
||||
)
|
||||
end_time_ns: Final = to_ns(end_time)
|
||||
if carrier is not None and carrier.span is not None:
|
||||
|
|
|
|||
|
|
@ -6,7 +6,8 @@ Langfuse ingests OTLP spans and reads from its own vendor namespace
|
|||
|
||||
Every attribute is declared as a ``key -> extractor`` table entry (one callable
|
||||
per mapping operation): ``_LLM_CALL_ATTRS`` for scalars and ``_BLOB_ATTRS`` for
|
||||
the JSON-serialized payloads. ``_llm_call`` just applies both tables.
|
||||
the JSON-serialized payloads. ``trace_attributes`` maps the caller's trace controls
|
||||
(shared with the root observation); ``_llm_call`` applies both tables plus it.
|
||||
"""
|
||||
|
||||
import json
|
||||
|
|
@ -16,6 +17,7 @@ from typing import Final
|
|||
from litellm.integrations.otel.mappers.base import AttributeMap, AttrValue, SpanData
|
||||
from litellm.integrations.otel.mappers.utils import (
|
||||
collect,
|
||||
drop_none_pairs,
|
||||
json_if,
|
||||
output_messages,
|
||||
serialize_messages,
|
||||
|
|
@ -25,10 +27,14 @@ from litellm.integrations.otel.model.payloads import (
|
|||
LLMRequestParams,
|
||||
LLMUsage,
|
||||
)
|
||||
from litellm.integrations.otel.model.trace_controls import TraceControls
|
||||
|
||||
LANGFUSE_OBSERVATION_INPUT: Final = "langfuse.observation.input"
|
||||
LANGFUSE_OBSERVATION_OUTPUT: Final = "langfuse.observation.output"
|
||||
LANGFUSE_TRACE_NAME: Final = "langfuse.trace.name"
|
||||
LANGFUSE_TRACE_USER_ID: Final = "user.id"
|
||||
LANGFUSE_TRACE_SESSION_ID: Final = "session.id"
|
||||
LANGFUSE_TRACE_TAGS: Final = "langfuse.trace.tags"
|
||||
|
||||
|
||||
class LangfuseMapper:
|
||||
|
|
@ -37,7 +43,6 @@ class LangfuseMapper:
|
|||
"langfuse.observation.model.name": lambda d: d.request_model or None,
|
||||
"langfuse.observation.metadata.provider": lambda d: d.provider or None,
|
||||
"langfuse.observation.id": lambda d: d.identity.call_id or None,
|
||||
LANGFUSE_TRACE_NAME: lambda d: d.trace_name or None,
|
||||
"langfuse.trace.metadata.team_id": lambda d: d.identity.team_id or None,
|
||||
"langfuse.trace.metadata.team_alias": lambda d: d.identity.team_alias or None,
|
||||
}
|
||||
|
|
@ -77,9 +82,21 @@ class LangfuseMapper:
|
|||
case _:
|
||||
return {}
|
||||
|
||||
@staticmethod
|
||||
def trace_attributes(trace: TraceControls) -> AttributeMap:
|
||||
return drop_none_pairs(
|
||||
(
|
||||
(LANGFUSE_TRACE_NAME, trace.name or None),
|
||||
(LANGFUSE_TRACE_USER_ID, trace.user_id or None),
|
||||
(LANGFUSE_TRACE_SESSION_ID, trace.session_id or None),
|
||||
(LANGFUSE_TRACE_TAGS, trace.tags or None),
|
||||
)
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def _llm_call(cls, data: LLMCallSpanData) -> AttributeMap:
|
||||
return {
|
||||
**collect(cls._LLM_CALL_ATTRS, data),
|
||||
**cls.trace_attributes(data.trace),
|
||||
**collect(cls._BLOB_ATTRS, data),
|
||||
}
|
||||
|
|
|
|||
|
|
@ -6,7 +6,7 @@ they live in one place.
|
|||
"""
|
||||
|
||||
import json
|
||||
from collections.abc import Callable, Mapping, Sequence
|
||||
from collections.abc import Callable, Iterable, Mapping, Sequence
|
||||
from typing import Final
|
||||
|
||||
from litellm.integrations.otel.mappers.base import AttributeMap, AttrValue
|
||||
|
|
@ -39,7 +39,12 @@ def tool_attr_budget(vocabularies: int) -> int:
|
|||
|
||||
def drop_none(values: Mapping[str, AttrValue | None]) -> AttributeMap:
|
||||
"""Return ``values`` with ``None``-valued entries removed."""
|
||||
return {k: v for k, v in values.items() if v is not None}
|
||||
return drop_none_pairs(values.items())
|
||||
|
||||
|
||||
def drop_none_pairs(pairs: Iterable[tuple[str, AttrValue | None]]) -> AttributeMap:
|
||||
"""Return ``pairs`` as a map with ``None``-valued entries removed."""
|
||||
return {k: v for k, v in pairs if v is not None}
|
||||
|
||||
|
||||
def tool_definition_attrs(
|
||||
|
|
|
|||
|
|
@ -43,12 +43,12 @@ from typing import TYPE_CHECKING, Any, Final, cast
|
|||
|
||||
from litellm.constants import LITELLM_LOGGING_NO_UPSTREAM_LLM_CALL
|
||||
from litellm.integrations.otel.model.semconv import resolve_operation
|
||||
from litellm.integrations.otel.model.utils import as_str, to_seconds
|
||||
from litellm.integrations.otel.model.trace_controls import TraceControls, caller_trace_controls
|
||||
from litellm.integrations.otel.model.utils import as_str, as_str_mapping, to_seconds
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from litellm.types.utils import StandardLoggingPayload
|
||||
|
||||
LANGFUSE_TRACE_NAME_HEADER: Final = "langfuse_trace_name"
|
||||
REQUESTER_METADATA_KEY: Final = "requester_metadata"
|
||||
REQUESTER_METADATA_PATH: Final = f"{REQUESTER_METADATA_KEY}."
|
||||
|
||||
|
|
@ -225,7 +225,7 @@ class LLMCallEvent:
|
|||
# needs to be reasonable for a span that never gets closed (a leak).
|
||||
provisional_span_name: str
|
||||
time_to_first_chunk_seconds: float | None
|
||||
trace_name: str | None
|
||||
trace: TraceControls
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, kwargs: Mapping[str, Any]) -> LLMCallEvent:
|
||||
|
|
@ -242,30 +242,10 @@ class LLMCallEvent:
|
|||
upstream_started=kwargs.get("api_call_start_time") is not None,
|
||||
provisional_span_name=f"{operation.value} {model}".strip(),
|
||||
time_to_first_chunk_seconds=time_to_first_chunk_seconds(kwargs),
|
||||
trace_name=caller_trace_name(kwargs),
|
||||
trace=caller_trace_controls(kwargs),
|
||||
)
|
||||
|
||||
|
||||
def caller_trace_name(kwargs: Mapping[str, object]) -> str | None:
|
||||
request: Final = _as_str_mapping(kwargs.get("litellm_params"))
|
||||
if request is None:
|
||||
return None
|
||||
proxy_request: Final = _as_str_mapping(request.get("proxy_server_request"))
|
||||
headers: Final = _as_str_mapping(proxy_request.get("headers")) if proxy_request is not None else None
|
||||
from_header: Final = as_str(headers.get(LANGFUSE_TRACE_NAME_HEADER)) if headers is not None else None
|
||||
if from_header:
|
||||
return from_header
|
||||
return next(
|
||||
(
|
||||
name
|
||||
for key in ("metadata", "litellm_metadata")
|
||||
if (metadata := _as_str_mapping(request.get(key))) is not None
|
||||
and (name := as_str(metadata.get("trace_name")))
|
||||
),
|
||||
None,
|
||||
)
|
||||
|
||||
|
||||
def time_to_first_chunk_seconds(kwargs: Mapping[str, Any]) -> float | None:
|
||||
"""Seconds from the upstream request being issued (``api_call_start_time``)
|
||||
to the first streamed chunk (``completion_start_time``); ``None`` for
|
||||
|
|
@ -300,15 +280,8 @@ def auth_metadata(payload: StandardLoggingPayload | None, kwargs: Mapping[str, o
|
|||
)
|
||||
|
||||
|
||||
def _as_str_mapping(value: object) -> Mapping[str, object] | None:
|
||||
"""A read-only view of ``value`` when it is a mapping, else ``None``."""
|
||||
if not isinstance(value, Mapping):
|
||||
return None
|
||||
return cast("Mapping[str, object]", value) # cast-ok: isinstance-guarded, JSON metadata has str keys
|
||||
|
||||
|
||||
def _string_entries(value: object) -> Mapping[str, str] | None:
|
||||
entries: Final = _as_str_mapping(value)
|
||||
entries: Final = as_str_mapping(value)
|
||||
if entries is None:
|
||||
return None
|
||||
typed: Final = MappingProxyType({key: item for key, item in entries.items() if isinstance(item, str)})
|
||||
|
|
@ -324,18 +297,18 @@ def _metadata_dicts(
|
|||
litellm copies it onto ``metadata``, but both are yielded so a route that
|
||||
populates only one is still covered.
|
||||
"""
|
||||
payload_view: Final = _as_str_mapping(payload)
|
||||
payload_view: Final = as_str_mapping(payload)
|
||||
if payload_view is not None:
|
||||
payload_metadata: Final = _as_str_mapping(payload_view.get("metadata"))
|
||||
payload_metadata: Final = as_str_mapping(payload_view.get("metadata"))
|
||||
if payload_metadata is not None:
|
||||
yield payload_metadata
|
||||
params: Final = _as_str_mapping(kwargs.get("litellm_params"))
|
||||
params: Final = as_str_mapping(kwargs.get("litellm_params"))
|
||||
if params is None:
|
||||
return
|
||||
yield from (
|
||||
metadata
|
||||
for key in ("metadata", "litellm_metadata")
|
||||
if (metadata := _as_str_mapping(params.get(key))) is not None
|
||||
if (metadata := as_str_mapping(params.get(key))) is not None
|
||||
)
|
||||
|
||||
|
||||
|
|
@ -365,14 +338,14 @@ def metadata_from_request_data(data: object) -> Mapping[str, object] | None:
|
|||
The proxy stores it under ``metadata`` or ``litellm_metadata`` depending on the route;
|
||||
the proxy-owned siblings (``user_api_key_*``, ``requester_ip_address``) are not read.
|
||||
"""
|
||||
top: Final = _as_str_mapping(data)
|
||||
top: Final = as_str_mapping(data)
|
||||
if top is None:
|
||||
return None
|
||||
snapshots: Final = tuple(
|
||||
snapshot
|
||||
for name in ("metadata", "litellm_metadata")
|
||||
if (nested := _as_str_mapping(top.get(name))) is not None
|
||||
and (snapshot := _as_str_mapping(nested.get(REQUESTER_METADATA_KEY))) is not None
|
||||
if (nested := as_str_mapping(top.get(name))) is not None
|
||||
and (snapshot := as_str_mapping(nested.get(REQUESTER_METADATA_KEY))) is not None
|
||||
)
|
||||
return MappingProxyType({REQUESTER_METADATA_KEY: snapshots[0]}) if snapshots else None
|
||||
|
||||
|
|
@ -382,7 +355,7 @@ def flatten_metadata(raw: Mapping[str, object]) -> Iterator[tuple[str, str]]:
|
|||
stack: Final = list(tuple(raw.items())[::-1]) # mutable-ok: iterative worklist keeps the walk off the call stack
|
||||
while stack:
|
||||
key, value = stack.pop()
|
||||
if (nested := _as_str_mapping(value)) is not None:
|
||||
if (nested := as_str_mapping(value)) is not None:
|
||||
stack.extend(tuple((f"{key}.{sub_key}", sub_value) for sub_key, sub_value in nested.items())[::-1])
|
||||
elif isinstance(value, (str, bool, int, float)):
|
||||
yield key, str(value)
|
||||
|
|
|
|||
|
|
@ -10,10 +10,7 @@ from types import MappingProxyType
|
|||
from typing import TYPE_CHECKING, ClassVar, Final, cast
|
||||
from urllib.parse import urlsplit
|
||||
|
||||
from litellm.integrations.otel.model.metadata import (
|
||||
RequestContext,
|
||||
RequestIdentity,
|
||||
)
|
||||
from litellm.integrations.otel.model.metadata import RequestContext, RequestIdentity
|
||||
from litellm.integrations.otel.model.semconv import (
|
||||
GenAIOperation,
|
||||
GenAIOutputType,
|
||||
|
|
@ -22,6 +19,7 @@ from litellm.integrations.otel.model.semconv import (
|
|||
resolve_output_type,
|
||||
resolve_provider,
|
||||
)
|
||||
from litellm.integrations.otel.model.trace_controls import TraceControls
|
||||
from litellm.integrations.otel.model.utils import (
|
||||
as_bool,
|
||||
as_float,
|
||||
|
|
@ -387,7 +385,7 @@ class LLMCallSpanData:
|
|||
output_type: GenAIOutputType | None = None
|
||||
call_type: str | None = None
|
||||
request_route: str | None = None
|
||||
trace_name: str | None = None
|
||||
trace: TraceControls = field(default_factory=TraceControls)
|
||||
|
||||
@classmethod
|
||||
def from_standard_logging_payload(
|
||||
|
|
@ -396,7 +394,7 @@ class LLMCallSpanData:
|
|||
capture_content: bool = False,
|
||||
time_to_first_chunk_seconds: float | None = None,
|
||||
request_route: str | None = None,
|
||||
trace_name: str | None = None,
|
||||
trace: TraceControls | None = None,
|
||||
) -> LLMCallSpanData:
|
||||
params: Final = cast(Mapping[str, object], payload.get("model_parameters") or {})
|
||||
# The single parse of the request's metadata — the request-vs-provider
|
||||
|
|
@ -438,7 +436,7 @@ class LLMCallSpanData:
|
|||
output_type=resolve_output_type(call_type),
|
||||
call_type=call_type or None,
|
||||
request_route=request_route or context.identity.request_route,
|
||||
trace_name=trace_name,
|
||||
trace=trace or TraceControls(),
|
||||
)
|
||||
|
||||
|
||||
|
|
|
|||
61
litellm/integrations/otel/model/trace_controls.py
Normal file
61
litellm/integrations/otel/model/trace_controls.py
Normal file
|
|
@ -0,0 +1,61 @@
|
|||
"""The caller's Langfuse trace controls, parsed from the live callback kwargs."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Mapping
|
||||
from dataclasses import dataclass
|
||||
from typing import Final
|
||||
|
||||
from pydantic import TypeAdapter, ValidationError
|
||||
|
||||
from litellm.integrations.otel.model.utils import as_str, as_str_mapping
|
||||
|
||||
LANGFUSE_HEADER_PREFIX: Final = "langfuse_"
|
||||
_ITEMS: Final = TypeAdapter(tuple[object, ...])
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class TraceControls:
|
||||
"""The caller's trace-level Langfuse controls: ``metadata.trace_name`` / ``trace_user_id`` / ``session_id`` /
|
||||
``tags`` on the request (SDK or proxy body), with the proxy's ``langfuse_<control>`` headers winning over the
|
||||
body for the scalar ones. Mutation controls (``trace_id``, ``existing_trace_id``, ``update_trace_keys``) are
|
||||
deliberately not carried."""
|
||||
|
||||
name: str | None = None
|
||||
user_id: str | None = None
|
||||
session_id: str | None = None
|
||||
tags: tuple[str, ...] = ()
|
||||
|
||||
|
||||
def caller_trace_controls(kwargs: Mapping[str, object]) -> TraceControls:
|
||||
request: Final = as_str_mapping(kwargs.get("litellm_params"))
|
||||
if request is None:
|
||||
return TraceControls()
|
||||
proxy_request: Final = as_str_mapping(request.get("proxy_server_request"))
|
||||
headers: Final = as_str_mapping(proxy_request.get("headers")) if proxy_request is not None else None
|
||||
bodies: Final = tuple(
|
||||
metadata
|
||||
for key in ("metadata", "litellm_metadata")
|
||||
if (metadata := as_str_mapping(request.get(key))) is not None
|
||||
)
|
||||
|
||||
def scalar(control: str) -> str | None:
|
||||
from_header: Final = as_str(headers.get(f"{LANGFUSE_HEADER_PREFIX}{control}")) if headers is not None else None
|
||||
if from_header:
|
||||
return from_header
|
||||
return next((value for body in bodies if (value := as_str(body.get(control)))), None)
|
||||
|
||||
return TraceControls(
|
||||
name=scalar("trace_name"),
|
||||
user_id=scalar("trace_user_id"),
|
||||
session_id=scalar("session_id"),
|
||||
tags=next((tags for body in bodies if (tags := _str_items(body.get("tags")))), ()),
|
||||
)
|
||||
|
||||
|
||||
def _str_items(value: object) -> tuple[str, ...]:
|
||||
try:
|
||||
items: Final = _ITEMS.validate_python(value)
|
||||
except ValidationError:
|
||||
return ()
|
||||
return tuple(item for item in items if isinstance(item, str) and item)
|
||||
|
|
@ -8,7 +8,13 @@ parsing lives in :mod:`litellm.integrations.otel.plumbing.providers` instead,
|
|||
because it delegates to the OTel SDK's own W3C Baggage parser.
|
||||
"""
|
||||
|
||||
from collections.abc import Mapping
|
||||
from datetime import datetime
|
||||
from typing import Final
|
||||
|
||||
from pydantic import TypeAdapter, ValidationError
|
||||
|
||||
_STR_MAPPING: Final = TypeAdapter(Mapping[str, object])
|
||||
|
||||
|
||||
def as_str(value: object) -> str | None:
|
||||
|
|
@ -55,6 +61,13 @@ def as_bool(value: object) -> bool | None:
|
|||
return bool(value)
|
||||
|
||||
|
||||
def as_str_mapping(value: object) -> Mapping[str, object] | None:
|
||||
try:
|
||||
return _STR_MAPPING.validate_python(value)
|
||||
except ValidationError:
|
||||
return None
|
||||
|
||||
|
||||
def as_str_tuple(value: object) -> tuple[str, ...] | None:
|
||||
if value is None:
|
||||
return None
|
||||
|
|
|
|||
|
|
@ -42,6 +42,7 @@ from litellm.types.utils import ( # noqa: E402
|
|||
INPUT_ATTR: Final = "langfuse.observation.input"
|
||||
OUTPUT_ATTR: Final = "langfuse.observation.output"
|
||||
TRACE_NAME_ATTR: Final = "langfuse.trace.name"
|
||||
TRACE_CONTROL_ATTRS: Final = (TRACE_NAME_ATTR, "user.id", "session.id", "langfuse.trace.tags")
|
||||
CHAT_DATA: Final = {"model": "gpt-5.4-mini", "messages": [{"role": "user", "content": "ping"}]}
|
||||
|
||||
|
||||
|
|
@ -374,6 +375,99 @@ def test_unnamed_request_leaves_the_trace_name_off_both_spans():
|
|||
assert TRACE_NAME_ATTR not in root_attrs and TRACE_NAME_ATTR not in generation_attrs
|
||||
|
||||
|
||||
@pytest.mark.parametrize("capture", ["span_only", "no_content"])
|
||||
def test_body_metadata_user_session_and_tags_land_on_the_root_and_the_generation(capture):
|
||||
logger, exporter = _logger(capture=capture)
|
||||
|
||||
root_attrs, generation_attrs = _run_named_request(
|
||||
logger,
|
||||
exporter,
|
||||
{
|
||||
"metadata": {
|
||||
"trace_user_id": "user-42",
|
||||
"session_id": "session-7",
|
||||
"tags": ["prod", "eval", "nightly"],
|
||||
"user_api_key_team_id": "team-from-proxy",
|
||||
},
|
||||
"proxy_server_request": {"headers": {}},
|
||||
},
|
||||
)
|
||||
|
||||
for attrs in (root_attrs, generation_attrs):
|
||||
assert attrs["user.id"] == "user-42"
|
||||
assert attrs["session.id"] == "session-7"
|
||||
assert tuple(attrs["langfuse.trace.tags"]) == ("prod", "eval", "nightly")
|
||||
assert TRACE_NAME_ATTR not in attrs
|
||||
|
||||
|
||||
def test_langfuse_user_and_session_headers_beat_body_metadata_on_both_spans():
|
||||
logger, exporter = _logger()
|
||||
|
||||
root_attrs, generation_attrs = _run_named_request(
|
||||
logger,
|
||||
exporter,
|
||||
{
|
||||
"metadata": {"trace_user_id": "from-body", "session_id": "from-body"},
|
||||
"proxy_server_request": {
|
||||
"headers": {"langfuse_trace_user_id": "from-header", "langfuse_session_id": "from-header-s"}
|
||||
},
|
||||
},
|
||||
)
|
||||
|
||||
for attrs in (root_attrs, generation_attrs):
|
||||
assert attrs["user.id"] == "from-header"
|
||||
assert attrs["session.id"] == "from-header-s"
|
||||
|
||||
|
||||
def test_caller_metadata_cannot_override_the_proxy_team_identity():
|
||||
logger, exporter = _logger()
|
||||
response: Final = ModelResponse(choices=[Choices(message=Message(role="assistant", content="pong"))])
|
||||
litellm_params: Final = {
|
||||
"metadata": {"trace_user_id": "u", "trace_metadata": {"team_id": "spoofed"}, "team_id": "spoofed"}
|
||||
}
|
||||
logger.log_pre_api_call(
|
||||
model="gpt-5.4-mini", messages=[], kwargs={"litellm_call_id": "call_1", "litellm_params": litellm_params}
|
||||
)
|
||||
payload: Final = {
|
||||
"call_type": "acompletion",
|
||||
"custom_llm_provider": "openai",
|
||||
"model": "gpt-5.4-mini",
|
||||
"messages": CHAT_DATA["messages"],
|
||||
"response": response.model_dump(),
|
||||
"status": "success",
|
||||
"litellm_call_id": "call_1",
|
||||
"metadata": {
|
||||
"user_api_key_team_id": "real-team",
|
||||
"user_api_key_team_alias": "real-alias",
|
||||
"team_id": "spoofed",
|
||||
"team_alias": "spoofed",
|
||||
},
|
||||
"hidden_params": {},
|
||||
}
|
||||
asyncio.run(
|
||||
logger.async_log_success_event(
|
||||
{"standard_logging_object": payload, "litellm_params": litellm_params}, response, None, None
|
||||
)
|
||||
)
|
||||
|
||||
attrs: Final = dict(exporter.get_finished_spans()[0].attributes or {})
|
||||
assert attrs["user.id"] == "u"
|
||||
assert attrs["langfuse.trace.metadata.team_id"] == "real-team"
|
||||
assert attrs["langfuse.trace.metadata.team_alias"] == "real-alias"
|
||||
assert "langfuse.trace.metadata" not in attrs and "langfuse.trace.id" not in attrs
|
||||
|
||||
|
||||
def test_a_request_without_trace_controls_stamps_none_of_them():
|
||||
logger, exporter = _logger()
|
||||
|
||||
root_attrs, generation_attrs = _run_named_request(
|
||||
logger, exporter, {"metadata": {"user_api_key_team_id": "t1", "tags": []}, "proxy_server_request": {"headers": {}}}
|
||||
)
|
||||
|
||||
assert set(TRACE_CONTROL_ATTRS).isdisjoint(root_attrs)
|
||||
assert set(TRACE_CONTROL_ATTRS).isdisjoint(generation_attrs)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("capture", "mappers"),
|
||||
[("no_content", ("genai", "langfuse")), ("span_only", ("genai",))],
|
||||
|
|
|
|||
|
|
@ -28,7 +28,8 @@ from litellm.integrations.otel import (
|
|||
)
|
||||
from litellm.integrations.otel.mappers.genai import GenAIMapper
|
||||
from litellm.integrations.otel.model import spans as spans_mod
|
||||
from litellm.integrations.otel.model.metadata import LLMCallEvent, caller_trace_name
|
||||
from litellm.integrations.otel.model.metadata import LLMCallEvent
|
||||
from litellm.integrations.otel.model.trace_controls import TraceControls, caller_trace_controls
|
||||
from litellm.integrations.otel.model.payloads import (
|
||||
LLMCallSpanData,
|
||||
RequestIdentity,
|
||||
|
|
@ -743,15 +744,62 @@ def test_request_identity_falls_back_to_legacy_team_keys():
|
|||
ids=["header", "body", "anthropic-body", "header-beats-body", "blank-header-falls-through", "neither", "empty"],
|
||||
)
|
||||
def test_caller_trace_name_prefers_the_langfuse_header_over_body_metadata(request_data, expected):
|
||||
assert caller_trace_name({"litellm_params": request_data}) == expected
|
||||
assert LLMCallEvent.from_dict({"litellm_params": request_data}).trace_name == expected
|
||||
assert caller_trace_controls({"litellm_params": request_data}).name == expected
|
||||
assert LLMCallEvent.from_dict({"litellm_params": request_data}).trace.name == expected
|
||||
|
||||
|
||||
def test_llm_span_data_carries_the_caller_trace_name():
|
||||
data: Final = LLMCallSpanData.from_standard_logging_payload(_sample_payload(), trace_name="nightly-eval")
|
||||
@pytest.mark.parametrize(
|
||||
("request_data", "expected"),
|
||||
[
|
||||
(
|
||||
{"metadata": {"trace_user_id": "u-body", "session_id": "s-body", "tags": ["a", "b", "c"]}},
|
||||
TraceControls(user_id="u-body", session_id="s-body", tags=("a", "b", "c")),
|
||||
),
|
||||
(
|
||||
{
|
||||
"proxy_server_request": {
|
||||
"headers": {"langfuse_trace_user_id": "u-header", "langfuse_session_id": "s-header"}
|
||||
},
|
||||
"metadata": {"trace_user_id": "u-body", "session_id": "s-body"},
|
||||
},
|
||||
TraceControls(user_id="u-header", session_id="s-header"),
|
||||
),
|
||||
(
|
||||
{"litellm_metadata": {"trace_user_id": "u-anthropic", "session_id": "s-anthropic", "tags": ["x"]}},
|
||||
TraceControls(user_id="u-anthropic", session_id="s-anthropic", tags=("x",)),
|
||||
),
|
||||
(
|
||||
{"metadata": {"tags": ["kept", 7, "", None, "also-kept"]}},
|
||||
TraceControls(tags=("kept", "also-kept")),
|
||||
),
|
||||
({"metadata": {"tags": "not-a-list", "trace_user_id": "", "session_id": 12}}, TraceControls(session_id="12")),
|
||||
(
|
||||
{
|
||||
"metadata": {
|
||||
"trace_id": "forced",
|
||||
"existing_trace_id": "forced",
|
||||
"update_trace_keys": ["name"],
|
||||
"trace_metadata": {"team_id": "spoofed"},
|
||||
"user_api_key_team_id": "t1",
|
||||
}
|
||||
},
|
||||
TraceControls(),
|
||||
),
|
||||
({}, TraceControls()),
|
||||
],
|
||||
ids=["body", "headers-beat-body", "anthropic-body", "non-string-tags-dropped", "scalar-coercion", "mutation-controls-ignored", "empty"],
|
||||
)
|
||||
def test_caller_trace_controls_carry_user_session_and_tags(request_data, expected):
|
||||
assert caller_trace_controls({"litellm_params": request_data}) == expected
|
||||
assert LLMCallEvent.from_dict({"litellm_params": request_data}).trace == expected
|
||||
|
||||
assert data.trace_name == "nightly-eval"
|
||||
assert LLMCallSpanData.from_standard_logging_payload(_sample_payload()).trace_name is None
|
||||
|
||||
def test_llm_span_data_carries_the_caller_trace_controls():
|
||||
controls: Final = TraceControls(name="nightly-eval", user_id="u1", session_id="s1", tags=("a", "b"))
|
||||
data: Final = LLMCallSpanData.from_standard_logging_payload(_sample_payload(), trace=controls)
|
||||
|
||||
assert data.trace == controls
|
||||
assert LLMCallSpanData.from_standard_logging_payload(_sample_payload()).trace == TraceControls()
|
||||
|
||||
|
||||
def test_llm_span_carries_proxy_request_route():
|
||||
|
|
|
|||
|
|
@ -18,6 +18,7 @@ from litellm.integrations.otel.mappers import (
|
|||
WeaveMapper,
|
||||
resolve_mappers,
|
||||
)
|
||||
from litellm.integrations.otel.model.trace_controls import TraceControls
|
||||
from litellm.integrations.otel.model.payloads import (
|
||||
LLMCallSpanData,
|
||||
LLMRequestParams,
|
||||
|
|
@ -135,8 +136,35 @@ def test_langfuse_mapper_observation_attrs():
|
|||
|
||||
|
||||
def test_langfuse_mapper_names_the_trace_from_the_caller():
|
||||
assert LangfuseMapper().map(_llm_call(trace_name="nightly-eval"))["langfuse.trace.name"] == "nightly-eval"
|
||||
assert "langfuse.trace.name" not in LangfuseMapper().map(_llm_call(trace_name=None))
|
||||
named = LangfuseMapper().map(_llm_call(trace=TraceControls(name="nightly-eval")))
|
||||
assert named["langfuse.trace.name"] == "nightly-eval"
|
||||
assert "langfuse.trace.name" not in LangfuseMapper().map(_llm_call(trace=TraceControls()))
|
||||
|
||||
|
||||
def test_langfuse_mapper_carries_the_caller_user_session_and_tags():
|
||||
controls = TraceControls(user_id="u-42", session_id="s-7", tags=("prod", "eval", "nightly"))
|
||||
attrs = LangfuseMapper().map(_llm_call(trace=controls))
|
||||
|
||||
assert attrs["user.id"] == "u-42"
|
||||
assert attrs["session.id"] == "s-7"
|
||||
assert attrs["langfuse.trace.tags"] == ("prod", "eval", "nightly")
|
||||
assert attrs["langfuse.trace.metadata.team_id"] == "t1"
|
||||
assert attrs["langfuse.trace.metadata.team_alias"] == "team one"
|
||||
|
||||
|
||||
def test_langfuse_mapper_omits_unset_trace_controls():
|
||||
attrs = LangfuseMapper().map(_llm_call(trace=TraceControls(user_id="", session_id=None, tags=())))
|
||||
|
||||
assert {"user.id", "session.id", "langfuse.trace.tags", "langfuse.trace.name"}.isdisjoint(attrs)
|
||||
|
||||
|
||||
def test_langfuse_trace_attributes_match_between_root_and_generation():
|
||||
controls = TraceControls(name="n", user_id="u", session_id="s", tags=("t",))
|
||||
generation = LangfuseMapper().map(_llm_call(trace=controls))
|
||||
|
||||
root = LangfuseMapper.trace_attributes(controls)
|
||||
assert root == {"langfuse.trace.name": "n", "user.id": "u", "session.id": "s", "langfuse.trace.tags": ("t",)}
|
||||
assert all(generation[key] == value for key, value in root.items())
|
||||
|
||||
|
||||
def test_langfuse_mapper_skips_when_no_messages():
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue