mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-28 06:23:07 +00:00
Merge c15e8c862b into b6f91a221b
This commit is contained in:
commit
39f2912642
12 changed files with 1997 additions and 19 deletions
|
|
@ -13,6 +13,7 @@ from opentelemetry.trace.status import Status, StatusCode
|
|||
|
||||
from litellm.integrations.otel.mappers import resolve_mappers
|
||||
from litellm.integrations.otel.mappers.base import AttributeMapper, AttrValue, SpanData
|
||||
from litellm.integrations.otel.mappers.langfuse import LANGFUSE_REQUEST_METADATA_ATTRIBUTES
|
||||
from litellm.integrations.otel.mappers.openinference import fit_indexed_messages
|
||||
from litellm.integrations.otel.model.config import OpenTelemetryV2Config
|
||||
from litellm.integrations.otel.model.payloads import (
|
||||
|
|
@ -300,9 +301,11 @@ class SpanEmitter:
|
|||
stamped_later: Final = error_attributes(error) if error else _NO_ATTRIBUTES
|
||||
reserved: Final = len(stamped_later.keys() - mapped.keys())
|
||||
overwrites: Final = carried_keys(span) & mapped.keys()
|
||||
# The Langfuse request metadata sheds first, so a span at its attribute limit loses nothing else to it.
|
||||
fitted: Final = fit_indexed_messages(
|
||||
MappingProxyType({key: value for key, value in mapped.items() if key not in overwrites}),
|
||||
attribute_budget(span, reserved),
|
||||
LANGFUSE_REQUEST_METADATA_ATTRIBUTES,
|
||||
)
|
||||
for key, value in mapped.items():
|
||||
if key in overwrites or key in fitted:
|
||||
|
|
|
|||
|
|
@ -74,6 +74,7 @@ from litellm.integrations.otel.plumbing.metrics import (
|
|||
create_genai_metrics,
|
||||
)
|
||||
from litellm.integrations.otel.plumbing.providers import (
|
||||
LITELLM_TRACER_NAME,
|
||||
attach_tenant_fan_out,
|
||||
build_tracer_provider,
|
||||
get_event_logger,
|
||||
|
|
@ -97,7 +98,6 @@ if TYPE_CHECKING:
|
|||
StandardLoggingPayload,
|
||||
)
|
||||
|
||||
LITELLM_TRACER_NAME: Final = "litellm"
|
||||
_published_v2_provider: ApiTracerProvider | None = None
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -13,6 +13,7 @@ the JSON-serialized payloads. ``trace_attributes`` maps the caller's trace contr
|
|||
import json
|
||||
from collections.abc import Callable
|
||||
from datetime import datetime, timezone
|
||||
from types import MappingProxyType
|
||||
from typing import Final
|
||||
|
||||
from litellm.integrations.langfuse.langfuse_otel_attributes import serialize_langfuse_datetime
|
||||
|
|
@ -37,6 +38,23 @@ 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"
|
||||
LANGFUSE_OBSERVATION_METADATA: Final = "langfuse.observation.metadata"
|
||||
LANGFUSE_TRACE_METADATA_PREFIX: Final = "langfuse.trace.metadata."
|
||||
TRACE_IDENTITY_FIELDS: Final = (
|
||||
"user_api_key_alias",
|
||||
"user_api_key_user_id",
|
||||
"user_api_key_end_user_id",
|
||||
"user_api_key_team_id",
|
||||
"user_api_key_team_alias",
|
||||
)
|
||||
# The request metadata attributes only a Langfuse destination receives.
|
||||
LANGFUSE_REQUEST_METADATA_ATTRIBUTES: Final[frozenset[str]] = frozenset(
|
||||
{LANGFUSE_OBSERVATION_METADATA, *(f"{LANGFUSE_TRACE_METADATA_PREFIX}{name}" for name in TRACE_IDENTITY_FIELDS)}
|
||||
)
|
||||
|
||||
|
||||
def _identity_field(name: str) -> Callable[[LLMCallSpanData], AttrValue | None]:
|
||||
return lambda d: d.request_metadata.get(name)
|
||||
|
||||
|
||||
class LangfuseMapper:
|
||||
|
|
@ -52,6 +70,9 @@ class LangfuseMapper:
|
|||
if d.completion_start_seconds is not None
|
||||
else None
|
||||
),
|
||||
**MappingProxyType(
|
||||
{f"{LANGFUSE_TRACE_METADATA_PREFIX}{name}": _identity_field(name) for name in TRACE_IDENTITY_FIELDS}
|
||||
),
|
||||
}
|
||||
|
||||
# Sub-tables folded into their respective JSON blobs.
|
||||
|
|
@ -75,6 +96,7 @@ class LangfuseMapper:
|
|||
|
||||
# JSON-payload attributes: each builder returns the serialized blob or None.
|
||||
_BLOB_ATTRS: dict[str, Callable[[LLMCallSpanData], AttrValue | None]] = {
|
||||
LANGFUSE_OBSERVATION_METADATA: lambda d: json_if(dict(d.request_metadata)),
|
||||
"langfuse.observation.model.parameters": lambda d: json_if(
|
||||
collect(LangfuseMapper._MODEL_PARAMS, d.request_params)
|
||||
),
|
||||
|
|
|
|||
|
|
@ -153,7 +153,9 @@ def _shed_order(groups: Mapping[tuple[str, int, int], tuple[str, ...]]) -> tuple
|
|||
_METADATA_KEY: Final = "metadata"
|
||||
|
||||
|
||||
def fit_indexed_messages(attrs: Mapping[str, AttrValue], budget: int | None) -> Mapping[str, AttrValue]:
|
||||
def fit_indexed_messages(
|
||||
attrs: Mapping[str, AttrValue], budget: int | None, shed_first: frozenset[str] = frozenset()
|
||||
) -> Mapping[str, AttrValue]:
|
||||
"""``attrs`` with indexed message attributes shed, least valuable first, until at most ``budget`` keys remain.
|
||||
|
||||
``None`` means the span has no attribute count limit. Every message still rides the ``input.value`` and
|
||||
|
|
@ -161,12 +163,14 @@ def fit_indexed_messages(attrs: Mapping[str, AttrValue], budget: int | None) ->
|
|||
after every indexed message attribute: message attributes are the indexed, queryable view (the blobs
|
||||
carry no per-index keys), so a count-squeezed span keeps them and the single metadata key absorbs only
|
||||
the residual shortfall. Shedding happens here, before ``span.set_attribute``, so the fit is exact and
|
||||
the SDK's dropped-attributes counter never silently masks the choice.
|
||||
the SDK's dropped-attributes counter never silently masks the choice. ``shed_first`` keys shed one at a
|
||||
time before anything else, so they never cost the span any other attribute.
|
||||
"""
|
||||
if budget is None or len(attrs) <= budget:
|
||||
return attrs
|
||||
groups: Final = _message_key_groups(attrs)
|
||||
sheddable: Final = (*(groups[group] for group in _shed_order(groups)),)
|
||||
leading: Final = tuple((key,) for key in attrs if key in shed_first)
|
||||
sheddable: Final = (*leading, *(groups[group] for group in _shed_order(groups)))
|
||||
flex: Final = (_METADATA_KEY,) if _METADATA_KEY in attrs else ()
|
||||
candidates: Final = (*sheddable, flex) if flex else sheddable
|
||||
running: Final = tuple(accumulate(len(candidate) for candidate in candidates))
|
||||
|
|
|
|||
|
|
@ -59,6 +59,17 @@ if TYPE_CHECKING:
|
|||
|
||||
_INTERNAL_CALL_ORIGINS: Final[frozenset[str]] = frozenset(get_args(InternalCallOrigin))
|
||||
_AUTH_ATTRIBUTE: Final = TypeAdapter(object)
|
||||
_EXPORTED_METADATA_ADAPTER: Final[TypeAdapter[Mapping[str, str]]] = TypeAdapter(Mapping[str, str])
|
||||
# The caller identity the Langfuse metadata keys export; the rest of the request metadata stays in the bare
|
||||
# ``metadata`` attribute.
|
||||
IDENTITY_METADATA_FIELDS: Final = (
|
||||
"user_api_key_alias",
|
||||
"user_api_key_user_id",
|
||||
"user_api_key_end_user_id",
|
||||
"user_api_key_team_id",
|
||||
"user_api_key_team_alias",
|
||||
"user_api_key_org_id",
|
||||
)
|
||||
|
||||
REQUESTER_METADATA_KEY: Final = "requester_metadata"
|
||||
REQUESTER_METADATA_PATH: Final = f"{REQUESTER_METADATA_KEY}."
|
||||
|
|
@ -471,6 +482,23 @@ def _model_info_id(model_info: object) -> str | None:
|
|||
return None
|
||||
|
||||
|
||||
def exported_request_metadata(metadata: Mapping[str, object]) -> Mapping[str, str]:
|
||||
"""The non-empty string ``IDENTITY_METADATA_FIELDS`` of ``metadata``, with ``user_api_key_*`` dropped when
|
||||
``litellm.redact_user_api_key_info`` is on, as the ``langfuse`` callback does."""
|
||||
from litellm.litellm_core_utils.redact_messages import (
|
||||
redact_user_api_key_info as _redact_user_api_key_info, # pyright: ignore[reportUnknownVariableType] # helper uses bare dict
|
||||
)
|
||||
|
||||
redact_metadata: Final = cast( # cast-ok: adapter validates the helper's output below
|
||||
Callable[[dict[str, str]], object],
|
||||
_redact_user_api_key_info,
|
||||
)
|
||||
present: Final = {
|
||||
field: value for field in IDENTITY_METADATA_FIELDS if isinstance(value := metadata.get(field), str) and value
|
||||
}
|
||||
return MappingProxyType(dict(_EXPORTED_METADATA_ADAPTER.validate_python(redact_metadata(present))))
|
||||
|
||||
|
||||
def _team_metadata_dict(value: object) -> Mapping[str, object] | None:
|
||||
"""The team's free-form metadata as a raw mapping, or ``None`` when missing
|
||||
or empty.
|
||||
|
|
|
|||
|
|
@ -12,7 +12,12 @@ from urllib.parse import urlsplit
|
|||
|
||||
from typing_extensions import ReadOnly, TypedDict
|
||||
|
||||
from litellm.integrations.otel.model.metadata import RequestContext, RequestIdentity, allowlisted_metadata
|
||||
from litellm.integrations.otel.model.metadata import (
|
||||
RequestContext,
|
||||
RequestIdentity,
|
||||
allowlisted_metadata,
|
||||
exported_request_metadata,
|
||||
)
|
||||
from litellm.integrations.otel.model.semconv import (
|
||||
GenAIOperation,
|
||||
GenAIOutputType,
|
||||
|
|
@ -417,6 +422,8 @@ class LLMCallSpanData:
|
|||
response_cost: float | None
|
||||
server: ServerInfo | None
|
||||
identity: RequestIdentity
|
||||
# The caller identity among the promoted metadata plus the team (see ``exported_request_metadata``).
|
||||
request_metadata: Mapping[str, str] = field(default_factory=lambda: _EMPTY_METADATA)
|
||||
is_streaming: bool | None = None
|
||||
cost: LLMCost = field(default_factory=LLMCost)
|
||||
tools: tuple[ToolDefinition, ...] = ()
|
||||
|
|
@ -467,6 +474,7 @@ class LLMCallSpanData:
|
|||
context: Final = RequestContext.from_standard_logging_payload(payload)
|
||||
metadata: Final = payload.get("metadata")
|
||||
decision: Final = metadata["routing_decision"] if metadata and "routing_decision" in metadata else None
|
||||
promoted_metadata: Final = allowlisted_metadata(context.identity.metadata, metadata_keys)
|
||||
# Normalize ``response`` to a dict once so the content/id reads below are a
|
||||
# plain ``.get`` — no repeated ``isinstance`` guards.
|
||||
raw_response: Final = payload.get("response")
|
||||
|
|
@ -498,7 +506,14 @@ class LLMCallSpanData:
|
|||
cost=LLMCost.from_breakdown(cast("Mapping[str, object] | None", payload.get("cost_breakdown"))),
|
||||
server=None if cache_hit else ServerInfo.from_api_base(context.api_base),
|
||||
identity=context.identity,
|
||||
promoted_metadata=allowlisted_metadata(context.identity.metadata, metadata_keys),
|
||||
request_metadata=exported_request_metadata(
|
||||
{
|
||||
"user_api_key_team_id": context.identity.team_id,
|
||||
"user_api_key_team_alias": context.identity.team_alias,
|
||||
**promoted_metadata,
|
||||
}
|
||||
),
|
||||
promoted_metadata=promoted_metadata,
|
||||
is_streaming=as_bool(payload.get("stream")),
|
||||
tools=_extract_tools(params),
|
||||
messages_in=_dicts(payload.get("messages")) if capture_content else (),
|
||||
|
|
|
|||
|
|
@ -43,6 +43,7 @@ from litellm._version import version as litellm_version
|
|||
from litellm.integrations.otel.mappers.langfuse import (
|
||||
LANGFUSE_OBSERVATION_INPUT,
|
||||
LANGFUSE_OBSERVATION_OUTPUT,
|
||||
LANGFUSE_REQUEST_METADATA_ATTRIBUTES,
|
||||
LANGFUSE_TRACE_NAME,
|
||||
)
|
||||
from litellm.integrations.otel.model.config import (
|
||||
|
|
@ -373,6 +374,7 @@ class _DrainPool:
|
|||
self._backlog -= 1
|
||||
|
||||
|
||||
LITELLM_TRACER_NAME: Final = "litellm"
|
||||
_NO_ATTRIBUTES: Final[Mapping[str, AttributeValue]] = MappingProxyType({})
|
||||
_DB_SYSTEM_KEYS: Final = frozenset({DB.SYSTEM_NAME, DB.SYSTEM_LEGACY})
|
||||
# Keys on a database span that describe the proxy's own datastore: its host, its
|
||||
|
|
@ -459,6 +461,48 @@ def without_message_content(span: ReadableSpan) -> ReadableSpan:
|
|||
return _SpanView(span, span.resource, kept, span.events, span.status, parent=span.parent)
|
||||
|
||||
|
||||
class _CountedSpanView(_SpanView):
|
||||
"""A ``_SpanView`` of ``inner`` with other attributes that reports the dropped counts ``inner`` reports."""
|
||||
|
||||
def __init__(self, inner: ReadableSpan, attributes: Attributes) -> None:
|
||||
super().__init__(inner, inner.resource, attributes, inner.events, inner.status, parent=inner.parent)
|
||||
self._dropped_counts: Final = (inner.dropped_attributes, inner.dropped_events, inner.dropped_links)
|
||||
|
||||
@property
|
||||
def dropped_attributes(self) -> int:
|
||||
return self._dropped_counts[0]
|
||||
|
||||
@property
|
||||
def dropped_events(self) -> int:
|
||||
return self._dropped_counts[1]
|
||||
|
||||
@property
|
||||
def dropped_links(self) -> int:
|
||||
return self._dropped_counts[2]
|
||||
|
||||
|
||||
def _is_litellm_span(span: ReadableSpan) -> bool:
|
||||
"""Whether LiteLLM's own tracer wrote ``span``, rather than application or instrumentation code."""
|
||||
scope: Final = span.instrumentation_scope
|
||||
return scope is not None and scope.name == LITELLM_TRACER_NAME
|
||||
|
||||
|
||||
def without_langfuse_request_metadata(span: ReadableSpan) -> ReadableSpan:
|
||||
"""``span`` without the request metadata the Langfuse mapper writes for a Langfuse destination.
|
||||
|
||||
Only LiteLLM's own spans lose it: any other span keeps every key its own code set.
|
||||
"""
|
||||
attributes: Final = span.attributes or _NO_ATTRIBUTES
|
||||
if not (_is_litellm_span(span) and any(key in LANGFUSE_REQUEST_METADATA_ATTRIBUTES for key in attributes)):
|
||||
return span
|
||||
return _CountedSpanView(
|
||||
span,
|
||||
MappingProxyType(
|
||||
{key: value for key, value in attributes.items() if key not in LANGFUSE_REQUEST_METADATA_ATTRIBUTES}
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def _is_database_span(attributes: Mapping[str, AttributeValue]) -> bool:
|
||||
return any(key in attributes for key in _DB_SYSTEM_KEYS)
|
||||
|
||||
|
|
@ -614,12 +658,16 @@ def _for_destination(
|
|||
captures on the server span, whose request side holds the caller's bearer token,
|
||||
and the query string of the request URL, which can hold the same key. A
|
||||
destination whose effective setting disables span content loses the prompt and
|
||||
response content too. The span itself stays, so the tenant still gets the whole
|
||||
trace tree.
|
||||
response content too, and a destination for any backend but Langfuse loses the
|
||||
request metadata the Langfuse mapper writes on LiteLLM's own spans. The span itself
|
||||
stays, so the tenant still gets the whole trace tree.
|
||||
"""
|
||||
extra: Final = destination.resource_attributes
|
||||
defaults: Final = destination.resource_defaults
|
||||
attributes: Final = span.attributes or _NO_ATTRIBUTES
|
||||
source: Final = (
|
||||
span if destination.callback_name == ExporterOwner.LANGFUSE_OTEL else without_langfuse_request_metadata(span)
|
||||
)
|
||||
attributes: Final = source.attributes or _NO_ATTRIBUTES
|
||||
database: Final = _is_database_span(attributes)
|
||||
owned: Final = _is_tenant_owned_span(attributes)
|
||||
unreachable: Final = _guardrail_unreachable(attributes)
|
||||
|
|
@ -631,14 +679,14 @@ def _for_destination(
|
|||
if _tenant_visible(key, database, owned, unreachable) and not (redacted and _is_message_content(key))
|
||||
}
|
||||
)
|
||||
recorded: Final = span.events
|
||||
recorded: Final = source.events
|
||||
events: Final = tuple(_without_stack_trace(event) for event in recorded) if owned else ()
|
||||
unchanged: Final = owned and _same_attributes(kept, attributes) and all(a is b for a, b in zip(events, recorded))
|
||||
if not extra and not defaults and unchanged:
|
||||
return span
|
||||
resource: Final = _tenant_resource(span.resource, defaults, extra)
|
||||
status: Final = span.status if owned else Status(span.status.status_code)
|
||||
return _SpanView(span, resource, kept, events, status, parent=span.parent)
|
||||
return source
|
||||
resource: Final = _tenant_resource(source.resource, defaults, extra)
|
||||
status: Final = source.status if owned else Status(source.status.status_code)
|
||||
return _SpanView(source, resource, kept, events, status, parent=source.parent)
|
||||
|
||||
|
||||
class TenantFanOutSpanProcessor(SpanProcessor):
|
||||
|
|
@ -1106,6 +1154,25 @@ class _OverriddenBackendFilter(SpanProcessor):
|
|||
return self._inner.force_flush(timeout_millis)
|
||||
|
||||
|
||||
class _LangfuseRequestMetadataFilter(SpanProcessor):
|
||||
"""Strip the Langfuse request metadata from LiteLLM's own spans before ``inner`` exports them."""
|
||||
|
||||
def __init__(self, inner: SpanProcessor) -> None:
|
||||
self._inner: Final = inner
|
||||
|
||||
def on_start(self, span: SDKSpan, parent_context: Context | None = None) -> None:
|
||||
self._inner.on_start(span, parent_context)
|
||||
|
||||
def on_end(self, span: ReadableSpan) -> None:
|
||||
self._inner.on_end(without_langfuse_request_metadata(span))
|
||||
|
||||
def shutdown(self) -> None:
|
||||
self._inner.shutdown()
|
||||
|
||||
def force_flush(self, timeout_millis: int = 30000) -> bool:
|
||||
return self._inner.force_flush(timeout_millis)
|
||||
|
||||
|
||||
class _MessageContentFilter(SpanProcessor):
|
||||
"""Strip prompt and response content from every span before ``inner`` exports it."""
|
||||
|
||||
|
|
@ -1377,13 +1444,17 @@ def build_resource(config: OpenTelemetryV2Config) -> Resource:
|
|||
|
||||
|
||||
def _spec_processor(
|
||||
spec: ExporterSpec, use_simple_processor: bool | None, strip_message_content: bool
|
||||
spec: ExporterSpec,
|
||||
use_simple_processor: bool | None,
|
||||
strip_message_content: bool,
|
||||
strip_langfuse_request_metadata: bool,
|
||||
) -> SpanProcessor:
|
||||
exporting: Final = _processor_for(
|
||||
_exporter_from_spec(spec),
|
||||
(spec.use_simple_processor if spec.use_simple_processor is not None else use_simple_processor),
|
||||
)
|
||||
return _MessageContentFilter(exporting) if strip_message_content else exporting
|
||||
scoped: Final = _LangfuseRequestMetadataFilter(exporting) if strip_langfuse_request_metadata else exporting
|
||||
return _MessageContentFilter(scoped) if strip_message_content else scoped
|
||||
|
||||
|
||||
def build_tracer_provider(
|
||||
|
|
@ -1411,6 +1482,8 @@ def build_tracer_provider(
|
|||
|
||||
``config.langfuse_span_scope`` narrows the exporter owned by ``langfuse_otel``
|
||||
alone; a collector or any other backend in the same config keeps the full tree.
|
||||
In a config that writes the Langfuse vocabulary, every exporter but ``langfuse_otel``'s
|
||||
goes without the request metadata the Langfuse mapper writes on LiteLLM's own spans.
|
||||
``strip_message_content`` removes prompt and response content before every
|
||||
exporter, for a routed provider whose spans can carry content another
|
||||
destination asked for.
|
||||
|
|
@ -1426,10 +1499,18 @@ def build_tracer_provider(
|
|||
|
||||
# ``config._normalize`` guarantees at least one spec (it folds the top-level
|
||||
# ``exporter``/``endpoint``/``headers`` fields in when ``exporters`` is empty).
|
||||
writes_langfuse: Final = "langfuse" in config.mapper_names or any(
|
||||
spec.owner is ExporterOwner.LANGFUSE_OTEL for spec in config.exporters
|
||||
)
|
||||
for spec in config.exporters:
|
||||
if spec.requires_headers and not spec.headers:
|
||||
continue
|
||||
processor = _spec_processor(spec, use_simple_processor, strip_message_content)
|
||||
processor = _spec_processor(
|
||||
spec,
|
||||
use_simple_processor,
|
||||
strip_message_content,
|
||||
strip_langfuse_request_metadata=writes_langfuse and spec.owner is not ExporterOwner.LANGFUSE_OTEL,
|
||||
)
|
||||
owner = spec.owner.value if tenant_overrides and spec.owner is not None else None
|
||||
scope = _operator_scope(config, spec)
|
||||
sink = _sink_key(spec.endpoint, parse_headers(spec.headers)) if _exports_to_the_wire(spec) else None
|
||||
|
|
@ -1577,7 +1658,7 @@ def _attached_processors(provider: trace.TracerProvider) -> "tuple[SpanProcessor
|
|||
return tuple(getattr(multi, "_span_processors", ()))
|
||||
|
||||
|
||||
def get_tracer(provider: TracerProvider, name: str = "litellm") -> Tracer:
|
||||
def get_tracer(provider: TracerProvider, name: str = LITELLM_TRACER_NAME) -> Tracer:
|
||||
# Stamp the instrumentation scope with the LiteLLM package version so every
|
||||
# emitted span carries a deterministic ``scope.version`` (the standard OTel
|
||||
# location for the emitting library's version) for downstream consumers.
|
||||
|
|
|
|||
1492
tests/integration/observability/test_langfuse_otel_metadata.py
Normal file
1492
tests/integration/observability/test_langfuse_otel_metadata.py
Normal file
File diff suppressed because it is too large
Load diff
|
|
@ -465,6 +465,27 @@ def test_arize_destination_drops_openinference_content(rig: Rig) -> None:
|
|||
assert model.get("llm.model_name"), sorted(model)
|
||||
|
||||
|
||||
def test_arize_destination_gets_no_langfuse_request_metadata(rig: Rig) -> None:
|
||||
team, key = rig.team_key(
|
||||
None, "arize-meta", callback="arize", variables={"arize_space_key": "space-meta", "arize_api_key": "arize-meta"}
|
||||
)
|
||||
cursors: Final = rig.cursors()
|
||||
sent: Final = _served(rig.send(key))
|
||||
operator: Final = _arrived(rig.sinks.operator, cursors.operator, sent, frozenset())
|
||||
assert any("langfuse.observation.metadata" in span["attributes"] for span in _model_spans(operator)), _names(
|
||||
operator
|
||||
)
|
||||
tenant: Final = _arrived(rig.sinks.arize, cursors.arize, sent, frozenset(), ("arize-space-id", "space-meta"))
|
||||
models: Final = [span["attributes"] for span in _model_spans(tenant)]
|
||||
assert any(attributes.get("langfuse.trace.metadata.team_id") == team for attributes in models), models
|
||||
assert not [
|
||||
name
|
||||
for attributes in models
|
||||
for name in attributes
|
||||
if name == "langfuse.observation.metadata" or name.startswith("langfuse.trace.metadata.user_api_key_")
|
||||
], models
|
||||
|
||||
|
||||
def test_weave_destination_drops_its_content(rig: Rig) -> None:
|
||||
_, key = rig.team_key(
|
||||
"no_content",
|
||||
|
|
|
|||
|
|
@ -18,7 +18,7 @@ from opentelemetry.sdk.resources import Resource
|
|||
from opentelemetry.sdk.trace import ReadableSpan, TracerProvider
|
||||
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
|
||||
from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
|
||||
from opentelemetry.trace import Status, StatusCode
|
||||
from opentelemetry.trace import Link, SpanContext, Status, StatusCode
|
||||
|
||||
import litellm
|
||||
from litellm.integrations.custom_logger import CustomLogger
|
||||
|
|
@ -5067,3 +5067,129 @@ class TestArizeProjectRouting:
|
|||
monkeypatch.setenv("ARIZE_PROJECT_NAME", "gateway-prod")
|
||||
|
||||
assert self._forwarded_projects() == {"gateway-prod"}
|
||||
|
||||
|
||||
_LANGFUSE_REQUEST_METADATA_SPAN: Final = MappingProxyType(
|
||||
{
|
||||
"langfuse.observation.metadata": '{"requester_metadata": {"headers": {"x-tenant": "acme"}}}',
|
||||
"langfuse.trace.metadata.user_api_key_alias": "prod-key",
|
||||
"langfuse.trace.metadata.user_api_key_user_id": "user-1",
|
||||
"langfuse.trace.metadata.team_id": "team-9",
|
||||
"gen_ai.operation.name": "chat",
|
||||
"gen_ai.request.model": "gpt-4o",
|
||||
}
|
||||
)
|
||||
_WITHOUT_LANGFUSE_REQUEST_METADATA: Final = MappingProxyType(
|
||||
{"langfuse.trace.metadata.team_id": "team-9", "gen_ai.operation.name": "chat", "gen_ai.request.model": "gpt-4o"}
|
||||
)
|
||||
|
||||
|
||||
class TestLangfuseRequestMetadataStaysWithLangfuse:
|
||||
"""The request metadata the Langfuse mapper writes reaches a Langfuse destination and no other."""
|
||||
|
||||
@staticmethod
|
||||
def _emit(
|
||||
provider: TracerProvider,
|
||||
destinations: tuple[OtelDestination, ...] = (),
|
||||
tracer_name: str = "litellm",
|
||||
overflow: bool = False,
|
||||
) -> None:
|
||||
def run() -> None:
|
||||
set_request_destinations(destinations)
|
||||
links: Final = (Link(SpanContext(trace_id=1, span_id=1, is_remote=True)),) if overflow else ()
|
||||
with get_tracer(provider, tracer_name).start_as_current_span("chat gpt-4o", links=links) as span:
|
||||
if overflow:
|
||||
span.set_attribute("app.evicted", "first")
|
||||
span.add_event("app.event")
|
||||
span.set_attributes(_LANGFUSE_REQUEST_METADATA_SPAN)
|
||||
|
||||
in_fresh_context(run)
|
||||
|
||||
@staticmethod
|
||||
def _provider(
|
||||
monkeypatch: pytest.MonkeyPatch, *owners: ExporterOwner | None
|
||||
) -> tuple[TracerProvider, dict[ExporterOwner | None, InMemorySpanExporter]]:
|
||||
exporters: dict[ExporterOwner | None, InMemorySpanExporter] = {} # mutable-ok: filled as specs build
|
||||
kind: Final = "lit8283_request_metadata"
|
||||
monkeypatch.setitem(
|
||||
otel_providers._EXPORTER_FACTORIES,
|
||||
kind,
|
||||
lambda spec: exporters.setdefault(spec.owner, InMemorySpanExporter()),
|
||||
)
|
||||
config: Final = OpenTelemetryV2Config(
|
||||
mapper_names=["genai", "langfuse"],
|
||||
exporters=[ExporterSpec(kind=kind, owner=owner) for owner in owners],
|
||||
)
|
||||
return build_tracer_provider(config, use_simple_processor=True, tenant_overrides=True), exporters
|
||||
|
||||
@staticmethod
|
||||
def _fan_out() -> tuple[TracerProvider, InMemorySpanExporter, InMemorySpanExporter]:
|
||||
langfuse_exporter, arize_exporter = InMemorySpanExporter(), InMemorySpanExporter()
|
||||
provider: Final = TracerProvider()
|
||||
provider.add_span_processor(
|
||||
TenantFanOutSpanProcessor(
|
||||
processor_factory=lambda destination: SimpleSpanProcessor(
|
||||
langfuse_exporter if destination.callback_name == "langfuse_otel" else arize_exporter
|
||||
)
|
||||
)
|
||||
)
|
||||
return provider, langfuse_exporter, arize_exporter
|
||||
|
||||
def test_a_collector_beside_the_langfuse_exporter_gets_no_langfuse_request_metadata(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
provider, exporters = self._provider(monkeypatch, None, ExporterOwner.LANGFUSE_OTEL)
|
||||
|
||||
self._emit(provider)
|
||||
|
||||
assert dict(by_name(exporters[ExporterOwner.LANGFUSE_OTEL])["chat gpt-4o"].attributes) == dict(
|
||||
_LANGFUSE_REQUEST_METADATA_SPAN
|
||||
)
|
||||
assert dict(by_name(exporters[None])["chat gpt-4o"].attributes) == dict(_WITHOUT_LANGFUSE_REQUEST_METADATA)
|
||||
|
||||
def test_a_collector_with_the_langfuse_mapper_and_no_langfuse_exporter_gets_no_langfuse_request_metadata(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
provider, exporters = self._provider(monkeypatch, None)
|
||||
|
||||
self._emit(provider)
|
||||
|
||||
assert dict(by_name(exporters[None])["chat gpt-4o"].attributes) == dict(_WITHOUT_LANGFUSE_REQUEST_METADATA)
|
||||
|
||||
def test_the_fan_out_forwards_langfuse_request_metadata_to_langfuse_destinations_only(self) -> None:
|
||||
provider, langfuse_exporter, arize_exporter = self._fan_out()
|
||||
|
||||
self._emit(provider, (LANGFUSE_DEST, INHERITING_DEST))
|
||||
|
||||
assert dict(by_name(langfuse_exporter)["chat gpt-4o"].attributes) == dict(_LANGFUSE_REQUEST_METADATA_SPAN)
|
||||
assert dict(by_name(arize_exporter)["chat gpt-4o"].attributes) == dict(_WITHOUT_LANGFUSE_REQUEST_METADATA)
|
||||
|
||||
def test_a_span_another_tracer_wrote_keeps_the_langfuse_metadata_keys_it_set(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
provider, exporters = self._provider(monkeypatch, None, ExporterOwner.LANGFUSE_OTEL)
|
||||
fanned, _langfuse_exporter, arize_exporter = self._fan_out()
|
||||
|
||||
self._emit(provider, tracer_name="app")
|
||||
self._emit(fanned, (LANGFUSE_DEST, INHERITING_DEST), tracer_name="app")
|
||||
|
||||
assert dict(by_name(exporters[None])["chat gpt-4o"].attributes) == dict(_LANGFUSE_REQUEST_METADATA_SPAN)
|
||||
assert dict(by_name(arize_exporter)["chat gpt-4o"].attributes) == dict(_LANGFUSE_REQUEST_METADATA_SPAN)
|
||||
|
||||
def test_a_span_without_the_langfuse_request_metadata_keeps_its_dropped_counts(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
monkeypatch.setenv("OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT", str(len(_LANGFUSE_REQUEST_METADATA_SPAN)))
|
||||
monkeypatch.setenv("OTEL_SPAN_EVENT_COUNT_LIMIT", "0")
|
||||
monkeypatch.setenv("OTEL_SPAN_LINK_COUNT_LIMIT", "0")
|
||||
provider, exporters = self._provider(monkeypatch, None, ExporterOwner.LANGFUSE_OTEL)
|
||||
fanned, _langfuse_exporter, arize_exporter = self._fan_out()
|
||||
|
||||
self._emit(provider, overflow=True)
|
||||
self._emit(fanned, (LANGFUSE_DEST, INHERITING_DEST), overflow=True)
|
||||
|
||||
stripped: Final = (by_name(exporters[None])["chat gpt-4o"], by_name(arize_exporter)["chat gpt-4o"])
|
||||
assert [dict(span.attributes or {}) for span in stripped] == [dict(_WITHOUT_LANGFUSE_REQUEST_METADATA)] * 2
|
||||
assert [(span.dropped_attributes, span.dropped_events, span.dropped_links) for span in stripped] == [
|
||||
(1, 1, 1)
|
||||
] * 2
|
||||
|
|
|
|||
|
|
@ -899,6 +899,37 @@ def test_indexed_messages_follow_the_span_limits_of_a_per_request_tracer_overrid
|
|||
assert span.attributes["llm.output_messages.0.message.content"] == "reply 0"
|
||||
|
||||
|
||||
def _team_conversation_span(monkeypatch, redact, max_span_attributes):
|
||||
monkeypatch.setattr("litellm.redact_user_api_key_info", redact)
|
||||
return _conversation_span(
|
||||
["genai", "openinference", "langfuse"],
|
||||
_conversation_payload(60),
|
||||
span_limits=SpanLimits(max_span_attributes=max_span_attributes),
|
||||
)
|
||||
|
||||
|
||||
def test_langfuse_identity_sheds_before_any_attribute_the_span_carries_without_it(monkeypatch):
|
||||
"""At its attribute limit a span drops the Langfuse identity keys first, then exactly what it drops without them."""
|
||||
plain = _team_conversation_span(monkeypatch, True, SpanLimits.UNSET).attributes
|
||||
full = _team_conversation_span(monkeypatch, False, SpanLimits.UNSET).attributes
|
||||
identity_keys = set(full) - set(plain)
|
||||
assert identity_keys == {
|
||||
"langfuse.observation.metadata",
|
||||
"langfuse.trace.metadata.user_api_key_team_id",
|
||||
"langfuse.trace.metadata.user_api_key_team_alias",
|
||||
}
|
||||
|
||||
squeezed_limit = len(plain) - 10
|
||||
squeezed_plain = _team_conversation_span(monkeypatch, True, squeezed_limit)
|
||||
squeezed = _team_conversation_span(monkeypatch, False, squeezed_limit)
|
||||
assert dict(squeezed.attributes) == dict(squeezed_plain.attributes)
|
||||
assert squeezed.dropped_attributes == squeezed_plain.dropped_attributes == 0
|
||||
|
||||
roomy = _team_conversation_span(monkeypatch, False, len(plain) + 2).attributes
|
||||
assert {key: roomy.get(key) for key in plain} == dict(plain)
|
||||
assert len(set(roomy) & identity_keys) == 2
|
||||
|
||||
|
||||
def test_span_attribute_limit_falls_back_to_the_environment_for_spans_outside_the_sdk(monkeypatch):
|
||||
monkeypatch.setenv("OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT", "48")
|
||||
assert span_attribute_limit(INVALID_SPAN) == 48
|
||||
|
|
|
|||
|
|
@ -22,6 +22,7 @@ from litellm.integrations.otel.mappers import (
|
|||
resolve_mappers,
|
||||
)
|
||||
from litellm.integrations.otel.mappers.openinference import fit_indexed_messages
|
||||
from litellm.integrations.otel.model.baggage import DEFAULT_BAGGAGE_METADATA_KEYS
|
||||
from litellm.integrations.otel.model.payloads import (
|
||||
EmbeddingOutput,
|
||||
LLMCallSpanData,
|
||||
|
|
@ -802,3 +803,157 @@ def test_resolve_mappers_composition_layers_vocabularies():
|
|||
def test_resolve_mappers_rejects_unknown_name():
|
||||
with pytest.raises(ValueError, match="unknown mapper name 'nope'"):
|
||||
resolve_mappers(["genai", "nope"])
|
||||
|
||||
|
||||
def _langfuse_attrs(
|
||||
metadata: Mapping[str, object], metadata_keys: tuple[str, ...] = DEFAULT_BAGGAGE_METADATA_KEYS
|
||||
) -> Mapping[str, object]:
|
||||
payload: Final[dict[str, object]] = {
|
||||
"call_type": "acompletion",
|
||||
"custom_llm_provider": "openai",
|
||||
"model": "gpt-4o",
|
||||
"metadata": metadata,
|
||||
"response": {},
|
||||
}
|
||||
return LangfuseMapper().map(LLMCallSpanData.from_standard_logging_payload(payload, metadata_keys=metadata_keys))
|
||||
|
||||
|
||||
def test_langfuse_mapper_emits_request_metadata_and_identity():
|
||||
attrs: Final = _langfuse_attrs(
|
||||
{
|
||||
"user_api_key_alias": "prod-key",
|
||||
"user_api_key_user_id": "user-1",
|
||||
"user_api_key_end_user_id": "end-1",
|
||||
"user_api_key_team_id": "team-9",
|
||||
"user_api_key_team_alias": "team nine",
|
||||
"user_api_key_max_budget": None,
|
||||
"requester_ip_address": "10.0.0.1",
|
||||
"spend_logs_metadata": {"ticket": "LIT-8283"},
|
||||
"requester_metadata": {"headers": {"x-tenant": "acme"}},
|
||||
}
|
||||
)
|
||||
|
||||
assert json.loads(attrs["langfuse.observation.metadata"]) == {
|
||||
"user_api_key_team_id": "team-9",
|
||||
"user_api_key_team_alias": "team nine",
|
||||
"user_api_key_user_id": "user-1",
|
||||
"user_api_key_alias": "prod-key",
|
||||
"user_api_key_end_user_id": "end-1",
|
||||
}
|
||||
assert attrs["langfuse.trace.metadata.user_api_key_alias"] == "prod-key"
|
||||
assert attrs["langfuse.trace.metadata.user_api_key_user_id"] == "user-1"
|
||||
assert attrs["langfuse.trace.metadata.user_api_key_end_user_id"] == "end-1"
|
||||
assert attrs["langfuse.trace.metadata.user_api_key_team_id"] == "team-9"
|
||||
assert attrs["langfuse.trace.metadata.user_api_key_team_alias"] == "team nine"
|
||||
assert attrs["langfuse.trace.metadata.team_id"] == "team-9"
|
||||
assert attrs["langfuse.trace.metadata.team_alias"] == "team nine"
|
||||
assert attrs["langfuse.observation.metadata.provider"] == "openai"
|
||||
|
||||
|
||||
def test_langfuse_mapper_exports_only_the_promoted_metadata_and_the_team():
|
||||
attrs: Final = _langfuse_attrs(
|
||||
{
|
||||
"user_api_key_alias": "prod-key",
|
||||
"user_api_key_team_id": "team-9",
|
||||
"requester_ip_address": "10.0.0.1",
|
||||
"user_api_key_user_email": "caller-email-secret",
|
||||
"user_api_key_auth_metadata": {"note": "auth-metadata-secret"},
|
||||
"user_agent": "user-agent-secret",
|
||||
"headers": {"x-tenant-secret": "header-secret"},
|
||||
"requester_custom_headers": {"x-tenant-secret": "custom-header-secret"},
|
||||
"requester_metadata": {"note": "requester-metadata-secret"},
|
||||
"spend_logs_metadata": {"note": "spend-logs-secret"},
|
||||
"prompt_management_metadata": {"prompt_variables": {"input": "prompt-variables-secret"}},
|
||||
"mcp_tool_call_metadata": {"arguments": {"location": "mcp-arguments-secret"}},
|
||||
"vector_store_request_metadata": [{"query": "vector-query-secret"}],
|
||||
"routing_decision": {"matched_keyword": "routing-prompt-secret"},
|
||||
}
|
||||
)
|
||||
|
||||
assert "secret" not in json.dumps(attrs)
|
||||
assert json.loads(attrs["langfuse.observation.metadata"]) == {
|
||||
"user_api_key_team_id": "team-9",
|
||||
"user_api_key_alias": "prod-key",
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("metadata_keys", "expected"),
|
||||
[
|
||||
((), {"user_api_key_team_id": "team-9"}),
|
||||
(("user_api_key_user_id",), {"user_api_key_team_id": "team-9", "user_api_key_user_id": "user-1"}),
|
||||
(("requester_metadata.trace_id", "requester_ip_address"), {"user_api_key_team_id": "team-9"}),
|
||||
],
|
||||
)
|
||||
def test_langfuse_mapper_follows_the_operator_metadata_allowlist(
|
||||
metadata_keys: tuple[str, ...], expected: Mapping[str, str]
|
||||
) -> None:
|
||||
attrs: Final = _langfuse_attrs(
|
||||
{
|
||||
"user_api_key_alias": "prod-key",
|
||||
"user_api_key_user_id": "user-1",
|
||||
"user_api_key_team_id": "team-9",
|
||||
"requester_ip_address": "10.0.0.1",
|
||||
"requester_metadata": {"trace_id": "trace-7", "note": "requester-metadata-secret"},
|
||||
},
|
||||
metadata_keys=metadata_keys,
|
||||
)
|
||||
|
||||
assert json.loads(attrs["langfuse.observation.metadata"]) == expected
|
||||
assert {key for key in attrs if key.startswith("langfuse.trace.metadata.user_api_key_")} == {
|
||||
f"langfuse.trace.metadata.{key}" for key in expected if key.startswith("user_api_key_")
|
||||
}
|
||||
|
||||
|
||||
def test_langfuse_mapper_skips_empty_identity_values():
|
||||
attrs: Final = _langfuse_attrs(
|
||||
{
|
||||
"user_api_key_alias": None,
|
||||
"user_api_key_user_id": "",
|
||||
"user_api_key_end_user_id": "end-1",
|
||||
}
|
||||
)
|
||||
|
||||
assert "langfuse.trace.metadata.user_api_key_alias" not in attrs
|
||||
assert "langfuse.trace.metadata.user_api_key_user_id" not in attrs
|
||||
assert attrs["langfuse.trace.metadata.user_api_key_end_user_id"] == "end-1"
|
||||
|
||||
|
||||
def test_langfuse_mapper_redacts_user_api_key_fields(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
import litellm
|
||||
|
||||
monkeypatch.setattr(litellm, "redact_user_api_key_info", True)
|
||||
attrs: Final = _langfuse_attrs(
|
||||
{
|
||||
"user_api_key_alias": "prod-key",
|
||||
"user_api_key_team_id": "team-9",
|
||||
"requester_ip_address": "10.0.0.1",
|
||||
}
|
||||
)
|
||||
|
||||
assert "langfuse.observation.metadata" not in attrs
|
||||
assert not [key for key in attrs if key.startswith("langfuse.trace.metadata.user_api_key_")]
|
||||
assert attrs["langfuse.trace.metadata.team_id"] == "team-9"
|
||||
|
||||
|
||||
def test_langfuse_mapper_omits_observation_metadata_without_payload_metadata():
|
||||
attrs: Final = _langfuse_attrs({})
|
||||
|
||||
assert "langfuse.observation.metadata" not in attrs
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"unexpected_shape",
|
||||
[
|
||||
{"prompt_management_metadata": {1: "one", "prompt_id": "prompt-1"}},
|
||||
{"mcp_tool_call_metadata": {1: "one", "name": "lookup_weather"}},
|
||||
{"vector_store_request_metadata": [{1: "one", "vector_store_id": "vs-1"}]},
|
||||
{"requester_metadata": 1},
|
||||
{"requester_metadata": True},
|
||||
{"requester_metadata": 1.5},
|
||||
],
|
||||
)
|
||||
def test_span_data_builds_from_request_metadata_of_unexpected_shape(unexpected_shape: dict[str, object]) -> None:
|
||||
attrs: Final = _langfuse_attrs({**unexpected_shape, "user_api_key_alias": "prod-key"})
|
||||
|
||||
assert json.loads(attrs["langfuse.observation.metadata"]) == {"user_api_key_alias": "prod-key"}
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue