mirror of
https://github.com/BerriAI/litellm.git
synced 2026-11-07 07:58:09 +00:00
fix(otel): give key/team Langfuse destinations the Langfuse fields when langfuse_otel is not a callback
With langfuse_otel only in a key's or team's logging, the startup logger is the generic one and its spans reached the key's Langfuse project without user, session or tags. The emitter now adds the Langfuse mapper for a request with a Langfuse destination, and every non-Langfuse exporter strips the Langfuse request metadata, since any logger can now write it.
This commit is contained in:
parent
3272253d05
commit
44d055e4d5
4 changed files with 213 additions and 13 deletions
|
|
@ -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,10 @@ class SpanEmitter:
|
|||
self._mappers: list[AttributeMapper] = (
|
||||
list(mappers) if mappers is not None else resolve_mappers(config.mapper_names)
|
||||
)
|
||||
# A key's or team's Langfuse destination reads the Langfuse vocabulary, which this logger may not write.
|
||||
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 +300,22 @@ 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 = (
|
||||
self._langfuse.map(data)
|
||||
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:
|
||||
|
|
|
|||
|
|
@ -1524,8 +1524,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 request metadata the Langfuse
|
||||
mapper writes on LiteLLM's own spans, 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.
|
||||
|
|
@ -1541,9 +1542,6 @@ 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
|
||||
|
|
@ -1551,7 +1549,7 @@ 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_langfuse_request_metadata=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)
|
||||
|
|
|
|||
|
|
@ -0,0 +1,131 @@
|
|||
"""A key's or team's own Langfuse project gets 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
|
||||
|
||||
import pytest
|
||||
from integration._support.client import Gateway, eventually, gateway_from_environment
|
||||
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"]
|
||||
|
||||
|
||||
def _chat(request: Request) -> Reply:
|
||||
marker: Final = json.loads(request.body)["messages"][-1]["content"]
|
||||
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": marker}, "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": request.param}),
|
||||
remove_environment=("LANGFUSE_PUBLIC_KEY", "LANGFUSE_SECRET_KEY", "LANGFUSE_HOST"),
|
||||
workers=2,
|
||||
) as owned,
|
||||
):
|
||||
yield owned.gateway
|
||||
|
||||
|
||||
def _model_spans(url: str, since: int, trace_id: str) -> tuple[Span, ...]:
|
||||
return tuple(
|
||||
span for span in spans_for_trace(recorded_spans(url, since)[1], trace_id) if OPERATION in span["attributes"]
|
||||
)
|
||||
|
||||
|
||||
@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}"
|
||||
response: Final = proxy.request(
|
||||
"POST",
|
||||
"/v1/chat/completions",
|
||||
{
|
||||
"model": model,
|
||||
"messages": [{"role": "user", "content": f"lf-tenant-only-{trace_id}"}],
|
||||
"metadata": {"tags": TAGS, "session_id": session, "trace_user_id": "user-42"},
|
||||
},
|
||||
key=key,
|
||||
headers={"traceparent": f"00-{trace_id}-{uuid.uuid4().hex[:16]}-01"},
|
||||
)
|
||||
assert response.status_code == 200, response.text
|
||||
|
||||
tenant: Final = eventually(
|
||||
lambda: _model_spans(audit_sinks.tenant, since[0], trace_id), lambda spans: bool(spans), seconds=40
|
||||
)
|
||||
operator: Final = eventually(
|
||||
lambda: _model_spans(audit_sinks.operator, since[1], trace_id), lambda spans: bool(spans), 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 "langfuse.observation.metadata" not in operator[0]["attributes"], sorted(operator[0]["attributes"])
|
||||
|
|
@ -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,
|
||||
|
|
@ -52,6 +53,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,
|
||||
|
|
@ -3519,6 +3522,64 @@ class TestLangfuseTenantOnly:
|
|||
assert later is at_startup
|
||||
assert loggers == [at_startup]
|
||||
|
||||
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)
|
||||
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]
|
||||
|
||||
def test_a_span_at_its_attribute_limit_sheds_the_destinations_langfuse_fields_first(self) -> None:
|
||||
def attributes(destinations: tuple[OtelDestination, ...], limit: int | None) -> 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, _MODEL_CALL_WITH_CONTENT)
|
||||
|
||||
in_fresh_context(run)
|
||||
return dict(exporter.get_finished_spans()[0].attributes or {})
|
||||
|
||||
generic: Final = attributes((), None)
|
||||
|
||||
assert "user.id" not in generic
|
||||
assert attributes((LANGFUSE_DEST,), len(generic)) == generic, "the generic attributes all survive"
|
||||
|
||||
|
||||
class TestContextIsolation:
|
||||
def test_destinations_do_not_leak_between_requests(self):
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue