From 8f0788486f4b4aa01a058f3f49731d57719bb80f Mon Sep 17 00:00:00 2001 From: yucheng Date: Wed, 23 Sep 2026 08:04:13 +0000 Subject: [PATCH] fix(langfuse_otel): restore preset cache key fallback and type the derived-field helpers Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- .../integrations/langfuse/langfuse_otel.py | 119 +++++++++++++----- .../test_langfuse_otel_trace_fields.py | 5 +- .../integrations/test_langfuse_otel.py | 36 ++++++ 3 files changed, 122 insertions(+), 38 deletions(-) diff --git a/litellm/integrations/langfuse/langfuse_otel.py b/litellm/integrations/langfuse/langfuse_otel.py index 883a95a39c1..e564766eabb 100644 --- a/litellm/integrations/langfuse/langfuse_otel.py +++ b/litellm/integrations/langfuse/langfuse_otel.py @@ -1,8 +1,9 @@ import base64 import json import os +from collections.abc import Iterable, Mapping, Sequence from datetime import datetime -from typing import TYPE_CHECKING, Any, Final, Optional +from typing import TYPE_CHECKING, Any, Final, Optional, Protocol, runtime_checkable from litellm._logging import verbose_logger from litellm.integrations.arize import _utils @@ -30,6 +31,27 @@ LANGFUSE_INGESTION_VERSION_HEADER: Final = "x-langfuse-ingestion-version" LANGFUSE_INGESTION_VERSION: Final = "4" +@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: + import litellm + + if litellm.cache is None: + return None + return litellm.cache._get_preset_cache_key_from_kwargs(**kwargs) + + class LangfuseOtelLogger(OpenTelemetry): def __init__(self, config=None, *args, **kwargs): # Prevent LangfuseOtelLogger from modifying global environment variables by constructing config manually @@ -125,27 +147,34 @@ class LangfuseOtelLogger(OpenTelemetry): safe_set_attribute(span, enum_attr.value, value) @staticmethod - def _observation_output(response_obj) -> str | None: + def _observation_output(response_obj: _Gettable | None) -> str | None: """Serialized observation output, or None when the response yields nothing.""" if not response_obj or not hasattr(response_obj, "get"): return None return _extract_output_items(response_obj) or _extract_choices_output(response_obj) @staticmethod - def _trace_tags(kwargs: dict, metadata: dict, derive_defaults: bool = True) -> tuple[str, ...]: + def _trace_tags( + kwargs: Mapping[str, object], metadata: Mapping[str, object], derive_defaults: bool = True + ) -> tuple[str, ...]: """Order-preserving dedupe of caller tags, request tags and langfuse_default_tags expansions.""" import litellm caller_tags: Final = metadata.get("tags") - request_tags: Final = (kwargs.get("standard_logging_object") or {}).get("request_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 = metadata.get("hidden_params", {}) or {} - return f"cache_key:{hidden_params.get('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 @@ -411,24 +440,22 @@ class LangfuseOtelLogger(OpenTelemetry): """ -def _extract_choices_output(response_obj) -> str | None: +def _extract_choices_output(response_obj: _Gettable) -> str | None: from litellm.litellm_core_utils.safe_json_dumps import safe_dumps choices: Final = response_obj.get("choices", []) - if not choices: + if not isinstance(choices, list) or not choices: + return None + first_choice: Final = _as_gettable(choices[0]) + if first_choice is None: + return None + message: Final = _as_gettable(first_choice.get("message", {})) + if message is None: return None - message: Final = choices[0].get("message", {}) tool_calls: Final = message.get("tool_calls") - if tool_calls: + if isinstance(tool_calls, list) and tool_calls: transformed_tool_calls: Final = [ - { - "id": response_obj.get("id", ""), - "name": tool_call.get("function", {}).get("name", ""), - "call_id": tool_call.get("id", ""), - "type": "function_call", - "arguments": _tool_call_arguments(tool_call.get("function", {}).get("arguments", "{}")), - } - for tool_call in tool_calls + entry for tool_call in tool_calls if (entry := _transformed_tool_call(response_obj, tool_call)) is not None ] return safe_dumps(transformed_tool_calls) output_data: Final = { @@ -442,6 +469,22 @@ def _extract_choices_output(response_obj) -> str | None: return safe_dumps(output_data) if output_data else None +def _transformed_tool_call(response_obj: _Gettable, tool_call: object) -> dict[str, object] | None: + call: Final = _as_gettable(tool_call) + if call is None: + return None + function: Final = _as_gettable(call.get("function", {})) + if function is None: + return None + return { + "id": response_obj.get("id", ""), + "name": function.get("name", ""), + "call_id": call.get("id", ""), + "type": "function_call", + "arguments": _tool_call_arguments(function.get("arguments", "{}")), + } + + def _tool_call_arguments(arguments: object) -> object: if not isinstance(arguments, str): return arguments @@ -451,39 +494,47 @@ def _tool_call_arguments(arguments: object) -> object: return {} -def _extract_output_items(response_obj) -> str | None: +def _extract_output_items(response_obj: _Gettable) -> str | None: from litellm.litellm_core_utils.safe_json_dumps import safe_dumps output: Final = response_obj.get("output", []) - if not output: + if not isinstance(output, list) or not output: return None rendered: Final = tuple(entry for item in output for entry in _output_items(item)) return safe_dumps(list(rendered)) if rendered else None -def _output_items(item) -> tuple[dict, ...]: - if not hasattr(item, "type"): - return () - if item.type == "reasoning" and hasattr(item, "summary"): +def _output_items(item: object) -> tuple[dict[str, object], ...]: + item_type: Final = _attr(item, "type") + if item_type == "reasoning": + summaries: Final = _attr(item, "summary") + if not isinstance(summaries, Iterable): + return () return tuple( - {"role": "reasoning_summary", "content": summary.text} - for summary in item.summary + {"role": "reasoning_summary", "content": _attr(summary, "text")} + for summary in summaries if hasattr(summary, "text") ) - if item.type == "message": + if item_type == "message": + content_items: Final = _attr(item, "content", [{}]) + first_content: Final = ( + content_items[0] + if isinstance(content_items, Sequence) and not isinstance(content_items, (str, bytes)) and content_items + else {} + ) return ( { - "role": getattr(item, "role", "assistant"), - "content": getattr(getattr(item, "content", [{}])[0], "text", ""), + "role": _attr(item, "role", "assistant"), + "content": _attr(first_content, "text", ""), }, ) - if item.type == "function_call": - arguments: Final = getattr(item, "arguments", "{}") + if item_type == "function_call": + arguments: Final = _attr(item, "arguments", "{}") return ( { - "id": getattr(item, "id", ""), - "name": getattr(item, "name", ""), - "call_id": getattr(item, "call_id", ""), + "id": _attr(item, "id", ""), + "name": _attr(item, "name", ""), + "call_id": _attr(item, "call_id", ""), "type": "function_call", "arguments": safe_json_loads(arguments, default={}) if isinstance(arguments, str) else arguments, }, diff --git a/tests/integration/observability/test_langfuse_otel_trace_fields.py b/tests/integration/observability/test_langfuse_otel_trace_fields.py index 3a1cacf6e97..44805197b2e 100644 --- a/tests/integration/observability/test_langfuse_otel_trace_fields.py +++ b/tests/integration/observability/test_langfuse_otel_trace_fields.py @@ -113,13 +113,10 @@ def _chat_stream(marker: str) -> tuple[bytes, ...]: def _generation_span(collector: Wire, marker: str) -> SpanRecord: - batches: list[Request] = [] # mutable-ok: accumulated across eventually() polls - def spans() -> tuple[SpanRecord, ...]: - batches.extend(collector.drain()) return tuple( record - for request in batches + for request in collector.drain() if request.target.endswith("/v1/traces") for record in _span_records(request.body) if record.attributes.get("langfuse.observation.type") == "generation" diff --git a/tests/test_litellm/integrations/test_langfuse_otel.py b/tests/test_litellm/integrations/test_langfuse_otel.py index 9945355505d..57600e2ebd2 100644 --- a/tests/test_litellm/integrations/test_langfuse_otel.py +++ b/tests/test_litellm/integrations/test_langfuse_otel.py @@ -1105,6 +1105,42 @@ class TestDerivedTraceFields: "user_api_key_alias:k1", ] + def test_cache_key_default_tag_falls_back_to_preset_cache_key(self, monkeypatch): + import litellm + + class _PresetCache: + @staticmethod + def _get_preset_cache_key_from_kwargs(**kwargs) -> str: + return "preset-abc" + + monkeypatch.setattr(litellm, "langfuse_default_tags", ["cache_key"]) + monkeypatch.setattr(litellm, "cache", _PresetCache()) + attributes = _emitted( + { + "call_type": "acompletion", + "litellm_params": {"metadata": {}}, + } + ) + assert json.loads(attributes["langfuse.trace.tags"]) == ["cache_key:preset-abc"] + + def test_cache_key_default_tag_prefers_hidden_params_over_preset(self, monkeypatch): + import litellm + + class _PresetCache: + @staticmethod + def _get_preset_cache_key_from_kwargs(**kwargs) -> str: + return "preset-abc" + + monkeypatch.setattr(litellm, "langfuse_default_tags", ["cache_key"]) + monkeypatch.setattr(litellm, "cache", _PresetCache()) + attributes = _emitted( + { + "call_type": "acompletion", + "litellm_params": {"metadata": {"hidden_params": {"cache_key": "explicit-key"}}}, + } + ) + assert json.loads(attributes["langfuse.trace.tags"]) == ["cache_key:explicit-key"] + def test_caller_tags_string_is_a_single_tag(self): attributes = _emitted( {