Merge 084be77643 into 36c64ce85d

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))
 

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

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,
                 )
             )

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.

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,

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,
     )
 
 

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()