mirror of
https://github.com/BerriAI/litellm.git
synced 2026-11-10 08:28:25 +00:00
fix(otel): keep the Langfuse fields written for a key's Langfuse destination off every other exporter
This commit is contained in:
parent
732de4340e
commit
f74c14ecc9
4 changed files with 126 additions and 26 deletions
|
|
@ -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():
|
||||
|
|
|
|||
|
|
@ -46,6 +46,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,
|
||||
|
|
@ -488,19 +489,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})
|
||||
)
|
||||
|
||||
|
||||
|
|
@ -656,6 +663,7 @@ def _for_destination(
|
|||
span: ReadableSpan,
|
||||
destination: "OtelDestination",
|
||||
default_capture: str,
|
||||
langfuse_mapped: bool,
|
||||
) -> ReadableSpan:
|
||||
"""The view of ``span`` a tenant destination receives.
|
||||
|
||||
|
|
@ -670,13 +678,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)
|
||||
|
|
@ -726,8 +737,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
|
||||
|
|
@ -820,7 +833,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
|
||||
|
|
@ -1196,17 +1210,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()
|
||||
|
|
@ -1489,13 +1505,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
|
||||
|
||||
|
||||
|
|
@ -1524,9 +1545,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.
|
||||
Every exporter but ``langfuse_otel``'s goes without the request metadata the Langfuse
|
||||
mapper writes on LiteLLM's own spans, which any logger writes for a request with a
|
||||
Langfuse destination.
|
||||
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.
|
||||
|
|
@ -1540,6 +1561,7 @@ 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).
|
||||
for spec in config.exporters:
|
||||
|
|
@ -1549,7 +1571,8 @@ def build_tracer_provider(
|
|||
spec,
|
||||
use_simple_processor,
|
||||
strip_message_content,
|
||||
strip_langfuse_request_metadata=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)
|
||||
|
|
@ -1586,8 +1609,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)):
|
||||
|
|
@ -1598,10 +1622,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(
|
||||
|
|
|
|||
|
|
@ -136,7 +136,9 @@ def test_a_tenant_langfuse_project_gets_user_session_and_tags_without_operator_k
|
|||
] == [{"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 len(operator) == 1, [span["name"] for span in operator]
|
||||
assert "langfuse.observation.metadata" not in 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(
|
||||
|
|
|
|||
|
|
@ -35,7 +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
|
||||
from litellm.integrations.otel.mappers.langfuse import LANGFUSE_REQUEST_METADATA_ATTRIBUTES, LangfuseMapper
|
||||
from litellm.integrations.otel.model.config import (
|
||||
CaptureMessageContent,
|
||||
ExporterOwner,
|
||||
|
|
@ -3481,6 +3481,62 @@ class TestLangfuseDestinationFields:
|
|||
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, ...]
|
||||
) -> 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_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"),
|
||||
[
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue