diff --git a/litellm/integrations/langfuse/langfuse_otel.py b/litellm/integrations/langfuse/langfuse_otel.py index 0117348da66..7d602f98c20 100644 --- a/litellm/integrations/langfuse/langfuse_otel.py +++ b/litellm/integrations/langfuse/langfuse_otel.py @@ -4,19 +4,25 @@ import os from collections.abc import Iterable, Mapping, Sequence from datetime import datetime from itertools import chain -from typing import TYPE_CHECKING, Any, Final, Optional, Protocol, runtime_checkable +from typing import TYPE_CHECKING, Any, Final, Optional from opentelemetry import trace -import litellm from litellm._logging import verbose_logger from litellm.constants import REDACTED_BY_LITELLM from litellm.integrations._types.open_inference import ErrorAttributes from litellm.integrations.arize import _utils +from litellm.integrations.langfuse.langfuse_otel_attributes import ( + Gettable as _Gettable, +) from litellm.integrations.langfuse.langfuse_otel_attributes import ( LangfuseLLMObsOTELAttributes, + default_langfuse_tags, serialize_langfuse_datetime, ) +from litellm.integrations.langfuse.langfuse_otel_attributes import ( + as_gettable as _as_gettable, +) from litellm.integrations.opentelemetry import ( HTTP_RESPONSE_STATUS_CODE_ATTRIBUTE, LITELLM_PROXY_REQUEST_SPAN_NAME, @@ -59,25 +65,10 @@ _ERROR_ATTRIBUTES: Final = frozenset( ) -@runtime_checkable -class _Gettable(Protocol): - def get(self, key: str, default: object = None) -> object: ... - - -def _as_gettable(value: object) -> _Gettable | None: - return value if isinstance(value, _Gettable) else None - - def _attr(item: object, name: str, default: object = None) -> object: return getattr(item, name, default) -def _preset_cache_key(kwargs: Mapping[str, object]) -> object: - if litellm.cache is None: - return None - return litellm.cache.get_preset_cache_key_from_kwargs(**kwargs) - - def _mask_flag(value: object) -> bool: parsed: Final = str_to_bool(value) if isinstance(value, str) else None return bool(value) if parsed is None else parsed @@ -254,31 +245,10 @@ class LangfuseOtelLogger(OpenTelemetry): def _trace_tags(kwargs: Mapping[str, object], metadata: Mapping[str, object]) -> tuple[str, ...]: caller_tags: Final = metadata.get("tags") request_tags: Final = (_as_gettable(kwargs.get("standard_logging_object")) or {}).get("request_tags") - default_tags: Final = litellm.langfuse_default_tags - - def _default_tag(key: str) -> str | None: - if key == "cache_hit": - return f"cache_hit:{kwargs.get('cache_hit', False)}" - if key == "cache_key": - hidden_params: Final = _as_gettable(metadata.get("hidden_params", {})) or {} - cache_key: Final = ( - hidden_params.get("cache_key") - if hidden_params.get("cache_key") is not None - else _preset_cache_key(kwargs) - ) - return f"cache_key:{cache_key}" - if key == "proxy_base_url": - proxy_base_url: Final = os.environ.get("PROXY_BASE_URL") - return f"proxy_base_url:{proxy_base_url}" if proxy_base_url is not None else None - if key in metadata and metadata[key] is not None: - return f"{key}:{metadata[key]}" - return None - - expanded: Final = tuple(_default_tag(key) for key in default_tags) if isinstance(default_tags, list) else () candidates: Final = ( _caller_tag_list(caller_tags) + (tuple(tag for tag in request_tags if isinstance(tag, str)) if isinstance(request_tags, list) else ()) - + tuple(tag for tag in expanded if tag is not None) + + default_langfuse_tags(kwargs, metadata) ) return tuple(dict.fromkeys(candidates)) diff --git a/litellm/integrations/langfuse/langfuse_otel_attributes.py b/litellm/integrations/langfuse/langfuse_otel_attributes.py index 2873dba93ed..fe7d356b95b 100644 --- a/litellm/integrations/langfuse/langfuse_otel_attributes.py +++ b/litellm/integrations/langfuse/langfuse_otel_attributes.py @@ -5,13 +5,15 @@ Relevant Issue: https://github.com/BerriAI/litellm/issues/13764 """ import json +import os from collections.abc import Mapping from datetime import datetime -from typing import TYPE_CHECKING, Any, Final +from typing import TYPE_CHECKING, Any, Final, Protocol, runtime_checkable from pydantic import BaseModel from typing_extensions import override +import litellm from litellm.integrations.opentelemetry_utils.base_otel_llm_obs_attributes import ( BaseLLMObsOTELAttributes, safe_set_attribute, @@ -30,10 +32,45 @@ if TYPE_CHECKING: from opentelemetry.trace import Span +@runtime_checkable +class Gettable(Protocol): + def get(self, key: str, default: object = None) -> object: ... + + +def as_gettable(value: object) -> Gettable | None: + return value if isinstance(value, Gettable) else None + + def serialize_langfuse_datetime(value: datetime) -> str: return json.dumps(value.astimezone().isoformat()) +def default_langfuse_tags( + kwargs: Mapping[str, object], metadata: Mapping[str, object], *, completed: bool = True +) -> tuple[str, ...]: + default_tags: Final = litellm.langfuse_default_tags + if not isinstance(default_tags, list): + return () + + def expand(key: str) -> str | None: + if key == "cache_hit": + return f"cache_hit:{kwargs.get('cache_hit', False)}" if completed else None + if key == "cache_key": + hidden_params: Final = as_gettable(metadata.get("hidden_params")) + hidden_key: Final = hidden_params.get("cache_key") if hidden_params else None + cache: Final = litellm.cache + cache_key: Final = ( + cache.get_preset_cache_key_from_kwargs(**kwargs) + if hidden_key is None and cache is not None + else hidden_key + ) + return f"cache_key:{cache_key}" if completed or cache_key is not None else None + value: Final = os.environ.get("PROXY_BASE_URL") if key == "proxy_base_url" else metadata.get(key) + return f"{key}:{value}" if value is not None else None + + return tuple(tag for key in default_tags if (tag := expand(key)) is not None) + + def get_output_content_by_type( response_obj: None | dict diff --git a/litellm/integrations/otel/langfuse_logger.py b/litellm/integrations/otel/langfuse_logger.py index 15b0f3f0cc2..4833f982f11 100644 --- a/litellm/integrations/otel/langfuse_logger.py +++ b/litellm/integrations/otel/langfuse_logger.py @@ -33,7 +33,7 @@ class LangfuseOpenTelemetryV2(OpenTelemetryV2): if root is not None and root.is_recording(): root.set_attributes( LangfuseMapper.trace_attributes( - caller_trace_controls(kwargs), + caller_trace_controls(kwargs, langfuse_defaults="request"), as_str(kwargs.get("call_type")) if self._derives_trace_name(root, kwargs) else None, ) ) diff --git a/litellm/integrations/otel/logger.py b/litellm/integrations/otel/logger.py index b5b65d299e0..f1827ec272d 100644 --- a/litellm/integrations/otel/logger.py +++ b/litellm/integrations/otel/logger.py @@ -532,7 +532,7 @@ class OpenTelemetryV2(CustomLogger): rejected at the gate or blocked by a pre-call guardrail before any upstream call — so there is nothing to record and no phantom span. """ - call: Final = LLMCallEvent.from_dict(kwargs) + call: Final = LLMCallEvent.from_dict(kwargs, langfuse_defaults="langfuse" in self.config.mapper_names) call_id: Final = call.call_id # ``pop`` is the dedup: this method runs from both the success and failure # paths, and whichever fires first removes the carrier and closes the span. diff --git a/litellm/integrations/otel/model/metadata.py b/litellm/integrations/otel/model/metadata.py index 28d6ad54607..ba5f3a5f48a 100644 --- a/litellm/integrations/otel/model/metadata.py +++ b/litellm/integrations/otel/model/metadata.py @@ -278,12 +278,12 @@ class LLMCallEvent: 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: + def from_dict(cls, kwargs: Mapping[str, object], *, langfuse_defaults: bool = False) -> LLMCallEvent: raw_payload: Final = kwargs.get("standard_logging_object") payload: Final = cast("StandardLoggingPayload", raw_payload) if raw_payload else None operation: Final = resolve_operation(as_str(kwargs.get("call_type")), kwargs.get(PASSTHROUGH_OPERATION_KEY)) model: Final = as_str(kwargs.get("model")) or "" - trace: Final = caller_trace_controls(kwargs) + trace: Final = caller_trace_controls(kwargs, langfuse_defaults="completed" if langfuse_defaults else None) return cls( call_id=_call_id(payload, kwargs), payload=payload, diff --git a/litellm/integrations/otel/model/trace_controls.py b/litellm/integrations/otel/model/trace_controls.py index eac7b5c897b..20ea4edf097 100644 --- a/litellm/integrations/otel/model/trace_controls.py +++ b/litellm/integrations/otel/model/trace_controls.py @@ -4,10 +4,11 @@ from __future__ import annotations from collections.abc import Mapping from dataclasses import dataclass -from typing import Final +from typing import Final, Literal from pydantic import TypeAdapter, ValidationError +from litellm.integrations.langfuse.langfuse_otel_attributes import default_langfuse_tags from litellm.integrations.otel.model.utils import as_str, as_str_mapping LANGFUSE_HEADER_PREFIX: Final = "langfuse_" @@ -27,17 +28,15 @@ class TraceControls: tags: tuple[str, ...] = () -def caller_trace_controls(kwargs: Mapping[str, object]) -> TraceControls: - request: Final = as_str_mapping(kwargs.get("litellm_params")) - if request is None: - return TraceControls() +def caller_trace_controls( + kwargs: Mapping[str, object], *, langfuse_defaults: Literal["request", "completed"] | None = None +) -> TraceControls: + request: Final = as_str_mapping(kwargs.get("litellm_params")) or {} proxy_request: Final = as_str_mapping(request.get("proxy_server_request")) headers: Final = as_str_mapping(proxy_request.get("headers")) if proxy_request is not None else None - bodies: Final = tuple( - metadata - for key in ("metadata", "litellm_metadata") - if (metadata := as_str_mapping(request.get(key))) is not None - ) + user_metadata: Final = as_str_mapping(request.get("metadata")) or {} + model_metadata: Final = as_str_mapping(request.get("litellm_metadata")) or {} + bodies: Final = (user_metadata, model_metadata) def scalar(control: str) -> str | None: from_header: Final = as_str(headers.get(f"{LANGFUSE_HEADER_PREFIX}{control}")) if headers is not None else None @@ -45,11 +44,21 @@ def caller_trace_controls(kwargs: Mapping[str, object]) -> TraceControls: return from_header return next((value for body in bodies if (value := as_str(body.get(control)))), None) + caller_tags: Final = next((tags for body in bodies if (tags := _str_items(body.get("tags")))), ()) + defaults: Final = ( + default_langfuse_tags( + kwargs, + {**model_metadata, **user_metadata}, + completed=langfuse_defaults == "completed", + ) + if langfuse_defaults is not None + else () + ) return TraceControls( name=scalar("trace_name"), user_id=scalar("trace_user_id"), session_id=scalar("session_id"), - tags=next((tags for body in bodies if (tags := _str_items(body.get("tags")))), ()), + tags=tuple(dict.fromkeys(caller_tags + defaults)) if defaults else caller_tags, ) diff --git a/tests/unit/integrations/otel/test_langfuse_logger.py b/tests/unit/integrations/otel/test_langfuse_logger.py index 1ac6635b8ee..f341618c3f2 100644 --- a/tests/unit/integrations/otel/test_langfuse_logger.py +++ b/tests/unit/integrations/otel/test_langfuse_logger.py @@ -348,6 +348,8 @@ def _run_named_request( exporter: InMemorySpanExporter, litellm_params: Mapping[str, object], parent_context: Context | None = None, + *, + cache_hit: bool = False, ) -> tuple[Mapping[str, object], Mapping[str, object]]: response: Final = ModelResponse(choices=[Choices(message=Message(role="assistant", content="pong"))]) root: Final = _start_root(logger, parent_context) @@ -370,7 +372,10 @@ def _run_named_request( } asyncio.run( logger.async_log_success_event( - {"standard_logging_object": payload, "litellm_params": litellm_params}, response, None, None + {"standard_logging_object": payload, "litellm_params": litellm_params, "cache_hit": cache_hit}, + response, + None, + None, ) ) generation: Final = next( @@ -540,6 +545,28 @@ def test_body_metadata_user_session_and_tags_land_on_the_root_and_the_generation assert TRACE_NAME_ATTR not in generation_attrs +@pytest.mark.parametrize("metadata_key", ["metadata", "litellm_metadata"]) +@pytest.mark.parametrize( + ("caller_tags", "cache_hit"), [([], False), (["caller-tag", "user_api_key_alias:team-a"], True)] +) +def test_langfuse_default_tags_are_added_to_the_request_and_generation( + monkeypatch, metadata_key, caller_tags, cache_hit +): + monkeypatch.setattr(litellm, "langfuse_default_tags", ["user_api_key_alias", "cache_hit", "missing"]) + logger, exporter = _logger(capture="no_content") + + root_attrs, generation_attrs = _run_named_request( + logger, + exporter, + {metadata_key: {"user_api_key_alias": "team-a", "tags": caller_tags}}, + cache_hit=cache_hit, + ) + + request_tags = tuple(dict.fromkeys([*caller_tags, "user_api_key_alias:team-a"])) + assert tuple(root_attrs["langfuse.trace.tags"]) == request_tags + assert tuple(generation_attrs["langfuse.trace.tags"]) == (*request_tags, f"cache_hit:{cache_hit}") + + def test_langfuse_user_and_session_headers_beat_body_metadata_on_both_spans(): logger, exporter = _logger()