mirror of
https://github.com/BerriAI/litellm.git
synced 2026-11-11 08:38:24 +00:00
Merge 4737d491d6 into 96fe78f96e
This commit is contained in:
yucheng-berri
GitHub
commit
2fc5398270
11 changed files with 937 additions and 47 deletions
|
|
@ -35,6 +35,8 @@ from litellm.types.utils import StandardCallbackDynamicParams
|
|||
if TYPE_CHECKING:
|
||||
from opentelemetry.trace import Span as _Span
|
||||
|
||||
from litellm.proxy._types import UserAPIKeyAuth
|
||||
|
||||
Span = _Span
|
||||
else:
|
||||
Span = Any
|
||||
|
|
@ -126,6 +128,26 @@ def _starts_its_own_trace(span: object, kwargs: Mapping[str, object]) -> bool:
|
|||
)
|
||||
|
||||
|
||||
def _a_destination_records_the_call(kwargs: Mapping[str, object]) -> bool:
|
||||
"""Whether a key's or team's ``langfuse_otel`` destination gets this call from the OTel v2 fan-out, which records
|
||||
every call but one whose logging payload shows it never reached a provider."""
|
||||
from litellm.integrations.otel.model.metadata import LLMCallEvent
|
||||
|
||||
if not _a_destination_carries_the_trace():
|
||||
return False
|
||||
call: Final = LLMCallEvent.from_dict(kwargs)
|
||||
return call.payload is None or call.reached_a_provider
|
||||
|
||||
|
||||
def _a_destination_carries_the_trace() -> bool:
|
||||
"""Whether this request resolved a key's or team's ``langfuse_otel`` destination, whose fan-out sends it the whole
|
||||
trace."""
|
||||
from litellm.integrations.otel.model.config import ExporterOwner
|
||||
from litellm.integrations.otel.plumbing.context import destination_backends
|
||||
|
||||
return ExporterOwner.LANGFUSE_OTEL in destination_backends()
|
||||
|
||||
|
||||
class LangfuseOtelLogger(OpenTelemetry):
|
||||
def __init__(self, config=None, *args, **kwargs):
|
||||
# Prevent LangfuseOtelLogger from modifying global environment variables by constructing config manually
|
||||
|
|
@ -541,6 +563,27 @@ class LangfuseOtelLogger(OpenTelemetry):
|
|||
"""
|
||||
return None
|
||||
|
||||
def _emit_once(
|
||||
self,
|
||||
kwargs: dict[str, object], # mutable-ok: the v1 dedupe marker is written into the request's kwargs
|
||||
*scope: object,
|
||||
) -> bool:
|
||||
"""Whether to emit this span: once per request, and never for a call a Langfuse destination records."""
|
||||
return not _a_destination_records_the_call(kwargs) and super()._emit_once(kwargs, *scope) # pyright: ignore[reportUnknownMemberType] # the v1 hook takes a bare dict
|
||||
|
||||
async def async_post_call_failure_hook(
|
||||
self,
|
||||
request_data: dict[str, object], # mutable-ok: the v1 hook contract
|
||||
original_exception: Exception,
|
||||
user_api_key_dict: "UserAPIKeyAuth",
|
||||
traceback_str: str | None = None,
|
||||
) -> None:
|
||||
"""Record the proxy failure of a request no Langfuse destination carries."""
|
||||
if not _a_destination_carries_the_trace():
|
||||
await super().async_post_call_failure_hook( # pyright: ignore[reportUnknownMemberType] # the v1 hook takes a bare dict
|
||||
request_data, original_exception, user_api_key_dict, traceback_str
|
||||
)
|
||||
|
||||
async def async_service_success_hook(self, *args, **kwargs):
|
||||
"""
|
||||
Langfuse should not receive service success logs.
|
||||
|
|
|
|||
|
|
@ -13,9 +13,9 @@ 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.langfuse import LANGFUSE_REQUEST_METADATA_ATTRIBUTES, LangfuseMapper
|
||||
from litellm.integrations.otel.mappers.openinference import fit_indexed_messages
|
||||
from litellm.integrations.otel.model.config import OpenTelemetryV2Config
|
||||
from litellm.integrations.otel.model.config import ExporterOwner, OpenTelemetryV2Config
|
||||
from litellm.integrations.otel.model.payloads import (
|
||||
GuardrailSpanData,
|
||||
LLMCallSpanData,
|
||||
|
|
@ -34,6 +34,7 @@ from litellm.integrations.otel.model.spans import (
|
|||
mcp_tool_call_span_name,
|
||||
service_span_name,
|
||||
)
|
||||
from litellm.integrations.otel.plumbing.context import destination_backends
|
||||
from litellm.integrations.otel.plumbing.events import GenAIEventRecorder
|
||||
from litellm.integrations.otel.plumbing.providers import to_otel_span_kind
|
||||
|
||||
|
|
@ -168,6 +169,9 @@ class SpanEmitter:
|
|||
self._mappers: list[AttributeMapper] = (
|
||||
list(mappers) if mappers is not None else resolve_mappers(config.mapper_names)
|
||||
)
|
||||
self._langfuse: Final = (
|
||||
None if any(isinstance(mapper, LangfuseMapper) for mapper in self._mappers) else LangfuseMapper()
|
||||
)
|
||||
# Bounded LRU (ordered by insertion / most-recent touch). Storing keys
|
||||
# only — the value is unused — so it behaves like a capped set.
|
||||
self._emitted: OrderedDict[tuple[str, SpanRole], None] = OrderedDict()
|
||||
|
|
@ -295,17 +299,27 @@ class SpanEmitter:
|
|||
)
|
||||
else None
|
||||
)
|
||||
mapped: Final = MappingProxyType(
|
||||
{key: value for mapper in self._mappers for key, value in mapper.map(data).items()}
|
||||
configured: Final = {key: value for mapper in self._mappers for key, value in mapper.map(data).items()}
|
||||
for_langfuse: Final = (
|
||||
dict(
|
||||
sorted(
|
||||
self._langfuse.map(data).items(),
|
||||
key=lambda item: item[0] not in LANGFUSE_REQUEST_METADATA_ATTRIBUTES,
|
||||
)
|
||||
)
|
||||
if self._langfuse is not None and ExporterOwner.LANGFUSE_OTEL in destination_backends()
|
||||
else _NO_ATTRIBUTES
|
||||
)
|
||||
mapped: Final = MappingProxyType({**configured, **for_langfuse})
|
||||
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.
|
||||
# The Langfuse request metadata, and the Langfuse vocabulary only a destination asked for, shed first,
|
||||
# so a span at its attribute limit loses nothing else to them.
|
||||
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,
|
||||
LANGFUSE_REQUEST_METADATA_ATTRIBUTES.union(for_langfuse),
|
||||
)
|
||||
for key, value in mapped.items():
|
||||
if key in overwrites or key in fitted:
|
||||
|
|
|
|||
|
|
@ -541,9 +541,7 @@ class OpenTelemetryV2(CustomLogger):
|
|||
# gate rejection carries ``is_no_upstream_call`` and gets no span. A cache
|
||||
# hit never hands off to a provider, so ``cache_hit`` substitutes for the
|
||||
# upstream stamp, since the payload proves the call completed.
|
||||
if carrier is None and (
|
||||
call.is_no_upstream_call or call.payload is None or (not call.upstream_started and not call.cache_hit)
|
||||
):
|
||||
if carrier is None and not call.reached_a_provider:
|
||||
return None
|
||||
try:
|
||||
return self._finish_carrier(carrier, call, start_time, end_time)
|
||||
|
|
|
|||
|
|
@ -103,6 +103,18 @@ class LangfuseMapper:
|
|||
),
|
||||
}
|
||||
|
||||
#: Every attribute this mapper can write on a model call.
|
||||
LLM_CALL_ATTRIBUTES: Final[frozenset[str]] = frozenset(
|
||||
{
|
||||
*_LLM_CALL_ATTRS,
|
||||
*_BLOB_ATTRS,
|
||||
LANGFUSE_TRACE_NAME,
|
||||
LANGFUSE_TRACE_USER_ID,
|
||||
LANGFUSE_TRACE_SESSION_ID,
|
||||
LANGFUSE_TRACE_TAGS,
|
||||
}
|
||||
)
|
||||
|
||||
def map(self, data: SpanData) -> AttributeMap:
|
||||
match data:
|
||||
case LLMCallSpanData():
|
||||
|
|
|
|||
|
|
@ -272,6 +272,11 @@ class LLMCallEvent:
|
|||
trace: TraceControls
|
||||
session_id: str | None
|
||||
|
||||
@property
|
||||
def reached_a_provider(self) -> bool:
|
||||
"""Whether the call was handed to a provider or answered from the response cache."""
|
||||
return not self.is_no_upstream_call and self.payload is not None and (self.upstream_started or self.cache_hit)
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, kwargs: Mapping[str, object]) -> LLMCallEvent:
|
||||
raw_payload: Final = kwargs.get("standard_logging_object")
|
||||
|
|
|
|||
|
|
@ -47,6 +47,7 @@ from litellm.integrations.otel.mappers.langfuse import (
|
|||
LANGFUSE_OBSERVATION_OUTPUT,
|
||||
LANGFUSE_REQUEST_METADATA_ATTRIBUTES,
|
||||
LANGFUSE_TRACE_NAME,
|
||||
LangfuseMapper,
|
||||
)
|
||||
from litellm.integrations.otel.model.config import (
|
||||
CaptureMessageContent,
|
||||
|
|
@ -490,19 +491,25 @@ def _is_litellm_span(span: ReadableSpan) -> bool:
|
|||
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.
|
||||
def without_destination_langfuse_attributes(span: ReadableSpan, langfuse_mapped: bool) -> ReadableSpan:
|
||||
"""``span`` without the Langfuse attributes only a Langfuse destination asked for.
|
||||
|
||||
Only LiteLLM's own spans lose it: any other span keeps every key its own code set.
|
||||
That is the request metadata the Langfuse mapper writes and, on a model call written by
|
||||
a logger whose own mappers leave the Langfuse vocabulary out (``langfuse_mapped`` off),
|
||||
every Langfuse attribute, since that logger wrote them for a key's or team's Langfuse
|
||||
destination alone. Only LiteLLM's own spans lose them: 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)):
|
||||
dropped: Final = (
|
||||
LANGFUSE_REQUEST_METADATA_ATTRIBUTES
|
||||
if langfuse_mapped or not is_llm_call_span(span)
|
||||
else LangfuseMapper.LLM_CALL_ATTRIBUTES
|
||||
)
|
||||
if not (_is_litellm_span(span) and any(key in dropped 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}
|
||||
),
|
||||
span, MappingProxyType({key: value for key, value in attributes.items() if key not in dropped})
|
||||
)
|
||||
|
||||
|
||||
|
|
@ -658,6 +665,7 @@ def _for_destination(
|
|||
span: ReadableSpan,
|
||||
destination: "OtelDestination",
|
||||
default_capture: str,
|
||||
langfuse_mapped: bool,
|
||||
) -> ReadableSpan:
|
||||
"""The view of ``span`` a tenant destination receives.
|
||||
|
||||
|
|
@ -672,13 +680,16 @@ def _for_destination(
|
|||
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, 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.
|
||||
Langfuse attributes only a Langfuse destination asked for (see
|
||||
:func:`without_destination_langfuse_attributes`). The span itself stays, so the
|
||||
tenant still gets the whole trace tree.
|
||||
"""
|
||||
extra: Final = destination.resource_attributes
|
||||
defaults: Final = destination.resource_defaults
|
||||
source: Final = (
|
||||
span if destination.callback_name == ExporterOwner.LANGFUSE_OTEL else without_langfuse_request_metadata(span)
|
||||
span
|
||||
if destination.callback_name == ExporterOwner.LANGFUSE_OTEL
|
||||
else without_destination_langfuse_attributes(span, langfuse_mapped)
|
||||
)
|
||||
attributes: Final = source.attributes or _NO_ATTRIBUTES
|
||||
database: Final = _is_database_span(attributes)
|
||||
|
|
@ -728,8 +739,10 @@ class TenantFanOutSpanProcessor(SpanProcessor):
|
|||
default_capture_by_backend: Mapping[str, str] = MappingProxyType({}),
|
||||
pending_drains: int = _MAX_PENDING_DRAINS,
|
||||
drain_pool: _DrainPool | None = None,
|
||||
langfuse_mapped: bool = True,
|
||||
) -> None:
|
||||
self._operator_sinks: Final = operator_sinks
|
||||
self._langfuse_mapped: Final = langfuse_mapped
|
||||
self._excluded_db_systems: Final = excluded_db_systems
|
||||
self._default_capture: Final = default_capture
|
||||
self._default_capture_by_backend: Final = default_capture_by_backend
|
||||
|
|
@ -822,7 +835,8 @@ class TenantFanOutSpanProcessor(SpanProcessor):
|
|||
try:
|
||||
processor.on_end(
|
||||
_scoped(
|
||||
_for_destination(span, destination, self._default_capture_for(destination)), destination.span_scope
|
||||
_for_destination(span, destination, self._default_capture_for(destination), self._langfuse_mapped),
|
||||
destination.span_scope,
|
||||
)
|
||||
)
|
||||
except Exception as exc: # noqa: BLE001 # one destination's failure must not cost the others their span
|
||||
|
|
@ -1198,17 +1212,19 @@ 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."""
|
||||
class _DestinationLangfuseAttributeFilter(SpanProcessor):
|
||||
"""Strip the Langfuse attributes only a Langfuse destination asked for from LiteLLM's own spans before ``inner``
|
||||
exports them."""
|
||||
|
||||
def __init__(self, inner: SpanProcessor) -> None:
|
||||
def __init__(self, inner: SpanProcessor, langfuse_mapped: bool) -> None:
|
||||
self._inner: Final = inner
|
||||
self._langfuse_mapped: Final = langfuse_mapped
|
||||
|
||||
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))
|
||||
self._inner.on_end(without_destination_langfuse_attributes(span, self._langfuse_mapped))
|
||||
|
||||
def shutdown(self) -> None:
|
||||
self._inner.shutdown()
|
||||
|
|
@ -1491,13 +1507,18 @@ def _spec_processor(
|
|||
spec: ExporterSpec,
|
||||
use_simple_processor: bool | None,
|
||||
strip_message_content: bool,
|
||||
strip_langfuse_request_metadata: bool,
|
||||
strip_destination_langfuse_attributes: bool,
|
||||
langfuse_mapped: 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),
|
||||
)
|
||||
scoped: Final = _LangfuseRequestMetadataFilter(exporting) if strip_langfuse_request_metadata else exporting
|
||||
scoped: Final = (
|
||||
_DestinationLangfuseAttributeFilter(exporting, langfuse_mapped)
|
||||
if strip_destination_langfuse_attributes
|
||||
else exporting
|
||||
)
|
||||
return _MessageContentFilter(scoped) if strip_message_content else scoped
|
||||
|
||||
|
||||
|
|
@ -1548,8 +1569,9 @@ 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.
|
||||
Every exporter but ``langfuse_otel``'s goes without the Langfuse attributes only a
|
||||
Langfuse destination asked for (see :func:`without_destination_langfuse_attributes`),
|
||||
which any logger writes for a request with a Langfuse destination.
|
||||
``strip_message_content`` removes prompt and response content before every
|
||||
exporter, for a routed provider whose spans can carry content another
|
||||
destination asked for.
|
||||
|
|
@ -1564,11 +1586,9 @@ def build_tracer_provider(
|
|||
provider.add_span_processor(_processor_for(exporter, use_simple_processor))
|
||||
return provider
|
||||
|
||||
langfuse_mapped: Final = _maps_langfuse(config)
|
||||
# ``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
|
||||
|
|
@ -1576,7 +1596,8 @@ def build_tracer_provider(
|
|||
spec,
|
||||
use_simple_processor,
|
||||
strip_message_content,
|
||||
strip_langfuse_request_metadata=writes_langfuse and spec.owner is not ExporterOwner.LANGFUSE_OTEL,
|
||||
strip_destination_langfuse_attributes=spec.owner is not ExporterOwner.LANGFUSE_OTEL,
|
||||
langfuse_mapped=langfuse_mapped,
|
||||
)
|
||||
owner = spec.owner.value if tenant_overrides and spec.owner is not None else None
|
||||
scope = _operator_scope(config, spec)
|
||||
|
|
@ -1613,8 +1634,9 @@ def attach_tenant_fan_out(
|
|||
so exactly one fan-out lands. ``configs`` name the operator's own exporters, one
|
||||
config per v2 logger since each keeps its own provider and still writes its
|
||||
account, so an additive destination pointing at any of them is delivered once
|
||||
rather than twice. ``excluded_db_systems`` only filters what the fan-out
|
||||
delivers, never the operator's own exporters.
|
||||
rather than twice. The first is the config of the logger that owns ``provider``,
|
||||
whose model calls the fan-out carries. ``excluded_db_systems`` only filters what
|
||||
the fan-out delivers, never the operator's own exporters.
|
||||
"""
|
||||
with _FAN_OUT_ATTACH_LOCK:
|
||||
if any(isinstance(processor, TenantFanOutSpanProcessor) for processor in _attached_processors(provider)):
|
||||
|
|
@ -1625,10 +1647,16 @@ def attach_tenant_fan_out(
|
|||
excluded_db_systems=excluded_db_systems,
|
||||
default_capture=_operator_capture_mode(configs),
|
||||
default_capture_by_backend=MappingProxyType(dict(_owned_capture_modes(configs))),
|
||||
langfuse_mapped=any(_maps_langfuse(config) for config in configs[:1]),
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def _maps_langfuse(config: OpenTelemetryV2Config) -> bool:
|
||||
"""Whether a logger built from ``config`` writes the Langfuse vocabulary on every model call."""
|
||||
return "langfuse" in config.mapper_names
|
||||
|
||||
|
||||
def _operator_capture_mode(configs: Sequence[OpenTelemetryV2Config]) -> str:
|
||||
"""The mode of the operator's generic ``otel`` callback, so ``callback_settings.otel`` counts as the global default"""
|
||||
return next(
|
||||
|
|
|
|||
|
|
@ -15,7 +15,7 @@ import litellm
|
|||
from litellm._logging import verbose_logger
|
||||
from litellm.integrations.otel.model.destination import OtelDestination
|
||||
from litellm.litellm_core_utils.url_utils import is_url_destination_allowed_by_host
|
||||
from litellm.types.utils import OtelSpanScope, StandardCallbackDynamicParams
|
||||
from litellm.types.utils import CaptureMessageContent, OtelSpanScope, StandardCallbackDynamicParams
|
||||
|
||||
#: An endpoint plus the OTLP transport to reach it with, or ``None`` when the backend
|
||||
#: names no destination. The transport is ``None`` where the backend has only one.
|
||||
|
|
@ -111,7 +111,7 @@ _REQUIRED_HEADERS_BY_CALLBACK: Final[Mapping[str, frozenset[str]]] = MappingProx
|
|||
_NO_ATTRS: Final[Mapping[str, str]] = MappingProxyType({})
|
||||
|
||||
|
||||
def _arize_project() -> Mapping[str, str]:
|
||||
def _arize_project(_params: StandardCallbackDynamicParams) -> Mapping[str, str]:
|
||||
from litellm.integrations.arize.arize import ArizeLogger
|
||||
|
||||
project: Final = ArizeLogger.get_arize_config().project_name
|
||||
|
|
@ -124,10 +124,27 @@ def _arize_default_project() -> Mapping[str, str]:
|
|||
return MappingProxyType({"model_id": ARIZE_DEFAULT_PROJECT})
|
||||
|
||||
|
||||
#: Resource attributes the operator chose for a backend, applied to every export the
|
||||
#: team receives over the span's own resource: ``ARIZE_PROJECT_NAME`` for Arize.
|
||||
_RESOURCE_ATTRIBUTES_BY_CALLBACK: Final[Mapping[str, Callable[[], Mapping[str, str]]]] = MappingProxyType(
|
||||
{"arize": _arize_project}
|
||||
def _stands_in_for_the_legacy_langfuse_logger() -> bool:
|
||||
"""Whether a Langfuse destination replaces the older Langfuse logger's copy, which a proxy without Langfuse keys
|
||||
of its own sends for a key's or team's ``langfuse_otel``."""
|
||||
return not (os.environ.get("LANGFUSE_PUBLIC_KEY") and os.environ.get("LANGFUSE_SECRET_KEY"))
|
||||
|
||||
|
||||
def _langfuse_environment(params: StandardCallbackDynamicParams) -> Mapping[str, str]:
|
||||
"""The key's or team's ``langfuse_environment``, else ``LANGFUSE_TRACING_ENVIRONMENT``, where the older Langfuse
|
||||
logger wrote it."""
|
||||
from litellm.types.integrations.langfuse_otel import LangfuseSpanAttributes
|
||||
|
||||
if not _stands_in_for_the_legacy_langfuse_logger():
|
||||
return _NO_ATTRS
|
||||
environment: Final = params.get("langfuse_environment") or os.environ.get("LANGFUSE_TRACING_ENVIRONMENT")
|
||||
return MappingProxyType({LangfuseSpanAttributes.LANGFUSE_ENVIRONMENT.value: environment} if environment else {})
|
||||
|
||||
|
||||
#: Resource attributes a backend gets on every export the team receives, over the
|
||||
#: span's own resource: ``ARIZE_PROJECT_NAME`` for Arize, the Langfuse environment.
|
||||
_RESOURCE_ATTRIBUTES_BY_CALLBACK: Final[Mapping[str, Callable[[StandardCallbackDynamicParams], Mapping[str, str]]]] = (
|
||||
MappingProxyType({"arize": _arize_project, "langfuse_otel": _langfuse_environment})
|
||||
)
|
||||
|
||||
#: Resource attributes a backend needs on every export to accept it, filled only where
|
||||
|
|
@ -137,12 +154,14 @@ _RESOURCE_DEFAULTS_BY_CALLBACK: Final[Mapping[str, Callable[[], Mapping[str, str
|
|||
)
|
||||
|
||||
|
||||
def _resource_attributes(callback_name: str, service_name: str | None) -> Mapping[str, str]:
|
||||
def _resource_attributes(
|
||||
callback_name: str, service_name: str | None, params: StandardCallbackDynamicParams
|
||||
) -> Mapping[str, str]:
|
||||
backend: Final = _RESOURCE_ATTRIBUTES_BY_CALLBACK.get(callback_name)
|
||||
return MappingProxyType(
|
||||
{
|
||||
**({"service.name": service_name} if service_name else {}),
|
||||
**(backend() if backend is not None else {}),
|
||||
**(backend(params) if backend is not None else {}),
|
||||
}
|
||||
)
|
||||
|
||||
|
|
@ -152,6 +171,14 @@ def _resource_defaults(callback_name: str) -> Mapping[str, str]:
|
|||
return backend() if backend is not None else _NO_ATTRS
|
||||
|
||||
|
||||
def _capture_message_content(callback_name: str, configured: str | None) -> str | None:
|
||||
"""The key's or team's own setting, else the prompt and response where the older Langfuse logger sent them
|
||||
whatever the global setting."""
|
||||
if configured is not None or callback_name != "langfuse_otel" or not _stands_in_for_the_legacy_langfuse_logger():
|
||||
return configured
|
||||
return CaptureMessageContent.SPAN_ONLY
|
||||
|
||||
|
||||
def _span_scope(callback_name: str, params: StandardCallbackDynamicParams) -> OtelSpanScope:
|
||||
if callback_name != "langfuse_otel":
|
||||
return "full"
|
||||
|
|
@ -214,12 +241,12 @@ def destination_for(
|
|||
return OtelDestination(
|
||||
endpoint=endpoint,
|
||||
headers=MappingProxyType(dict(headers)),
|
||||
resource_attributes=_resource_attributes(callback_name, service_name),
|
||||
resource_attributes=_resource_attributes(callback_name, service_name, params),
|
||||
resource_defaults=_resource_defaults(callback_name),
|
||||
callback_name=callback_name,
|
||||
protocol=protocol,
|
||||
span_scope=_span_scope(callback_name, params),
|
||||
capture_message_content=capture_message_content,
|
||||
capture_message_content=_capture_message_content(callback_name, capture_message_content),
|
||||
excluded_db_systems=excluded_db_systems,
|
||||
success_sampling_rate=success_sampling_rate,
|
||||
error_sampling_rate=error_sampling_rate,
|
||||
|
|
|
|||
|
|
@ -0,0 +1,224 @@
|
|||
"""Without operator Langfuse keys on OTel v2, Langfuse logging the OTel v2 fan-out does not carry still reaches the
|
||||
tenant's own Langfuse project: a key's failure-only entry on a failed call, a config ``default_team_settings`` entry,
|
||||
and a key destination's call the proxy rejected before any provider. A ``success_and_failure`` key entry sends that
|
||||
project exactly one copy of a failed call."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import uuid
|
||||
from collections.abc import Callable, Iterator
|
||||
from typing import Final, Literal
|
||||
|
||||
import pytest
|
||||
from integration._support.client import Gateway, gateway_from_environment, held
|
||||
from integration._support.otlp_sink import (
|
||||
AuditConfigWriter,
|
||||
Span,
|
||||
SpanSinks,
|
||||
langfuse_callback_variables,
|
||||
recorded_spans,
|
||||
spans_for_trace,
|
||||
)
|
||||
from integration._support.process import owned_proxy_process
|
||||
from integration._support.wire import Reply, Request, Wire, wire_server
|
||||
from pydantic import JsonValue
|
||||
|
||||
OPERATION: Final = "gen_ai.operation.name"
|
||||
TAGS: Final = ["agent:orders", "run:r1"]
|
||||
CONFIG_TEAM: Final = f"lf-config-team-{uuid.uuid4().hex[:8]}"
|
||||
FAIL: Final = "please-fail"
|
||||
|
||||
|
||||
def _chat(request: Request) -> Reply:
|
||||
if FAIL.encode() in request.body:
|
||||
return Reply(
|
||||
status=500, body=json.dumps({"error": {"message": "upstream boom", "type": "server_error"}}).encode()
|
||||
)
|
||||
return Reply(
|
||||
body=json.dumps(
|
||||
{
|
||||
"id": f"chatcmpl-{uuid.uuid4().hex}",
|
||||
"object": "chat.completion",
|
||||
"created": 1,
|
||||
"model": "gpt-4o-mini",
|
||||
"choices": [{"index": 0, "message": {"role": "assistant", "content": "ok"}, "finish_reason": "stop"}],
|
||||
"usage": {"prompt_tokens": 7, "completion_tokens": 2, "total_tokens": 9},
|
||||
}
|
||||
).encode()
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def provider() -> Iterator[Wire]:
|
||||
with wire_server(_chat) as wire:
|
||||
yield wire
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def proxy(
|
||||
audit_sinks: SpanSinks,
|
||||
otel_audit_config: AuditConfigWriter,
|
||||
tmp_path_factory: pytest.TempPathFactory,
|
||||
) -> Iterator[Gateway]:
|
||||
directory: Final = tmp_path_factory.mktemp("langfuse-legacy-destinations")
|
||||
settings: Final[dict[str, JsonValue]] = {
|
||||
"callbacks": [],
|
||||
"default_team_settings": [
|
||||
{
|
||||
"team_id": CONFIG_TEAM,
|
||||
"success_callback": ["langfuse_otel"],
|
||||
"callback_vars": dict(langfuse_callback_variables(audit_sinks)),
|
||||
}
|
||||
],
|
||||
}
|
||||
with (
|
||||
gateway_from_environment() as gateway,
|
||||
owned_proxy_process(
|
||||
gateway,
|
||||
directory,
|
||||
{
|
||||
"LITELLM_OTEL_V2": "1",
|
||||
"OTEL_BSP_SCHEDULE_DELAY": "300",
|
||||
"OTEL_EXPORTER": "otlp_http",
|
||||
"OTEL_ENDPOINT": audit_sinks.operator,
|
||||
},
|
||||
config=otel_audit_config(directory, settings),
|
||||
remove_environment=("LANGFUSE_PUBLIC_KEY", "LANGFUSE_SECRET_KEY", "LANGFUSE_HOST"),
|
||||
workers=2,
|
||||
) as owned,
|
||||
):
|
||||
yield owned.gateway
|
||||
|
||||
|
||||
def _tenant_spans(url: str, since: int, trace_id: str) -> tuple[Span, ...]:
|
||||
return spans_for_trace(recorded_spans(url, since)[1], trace_id)
|
||||
|
||||
|
||||
def _tags(value: object) -> object:
|
||||
"""The tags a span carries: the legacy logger writes them as a JSON string, OTel v2 as a list."""
|
||||
return json.loads(value) if isinstance(value, str) else value
|
||||
|
||||
|
||||
def _settled(spans: tuple[Span, ...]) -> bool:
|
||||
"""The model span and the request's root span have both arrived."""
|
||||
return any(OPERATION in span["attributes"] for span in spans) and any(
|
||||
str(span["name"]).startswith("POST ") for span in spans
|
||||
)
|
||||
|
||||
|
||||
def _logging_entry(audit_sinks: SpanSinks, callback_type: str) -> list[JsonValue]:
|
||||
return [
|
||||
{
|
||||
"callback_name": "langfuse_otel",
|
||||
"callback_type": callback_type,
|
||||
"callback_vars": langfuse_callback_variables(audit_sinks),
|
||||
}
|
||||
]
|
||||
|
||||
|
||||
def _chat_status(proxy: Gateway, model: str, key: str, trace_id: str, content: str) -> int:
|
||||
body: Final[dict[str, JsonValue]] = {
|
||||
"model": model,
|
||||
"messages": [{"role": "user", "content": content}],
|
||||
"metadata": {"tags": [*TAGS], "session_id": f"sess-{trace_id}", "trace_user_id": "user-42"},
|
||||
}
|
||||
return proxy.request(
|
||||
"POST",
|
||||
"/v1/chat/completions",
|
||||
body,
|
||||
key=key,
|
||||
headers={"traceparent": f"00-{trace_id}-{uuid.uuid4().hex[:16]}-01"},
|
||||
).status_code
|
||||
|
||||
|
||||
def _trace_fields(spans: tuple[Span, ...]) -> list[tuple[object, object]]:
|
||||
return [
|
||||
(span["attributes"]["user.id"], _tags(span["attributes"].get("langfuse.trace.tags")))
|
||||
for span in spans
|
||||
if "user.id" in span["attributes"]
|
||||
]
|
||||
|
||||
|
||||
def _carries_the_trace_fields_once(spans: tuple[Span, ...]) -> bool:
|
||||
return _trace_fields(spans) == [("user-42", TAGS)]
|
||||
|
||||
|
||||
def _one_failed_call(spans: tuple[Span, ...]) -> bool:
|
||||
return (
|
||||
_settled(spans)
|
||||
and len([span for span in spans if OPERATION in span["attributes"]]) == 1
|
||||
and "litellm_request" not in {span["name"] for span in spans}
|
||||
)
|
||||
|
||||
|
||||
def _send(
|
||||
proxy: Gateway,
|
||||
provider: Wire,
|
||||
audit_sinks: SpanSinks,
|
||||
setup: Literal["failure", "config_team", "both"],
|
||||
expected: Callable[[tuple[Span, ...]], bool],
|
||||
) -> tuple[int, tuple[Span, ...]]:
|
||||
logging_entry: Final = _logging_entry(audit_sinks, "failure" if setup == "failure" else "success_and_failure")
|
||||
with proxy.scenario() as scenario:
|
||||
model: Final = scenario.model(api_base=provider.url + "/v1", num_retries=0)
|
||||
alias: Final = f"lf-legacy-{setup}-{uuid.uuid4().hex}"
|
||||
key: Final = (
|
||||
scenario.key(key_alias=alias, team_id=scenario.team(team_id=CONFIG_TEAM))
|
||||
if setup == "config_team"
|
||||
else scenario.key(key_alias=alias, metadata={"logging": logging_entry})
|
||||
)
|
||||
since: Final = recorded_spans(audit_sinks.tenant)[0]
|
||||
trace_id: Final = uuid.uuid4().hex
|
||||
content: Final = f"lf-legacy-{setup}-{trace_id}" + ("" if setup == "config_team" else f" {FAIL}")
|
||||
status: Final = _chat_status(proxy, model, key, trace_id, content)
|
||||
tenant: Final = held(
|
||||
lambda: _tenant_spans(audit_sinks.tenant, since, trace_id), expected, holding=3, seconds=40
|
||||
)
|
||||
return status, tenant
|
||||
|
||||
|
||||
@pytest.mark.parametrize("setup", ["failure", "config_team"])
|
||||
def test_a_tenant_langfuse_project_without_a_destination_still_gets_the_trace(
|
||||
setup: Literal["failure", "config_team"], proxy: Gateway, provider: Wire, audit_sinks: SpanSinks
|
||||
) -> None:
|
||||
status, tenant = _send(proxy, provider, audit_sinks, setup, _carries_the_trace_fields_once)
|
||||
|
||||
assert status == (500 if setup == "failure" else 200)
|
||||
assert _trace_fields(tenant) == [("user-42", TAGS)], [span["name"] for span in tenant]
|
||||
|
||||
|
||||
def test_a_key_destinations_call_rejected_before_any_provider_still_gets_the_trace_fields(
|
||||
proxy: Gateway, provider: Wire, audit_sinks: SpanSinks
|
||||
) -> None:
|
||||
with proxy.scenario() as scenario:
|
||||
model: Final = scenario.model(api_base=provider.url + "/v1", num_retries=0)
|
||||
key: Final = scenario.key(
|
||||
key_alias=f"lf-legacy-rejected-{uuid.uuid4().hex}",
|
||||
metadata={"logging": _logging_entry(audit_sinks, "success_and_failure")},
|
||||
rpm_limit=1,
|
||||
)
|
||||
since: Final = recorded_spans(audit_sinks.tenant)[0]
|
||||
statuses: Final = {
|
||||
trace_id: _chat_status(proxy, model, key, trace_id, f"lf-legacy-rejected-{trace_id}")
|
||||
for trace_id in [uuid.uuid4().hex for _ in range(4)]
|
||||
}
|
||||
rejected: Final = [trace_id for trace_id, status in statuses.items() if status == 429]
|
||||
assert rejected, statuses
|
||||
tenant: Final = held(
|
||||
lambda: _tenant_spans(audit_sinks.tenant, since, rejected[0]),
|
||||
_carries_the_trace_fields_once,
|
||||
holding=3,
|
||||
seconds=40,
|
||||
)
|
||||
|
||||
assert _trace_fields(tenant) == [("user-42", TAGS)], [span["name"] for span in tenant]
|
||||
|
||||
|
||||
def test_a_failed_call_reaches_a_success_and_failure_destination_once(
|
||||
proxy: Gateway, provider: Wire, audit_sinks: SpanSinks
|
||||
) -> None:
|
||||
status, tenant = _send(proxy, provider, audit_sinks, "both", _one_failed_call)
|
||||
|
||||
assert status == 500
|
||||
assert _one_failed_call(tenant), [span["name"] for span in tenant]
|
||||
|
|
@ -0,0 +1,182 @@
|
|||
"""A key's or team's own Langfuse project gets one copy of the call with the Langfuse trace fields when the operator
|
||||
has no Langfuse keys, whether or not ``langfuse_otel`` is listed in ``litellm_settings.callbacks``."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import uuid
|
||||
from collections.abc import Iterator
|
||||
from typing import Final, Literal, cast
|
||||
|
||||
import pytest
|
||||
from integration._support.client import Gateway, gateway_from_environment, held
|
||||
from integration._support.otlp_sink import (
|
||||
AuditConfigWriter,
|
||||
Span,
|
||||
SpanSinks,
|
||||
langfuse_callback_variables,
|
||||
recorded_spans,
|
||||
spans_for_trace,
|
||||
)
|
||||
from integration._support.process import owned_proxy_process
|
||||
from integration._support.wire import Reply, Request, Wire, wire_server
|
||||
from pydantic import JsonValue
|
||||
|
||||
OPERATION: Final = "gen_ai.operation.name"
|
||||
TAGS: Final = ["agent:orders", "run:r1"]
|
||||
LEGACY_SPANS: Final = frozenset({"litellm_request", "raw_gen_ai_request"})
|
||||
|
||||
|
||||
def _chat(_request: Request) -> Reply:
|
||||
return Reply(
|
||||
body=json.dumps(
|
||||
{
|
||||
"id": f"chatcmpl-{uuid.uuid4().hex}",
|
||||
"object": "chat.completion",
|
||||
"created": 1,
|
||||
"model": "gpt-4o-mini",
|
||||
"choices": [{"index": 0, "message": {"role": "assistant", "content": "ok"}, "finish_reason": "stop"}],
|
||||
"usage": {"prompt_tokens": 7, "completion_tokens": 2, "total_tokens": 9},
|
||||
}
|
||||
).encode()
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def provider() -> Iterator[Wire]:
|
||||
with wire_server(_chat) as wire:
|
||||
yield wire
|
||||
|
||||
|
||||
@pytest.fixture(scope="module", params=[["langfuse_otel"], []], ids=["listed", "not-listed"])
|
||||
def proxy(
|
||||
request: pytest.FixtureRequest,
|
||||
audit_sinks: SpanSinks,
|
||||
otel_audit_config: AuditConfigWriter,
|
||||
tmp_path_factory: pytest.TempPathFactory,
|
||||
) -> Iterator[Gateway]:
|
||||
directory: Final = tmp_path_factory.mktemp("langfuse-tenant-only")
|
||||
with (
|
||||
gateway_from_environment() as gateway,
|
||||
owned_proxy_process(
|
||||
gateway,
|
||||
directory,
|
||||
{
|
||||
"LITELLM_OTEL_V2": "1",
|
||||
"OTEL_BSP_SCHEDULE_DELAY": "300",
|
||||
"OTEL_EXPORTER": "otlp_http",
|
||||
"OTEL_ENDPOINT": audit_sinks.operator,
|
||||
},
|
||||
config=otel_audit_config(directory, {"callbacks": cast("list[JsonValue]", request.param)}),
|
||||
remove_environment=("LANGFUSE_PUBLIC_KEY", "LANGFUSE_SECRET_KEY", "LANGFUSE_HOST"),
|
||||
workers=2,
|
||||
) as owned,
|
||||
):
|
||||
yield owned.gateway
|
||||
|
||||
|
||||
def _trace_spans(url: str, since: int, trace_id: str) -> tuple[Span, ...]:
|
||||
return spans_for_trace(recorded_spans(url, since)[1], trace_id)
|
||||
|
||||
|
||||
def _model_spans(spans: tuple[Span, ...]) -> tuple[Span, ...]:
|
||||
return tuple(span for span in spans if OPERATION in span["attributes"])
|
||||
|
||||
|
||||
def _one_model_span(spans: tuple[Span, ...]) -> bool:
|
||||
return len(_model_spans(spans)) == 1 and not LEGACY_SPANS & {str(span["name"]) for span in spans}
|
||||
|
||||
|
||||
@pytest.mark.parametrize("level", ["key", "team"])
|
||||
def test_a_tenant_langfuse_project_gets_user_session_and_tags_without_operator_keys(
|
||||
level: Literal["key", "team"], proxy: Gateway, provider: Wire, audit_sinks: SpanSinks
|
||||
) -> None:
|
||||
logging: Final[list[JsonValue]] = [
|
||||
{
|
||||
"callback_name": "langfuse_otel",
|
||||
"callback_type": "success",
|
||||
"callback_vars": langfuse_callback_variables(audit_sinks),
|
||||
}
|
||||
]
|
||||
with proxy.scenario() as scenario:
|
||||
model: Final = scenario.model(api_base=provider.url + "/v1")
|
||||
alias: Final = f"lf-tenant-only-{uuid.uuid4().hex}"
|
||||
key: Final = (
|
||||
scenario.key(key_alias=alias, metadata={"logging": logging})
|
||||
if level == "key"
|
||||
else scenario.key(key_alias=alias, team_id=scenario.team(metadata={"logging": logging}))
|
||||
)
|
||||
since: Final = (recorded_spans(audit_sinks.tenant)[0], recorded_spans(audit_sinks.operator)[0])
|
||||
trace_id: Final = uuid.uuid4().hex
|
||||
session: Final = f"sess-{trace_id}"
|
||||
body: Final[dict[str, JsonValue]] = {
|
||||
"model": model,
|
||||
"messages": [{"role": "user", "content": f"lf-tenant-only-{trace_id}"}],
|
||||
"metadata": {"tags": [*TAGS], "session_id": session, "trace_user_id": "user-42"},
|
||||
}
|
||||
response: Final = proxy.request(
|
||||
"POST",
|
||||
"/v1/chat/completions",
|
||||
body,
|
||||
key=key,
|
||||
headers={"traceparent": f"00-{trace_id}-{uuid.uuid4().hex[:16]}-01"},
|
||||
)
|
||||
assert response.status_code == 200, response.text
|
||||
|
||||
tenant: Final = _model_spans(
|
||||
held(lambda: _trace_spans(audit_sinks.tenant, since[0], trace_id), _one_model_span, holding=3, seconds=40)
|
||||
)
|
||||
operator: Final = _model_spans(
|
||||
held(lambda: _trace_spans(audit_sinks.operator, since[1], trace_id), _one_model_span, holding=3, seconds=40)
|
||||
)
|
||||
|
||||
assert [
|
||||
{name: span["attributes"].get(name) for name in ("user.id", "session.id", "langfuse.trace.tags")}
|
||||
for span in tenant
|
||||
] == [{"user.id": "user-42", "session.id": session, "langfuse.trace.tags": TAGS}], [s["name"] for s in tenant]
|
||||
assert "langfuse.observation.metadata" in tenant[0]["attributes"], sorted(tenant[0]["attributes"])
|
||||
assert f"lf-tenant-only-{trace_id}" in str(tenant[0]["attributes"].get("langfuse.observation.input")), sorted(
|
||||
tenant[0]["attributes"]
|
||||
)
|
||||
assert len(operator) == 1, [span["name"] for span in operator]
|
||||
assert f"lf-tenant-only-{trace_id}" not in json.dumps(operator[0]["attributes"]), sorted(operator[0]["attributes"])
|
||||
assert [
|
||||
name for name in operator[0]["attributes"] if name in ("user.id", "session.id") or name.startswith("langfuse.")
|
||||
] == [], sorted(operator[0]["attributes"])
|
||||
|
||||
|
||||
def test_a_tenant_langfuse_project_gets_the_keys_langfuse_environment(
|
||||
proxy: Gateway, provider: Wire, audit_sinks: SpanSinks
|
||||
) -> None:
|
||||
logging: Final[list[JsonValue]] = [
|
||||
{
|
||||
"callback_name": "langfuse_otel",
|
||||
"callback_type": "success",
|
||||
"callback_vars": {**langfuse_callback_variables(audit_sinks), "langfuse_environment": "prod"},
|
||||
}
|
||||
]
|
||||
with proxy.scenario() as scenario:
|
||||
model: Final = scenario.model(api_base=provider.url + "/v1")
|
||||
key: Final = scenario.key(key_alias=f"lf-tenant-env-{uuid.uuid4().hex}", metadata={"logging": logging})
|
||||
since: Final = (recorded_spans(audit_sinks.tenant)[0], recorded_spans(audit_sinks.operator)[0])
|
||||
trace_id: Final = uuid.uuid4().hex
|
||||
response: Final = proxy.request(
|
||||
"POST",
|
||||
"/v1/chat/completions",
|
||||
{"model": model, "messages": [{"role": "user", "content": f"lf-tenant-env-{trace_id}"}]},
|
||||
key=key,
|
||||
headers={"traceparent": f"00-{trace_id}-{uuid.uuid4().hex[:16]}-01"},
|
||||
)
|
||||
assert response.status_code == 200, response.text
|
||||
|
||||
tenant: Final = held(
|
||||
lambda: _trace_spans(audit_sinks.tenant, since[0], trace_id), _one_model_span, holding=3, seconds=40
|
||||
)
|
||||
operator: Final = held(
|
||||
lambda: _trace_spans(audit_sinks.operator, since[1], trace_id), _one_model_span, holding=3, seconds=40
|
||||
)
|
||||
|
||||
assert [span["resource"].get("langfuse.environment") for span in tenant] == ["prod"] * len(tenant), [
|
||||
span["name"] for span in tenant
|
||||
]
|
||||
assert not any("langfuse.environment" in span["resource"] for span in operator), [span["name"] for span in operator]
|
||||
|
|
@ -17,7 +17,7 @@ import pytest
|
|||
from opentelemetry import trace as trace_api
|
||||
from opentelemetry.context import Context
|
||||
from opentelemetry.sdk.resources import Resource
|
||||
from opentelemetry.sdk.trace import ReadableSpan, TracerProvider
|
||||
from opentelemetry.sdk.trace import ReadableSpan, SpanLimits, TracerProvider
|
||||
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
|
||||
from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
|
||||
from opentelemetry.sdk.trace.id_generator import RandomIdGenerator
|
||||
|
|
@ -27,6 +27,7 @@ import litellm
|
|||
from litellm.integrations.custom_logger import CustomLogger
|
||||
from litellm.integrations.otel import GenAIOperation
|
||||
from litellm.integrations.otel import logger as otel_logger
|
||||
from litellm.integrations.otel.emitter import SpanEmitter
|
||||
from litellm.integrations.otel.logger import (
|
||||
OpenTelemetryV2,
|
||||
build_otel_v2_logger,
|
||||
|
|
@ -34,6 +35,7 @@ from litellm.integrations.otel.logger import (
|
|||
publish_global_otel_v2_provider,
|
||||
)
|
||||
from litellm.integrations.otel.mappers import resolve_mappers
|
||||
from litellm.integrations.otel.mappers.langfuse import LANGFUSE_REQUEST_METADATA_ATTRIBUTES, LangfuseMapper
|
||||
from litellm.integrations.otel.model.config import (
|
||||
CaptureMessageContent,
|
||||
ExporterOwner,
|
||||
|
|
@ -52,6 +54,8 @@ from litellm.integrations.otel.model.payloads import (
|
|||
ServerInfo,
|
||||
ToolDefinition,
|
||||
)
|
||||
from litellm.integrations.otel.model.spans import SpanRole
|
||||
from litellm.integrations.otel.model.trace_controls import TraceControls
|
||||
from litellm.integrations.otel.plumbing import providers as otel_providers
|
||||
from litellm.integrations.otel.plumbing.context import (
|
||||
destination_backends,
|
||||
|
|
@ -3435,6 +3439,276 @@ class TestArizeTenantOnly:
|
|||
assert loggers == [at_startup]
|
||||
|
||||
|
||||
class TestLangfuseDestinationFields:
|
||||
"""A key's or team's Langfuse destination gets the Langfuse fields from whichever logger the proxy published."""
|
||||
|
||||
def test_a_key_destination_gets_the_langfuse_fields_from_the_generic_startup_logger(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""With ``langfuse_otel`` only in a key's or team's logging, the startup logger is the
|
||||
generic one; the key's project still gets the user, session and tags, and the
|
||||
operator's stdout placeholder still gets no Langfuse request metadata."""
|
||||
credential_less_proxy(monkeypatch)
|
||||
capture: Final = _ExporterCapture()
|
||||
for kind in ("otlp_grpc", "otlp_http", "console"):
|
||||
monkeypatch.setitem(otel_providers._EXPORTER_FACTORIES, kind, capture.build) # pyright: ignore[reportPrivateUsage] # the factory table is the seam for capturing each exporter
|
||||
config: Final = OpenTelemetryV2Config()
|
||||
provider: Final = build_tracer_provider(config, tenant_overrides=True)
|
||||
attach_tenant_fan_out(provider, config)
|
||||
emitter: Final = SpanEmitter(get_tracer(provider, "litellm"), config)
|
||||
call: Final = replace(
|
||||
_MODEL_CALL_WITHOUT_CONTENT,
|
||||
trace=TraceControls(user_id="user-42", session_id="sess-1", tags=("agent:orders",)),
|
||||
request_metadata=MappingProxyType({"user_api_key_alias": "prod-key"}),
|
||||
)
|
||||
|
||||
def with_destination() -> None:
|
||||
set_request_destinations(deliverable_destinations((LANGFUSE_DEST,), provider))
|
||||
emitter.emit(SpanRole.LLM_CALL, call)
|
||||
|
||||
in_fresh_context(with_destination)
|
||||
provider.force_flush()
|
||||
|
||||
received: Final = {
|
||||
headers: exporter.get_finished_spans()[0].attributes or {} for headers, exporter in capture.built
|
||||
}
|
||||
to_langfuse: Final = received[LANGFUSE_DEST.header_string()]
|
||||
assert {key: to_langfuse.get(key) for key in ("user.id", "session.id", "langfuse.trace.tags")} == {
|
||||
"user.id": "user-42",
|
||||
"session.id": "sess-1",
|
||||
"langfuse.trace.tags": ("agent:orders",),
|
||||
}
|
||||
assert "langfuse.observation.metadata" in to_langfuse
|
||||
assert "langfuse.observation.metadata" not in received[None]
|
||||
|
||||
@staticmethod
|
||||
def _received_by_account(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
config: OpenTelemetryV2Config,
|
||||
destinations: tuple[OtelDestination, ...],
|
||||
with_content: bool = False,
|
||||
) -> Mapping[str | None, Mapping[str, object]]:
|
||||
"""The attributes of the model call each account received, keyed by the headers it was sent with."""
|
||||
credential_less_proxy(monkeypatch)
|
||||
capture: Final = _ExporterCapture()
|
||||
for kind in ("otlp_grpc", "otlp_http", "console"):
|
||||
monkeypatch.setitem(otel_providers._EXPORTER_FACTORIES, kind, capture.build) # pyright: ignore[reportPrivateUsage] # the factory table is the seam for capturing each exporter
|
||||
provider: Final = build_tracer_provider(config, tenant_overrides=True)
|
||||
attach_tenant_fan_out(provider, config)
|
||||
emitter: Final = SpanEmitter(get_tracer(provider, "litellm"), config)
|
||||
call: Final = replace(
|
||||
_MODEL_CALL_WITH_CONTENT if with_content else _MODEL_CALL_WITHOUT_CONTENT,
|
||||
trace=TraceControls(user_id="user-42", session_id="sess-1", tags=("agent:orders",)),
|
||||
request_metadata=MappingProxyType({"user_api_key_alias": "prod-key"}),
|
||||
)
|
||||
|
||||
def with_destinations() -> None:
|
||||
set_request_destinations(deliverable_destinations(destinations, provider))
|
||||
emitter.emit(SpanRole.LLM_CALL, call)
|
||||
|
||||
in_fresh_context(with_destinations)
|
||||
provider.force_flush()
|
||||
return MappingProxyType(
|
||||
{headers: dict(exporter.get_finished_spans()[0].attributes or {}) for headers, exporter in capture.built}
|
||||
)
|
||||
|
||||
def test_only_the_langfuse_destination_gets_the_langfuse_fields_written_for_it(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""The operator's own exporter and the key's Arize destination get the model call as they would with no
|
||||
Langfuse destination on the request."""
|
||||
received: Final = self._received_by_account(monkeypatch, OpenTelemetryV2Config(), (LANGFUSE_DEST, ARIZE_DEST))
|
||||
|
||||
assert received[LANGFUSE_DEST.header_string()]["user.id"] == "user-42"
|
||||
assert {
|
||||
headers: sorted(LangfuseMapper.LLM_CALL_ATTRIBUTES & attributes.keys())
|
||||
for headers, attributes in received.items()
|
||||
if headers != LANGFUSE_DEST.header_string()
|
||||
} == {None: [], ARIZE_DEST.header_string(): []}
|
||||
|
||||
def test_an_exporter_whose_config_maps_langfuse_keeps_the_langfuse_fields_but_the_request_metadata(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
received: Final = self._received_by_account(
|
||||
monkeypatch, OpenTelemetryV2Config(mapper_names=["genai", "langfuse"]), (LANGFUSE_DEST,)
|
||||
)
|
||||
|
||||
assert {key: received[None].get(key) for key in ("user.id", "session.id", "langfuse.trace.tags")} == {
|
||||
"user.id": "user-42",
|
||||
"session.id": "sess-1",
|
||||
"langfuse.trace.tags": ("agent:orders",),
|
||||
}
|
||||
assert not LANGFUSE_REQUEST_METADATA_ATTRIBUTES & received[None].keys()
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("key_environment", "tracing_environment", "expected"),
|
||||
[
|
||||
("prod", None, {"langfuse.environment": "prod"}),
|
||||
("prod", "staging", {"langfuse.environment": "prod"}),
|
||||
(None, "staging", {"langfuse.environment": "staging"}),
|
||||
(None, None, {}),
|
||||
],
|
||||
ids=["key", "key-over-env", "env", "neither"],
|
||||
)
|
||||
def test_a_langfuse_destination_names_the_environment_the_legacy_logger_wrote(
|
||||
self,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
key_environment: str | None,
|
||||
tracing_environment: str | None,
|
||||
expected: Mapping[str, str],
|
||||
) -> None:
|
||||
credential_less_proxy(monkeypatch)
|
||||
if tracing_environment is None:
|
||||
monkeypatch.delenv("LANGFUSE_TRACING_ENVIRONMENT", raising=False)
|
||||
else:
|
||||
monkeypatch.setenv("LANGFUSE_TRACING_ENVIRONMENT", tracing_environment)
|
||||
|
||||
destination: Final = destination_for(
|
||||
"langfuse_otel",
|
||||
StandardCallbackDynamicParams(
|
||||
langfuse_public_key="pk", langfuse_secret_key="sk", langfuse_environment=key_environment
|
||||
),
|
||||
)
|
||||
|
||||
assert destination is not None
|
||||
assert dict(destination.resource_attributes) == expected
|
||||
|
||||
def test_a_langfuse_destination_on_a_proxy_with_langfuse_keys_names_no_environment(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""With operator keys the older Langfuse logger never runs, so there is no environment of its to keep."""
|
||||
monkeypatch.setenv("LANGFUSE_PUBLIC_KEY", "pk-operator")
|
||||
monkeypatch.setenv("LANGFUSE_SECRET_KEY", "sk-operator")
|
||||
|
||||
destination: Final = destination_for(
|
||||
"langfuse_otel",
|
||||
StandardCallbackDynamicParams(
|
||||
langfuse_public_key="pk", langfuse_secret_key="sk", langfuse_environment="prod"
|
||||
),
|
||||
)
|
||||
|
||||
assert destination is not None
|
||||
assert dict(destination.resource_attributes) == {}
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("configured", "operator_keys", "expected"),
|
||||
[
|
||||
(None, False, "span_only"),
|
||||
("no_content", False, "no_content"),
|
||||
(None, True, None),
|
||||
("span_only", True, "span_only"),
|
||||
],
|
||||
ids=["unset", "key-no-content", "operator-keys", "operator-keys-key-span-only"],
|
||||
)
|
||||
def test_a_langfuse_destination_captures_the_prompt_where_the_legacy_logger_sent_it(
|
||||
self,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
configured: str | None,
|
||||
operator_keys: bool,
|
||||
expected: str | None,
|
||||
) -> None:
|
||||
credential_less_proxy(monkeypatch)
|
||||
if operator_keys:
|
||||
monkeypatch.setenv("LANGFUSE_PUBLIC_KEY", "pk-operator")
|
||||
monkeypatch.setenv("LANGFUSE_SECRET_KEY", "sk-operator")
|
||||
|
||||
destination: Final = destination_for(
|
||||
"langfuse_otel",
|
||||
StandardCallbackDynamicParams(langfuse_public_key="pk", langfuse_secret_key="sk"),
|
||||
capture_message_content=configured,
|
||||
)
|
||||
|
||||
assert destination is not None
|
||||
assert destination.capture_message_content == expected
|
||||
|
||||
def test_only_the_langfuse_destination_gets_the_prompt_it_captures_by_default(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
credential_less_proxy(monkeypatch)
|
||||
destination: Final = destination_for(
|
||||
"langfuse_otel", StandardCallbackDynamicParams(langfuse_public_key="pk", langfuse_secret_key="sk")
|
||||
)
|
||||
arize: Final = destination_for("arize", StandardCallbackDynamicParams(arize_space_id="s", arize_api_key="k"))
|
||||
assert destination is not None
|
||||
assert arize is not None
|
||||
|
||||
received: Final = self._received_by_account(
|
||||
monkeypatch, OpenTelemetryV2Config(), (destination, arize), with_content=True
|
||||
)
|
||||
|
||||
assert {headers: "langfuse.observation.input" in attributes for headers, attributes in received.items()} == {
|
||||
destination.header_string(): True,
|
||||
arize.header_string(): False,
|
||||
None: False,
|
||||
}
|
||||
assert not [key for key in received[None] if key.startswith("gen_ai.input")]
|
||||
|
||||
def test_only_the_langfuse_destination_gets_the_keys_environment(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
credential_less_proxy(monkeypatch)
|
||||
capture: Final = _ExporterCapture()
|
||||
for kind in ("otlp_grpc", "otlp_http", "console"):
|
||||
monkeypatch.setitem(otel_providers._EXPORTER_FACTORIES, kind, capture.build) # pyright: ignore[reportPrivateUsage] # the factory table is the seam for capturing each exporter
|
||||
config: Final = OpenTelemetryV2Config()
|
||||
provider: Final = build_tracer_provider(config, tenant_overrides=True)
|
||||
attach_tenant_fan_out(provider, config)
|
||||
emitter: Final = SpanEmitter(get_tracer(provider, "litellm"), config)
|
||||
destination: Final = destination_for(
|
||||
"langfuse_otel", {"langfuse_public_key": "pk", "langfuse_secret_key": "sk", "langfuse_environment": "prod"}
|
||||
)
|
||||
assert destination is not None
|
||||
|
||||
def with_destination() -> None:
|
||||
set_request_destinations(deliverable_destinations((destination,), provider))
|
||||
emitter.emit(SpanRole.LLM_CALL, _MODEL_CALL_WITHOUT_CONTENT)
|
||||
|
||||
in_fresh_context(with_destination)
|
||||
provider.force_flush()
|
||||
|
||||
environments: Final = {
|
||||
headers: exporter.get_finished_spans()[0].resource.attributes.get("langfuse.environment")
|
||||
for headers, exporter in capture.built
|
||||
}
|
||||
assert environments == {destination.header_string(): "prod", None: None}
|
||||
|
||||
@staticmethod
|
||||
def _attributes(
|
||||
destinations: tuple[OtelDestination, ...], limit: int | None, call: LLMCallSpanData
|
||||
) -> dict[str, object]:
|
||||
exporter: Final = InMemorySpanExporter()
|
||||
provider: Final = TracerProvider(span_limits=SpanLimits(max_span_attributes=limit))
|
||||
provider.add_span_processor(SimpleSpanProcessor(exporter))
|
||||
emitter: Final = SpanEmitter(get_tracer(provider, "litellm"), OpenTelemetryV2Config())
|
||||
|
||||
def run() -> None:
|
||||
set_request_destinations(destinations)
|
||||
emitter.emit(SpanRole.LLM_CALL, call)
|
||||
|
||||
in_fresh_context(run)
|
||||
return dict(exporter.get_finished_spans()[0].attributes or {})
|
||||
|
||||
def test_a_span_at_its_attribute_limit_sheds_the_destinations_langfuse_fields_first(self) -> None:
|
||||
generic: Final = self._attributes((), None, _MODEL_CALL_WITH_CONTENT)
|
||||
|
||||
assert "user.id" not in generic
|
||||
assert self._attributes((LANGFUSE_DEST,), len(generic), _MODEL_CALL_WITH_CONTENT) == generic, (
|
||||
"the generic attributes all survive"
|
||||
)
|
||||
|
||||
def test_a_span_at_its_attribute_limit_sheds_the_destinations_request_metadata_before_its_other_fields(
|
||||
self,
|
||||
) -> None:
|
||||
call: Final = replace(
|
||||
_MODEL_CALL_WITH_CONTENT, request_metadata=MappingProxyType({"user_api_key_alias": "prod-key"})
|
||||
)
|
||||
whole: Final = self._attributes((LANGFUSE_DEST,), None, call)
|
||||
|
||||
squeezed: Final = self._attributes((LANGFUSE_DEST,), len(whole) - 1, call)
|
||||
|
||||
assert len(squeezed) == len(whole) - 1
|
||||
assert whole.keys() - squeezed.keys() <= LANGFUSE_REQUEST_METADATA_ATTRIBUTES
|
||||
assert squeezed["langfuse.observation.type"] == "generation"
|
||||
|
||||
|
||||
class TestContextIsolation:
|
||||
def test_destinations_do_not_leak_between_requests(self):
|
||||
def first():
|
||||
|
|
@ -5103,7 +5377,10 @@ class TestCaptureMessageContent:
|
|||
def test_the_setting_rides_the_teams_destination_and_omission_stays_omitted(
|
||||
self, monkeypatch: pytest.MonkeyPatch, setting: str | None
|
||||
) -> None:
|
||||
"""On a proxy with Langfuse keys of its own; without them see ``TestLangfuseDestinationFields``."""
|
||||
monkeypatch.setenv("LITELLM_OTEL_V2", "true")
|
||||
monkeypatch.setenv("LANGFUSE_PUBLIC_KEY", "pk-operator")
|
||||
monkeypatch.setenv("LANGFUSE_SECRET_KEY", "sk-operator")
|
||||
is_otel_v2_enabled.cache_clear()
|
||||
callback_vars: Final = {
|
||||
"langfuse_public_key": "pk-team",
|
||||
|
|
|
|||
|
|
@ -1,9 +1,12 @@
|
|||
import asyncio
|
||||
import base64
|
||||
import contextvars
|
||||
import json
|
||||
import os
|
||||
from collections.abc import Mapping, Sequence
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime, timezone
|
||||
from types import MappingProxyType
|
||||
from typing import Final
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
|
|
@ -12,12 +15,17 @@ from openai.types.responses import ResponseFunctionToolCall
|
|||
from opentelemetry import trace
|
||||
from opentelemetry.context import Context
|
||||
from opentelemetry.sdk.trace import SpanLimits, TracerProvider
|
||||
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
|
||||
from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
|
||||
from opentelemetry.trace import Span
|
||||
|
||||
import litellm
|
||||
from litellm.constants import REDACTED_BY_LITELLM
|
||||
from litellm.constants import LITELLM_LOGGING_NO_UPSTREAM_LLM_CALL, REDACTED_BY_LITELLM
|
||||
from litellm.integrations.langfuse.langfuse_otel import LangfuseOtelLogger
|
||||
from litellm.integrations.opentelemetry import LITELLM_PROXY_REQUEST_SPAN_NAME, OpenTelemetry, OpenTelemetryConfig
|
||||
from litellm.integrations.otel.model.destination import OtelDestination
|
||||
from litellm.integrations.otel.plumbing.context import set_request_destinations
|
||||
from litellm.proxy._types import UserAPIKeyAuth
|
||||
from litellm.types.llms.openai import ResponsesAPIResponse
|
||||
from litellm.types.utils import Choices, ModelResponse, StandardCallbackDynamicParams
|
||||
|
||||
|
|
@ -1507,3 +1515,75 @@ def test_langfuse_otel_does_not_start_proxy_request_span() -> None:
|
|||
finally:
|
||||
langfuse_provider.shutdown()
|
||||
generic_provider.shutdown()
|
||||
|
||||
|
||||
_HOOK_CALL: Final[Mapping[str, object]] = MappingProxyType({})
|
||||
|
||||
|
||||
def _spans_the_legacy_logger_exports(
|
||||
destinations: tuple[OtelDestination, ...], call: Mapping[str, object] = _HOOK_CALL
|
||||
) -> tuple[str, ...]:
|
||||
exporter: Final = InMemorySpanExporter()
|
||||
provider: Final = TracerProvider()
|
||||
provider.add_span_processor(SimpleSpanProcessor(exporter))
|
||||
with patch.dict(os.environ, LANGFUSE_ENV_ONLY, clear=True):
|
||||
logger: Final = LangfuseOtelLogger(tracer_provider=provider)
|
||||
at: Final = datetime(2026, 1, 1, tzinfo=timezone.utc)
|
||||
kwargs: Final[dict[str, object]] = {**CHAT_KWARGS, **call, "litellm_params": {"metadata": {}}}
|
||||
server_span: Final = provider.get_tracer("proxy").start_span("POST /v1/chat/completions")
|
||||
|
||||
async def log() -> None:
|
||||
set_request_destinations(destinations)
|
||||
logger.log_success_event(dict(kwargs), _chat_response("ok"), at, at) # pyright: ignore[reportUnknownMemberType] # the v1 logger hooks are untyped
|
||||
await logger.async_log_failure_event({**kwargs, "exception": ValueError("boom")}, None, at, at) # pyright: ignore[reportUnknownMemberType] # the v1 logger hooks are untyped
|
||||
await logger.async_post_call_failure_hook({}, ValueError("boom"), UserAPIKeyAuth(parent_otel_span=server_span))
|
||||
|
||||
try:
|
||||
contextvars.copy_context().run(asyncio.run, log())
|
||||
return tuple(span.name for span in exporter.get_finished_spans())
|
||||
finally:
|
||||
provider.shutdown()
|
||||
|
||||
|
||||
_LANGFUSE_DESTINATION: Final = OtelDestination(
|
||||
endpoint="http://tenant.local/api/public/otel",
|
||||
headers={"Authorization": "Basic dGVuYW50"},
|
||||
callback_name="langfuse_otel",
|
||||
)
|
||||
|
||||
|
||||
_PROVIDER_CALL: Final[Mapping[str, object]] = MappingProxyType(
|
||||
{"standard_logging_object": {"id": "c1"}, "api_call_start_time": datetime(2026, 1, 1, tzinfo=timezone.utc)}
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("call", [_HOOK_CALL, _PROVIDER_CALL], ids=["hook", "provider-call"])
|
||||
def test_a_request_with_a_langfuse_destination_exports_nothing_through_the_legacy_logger(
|
||||
call: Mapping[str, object],
|
||||
) -> None:
|
||||
assert _spans_the_legacy_logger_exports((_LANGFUSE_DESTINATION,), call) == ()
|
||||
|
||||
|
||||
def test_a_call_rejected_before_any_provider_still_exports_through_the_legacy_logger() -> None:
|
||||
rejected: Final = {"standard_logging_object": {"id": "c1"}, LITELLM_LOGGING_NO_UPSTREAM_LLM_CALL: True}
|
||||
|
||||
assert _spans_the_legacy_logger_exports((_LANGFUSE_DESTINATION,), rejected) == (
|
||||
"litellm_request",
|
||||
"raw_gen_ai_request",
|
||||
"litellm_request",
|
||||
)
|
||||
|
||||
|
||||
def test_a_request_without_a_langfuse_destination_still_exports_through_the_legacy_logger() -> None:
|
||||
logged: Final = (
|
||||
"litellm_request",
|
||||
"raw_gen_ai_request",
|
||||
"litellm_request",
|
||||
"Failed Proxy Server Request",
|
||||
"POST /v1/chat/completions",
|
||||
)
|
||||
|
||||
assert _spans_the_legacy_logger_exports(()) == logged
|
||||
assert _spans_the_legacy_logger_exports((_LANGFUSE_DESTINATION.model_copy(update={"callback_name": "arize"}),)) == (
|
||||
logged
|
||||
)
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue