scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-11-11 08:38:24 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 359

fix(langfuse otel): share request metadata export across callbacks

Browse source
This commit is contained in:
yucheng 2026-10-16 14:04:39 +00:00
parent 19c8b4b05a
commit c12102dd92
7 changed files with 78 additions and 84 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

48
litellm/integrations/langfuse/langfuse_otel.py
View file

@ -1,10 +1,12 @@
import base64
import json
import os
from collections.abc import Sequence
from collections.abc import Mapping, Sequence
from datetime import datetime
from typing import TYPE_CHECKING, Any, Final, Optional
from pydantic import TypeAdapter
from litellm._logging import verbose_logger
from litellm.integrations.arize import _utils
from litellm.integrations.langfuse.langfuse_otel_attributes import (
@ -25,19 +27,13 @@ else:
Span = Any
_STANDARD_LOGGING_OBJECT_ADAPTER: Final[TypeAdapter[Mapping[str, object]]] = TypeAdapter(Mapping[str, object])
LANGFUSE_CLOUD_EU_ENDPOINT: Final = "https://cloud.langfuse.com/api/public/otel"
LANGFUSE_CLOUD_US_ENDPOINT: Final = "https://us.cloud.langfuse.com/api/public/otel"
LANGFUSE_INGESTION_VERSION_HEADER: Final = "x-langfuse-ingestion-version"
LANGFUSE_INGESTION_VERSION: Final = "4"
_TRACE_IDENTITY_FIELDS: Final = (
"user_api_key_alias",
"user_api_key_user_id",
"user_api_key_end_user_id",
"user_api_key_team_id",
"user_api_key_team_alias",
)
class LangfuseOtelLogger(OpenTelemetry):
def __init__(self, config=None, *args, **kwargs):
@ -135,26 +131,30 @@ class LangfuseOtelLogger(OpenTelemetry):
safe_set_attribute(span, enum_attr.value, value)
@staticmethod
def _set_request_metadata_attributes(span: Span, kwargs: dict[str, object]) -> None:
def _set_request_metadata_attributes(
span: Span,
kwargs: dict[str, object], # mutable-ok: required callback signature accepts kwargs as a dict
) -> None:
from litellm.integrations.arize._utils import safe_set_attribute
from litellm.integrations.langfuse.langfuse import log_requester_metadata
from litellm.litellm_core_utils.redact_messages import redact_user_api_key_info
from litellm.integrations.otel.mappers.langfuse import TRACE_IDENTITY_FIELDS
from litellm.integrations.otel.model.metadata import exported_request_metadata
from litellm.litellm_core_utils.safe_json_dumps import safe_dumps
standard_logging_object: Final = kwargs.get("standard_logging_object")
request_metadata: Final = (
standard_logging_object.get("metadata") if isinstance(standard_logging_object, dict) else None
)
if not isinstance(request_metadata, dict):
raw_standard_logging_object: Final = kwargs.get("standard_logging_object")
if not isinstance(raw_standard_logging_object, dict):
return
observation_metadata: Final = { # mutable-ok: stays a real dict for safe_dumps and .get reads below
key: value
for key, value in log_requester_metadata(redact_user_api_key_info(metadata=request_metadata)).items()
if value is not None
}
safe_set_attribute(span, LangfuseSpanAttributes.OBSERVATION_METADATA.value, safe_dumps(observation_metadata))
standard_logging_object: Final = _STANDARD_LOGGING_OBJECT_ADAPTER.validate_python(raw_standard_logging_object)
if not isinstance(standard_logging_object.get("metadata"), dict):
return
observation_metadata: Final = exported_request_metadata(standard_logging_object)
if observation_metadata:
safe_set_attribute(
span,
LangfuseSpanAttributes.OBSERVATION_METADATA.value,
safe_dumps(dict(observation_metadata)),
)
trace_prefix: Final = LangfuseSpanAttributes.TRACE_METADATA.value
for field in _TRACE_IDENTITY_FIELDS:
for field in TRACE_IDENTITY_FIELDS:
if value := observation_metadata.get(field):
safe_set_attribute(span, f"{trace_prefix}.{field}", value)

7
litellm/integrations/otel/mappers/langfuse.py
View file

@ -11,10 +11,12 @@ the JSON-serialized payloads. ``trace_attributes`` maps the caller's trace contr
"""
import json
from collections.abc import Callable
from collections.abc import Callable, Mapping
from types import MappingProxyType
from typing import Final
from pydantic import TypeAdapter
from litellm.integrations.otel.mappers.base import AttributeMap, AttrValue, SpanData
from litellm.integrations.otel.mappers.utils import (
collect,
@ -40,6 +42,7 @@ LANGFUSE_TRACE_SESSION_ID: Final = "session.id"
LANGFUSE_TRACE_TAGS: Final = "langfuse.trace.tags"
LANGFUSE_OBSERVATION_METADATA: Final = "langfuse.observation.metadata"
LANGFUSE_TRACE_METADATA_PREFIX: Final = "langfuse.trace.metadata."
_REQUEST_METADATA_ADAPTER: Final[TypeAdapter[Mapping[str, object]]] = TypeAdapter(Mapping[str, object])
TRACE_IDENTITY_FIELDS: Final = (
"user_api_key_alias",
"user_api_key_user_id",
@ -88,7 +91,7 @@ class LangfuseMapper:
# JSON-payload attributes: each builder returns the serialized blob or None.
_BLOB_ATTRS: dict[str, Callable[[LLMCallSpanData], AttrValue | None]] = {
LANGFUSE_OBSERVATION_METADATA: lambda d: (
safe_dumps(dict(d.request_metadata)) # mutable-ok: safe_dumps only serializes real dicts
safe_dumps({**_REQUEST_METADATA_ADAPTER.validate_python(d.request_metadata)})
if d.request_metadata
else None
),

4
litellm/integrations/otel/mappers/utils.py
View file

@ -135,9 +135,9 @@ def message_content(message: object) -> str | None:
return None
def output_messages(data: LLMCallSpanData) -> list:
def output_messages(data: LLMCallSpanData) -> tuple[object, ...]:
"""The ``message`` payload of each response choice."""
return [c.get("message") for c in data.choices_out if isinstance(c, dict)]
return tuple(c.get("message") for c in data.choices_out if isinstance(c, dict))
def message_tool_calls(message: object) -> tuple[MessageToolCall, ...]:

24
litellm/integrations/otel/model/metadata.py
View file

@ -49,11 +49,9 @@ from litellm.constants import (
LITELLM_LOGGING_NO_UPSTREAM_LLM_CALL,
SESSION_ID_GENERATED_METADATA_KEY,
)
from litellm.integrations.langfuse.langfuse import log_requester_metadata
from litellm.integrations.otel.model.semconv import resolve_operation
from litellm.integrations.otel.model.trace_controls import TraceControls, caller_trace_controls
from litellm.integrations.otel.model.utils import as_str, as_str_mapping, to_seconds
from litellm.litellm_core_utils.redact_messages import redact_user_api_key_info
from litellm.types.utils import InternalCallOrigin
if TYPE_CHECKING:
@ -61,6 +59,7 @@ if TYPE_CHECKING:
_INTERNAL_CALL_ORIGINS: Final[frozenset[str]] = frozenset(get_args(InternalCallOrigin))
_AUTH_ATTRIBUTE: Final = TypeAdapter(object)
_REQUEST_METADATA_MAPPING_ADAPTER: Final[TypeAdapter[Mapping[str, object]]] = TypeAdapter(Mapping[str, object])
REQUESTER_METADATA_KEY: Final = "requester_metadata"
REQUESTER_METADATA_PATH: Final = f"{REQUESTER_METADATA_KEY}."
@ -466,20 +465,25 @@ def _model_info_id(model_info: object) -> str | None:
return None
def exported_request_metadata(payload: StandardLoggingPayload) -> Mapping[str, object]:
def exported_request_metadata(payload: Mapping[str, object]) -> Mapping[str, object]:
"""The request metadata as the logging callbacks export it: ``user_api_key_*``
dropped when ``litellm.redact_user_api_key_info`` is on, header keys nested
under ``requester_metadata``, ``None`` values dropped."""
raw_meta: Final = cast( # cast-ok: StandardLoggingPayload.get returns Any | None
Mapping[str, object], payload.get("metadata") or MappingProxyType({})
from litellm.integrations.langfuse.langfuse import log_requester_metadata
from litellm.litellm_core_utils.redact_messages import ( # pyright: ignore[reportUnknownVariableType] # helper uses bare dict
redact_user_api_key_info as _redact_user_api_key_info,
)
redacted: Final = cast( # cast-ok: redact_user_api_key_info is untyped
Mapping[str, object],
redact_user_api_key_info(metadata=dict(raw_meta)), # mutable-ok: the redactor requires a real dict
raw_metadata: Final = payload.get("metadata")
metadata: Final = _REQUEST_METADATA_MAPPING_ADAPTER.validate_python(
raw_metadata if isinstance(raw_metadata, Mapping) else MappingProxyType({})
)
exported: Final = cast( # cast-ok: log_requester_metadata is untyped
Mapping[str, object], log_requester_metadata(redacted)
redact_metadata: Final = cast( # cast-ok: adapter validates the helper's output below
Callable[[dict[str, object]], object],
_redact_user_api_key_info,
)
redacted: Final = _REQUEST_METADATA_MAPPING_ADAPTER.validate_python(redact_metadata(dict(metadata)))
exported: Final = _REQUEST_METADATA_MAPPING_ADAPTER.validate_python(log_requester_metadata(redacted))
return MappingProxyType({key: value for key, value in exported.items() if value is not None})

15
tests/integration/observability/test_langfuse_otel_metadata.py
View file

@ -11,7 +11,6 @@ from pathlib import Path
from typing import Final
import httpx
import psutil
import yaml
from anthropic import Anthropic, AsyncAnthropic
from integration._support.client import Gateway, Scenario, eventually
@ -928,20 +927,16 @@ def test_langfuse_otel_worker_kill_mid_burst_keeps_serving_and_exports(gateway:
return 0, repr(error)
return response.status_code, response.json()["id"] if response.status_code == 200 else response.text
workers: Final = tuple(
child
for child in psutil.Process(rig.proxy.process.pid).children(recursive=True)
if "spawn_main" in " ".join(child.cmdline())
)
assert len(workers) >= 2, f"expected multiple uvicorn workers, got {workers!r}"
victim: Final = workers[0]
workers: Final = rig.proxy.worker_pids()
assert len(workers) >= 2, f"expected multiple uvicorn workers, got {sorted(workers)!r}"
victim: Final = min(workers)
with ThreadPoolExecutor(max_workers=20) as pool:
futures: Final = tuple(pool.submit(call, index) for index in range(20))
while sum(f.done() for f in futures) < 5:
wait(futures, timeout=0.05)
victim.kill()
replacement: Final = rig.proxy.kill_worker(victim)
outcomes: Final = tuple(f.result() for f in futures)
assert not victim.is_running() or victim.status() == psutil.STATUS_ZOMBIE
assert replacement != victim
succeeded: Final = tuple(identifier for status, identifier in outcomes if status == 200)
assert succeeded, outcomes
post_kill: Final = tuple(call(20 + index) for index in range(5))

32
tests/unit/integrations/otel/test_otel_v2_vendor_mappers.py
View file

@ -834,26 +834,22 @@ def test_langfuse_mapper_skips_empty_identity_values():
assert attrs["langfuse.trace.metadata.user_api_key_end_user_id"] == "end-1"
def test_langfuse_mapper_redacts_user_api_key_fields():
def test_langfuse_mapper_redacts_user_api_key_fields(monkeypatch: pytest.MonkeyPatch) -> None:
import litellm
saved: Final = litellm.redact_user_api_key_info
try:
litellm.redact_user_api_key_info = True
payload: Final[dict[str, object]] = {
"call_type": "acompletion",
"custom_llm_provider": "openai",
"model": "gpt-4o",
"metadata": {
"user_api_key_alias": "prod-key",
"user_api_key_team_id": "team-9",
"spend_logs_metadata": {"ticket": "LIT-8283"},
},
"response": {},
}
attrs: Final = LangfuseMapper().map(LLMCallSpanData.from_standard_logging_payload(payload))
finally:
litellm.redact_user_api_key_info = saved
monkeypatch.setattr(litellm, "redact_user_api_key_info", True)
payload: Final[dict[str, object]] = {
"call_type": "acompletion",
"custom_llm_provider": "openai",
"model": "gpt-4o",
"metadata": {
"user_api_key_alias": "prod-key",
"user_api_key_team_id": "team-9",
"spend_logs_metadata": {"ticket": "LIT-8283"},
},
"response": {},
}
attrs: Final = LangfuseMapper().map(LLMCallSpanData.from_standard_logging_payload(payload))
exported: Final = json.loads(attrs["langfuse.observation.metadata"])
assert not [key for key in exported if key.startswith("user_api_key")]

32
tests/unit/integrations/test_langfuse_otel.py
View file

@ -500,7 +500,7 @@ class TestLangfuseOtelIntegration:
"requester_ip_address": "10.0.0.1",
"requester_metadata": {"ticket": "LIT-8283"},
}
kwargs: Final = {
kwargs: Final[dict[str, object]] = {
"litellm_params": {"metadata": {}},
"standard_logging_object": {"metadata": request_metadata},
}
@ -519,9 +519,8 @@ class TestLangfuseOtelIntegration:
"langfuse.trace.metadata.user_api_key_team_alias": "team-a",
}
def test_request_metadata_redaction_matches_vanilla_langfuse(self) -> None:
previous_flag: Final = litellm.redact_user_api_key_info
litellm.redact_user_api_key_info = True
def test_request_metadata_redaction_matches_vanilla_langfuse(self, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(litellm, "redact_user_api_key_info", True)
request_metadata: Final = {
"user_api_key_hash": "hash123",
"user_api_key_alias": "prod-key",
@ -533,25 +532,22 @@ class TestLangfuseOtelIntegration:
"requester_ip_address": "10.0.0.1",
"requester_metadata": {"ticket": "LIT-8283"},
}
kwargs: Final = {
kwargs: Final[dict[str, object]] = {
"litellm_params": {"metadata": {}},
"standard_logging_object": {"metadata": request_metadata},
}
try:
with patch("litellm.integrations.arize._utils.safe_set_attribute") as mock_safe_set_attribute:
LangfuseOtelLogger._set_langfuse_specific_attributes(MagicMock(), kwargs, None)
with patch("litellm.integrations.arize._utils.safe_set_attribute") as mock_safe_set_attribute:
LangfuseOtelLogger._set_langfuse_specific_attributes(MagicMock(), kwargs, None)
actual: Final = {call.args[1]: call.args[2] for call in mock_safe_set_attribute.call_args_list}
actual: Final = {call.args[1]: call.args[2] for call in mock_safe_set_attribute.call_args_list}
assert json.loads(actual["langfuse.observation.metadata"]) == {
"spend_logs_metadata": {"env": "prod"},
"requester_ip_address": "10.0.0.1",
"requester_metadata": {"ticket": "LIT-8283"},
}
assert not [key for key in actual if key.startswith("langfuse.trace.metadata.")]
finally:
litellm.redact_user_api_key_info = previous_flag
assert json.loads(actual["langfuse.observation.metadata"]) == {
"spend_logs_metadata": {"env": "prod"},
"requester_ip_address": "10.0.0.1",
"requester_metadata": {"ticket": "LIT-8283"},
}
assert not [key for key in actual if key.startswith("langfuse.trace.metadata.")]
def test_request_metadata_drops_null_fields_and_skips_empty_trace_identities(self) -> None:
request_metadata: Final = {
@ -564,7 +560,7 @@ class TestLangfuseOtelIntegration:
"team_alias": None,
"spend_logs_metadata": {"env": "prod"},
}
kwargs: Final = {
kwargs: Final[dict[str, object]] = {
"litellm_params": {"metadata": {}},
"standard_logging_object": {"metadata": request_metadata},
}