diff --git a/litellm/integrations/langfuse/langfuse_otel.py b/litellm/integrations/langfuse/langfuse_otel.py index 9364a3d4216..34d2c9fce0d 100644 --- a/litellm/integrations/langfuse/langfuse_otel.py +++ b/litellm/integrations/langfuse/langfuse_otel.py @@ -1,19 +1,32 @@ import base64 import json import os -from collections.abc import Mapping, Sequence +from collections.abc import Iterable, Mapping, Sequence from datetime import datetime -from typing import TYPE_CHECKING, Any, Final, Optional +from itertools import chain +from typing import TYPE_CHECKING, Any, Final, Optional, Protocol, runtime_checkable +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 ( LangfuseLLMObsOTELAttributes, serialize_langfuse_datetime, ) -from litellm.integrations.opentelemetry import OpenTelemetry, OpenTelemetryConfig +from litellm.integrations.opentelemetry import ( + HTTP_RESPONSE_STATUS_CODE_ATTRIBUTE, + LITELLM_PROXY_REQUEST_SPAN_NAME, + OpenTelemetry, + OpenTelemetryConfig, +) from litellm.litellm_core_utils.core_helpers import as_str_mapping +from litellm.litellm_core_utils.safe_json_dumps import safe_dumps from litellm.litellm_core_utils.safe_json_loads import safe_json_loads +from litellm.secret_managers.main import str_to_bool from litellm.types.integrations.langfuse_otel import ( LangfuseSpanAttributes, ) @@ -31,6 +44,86 @@ 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" +_ERROR_ATTRIBUTES: Final = frozenset( + { + ErrorAttributes.ERROR_CODE, + HTTP_RESPONSE_STATUS_CODE_ATTRIBUTE, + ErrorAttributes.ERROR_TYPE, + ErrorAttributes.ERROR_MESSAGE, + ErrorAttributes.ERROR_LLM_PROVIDER, + ErrorAttributes.ERROR_STACK_TRACE, + } +) + + +@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 + + +def _caller_tag_list(value: object) -> tuple[str, ...]: + parsed: Final = ( + safe_json_loads(value, default=value) if isinstance(value, str) and value.strip().startswith("[") else value + ) + if isinstance(parsed, list): + return tuple(tag if isinstance(tag, str) else json.dumps(tag, default=str) for tag in parsed) + if isinstance(parsed, str) and parsed: + return tuple(tag.strip() for tag in parsed.split(",")) if "," in parsed else (parsed,) + return () + + +def _parent_of(span: object) -> trace.SpanContext | None: + parent: Final = getattr(span, "parent", None) + return parent if isinstance(parent, trace.SpanContext) else None + + +def _litellm_proxy_span(kwargs: Mapping[str, object]) -> trace.Span | None: + litellm_params: Final = _as_gettable(kwargs.get("litellm_params")) or {} + carried: Final = ( + (_as_gettable(litellm_params.get(key)) or {}).get("litellm_parent_otel_span") + for key in ("metadata", "litellm_metadata") + ) + return next( + ( + span + for span in carried + if isinstance(span, trace.Span) and getattr(span, "name", None) == LITELLM_PROXY_REQUEST_SPAN_NAME + ), + None, + ) + + +def _starts_its_own_trace(span: object, kwargs: Mapping[str, object]) -> bool: + if span is trace.get_current_span(): + return False + parent: Final = _parent_of(span) + if parent is None: + return True + proxy_span: Final = _litellm_proxy_span(kwargs) + return ( + proxy_span is not None + and _parent_of(proxy_span) is None + and proxy_span.get_span_context().span_id == parent.span_id + ) class LangfuseOtelLogger(OpenTelemetry): @@ -129,98 +222,42 @@ class LangfuseOtelLogger(OpenTelemetry): safe_set_attribute(span, enum_attr.value, value) @staticmethod - def _set_observation_output(span: Span, response_obj): - """Helper to set observation output attributes.""" - from litellm.integrations.arize._utils import safe_set_attribute - from litellm.litellm_core_utils.safe_json_dumps import safe_dumps - + def _observation_output(response_obj: _Gettable | None) -> str | None: if not response_obj or not hasattr(response_obj, "get"): - return + return None + return _extract_output_items(response_obj) or _extract_choices_output(response_obj) - choices: Final = response_obj.get("choices", []) - if choices: - first_choice: Final = choices[0] - message: Final = first_choice.get("message", {}) - tool_calls: Final = message.get("tool_calls") - if tool_calls: - transformed_tool_calls: Final = [] - for tool_call in tool_calls: - function = tool_call.get("function", {}) - arguments_str = function.get("arguments", "{}") - try: - arguments_obj = json.loads(arguments_str) if isinstance(arguments_str, str) else arguments_str - except json.JSONDecodeError: - arguments_obj = {} - langfuse_tool_call = { - "id": response_obj.get("id", ""), - "name": function.get("name", ""), - "call_id": tool_call.get("id", ""), - "type": "function_call", - "arguments": arguments_obj, - } - transformed_tool_calls.append(langfuse_tool_call) - safe_set_attribute( - span, - LangfuseSpanAttributes.OBSERVATION_OUTPUT.value, - safe_dumps(transformed_tool_calls), - ) - else: - output_data: Final = {} - if message.get("role"): - output_data["role"] = message.get("role") - if message.get("content") is not None: - output_data["content"] = message.get("content") - if output_data: - safe_set_attribute( - span, - LangfuseSpanAttributes.OBSERVATION_OUTPUT.value, - safe_dumps(output_data), - ) + @staticmethod + 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 - output: Final = response_obj.get("output", []) - if output: - output_items_data: Final[list[dict]] = [] - for item in output: - if hasattr(item, "type"): - item_type = item.type - if item_type == "reasoning" and hasattr(item, "summary"): - for summary in item.summary: - if hasattr(summary, "text"): - output_items_data.append( - { - "role": "reasoning_summary", - "content": summary.text, - } - ) - elif item_type == "message": - item_content: Sequence[object] = getattr(item, "content", [{}]) - output_items_data.append( - { - "role": getattr(item, "role", "assistant"), - "content": getattr(item_content[0], "text", ""), - } - ) - elif item_type == "function_call": - item_arguments: object = getattr(item, "arguments", "{}") - arguments_obj = ( - safe_json_loads(item_arguments, default={}) - if isinstance(item_arguments, str) - else item_arguments - ) - langfuse_tool_call = { - "id": getattr(item, "id", ""), - "name": getattr(item, "name", ""), - "call_id": getattr(item, "call_id", ""), - "type": "function_call", - "arguments": arguments_obj, - } - output_items_data.append(langfuse_tool_call) - if output_items_data: - safe_set_attribute( - span, - LangfuseSpanAttributes.OBSERVATION_OUTPUT.value, - safe_dumps(output_items_data), + 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) + ) + return tuple(dict.fromkeys(candidates)) @staticmethod def _set_langfuse_specific_attributes(span: Span, kwargs: Mapping[str, object], response_obj): @@ -249,13 +286,12 @@ class LangfuseOtelLogger(OpenTelemetry): metadata: Final = LangfuseOtelLogger._extract_langfuse_metadata(kwargs) LangfuseOtelLogger._set_metadata_attributes(span=span, metadata=metadata) - messages: Final = kwargs.get("messages") - if messages: - safe_set_attribute( - span, - LangfuseSpanAttributes.OBSERVATION_INPUT.value, - safe_dumps(messages), - ) + joins_existing_trace: Final = metadata.get("existing_trace_id") is not None or not _starts_its_own_trace( + span, kwargs + ) + input_json: Final = safe_dumps(kwargs.get("messages")) if kwargs.get("messages") else None + if input_json is not None: + safe_set_attribute(span, LangfuseSpanAttributes.OBSERVATION_INPUT.value, input_json) completion_start_time: Final = kwargs.get("completion_start_time") if isinstance(completion_start_time, datetime): @@ -265,7 +301,64 @@ class LangfuseOtelLogger(OpenTelemetry): serialize_langfuse_datetime(completion_start_time), ) - LangfuseOtelLogger._set_observation_output(span=span, response_obj=response_obj) + output_json: Final = LangfuseOtelLogger._observation_output(response_obj) + if output_json is not None: + safe_set_attribute(span, LangfuseSpanAttributes.OBSERVATION_OUTPUT.value, output_json) + + if not joins_existing_trace: + LangfuseOtelLogger._set_trace_fields( + span=span, kwargs=kwargs, metadata=metadata, input_json=input_json, output_json=output_json + ) + + @staticmethod + def _set_trace_fields( + span: Span, + kwargs: Mapping[str, object], + metadata: Mapping[str, object], + input_json: str | None, + output_json: str | None, + ) -> None: + from litellm.integrations.otel.emitter import carried_keys, span_attribute_limit + + tags: Final = LangfuseOtelLogger._trace_tags(kwargs, metadata) + fields: Final = ( + *( + ((LangfuseSpanAttributes.TRACE_NAME.value, f"litellm-{kwargs.get('call_type') or 'completion'}"),) + if metadata.get("trace_name") is None + else () + ), + *(((LangfuseSpanAttributes.TAGS.value, json.dumps(list(tags))),) if tags else ()), + *( + ( + ( + LangfuseSpanAttributes.TRACE_INPUT.value, + REDACTED_BY_LITELLM if _mask_flag(metadata.get("mask_input")) else input_json, + ), + ) + if input_json is not None + else () + ), + *( + ( + ( + LangfuseSpanAttributes.TRACE_OUTPUT.value, + REDACTED_BY_LITELLM if _mask_flag(metadata.get("mask_output")) else output_json, + ), + ) + if output_json is not None + else () + ), + ) + limit: Final = span_attribute_limit(span) + carried: Final = carried_keys(span) + new_keys: Final = tuple(key for key, _value in fields if key not in carried) + room: Final = ( + len(new_keys) if limit is None else max(limit - len(carried) - len(_ERROR_ATTRIBUTES - carried), 0) + ) + written: Final = carried | frozenset(new_keys[:room]) + for key, value in fields: + if key in written: + _utils.safe_set_attribute(span, key, value) @staticmethod def _get_langfuse_otel_host() -> str | None: @@ -457,3 +550,101 @@ class LangfuseOtelLogger(OpenTelemetry): """ Langfuse should not receive service failure logs. """ + + +def _extract_choices_output(response_obj: _Gettable) -> str | None: + choices: Final = response_obj.get("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 + tool_calls: Final = message.get("tool_calls") + if isinstance(tool_calls, list) and tool_calls: + transformed_tool_calls: Final = [ + 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 = { + key: value + for key, value in ( + ("role", message.get("role") or None), + ("content", message.get("content")), + ) + if value is not 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 + try: + return json.loads(arguments) + except json.JSONDecodeError: + return {} + + +def _extract_output_items(response_obj: _Gettable) -> str | None: + output: Final = response_obj.get("output", []) + if not isinstance(output, list) or not output: + return None + rendered: Final = tuple(chain.from_iterable(map(_output_items, output))) + return safe_dumps(list(rendered)) if rendered else None + + +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": _attr(summary, "text")} + for summary in summaries + if hasattr(summary, "text") + ) + 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": _attr(item, "role", "assistant"), + "content": _attr(first_content, "text", ""), + }, + ) + if item_type == "function_call": + arguments: Final = _attr(item, "arguments", "{}") + return ( + { + "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, + }, + ) + return () diff --git a/litellm/integrations/otel/langfuse_logger.py b/litellm/integrations/otel/langfuse_logger.py index c869474f532..15b0f3f0cc2 100644 --- a/litellm/integrations/otel/langfuse_logger.py +++ b/litellm/integrations/otel/langfuse_logger.py @@ -1,18 +1,25 @@ from collections.abc import AsyncGenerator, AsyncIterator, Callable, Mapping from typing import TYPE_CHECKING, Final +from opentelemetry.sdk.trace import ReadableSpan + from litellm._logging import verbose_logger from litellm.integrations.otel.logger import OpenTelemetryV2 from litellm.integrations.otel.mappers.langfuse import ( LANGFUSE_OBSERVATION_INPUT, LANGFUSE_OBSERVATION_OUTPUT, + LANGFUSE_TRACE_NAME, LangfuseMapper, ) +from litellm.integrations.otel.model.metadata import internal_call_origin from litellm.integrations.otel.model.request_io import request_input, response_output, stream_output from litellm.integrations.otel.model.trace_controls import caller_trace_controls +from litellm.integrations.otel.model.utils import as_str from litellm.integrations.otel.plumbing.context import request_root_span if TYPE_CHECKING: + from opentelemetry.trace import Span + from litellm.types.proxy.auth.user_api_key_auth import UserAPIKeyAuth from litellm.types.utils import ModelResponseStream @@ -24,9 +31,23 @@ class LangfuseOpenTelemetryV2(OpenTelemetryV2): def log_pre_api_call(self, model: str, messages: object, kwargs: Mapping[str, object]) -> None: root: Final = request_root_span() if root is not None and root.is_recording(): - root.set_attributes(LangfuseMapper.trace_attributes(caller_trace_controls(kwargs))) + root.set_attributes( + LangfuseMapper.trace_attributes( + caller_trace_controls(kwargs), + as_str(kwargs.get("call_type")) if self._derives_trace_name(root, kwargs) else None, + ) + ) super().log_pre_api_call(model, messages, kwargs) + def _derives_trace_name(self, root: "Span", kwargs: Mapping[str, object]) -> bool: + return ( + internal_call_origin(None, kwargs) is None + and isinstance(root, ReadableSpan) + and root.parent is None + and root.resource is self.tracer_provider.resource + and LANGFUSE_TRACE_NAME not in (root.attributes or {}) + ) + class LangfuseContentOpenTelemetryV2(LangfuseOpenTelemetryV2): """Stamps the request's input and output on the root observation while it is still recording. diff --git a/litellm/integrations/otel/mappers/langfuse.py b/litellm/integrations/otel/mappers/langfuse.py index e60ea539188..230a2e261fe 100644 --- a/litellm/integrations/otel/mappers/langfuse.py +++ b/litellm/integrations/otel/mappers/langfuse.py @@ -96,10 +96,10 @@ class LangfuseMapper: return {} @staticmethod - def trace_attributes(trace: TraceControls) -> AttributeMap: + def trace_attributes(trace: TraceControls, call_type: str | None = None) -> AttributeMap: return drop_none_pairs( ( - (LANGFUSE_TRACE_NAME, trace.name or None), + (LANGFUSE_TRACE_NAME, trace.name or (f"litellm-{call_type}" if call_type else None)), (LANGFUSE_TRACE_USER_ID, trace.user_id or None), (LANGFUSE_TRACE_SESSION_ID, trace.session_id or None), (LANGFUSE_TRACE_TAGS, trace.tags or None), diff --git a/litellm/types/integrations/langfuse_otel.py b/litellm/types/integrations/langfuse_otel.py index 5647f28c757..459f2e984a4 100644 --- a/litellm/types/integrations/langfuse_otel.py +++ b/litellm/types/integrations/langfuse_otel.py @@ -32,6 +32,8 @@ class LangfuseSpanAttributes(str, Enum): OBSERVATION_COMPLETION_START_TIME = "langfuse.observation.completion_start_time" # ---- Trace-level metadata ---- + TRACE_INPUT = "langfuse.trace.input" + TRACE_OUTPUT = "langfuse.trace.output" TRACE_USER_ID = "user.id" SESSION_ID = "session.id" TAGS = "langfuse.trace.tags" diff --git a/tests/integration/observability/test_langfuse_otel_trace_fields.py b/tests/integration/observability/test_langfuse_otel_trace_fields.py new file mode 100644 index 00000000000..2b78b402926 --- /dev/null +++ b/tests/integration/observability/test_langfuse_otel_trace_fields.py @@ -0,0 +1,660 @@ +import json +import threading +import uuid +from itertools import product +from collections.abc import Iterator, Mapping, Sequence +from dataclasses import dataclass +from pathlib import Path +from typing import Final + +import pytest +import yaml +from _s3_v2_support import SURFACES, call_surface, mixed_burst, surface_reply +from integration._support.client import Gateway, eventually, gateway_from_environment +from integration._support.process import owned_proxy +from integration._support.wire import Reply, Request, wire_server +from opentelemetry.proto.collector.trace.v1.trace_service_pb2 import ExportTraceServiceRequest +from opentelemetry.proto.common.v1.common_pb2 import AnyValue +from opentelemetry.proto.trace.v1.trace_pb2 import Span + +REDACTED: Final = "redacted-by-litellm" +TRACE_FIELDS: Final = ("langfuse.trace.name", "langfuse.trace.input", "langfuse.trace.output", "langfuse.trace.tags") +SURFACE_CALL_TYPES: Final = { + "chat": "acompletion", + "chat_stream": "acompletion", + "messages": "anthropic_messages", + "messages_stream": "anthropic_messages", + "responses": "aresponses", + "responses_stream": "aresponses", +} +ANTHROPIC_MODEL: Final = "anthropic/claude-sonnet-4-5-20250929" +JUDGE_MODEL: Final = "lit8281-judge-model" +JUDGE_GUARDRAIL: Final = "lit8281-judge" + + +@dataclass(frozen=True, slots=True) +class SpanRecord: + name: str + trace_id: str + parent_span_id: str + attributes: Mapping[str, object] + dropped_attributes: int = 0 + + def tags(self) -> tuple[object, ...]: + raw: Final = self.attributes.get("langfuse.trace.tags") + if isinstance(raw, tuple): + return raw + if not isinstance(raw, str) or not raw: + return () + if raw.strip().startswith("["): + return tuple(json.loads(raw)) + return tuple(tag.strip() for tag in raw.split(",")) if "," in raw else (raw,) + + +def _attribute_value(value: AnyValue) -> object: + kind: Final = value.WhichOneof("value") + if kind == "array_value": + return tuple(_attribute_value(item) for item in value.array_value.values) + return getattr(value, kind) if kind is not None else None + + +def _exported_spans(body: bytes) -> Iterator[Span]: + for resource in ExportTraceServiceRequest.FromString(body).resource_spans: + for scope in resource.scope_spans: + yield from scope.spans + + +def _span_records(body: bytes) -> tuple[SpanRecord, ...]: + return tuple( + SpanRecord( + name=span.name, + trace_id=span.trace_id.hex(), + parent_span_id=span.parent_span_id.hex(), + attributes={pair.key: _attribute_value(pair.value) for pair in span.attributes}, + dropped_attributes=span.dropped_attributes_count, + ) + for span in _exported_spans(body) + ) + + +class LangfuseSink: + def __init__(self) -> None: + self.lock: Final = threading.Lock() + self.outage: Final = threading.Event() + self.accepted: Final[list[SpanRecord]] = [] # mutable-ok: spans accumulate as the exporter delivers them + self.roots: Final[list[SpanRecord]] = [] # mutable-ok: root spans accumulate as the exporter delivers them + self.spans: Final[list[SpanRecord]] = [] # mutable-ok: every span accumulates as the exporter delivers it + self.refusals: Final[list[int]] = [] # mutable-ok: one entry per batch refused during the outage + + def respond(self, request: Request) -> Reply: + if not request.target.endswith("/v1/traces"): + return Reply(body=b"") + assert request.headers.get("x-langfuse-ingestion-version") == "4", request.headers + if self.outage.is_set(): + with self.lock: + self.refusals.append(1) + return Reply(status=503, body=b'{"error": "scripted outage"}') + records: Final = _span_records(request.body) + with self.lock: + self.accepted.extend( + record for record in records if record.attributes.get("langfuse.observation.type") == "generation" + ) + self.roots.extend(record for record in records if not record.parent_span_id) + self.spans.extend(records) + return Reply(body=b"") + + def matching(self, marker: str) -> tuple[SpanRecord, ...]: + with self.lock: + snapshot: Final = tuple(self.accepted) + return tuple(record for record in snapshot if marker in str(tuple(record.attributes.values()))) + + def one(self, marker: str) -> SpanRecord: + return eventually(lambda: self.matching(marker), lambda found: len(found) == 1, seconds=70)[0] + + def in_trace(self, trace_id: str) -> tuple[SpanRecord, ...]: + with self.lock: + return tuple(record for record in self.spans if record.trace_id == trace_id) + + def root_of(self, span: SpanRecord) -> SpanRecord: + def roots() -> tuple[SpanRecord, ...]: + with self.lock: + return tuple(root for root in self.roots if root.trace_id == span.trace_id) + + return eventually(roots, lambda found: len(found) == 1, seconds=70)[0] + + +class OtlpSink: + def __init__(self) -> None: + self.lock: Final = threading.Lock() + self.spans: Final[list[SpanRecord]] = [] # mutable-ok: spans accumulate as the exporter delivers them + + def respond(self, request: Request) -> Reply: + if request.target.endswith("/v1/traces"): + records: Final = _span_records(request.body) + with self.lock: + self.spans.extend(records) + return Reply(body=b"") + + def root_of_request(self, marker: str) -> SpanRecord: + + def roots() -> tuple[SpanRecord, ...]: + with self.lock: + snapshot: Final = tuple(self.spans) + traces: Final = {record.trace_id for record in snapshot if marker in str(tuple(record.attributes.values()))} + return tuple(record for record in snapshot if record.trace_id in traces and not record.parent_span_id) + + return eventually(roots, lambda found: len(found) == 1, seconds=70)[0] + + +def _chat_completion(marker: str, text: str = "langfuse-otel scripted reply") -> bytes: + return json.dumps( + { + "id": marker, + "object": "chat.completion", + "created": 1, + "model": "gpt-4o-mini", + "choices": [{"index": 0, "message": {"role": "assistant", "content": text}, "finish_reason": "stop"}], + "usage": {"prompt_tokens": 5, "completion_tokens": 4, "total_tokens": 9}, + } + ).encode() + + +def _provider(request: Request) -> Reply: + if request.method != "POST" or not request.body: + return Reply(body=b'{"object": "list", "data": []}') + body: Final = json.loads(request.body) + if request.target.endswith("/embeddings"): + return Reply( + body=json.dumps( + { + "object": "list", + "data": [{"object": "embedding", "index": 0, "embedding": [0.25, 0.5]}], + "model": "text-embedding-3-small", + "usage": {"prompt_tokens": 3, "total_tokens": 3}, + } + ).encode() + ) + if body.get("response_format") == {"type": "json_object"}: + return Reply(body=_chat_completion("lit8281-judge-verdict", json.dumps({"overall_score": 100}))) + messages: Final = body.get("messages") + first: Final = messages[0]["content"] if isinstance(messages, list) and messages else None + if isinstance(first, str) and first.startswith("fail-"): + return Reply(status=500, body=b'{"error": {"message": "scripted provider failure", "type": "server_error"}}') + return surface_reply(request) + + +def _langfuse_config(directory: Path, name: str, settings: Mapping[str, object], provider_url: str) -> Path: + base: Final = yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text()) + judge: Final = { + "model_name": JUDGE_MODEL, + "litellm_params": { + "model": "openai/gpt-4o-mini", + "api_base": provider_url + "/v1", + "api_key": "synthetic-provider-key", + }, + } + config: Final = { + **base, + "model_list": [*base["model_list"], judge], + "guardrails": [ + { + "guardrail_name": JUDGE_GUARDRAIL, + "litellm_params": { + "guardrail": "llm_as_a_judge", + "mode": "post_call", + "judge_model": JUDGE_MODEL, + "on_failure": "log", + "criteria": [{"name": "helpful", "description": "The reply answers the user", "weight": 100}], + }, + } + ], + "litellm_settings": {**base["litellm_settings"], "callbacks": ["langfuse_otel"], **settings}, + } + path: Final = directory / f"{name}.yaml" + path.write_text(yaml.safe_dump(config)) + return path + + +@dataclass(frozen=True, slots=True) +class Rig: + proxy: Gateway + sink: LangfuseSink + openai_model: str + anthropic_model: str + embedding_model: str + key: str + + +def _rig( + tmp_path_factory: pytest.TempPathFactory, name: str, environment: Mapping[str, str], settings: Mapping[str, object] +) -> Iterator[Rig]: + directory: Final = tmp_path_factory.mktemp(name) + sink: Final = LangfuseSink() + with ( + wire_server(sink.respond) as destination, + wire_server(_provider) as provider, + gateway_from_environment() as gateway, + owned_proxy( + gateway, + directory, + { + "LANGFUSE_PUBLIC_KEY": "pk-lf-test", + "LANGFUSE_SECRET_KEY": "sk-lf-test", + "LANGFUSE_HOST": destination.url, + **environment, + }, + config=_langfuse_config(directory, name, settings, provider.url), + workers=2, + ) as proxy, + proxy.scenario() as scenario, + ): + openai_model: Final = scenario.model(api_base=provider.url + "/v1", api_key="synthetic-provider-key") + anthropic_model: Final = scenario.model( + model=ANTHROPIC_MODEL, api_base=provider.url, api_key="synthetic-provider-key" + ) + embedding_model: Final = scenario.model( + model="openai/text-embedding-3-small", + api_base=provider.url + "/v1", + api_key="synthetic-provider-key", + model_info={"mode": "embedding"}, + ) + key: Final = scenario.key(models=[openai_model, anthropic_model, embedding_model]) + yield Rig(proxy, sink, openai_model, anthropic_model, embedding_model, key) + + +SPAN_ATTRIBUTE_LIMIT: Final = 128 + + +@pytest.fixture(scope="module") +def v1(tmp_path_factory: pytest.TempPathFactory) -> Iterator[Rig]: + yield from _rig( + tmp_path_factory, + "langfuse-v1", + {"OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT": str(SPAN_ATTRIBUTE_LIMIT)}, + {"langfuse_default_tags": ["cache_hit"]}, + ) + + +V2_ENVIRONMENT: Final = {"LITELLM_OTEL_V2": "1", "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT": "span_only"} + + +@pytest.fixture(scope="module") +def v2(tmp_path_factory: pytest.TempPathFactory) -> Iterator[Rig]: + yield from _rig(tmp_path_factory, "langfuse-v2", V2_ENVIRONMENT, {}) + + +@pytest.fixture(scope="module") +def v2_llm_only(tmp_path_factory: pytest.TempPathFactory) -> Iterator[Rig]: + yield from _rig( + tmp_path_factory, "langfuse-v2-llm-only", {**V2_ENVIRONMENT, "LITELLM_OTEL_LANGFUSE_SPAN_SCOPE": "llm_only"}, {} + ) + + +@dataclass(frozen=True, slots=True) +class SharedRig: + rig: Rig + otel: OtlpSink + + +@pytest.fixture(scope="module") +def v2_after_otel(tmp_path_factory: pytest.TempPathFactory) -> Iterator[SharedRig]: + otel: Final = OtlpSink() + with wire_server(otel.respond) as destination: + environment: Final = { + **V2_ENVIRONMENT, + "OTEL_EXPORTER": "otlp_http", + "OTEL_ENDPOINT": destination.url + "/v1/traces", + } + for rig in _rig( + tmp_path_factory, "langfuse-v2-after-otel", environment, {"callbacks": ["otel", "langfuse_otel"]} + ): + yield SharedRig(rig, otel) + + +def _chat( + rig: Rig, + content: str, + metadata: Mapping[str, object] | None = None, + headers: Mapping[str, str] | None = None, + guardrails: Sequence[str] = (), +) -> str: + response: Final = rig.proxy.request( + "POST", + "/v1/chat/completions", + { + "model": rig.openai_model, + "messages": [{"role": "user", "content": content}], + **({"metadata": dict(metadata)} if metadata is not None else {}), + **({"guardrails": list(guardrails)} if guardrails else {}), + }, + key=rig.key, + headers=headers, + ) + assert response.status_code == 200, response.text + return response.text + + +def _trace_fields(spans: tuple[SpanRecord, ...]) -> Iterator[str]: + for record in spans: + yield from (key for key in record.attributes if key in TRACE_FIELDS) + + +def _assert_derived(span: SpanRecord, call_type: str) -> None: + attributes: Final = span.attributes + assert attributes.get("langfuse.trace.name") == f"litellm-{call_type}", attributes + assert attributes.get("langfuse.trace.input") == attributes.get("langfuse.observation.input"), attributes + assert attributes.get("langfuse.trace.output") == attributes.get("langfuse.observation.output"), attributes + tags: Final = span.tags() + assert any(str(tag).startswith("User-Agent: ") for tag in tags), tags + assert len(tags) == len(set(map(str, tags))), tags + + +@pytest.mark.parametrize("surface", SURFACES) +def test_every_surface_derives_trace_name_io_and_tags(v1: Rig, surface: str) -> None: + marker: Final = f"lit8281-{surface}-{uuid.uuid4().hex}" + call_surface(v1.proxy, surface, v1.openai_model, v1.anthropic_model, v1.key, marker) + span: Final = v1.sink.one(marker) + _assert_derived(span, SURFACE_CALL_TYPES[surface]) + assert marker in str(span.attributes["langfuse.trace.input"]), span.attributes + assert span.attributes.get("langfuse.trace.output") is not None, span.attributes + assert any(str(tag).startswith("cache_hit:") for tag in span.tags()), span.tags() + + +def test_inbound_traceparent_joins_the_caller_trace_without_derived_trace_fields(v1: Rig) -> None: + marker: Final = "lit8281-parent-" + uuid.uuid4().hex + trace_id: Final = uuid.uuid4().hex + parent_id: Final = uuid.uuid4().hex[:16] + _chat(v1, marker, headers={"traceparent": f"00-{trace_id}-{parent_id}-01"}) + span: Final = v1.sink.one(marker) + assert (span.trace_id, span.parent_span_id) == (trace_id, parent_id), span + assert not set(TRACE_FIELDS) & set(span.attributes), span.attributes + assert marker in str(span.attributes.get("langfuse.observation.input")), span.attributes + assert span.attributes.get("langfuse.observation.output") is not None, span.attributes + + +def test_inbound_traceparent_still_carries_the_caller_trace_name_and_tags(v1: Rig) -> None: + marker: Final = "lit8281-parent-caller-" + uuid.uuid4().hex + trace_id: Final = uuid.uuid4().hex + _chat( + v1, + marker, + metadata={"trace_name": "caller-trace", "tags": ["caller-tag"]}, + headers={"traceparent": f"00-{trace_id}-{uuid.uuid4().hex[:16]}-01"}, + ) + span: Final = v1.sink.one(marker) + assert span.trace_id == trace_id, span + assert span.attributes.get("langfuse.trace.name") == "caller-trace", span.attributes + assert span.tags() == ("caller-tag",), span.attributes + assert not {"langfuse.trace.input", "langfuse.trace.output"} & set(span.attributes), span.attributes + + +def test_caller_trace_name_and_tags_win_over_derived_values(v1: Rig) -> None: + marker: Final = "lit8281-caller-" + uuid.uuid4().hex + _chat(v1, marker, metadata={"trace_name": "caller-trace", "tags": ["caller-tag", "second-tag"]}) + span: Final = v1.sink.one(marker) + assert span.attributes["langfuse.trace.name"] == "caller-trace", span.attributes + assert span.tags()[:2] == ("caller-tag", "second-tag"), span.tags() + assert span.attributes.get("langfuse.trace.input") == span.attributes.get("langfuse.observation.input") + + +@pytest.mark.parametrize( + ("metadata", "headers", "expected"), + [ + pytest.param({"tags": "c1, c2"}, None, ("c1", "c2"), id="body_comma_string"), + pytest.param({"tags": '["j1", "j2"]'}, None, ("j1", "j2"), id="body_json_array_string"), + pytest.param(None, {"langfuse_tags": "h1,h2"}, ("h1", "h2"), id="header_comma_string"), + ], +) +def test_string_caller_tags_keep_the_tags_langfuse_split_them_into( + v1: Rig, metadata: Mapping[str, object] | None, headers: Mapping[str, str] | None, expected: tuple[str, ...] +) -> None: + marker: Final = "lit8281-strtags-" + uuid.uuid4().hex + _chat(v1, marker, metadata=metadata, headers=headers) + span: Final = v1.sink.one(marker) + assert span.tags()[: len(expected)] == expected, span.attributes + + +def test_trace_name_header_wins_over_the_derived_name(v1: Rig) -> None: + marker: Final = "lit8281-header-" + uuid.uuid4().hex + _chat(v1, marker, headers={"langfuse_trace_name": "header-trace"}) + span: Final = v1.sink.one(marker) + assert span.attributes["langfuse.trace.name"] == "header-trace", span.attributes + + +def test_joined_trace_keeps_caller_tags_and_gets_no_derived_trace_fields(v1: Rig) -> None: + marker: Final = "lit8281-join-" + uuid.uuid4().hex + existing: Final = uuid.uuid4().hex + _chat(v1, marker, metadata={"existing_trace_id": existing, "tags": ["join-tag"]}) + span: Final = v1.sink.one(marker) + attributes: Final = span.attributes + assert attributes.get("langfuse.trace.existing_id") == existing, attributes + assert span.tags() == ("join-tag",), attributes + assert "langfuse.trace.name" not in attributes, attributes + assert "langfuse.trace.input" not in attributes, attributes + assert "langfuse.trace.output" not in attributes, attributes + assert marker in str(attributes.get("langfuse.observation.input")), attributes + + +def test_joined_trace_without_caller_tags_gets_no_trace_tags(v1: Rig) -> None: + marker: Final = "lit8281-join-bare-" + uuid.uuid4().hex + _chat(v1, marker, metadata={"existing_trace_id": uuid.uuid4().hex}) + attributes: Final = v1.sink.one(marker).attributes + assert not { + key for key in attributes if key.startswith("langfuse.trace.") and key != "langfuse.trace.existing_id" + }, attributes + + +def test_mask_flags_redact_the_derived_trace_io(v1: Rig) -> None: + marker: Final = "lit8281-mask-" + uuid.uuid4().hex + _chat(v1, marker, metadata={"mask_input": True, "mask_output": True}) + attributes: Final = v1.sink.one(marker).attributes + assert attributes.get("langfuse.trace.input") == REDACTED, attributes + assert attributes.get("langfuse.trace.output") == REDACTED, attributes + assert attributes.get("langfuse.trace.name") == "litellm-acompletion", attributes + + +@pytest.mark.parametrize( + ("headers", "masked"), + [ + pytest.param({"langfuse_mask_input": "true", "langfuse_mask_output": "true"}, True, id="true"), + pytest.param({"langfuse_mask_input": "false", "langfuse_mask_output": "false"}, False, id="false"), + ], +) +def test_mask_flag_headers_parse_as_booleans(v1: Rig, headers: Mapping[str, str], masked: bool) -> None: + marker: Final = "lit8281-mask-header-" + uuid.uuid4().hex + _chat(v1, marker, headers=headers) + attributes: Final = v1.sink.one(marker).attributes + expected: Final = ( + (REDACTED, REDACTED) + if masked + else (attributes.get("langfuse.observation.input"), attributes.get("langfuse.observation.output")) + ) + assert (attributes.get("langfuse.trace.input"), attributes.get("langfuse.trace.output")) == expected, attributes + assert marker in str(attributes.get("langfuse.observation.input")), attributes + + +@pytest.mark.parametrize( + "tags", + [ + pytest.param(5, id="int"), + pytest.param(["ok-tag", 7, None], id="mixed_list"), + pytest.param("", id="empty_string"), + pytest.param("x" * 5000, id="five_kb_string"), + ], +) +def test_malformed_caller_tags_still_log_the_generation_with_derived_fields(v1: Rig, tags: object) -> None: + marker: Final = "lit8281-badtags-" + uuid.uuid4().hex + _chat(v1, marker, metadata={"tags": tags}) + span: Final = v1.sink.one(marker) + assert span.attributes.get("langfuse.trace.name") == "litellm-acompletion", span.attributes + assert span.attributes.get("langfuse.trace.input") == span.attributes.get("langfuse.observation.input") + assert all(isinstance(tag, str) for tag in span.tags()), span.tags() + + +def test_provider_failure_span_derives_name_and_input_without_output(v1: Rig) -> None: + marker: Final = "fail-lit8281-" + uuid.uuid4().hex + response: Final = v1.proxy.request( + "POST", + "/v1/chat/completions", + {"model": v1.openai_model, "messages": [{"role": "user", "content": marker}]}, + key=v1.key, + ) + assert response.status_code >= 500, response.text + span: Final = v1.sink.one(marker) + assert span.attributes.get("langfuse.trace.name") == "litellm-acompletion", span.attributes + assert span.attributes.get("langfuse.trace.input") == span.attributes.get("langfuse.observation.input") + assert "langfuse.trace.output" not in span.attributes, span.attributes + + +def test_embedding_call_derives_the_trace_name(v1: Rig) -> None: + marker: Final = "lit8281-embed-" + uuid.uuid4().hex + response: Final = v1.proxy.request( + "POST", "/v1/embeddings", {"model": v1.embedding_model, "input": marker, "user": marker}, key=v1.key + ) + assert response.status_code == 200, response.text + span: Final = v1.sink.one(marker) + assert span.attributes.get("langfuse.trace.name") == "litellm-aembedding", span.attributes + assert span.attributes.get("langfuse.trace.output") == span.attributes.get("langfuse.observation.output") + + +def test_identical_requests_each_log_exactly_one_span_with_derived_fields(v1: Rig) -> None: + marker: Final = "lit8281-twice-" + uuid.uuid4().hex + _chat(v1, marker) + _chat(v1, marker) + spans: Final = eventually(lambda: v1.sink.matching(marker), lambda found: len(found) == 2, seconds=70) + for span in spans: + _assert_derived(span, "acompletion") + + +def test_otel_v2_names_the_request_root_and_leaves_trace_io_to_the_root_observation(v2: Rig) -> None: + derived: Final = "lit8281-v2-" + uuid.uuid4().hex + caller: Final = "lit8281-v2-caller-" + uuid.uuid4().hex + _chat(v2, derived) + _chat(v2, caller, metadata={"trace_name": "caller-trace", "tags": ["caller-tag"]}) + derived_span: Final = v2.sink.one(derived) + assert not set(_trace_fields((derived_span,))), derived_span.attributes + root: Final = v2.sink.root_of(derived_span) + assert root.attributes.get("langfuse.trace.name") == "litellm-acompletion", root.attributes + assert derived in str(root.attributes.get("langfuse.observation.input")), root.attributes + assert root.attributes.get("langfuse.observation.output") is not None, root.attributes + caller_span: Final = v2.sink.one(caller) + assert caller_span.attributes.get("langfuse.trace.name") == "caller-trace", caller_span.attributes + assert caller_span.tags() == ("caller-tag",), caller_span.attributes + + +def test_otel_v2_inbound_traceparent_leaves_the_trace_name_to_the_caller(v2: Rig) -> None: + marker: Final = "lit8281-v2-parent-" + uuid.uuid4().hex + trace_id: Final = uuid.uuid4().hex + _chat(v2, marker, headers={"traceparent": f"00-{trace_id}-{uuid.uuid4().hex[:16]}-01"}) + span: Final = v2.sink.one(marker) + assert span.trace_id == trace_id, span + spans: Final = eventually( + lambda: v2.sink.in_trace(trace_id), + lambda found: any(record.parent_span_id != span.parent_span_id for record in found), + seconds=70, + ) + assert not set(_trace_fields(spans)), spans + + +@pytest.mark.parametrize( + ("metadata", "headers"), + [({"trace_name": "caller-trace"}, None), (None, {"langfuse_trace_name": "caller-trace"})], + ids=["body", "header"], +) +def test_otel_v2_post_call_judge_keeps_the_caller_trace_name_on_the_root( + v2: Rig, metadata: Mapping[str, object] | None, headers: Mapping[str, str] | None +) -> None: + marker: Final = "lit8281-v2-judge-" + uuid.uuid4().hex + _chat(v2, marker, metadata=metadata, headers=headers, guardrails=[JUDGE_GUARDRAIL]) + generations: Final = eventually(lambda: v2.sink.matching(marker), lambda found: len(found) == 2, seconds=70) + assert len({generation.trace_id for generation in generations}) == 1, generations + root: Final = v2.sink.root_of(generations[0]) + assert root.attributes.get("langfuse.trace.name") == "caller-trace", root.attributes + + +def test_otel_v2_generic_otel_root_gets_no_derived_langfuse_trace_name(v2_after_otel: SharedRig) -> None: + derived: Final = "lit8281-v2-after-otel-" + uuid.uuid4().hex + caller: Final = "lit8281-v2-after-otel-caller-" + uuid.uuid4().hex + _chat(v2_after_otel.rig, derived) + _chat(v2_after_otel.rig, caller, metadata={"trace_name": "caller-trace"}) + root: Final = v2_after_otel.otel.root_of_request(derived) + assert "langfuse.trace.name" not in root.attributes, root.attributes + caller_root: Final = v2_after_otel.otel.root_of_request(caller) + assert caller_root.attributes.get("langfuse.trace.name") == "caller-trace", caller_root.attributes + + +def test_otel_v2_llm_only_root_generation_keeps_its_span_name_as_the_trace_name(v2_llm_only: Rig) -> None: + marker: Final = "lit8281-v2-llm-only-" + uuid.uuid4().hex + _chat(v2_llm_only, marker) + span: Final = v2_llm_only.sink.one(marker) + assert span.parent_span_id == "", span + assert span.attributes.get("langfuse.trace.name") == span.name, span + assert marker in str(span.attributes.get("langfuse.observation.input")), span.attributes + + +def test_destination_outage_mid_burst_lands_every_span_once_with_derived_fields(v1: Rig) -> None: + marker: Final = "lit8281-chaos-" + uuid.uuid4().hex[:12] + v1.sink.outage.set() + try: + answered: Final = mixed_burst(v1.proxy, v1.openai_model, v1.anthropic_model, v1.key, marker, per_surface=4) + eventually(lambda: len(v1.sink.refusals), lambda refused: refused >= 1, seconds=70) + finally: + v1.sink.outage.clear() + assert len(answered) == 4 * len(SURFACES) + expected: Final = {f"{marker}-{surface}-{index}": surface for surface, index in product(SURFACES, range(4))} + + def landed() -> Mapping[str, tuple[SpanRecord, ...]]: + return {identity: v1.sink.matching(identity) for identity in expected} + + spans: Final = eventually(landed, lambda found: all(len(records) >= 1 for records in found.values()), seconds=120) + assert {identity: len(records) for identity, records in spans.items()} == dict.fromkeys(expected, 1) + for identity, records in spans.items(): + _assert_derived(records[0], SURFACE_CALL_TYPES[expected[identity]]) + + +def _tool(index: int) -> Mapping[str, object]: + return { + "type": "function", + "function": { + "name": f"lookup_{index}", + "description": f"Looks up record {index}", + "parameters": {"type": "object", "properties": {"id": {"type": "string"}}}, + }, + } + + +def _chat_with_tools( + rig: Rig, + content: str, + tools: int, + metadata: Mapping[str, object] | None = None, + headers: Mapping[str, str] | None = None, +) -> SpanRecord: + response: Final = rig.proxy.request( + "POST", + "/v1/chat/completions", + { + "model": rig.openai_model, + "messages": [{"role": "user", "content": content}], + "tools": [_tool(index) for index in range(tools)], + **({"metadata": dict(metadata)} if metadata is not None else {}), + }, + key=rig.key, + headers=headers, + ) + assert response.status_code == (500 if content.startswith("fail-") else 200), response.text + return rig.sink.one(content) + + +@pytest.mark.parametrize(("tools", "prefix"), product((30, 33, 34, 35, 36, 40), ("", "fail-"))) +def test_trace_fields_never_evict_or_drop_another_attribute_of_a_full_span(v1: Rig, tools: int, prefix: str) -> None: + run: Final = uuid.uuid4().hex + derived: Final = _chat_with_tools(v1, f"{prefix}lit8281-full-{tools}-{run}", tools, {"trace_id": uuid.uuid4().hex}) + joined: Final = _chat_with_tools( + v1, + f"{prefix}lit8281-full-joined-{tools}-{run}", + tools, + headers={"traceparent": f"00-{uuid.uuid4().hex}-{uuid.uuid4().hex[:16]}-01"}, + ) + assert set(joined.attributes) <= set(derived.attributes), set(joined.attributes) - set(derived.attributes) + assert derived.dropped_attributes == joined.dropped_attributes, (derived.dropped_attributes, joined.attributes) diff --git a/tests/integration/sdk/test_langfuse_otel_trace_fields_sdk.py b/tests/integration/sdk/test_langfuse_otel_trace_fields_sdk.py new file mode 100644 index 00000000000..33846960ab2 --- /dev/null +++ b/tests/integration/sdk/test_langfuse_otel_trace_fields_sdk.py @@ -0,0 +1,362 @@ +import json +import os +import subprocess +import sys +import threading +import uuid +from collections.abc import Iterator, Mapping +from dataclasses import dataclass +from typing import Final + +import pytest +from integration._support.client import eventually +from integration._support.wire import Reply, Request, wire_server +from opentelemetry.proto.collector.trace.v1.trace_service_pb2 import ExportTraceServiceRequest +from opentelemetry.proto.trace.v1.trace_pb2 import Span + +CALLS: Final = { + "sync": "completion", + "sync_stream": "completion", + "async": "acompletion", + "async_stream": "acompletion", +} +SDK_CALLS: Final = """ +import asyncio +import sys + +import litellm + +litellm.callbacks = ["langfuse_otel"] +api_base, marker = sys.argv[1], sys.argv[2] + + +def request(call): + return { + "model": "openai/gpt-4o-mini", + "api_base": api_base, + "api_key": "synthetic-provider-key", + "messages": [{"role": "user", "content": f"{marker}-{call}"}], + "stream": call.endswith("_stream"), + } + + +async def async_calls(): + await litellm.acompletion(**request("async")) + async for _ in await litellm.acompletion(**request("async_stream")): + pass + + +litellm.completion(**request("sync")) +for _ in litellm.completion(**request("sync_stream")): + pass +asyncio.run(async_calls()) +print("calls done", flush=True) +sys.stdin.read() +""" + + +APP_TRACE_CALLS: Final = """ +import asyncio +import json +import sys +import threading +import time + +import litellm +from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.trace.export import SimpleSpanProcessor + +litellm.callbacks = ["langfuse_otel"] +api_base, marker, sink, writes_onto_app_span = sys.argv[1], sys.argv[2], sys.argv[3], sys.argv[4] == "1" +app = TracerProvider() +app.add_span_processor(SimpleSpanProcessor(OTLPSpanExporter(endpoint=sink + "/api/public/otel/v1/traces"))) +tracer = app.get_tracer("checkout-app") + + +def request(call, **extra): + return { + "model": "openai/gpt-4o-mini", + "api_base": api_base, + "api_key": "synthetic-provider-key", + "messages": [{"role": "user", "content": f"{marker}-{call}"}], + **extra, + } + + +def name_trace(root, call): + root.set_attribute("app.call", f"{marker}-{call}") + root.set_attribute("langfuse.trace.name", f"checkout-{marker}-{call}") + root.set_attribute("langfuse.trace.input", json.dumps({"cart": f"{marker}-{call}"})) + + +def close_trace(root, call): + root.set_attribute("langfuse.trace.output", json.dumps({"order": f"{marker}-{call}"})) + root.end() + + +def litellm_wrote_onto(root): + return not writes_onto_app_span or "langfuse.observation.type" in root.attributes + + +def ambient_sync(): + with tracer.start_as_current_span("checkout", end_on_exit=False) as root: + name_trace(root, "ambient_sync") + litellm.completion(**request("ambient_sync")) + deadline = time.monotonic() + 30 + while not litellm_wrote_onto(root) and time.monotonic() < deadline: + threading.Event().wait(0.05) + close_trace(root, "ambient_sync") + + +async def ambient_async(): + with tracer.start_as_current_span("checkout", end_on_exit=False) as root: + name_trace(root, "ambient_async") + await litellm.acompletion(**request("ambient_async")) + deadline = time.monotonic() + 30 + while not litellm_wrote_onto(root) and time.monotonic() < deadline: + await asyncio.sleep(0.05) + close_trace(root, "ambient_async") + + +def caller_span_sync(): + root = tracer.start_span("checkout") + name_trace(root, "caller_span_sync") + litellm.completion(**request("caller_span_sync", metadata={"litellm_parent_otel_span": root})) + close_trace(root, "caller_span_sync") + + +async def caller_span_async(): + root = tracer.start_span("checkout") + name_trace(root, "caller_span_async") + await litellm.acompletion(**request("caller_span_async", metadata={"litellm_parent_otel_span": root})) + close_trace(root, "caller_span_async") + + +async def async_calls(): + await ambient_async() + await caller_span_async() + + +ambient_sync() +caller_span_sync() +asyncio.run(async_calls()) +print("calls done", flush=True) +sys.stdin.read() +""" +APP_CALLS: Final = ("ambient_sync", "ambient_async", "caller_span_sync", "caller_span_async") + + +def _exported_spans(body: bytes) -> Iterator[Span]: + for resource in ExportTraceServiceRequest.FromString(body).resource_spans: + for scope in resource.scope_spans: + yield from scope.spans + + +def _attributes(span: Span) -> Mapping[str, str]: + return {pair.key: pair.value.string_value for pair in span.attributes} + + +@dataclass(frozen=True, slots=True) +class TraceSpan: + trace_id: str + parent_span_id: str + attributes: Mapping[str, str] + + +class TraceSink: + def __init__(self) -> None: + self.lock: Final = threading.Lock() + self.spans: Final[list[TraceSpan]] = [] # mutable-ok: spans accumulate as both exporters deliver + + def respond(self, request: Request) -> Reply: + if not request.target.endswith("/v1/traces"): + return Reply(body=b"") + spans: Final = tuple( + TraceSpan(span.trace_id.hex(), span.parent_span_id.hex(), _attributes(span)) + for span in _exported_spans(request.body) + ) + with self.lock: + self.spans.extend(spans) + return Reply(body=b"") + + def traces(self, marker: str, calls: tuple[str, ...]) -> Mapping[str, tuple[TraceSpan, ...]]: + with self.lock: + snapshot: Final = tuple(self.spans) + roots: Final = { + call: next( + ( + span + for span in snapshot + if not span.parent_span_id and span.attributes.get("app.call") == f"{marker}-{call}" + ), + None, + ) + for call in calls + } + return { + call: tuple(span for span in snapshot if span.trace_id == root.trace_id) + for call, root in roots.items() + if root is not None + } + + +def _trace_values(spans: tuple[TraceSpan, ...], keys: tuple[str, ...]) -> Iterator[tuple[str, str]]: + for span in spans: + yield from ((key, span.attributes[key]) for key in keys if key in span.attributes) + + +def _logged(spans: tuple[TraceSpan, ...]) -> bool: + return any(span.attributes.get("langfuse.observation.type") == "generation" for span in spans) + + +def _chat_reply(request: Request) -> Reply: + body: Final = json.loads(request.body) + content: Final = body["messages"][0]["content"] + if body.get("stream"): + delta: Final = {"id": content, "object": "chat.completion.chunk", "created": 1, "model": "gpt-4o-mini"} + return Reply( + content_type="text/event-stream", + chunks=( + f"data: {json.dumps({**delta, 'choices': [{'index': 0, 'delta': {'role': 'assistant', 'content': 'ok'}}]})}\n\n".encode(), + f"data: {json.dumps({**delta, 'choices': [{'index': 0, 'delta': {}, 'finish_reason': 'stop'}]})}\n\n".encode(), + b"data: [DONE]\n\n", + ), + ) + return Reply( + body=json.dumps( + { + "id": content, + "object": "chat.completion", + "created": 1, + "model": "gpt-4o-mini", + "choices": [{"index": 0, "message": {"role": "assistant", "content": "ok"}, "finish_reason": "stop"}], + "usage": {"prompt_tokens": 5, "completion_tokens": 1, "total_tokens": 6}, + } + ).encode() + ) + + +class GenerationSink: + def __init__(self) -> None: + self.lock: Final = threading.Lock() + self.generations: Final[list[Mapping[str, str]]] = [] # mutable-ok: spans accumulate as the exporter delivers + + def respond(self, request: Request) -> Reply: + if not request.target.endswith("/v1/traces"): + return Reply(body=b"") + spans: Final = tuple(_attributes(span) for span in _exported_spans(request.body)) + with self.lock: + self.generations.extend(span for span in spans if span.get("langfuse.observation.type") == "generation") + return Reply(body=b"") + + def by_call(self, marker: str) -> Mapping[str, tuple[Mapping[str, str], ...]]: + with self.lock: + snapshot: Final = tuple(self.generations) + return { + call: tuple(span for span in snapshot if f'"{marker}-{call}"' in span.get("langfuse.observation.input", "")) + for call in CALLS + } + + +def test_sdk_sync_and_async_calls_derive_trace_name_and_io_from_their_call_type() -> None: + marker: Final = "lit8281-sdk-" + uuid.uuid4().hex + sink: Final = GenerationSink() + with wire_server(sink.respond) as destination, wire_server(_chat_reply) as provider: + child: Final = subprocess.Popen( + [sys.executable, "-P", "-c", SDK_CALLS, provider.url + "/v1", marker], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + env={ + **os.environ, + "LANGFUSE_PUBLIC_KEY": "pk-lf-sdk", + "LANGFUSE_SECRET_KEY": "sk-lf-sdk", + "LANGFUSE_HOST": destination.url, + "OTEL_BSP_SCHEDULE_DELAY": "200", + }, + ) + try: + spans: Final = eventually( + lambda: sink.by_call(marker), + lambda found: all(len(records) == 1 for records in found.values()) or child.poll() is not None, + seconds=70, + ) + finally: + output: Final = child.communicate(input=b"", timeout=60)[0].decode() + assert {call: len(records) for call, records in spans.items()} == dict.fromkeys(CALLS, 1), output + for call, call_type in CALLS.items(): + attributes = spans[call][0] + assert attributes["langfuse.trace.name"] == f"litellm-{call_type}", (call, attributes) + assert attributes["langfuse.trace.input"] == attributes["langfuse.observation.input"], (call, attributes) + assert attributes["langfuse.trace.output"] == attributes["langfuse.observation.output"], (call, attributes) + assert json.loads(attributes["langfuse.trace.output"]) == {"role": "assistant", "content": "ok"}, call + + +@pytest.mark.parametrize( + ("environment", "writes_onto_app_span", "calls"), + [ + pytest.param({}, True, APP_CALLS, id="litellm_writes_onto_the_app_span"), + pytest.param({"USE_OTEL_LITELLM_REQUEST_SPAN": "true"}, False, APP_CALLS, id="litellm_request_child_span"), + pytest.param( + {"LITELLM_OTEL_V2": "1", "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT": "span_only"}, + False, + ("ambient_async",), + id="otel_v2_llm_call_child_span", + ), + ], +) +def test_sdk_call_inside_an_app_trace_leaves_the_app_trace_name_and_io_alone( + environment: Mapping[str, str], writes_onto_app_span: bool, calls: tuple[str, ...] +) -> None: + marker: Final = "lit8281-app-" + uuid.uuid4().hex + sink: Final = TraceSink() + with wire_server(sink.respond) as destination, wire_server(_chat_reply) as provider: + child: Final = subprocess.Popen( + [ + sys.executable, + "-P", + "-c", + APP_TRACE_CALLS, + provider.url + "/v1", + marker, + destination.url, + "1" if writes_onto_app_span else "0", + ], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + env={ + **os.environ, + "LANGFUSE_PUBLIC_KEY": "pk-lf-sdk", + "LANGFUSE_SECRET_KEY": "sk-lf-sdk", + "LANGFUSE_HOST": destination.url, + "OTEL_BSP_SCHEDULE_DELAY": "200", + **environment, + }, + ) + try: + traces: Final = eventually( + lambda: sink.traces(marker, calls), + lambda found: ( + (len(found) == len(calls) and all(map(_logged, found.values()))) or child.poll() is not None + ), + seconds=70, + ) + finally: + output: Final = child.communicate(input=b"", timeout=60)[0].decode() + assert sorted(call for call, spans in traces.items() if _logged(spans)) == sorted(calls), output + for call, spans in traces.items(): + _assert_the_app_trace_fields_stand(marker, call, spans) + + +def _assert_the_app_trace_fields_stand(marker: str, call: str, spans: tuple[TraceSpan, ...]) -> None: + app: Final = { + "langfuse.trace.name": f"checkout-{marker}-{call}", + "langfuse.trace.input": json.dumps({"cart": f"{marker}-{call}"}), + "langfuse.trace.output": json.dumps({"order": f"{marker}-{call}"}), + } + root: Final = next(span for span in spans if not span.parent_span_id) + assert {key: root.attributes.get(key) for key in app} == app, (call, root.attributes) + written: Final = set(_trace_values(spans, (*app, "langfuse.trace.tags"))) + assert written == set(app.items()), (call, written) diff --git a/tests/unit/integrations/otel/test_langfuse_logger.py b/tests/unit/integrations/otel/test_langfuse_logger.py index d55f0577e03..1ac6635b8ee 100644 --- a/tests/unit/integrations/otel/test_langfuse_logger.py +++ b/tests/unit/integrations/otel/test_langfuse_logger.py @@ -11,6 +11,8 @@ import pytest pytest.importorskip("opentelemetry") +from opentelemetry import trace # noqa: E402 # needs the importorskip above +from opentelemetry.context import Context # noqa: E402 # needs the importorskip above from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter # noqa: E402 from opentelemetry.trace import SpanKind # noqa: E402 @@ -64,8 +66,8 @@ def _logger(*, capture: str = "span_only", mappers: Sequence[str] = ("genai", "l return build_otel_v2_logger(config=cfg, tracer_provider=tracer_provider), exporter -def _start_root(logger): - root = logger._emitter.start_span(SpanRole.PROXY_REQUEST, LITELLM_PROXY_REQUEST_SPAN_NAME) +def _start_root(logger: OpenTelemetryV2, parent_context: Context | None = None): + root: Final = logger._emitter.start_span(SpanRole.PROXY_REQUEST, LITELLM_PROXY_REQUEST_SPAN_NAME, parent_context) set_request_root_span(root) return root @@ -331,13 +333,28 @@ def test_unrenderable_output_never_raises_into_the_request(): assert INPUT_ATTR not in attrs and OUTPUT_ATTR not in attrs +def _inbound_traceparent() -> Context: + caller: Final = trace.SpanContext( + trace_id=0x4BF92F3577B34DA6A3CE929D0E0E4736, + span_id=0x00F067AA0BA902B7, + is_remote=True, + trace_flags=trace.TraceFlags(trace.TraceFlags.SAMPLED), + ) + return trace.set_span_in_context(trace.NonRecordingSpan(caller)) + + def _run_named_request( - logger: OpenTelemetryV2, exporter: InMemorySpanExporter, litellm_params: Mapping[str, object] + logger: OpenTelemetryV2, + exporter: InMemorySpanExporter, + litellm_params: Mapping[str, object], + parent_context: Context | None = None, ) -> tuple[Mapping[str, object], Mapping[str, object]]: response: Final = ModelResponse(choices=[Choices(message=Message(role="assistant", content="pong"))]) - root: Final = _start_root(logger) + root: Final = _start_root(logger, parent_context) logger.log_pre_api_call( - model="gpt-5.4-mini", messages=[], kwargs={"litellm_call_id": "call_1", "litellm_params": litellm_params} + model="gpt-5.4-mini", + messages=[], + kwargs={"litellm_call_id": "call_1", "litellm_params": litellm_params, "call_type": "acompletion"}, ) root.end() payload: Final = { @@ -390,14 +407,113 @@ def test_body_metadata_trace_name_names_the_root_and_the_generation(): assert generation_attrs[TRACE_NAME_ATTR] == "from-body" -def test_unnamed_request_leaves_the_trace_name_off_both_spans(): +def test_unnamed_request_derives_the_trace_name_on_the_root_only(): logger, exporter = _logger() root_attrs, generation_attrs = _run_named_request(logger, exporter, {"proxy_server_request": {"headers": {}}}) + assert root_attrs[TRACE_NAME_ATTR] == "litellm-acompletion" + assert TRACE_NAME_ATTR not in generation_attrs + + +def test_unnamed_request_under_an_inbound_traceparent_leaves_the_trace_name_to_the_caller(): + logger, exporter = _logger() + + root_attrs, generation_attrs = _run_named_request( + logger, exporter, {"proxy_server_request": {"headers": {}}}, _inbound_traceparent() + ) + assert TRACE_NAME_ATTR not in root_attrs and TRACE_NAME_ATTR not in generation_attrs +def test_named_request_under_an_inbound_traceparent_keeps_the_caller_trace_name(): + logger, exporter = _logger() + + root_attrs, generation_attrs = _run_named_request( + logger, + exporter, + {"metadata": {"trace_name": "from-body"}, "proxy_server_request": {"headers": {}}}, + _inbound_traceparent(), + ) + + assert root_attrs[TRACE_NAME_ATTR] == "from-body" + assert generation_attrs[TRACE_NAME_ATTR] == "from-body" + + +UNNAMED: Final = {"proxy_server_request": {"headers": {}}} +JUDGE_PARAMS: Final = {"metadata": {"internal_call_origin": "llm_as_a_judge_guardrail"}} +CALLER_NAMES: Final = ( + {"metadata": {"trace_name": "from-caller"}, "proxy_server_request": {"headers": {}}}, + {"proxy_server_request": {"headers": {"langfuse_trace_name": "from-caller"}}}, +) + + +def _root_trace_name_after( + logger: OpenTelemetryV2, + calls: Sequence[tuple[str, Mapping[str, object]]], + root_logger: OpenTelemetryV2 | None = None, +) -> object: + root: Final = _start_root(root_logger or logger) + for index, (call_type, litellm_params) in enumerate(calls): + logger.log_pre_api_call( + model="gpt-5.4-mini", + messages=[], + kwargs={"litellm_call_id": f"call_{index}", "litellm_params": litellm_params, "call_type": call_type}, + ) + root.end() + return dict(root.attributes or {}).get(TRACE_NAME_ATTR) + + +@pytest.mark.parametrize("later", [JUDGE_PARAMS, UNNAMED], ids=["litellm-sub-call", "untagged-call"]) +@pytest.mark.parametrize("caller", CALLER_NAMES, ids=["body", "header"]) +def test_a_later_call_in_the_request_keeps_the_caller_trace_name_on_the_root( + caller: Mapping[str, object], later: Mapping[str, object] +) -> None: + logger, _ = _logger() + + assert _root_trace_name_after(logger, (("acompletion", caller), ("acompletion", later))) == "from-caller" + + +@pytest.mark.parametrize("caller", CALLER_NAMES, ids=["body", "header"]) +def test_a_caller_trace_name_replaces_the_name_an_earlier_call_derived(caller: Mapping[str, object]) -> None: + logger, _ = _logger() + + assert _root_trace_name_after(logger, (("acompletion", JUDGE_PARAMS), ("acompletion", caller))) == "from-caller" + + +def test_a_later_call_keeps_the_trace_name_the_first_call_derived(): + logger, _ = _logger() + + assert ( + _root_trace_name_after(logger, (("anthropic_messages", UNNAMED), ("acompletion", UNNAMED))) + == "litellm-anthropic_messages" + ) + + +def test_a_litellm_sub_call_before_the_request_call_leaves_the_trace_name_to_the_request_call(): + logger, _ = _logger() + + assert ( + _root_trace_name_after(logger, (("acompletion", JUDGE_PARAMS), ("anthropic_messages", UNNAMED))) + == "litellm-anthropic_messages" + ) + + +def test_a_root_another_vendor_started_gets_no_derived_trace_name(): + logger, _ = _logger() + other_vendor, _ = _logger(mappers=("genai",)) + + assert _root_trace_name_after(logger, (("acompletion", UNNAMED),), root_logger=other_vendor) is None + + +@pytest.mark.parametrize("caller", CALLER_NAMES, ids=["body", "header"]) +def test_a_root_another_vendor_started_still_gets_the_caller_trace_name(caller: Mapping[str, object]) -> None: + logger, _ = _logger() + other_vendor, _ = _logger(mappers=("genai",)) + + assert _root_trace_name_after(logger, (("acompletion", caller),), root_logger=other_vendor) == "from-caller" + + @pytest.mark.parametrize("capture", ["span_only", "no_content"]) def test_body_metadata_user_session_and_tags_land_on_the_root_and_the_generation(capture): logger, exporter = _logger(capture=capture) @@ -420,7 +536,8 @@ def test_body_metadata_user_session_and_tags_land_on_the_root_and_the_generation assert attrs["user.id"] == "user-42" assert attrs["session.id"] == "session-7" assert tuple(attrs["langfuse.trace.tags"]) == ("prod", "eval", "nightly") - assert TRACE_NAME_ATTR not in attrs + assert root_attrs[TRACE_NAME_ATTR] == "litellm-acompletion" + assert TRACE_NAME_ATTR not in generation_attrs def test_langfuse_user_and_session_headers_beat_body_metadata_on_both_spans(): @@ -489,7 +606,9 @@ def test_a_request_without_trace_controls_stamps_none_of_them(): {"metadata": {"user_api_key_team_id": "t1", "tags": []}, "proxy_server_request": {"headers": {}}}, ) - assert set(TRACE_CONTROL_ATTRS).isdisjoint(root_attrs) + caller_only: Final = tuple(attr for attr in TRACE_CONTROL_ATTRS if attr != TRACE_NAME_ATTR) + assert set(caller_only).isdisjoint(root_attrs) + assert root_attrs[TRACE_NAME_ATTR] == "litellm-acompletion" assert set(TRACE_CONTROL_ATTRS).isdisjoint(generation_attrs) diff --git a/tests/unit/integrations/otel/test_otel_v2_vendor_mappers.py b/tests/unit/integrations/otel/test_otel_v2_vendor_mappers.py index 7e112fcf6a4..98401fd17d3 100644 --- a/tests/unit/integrations/otel/test_otel_v2_vendor_mappers.py +++ b/tests/unit/integrations/otel/test_otel_v2_vendor_mappers.py @@ -802,3 +802,29 @@ def test_resolve_mappers_composition_layers_vocabularies(): def test_resolve_mappers_rejects_unknown_name(): with pytest.raises(ValueError, match="unknown mapper name 'nope'"): resolve_mappers(["genai", "nope"]) + + +def test_langfuse_mapper_leaves_the_trace_name_off_an_unnamed_generation() -> None: + attrs: Final = LangfuseMapper().map(_llm_call(call_type="acompletion", trace=TraceControls())) + + assert "langfuse.trace.name" not in attrs + + +def test_langfuse_mapper_carries_the_caller_trace_name_on_the_generation() -> None: + attrs: Final = LangfuseMapper().map(_llm_call(call_type="acompletion", trace=TraceControls(name="caller-trace"))) + + assert attrs["langfuse.trace.name"] == "caller-trace" + + +@pytest.mark.parametrize( + ("trace", "call_type", "expected"), + [ + (TraceControls(), "aresponses", {"langfuse.trace.name": "litellm-aresponses"}), + (TraceControls(name="caller-trace"), "aresponses", {"langfuse.trace.name": "caller-trace"}), + (TraceControls(), None, {}), + ], +) +def test_langfuse_root_trace_attributes_derive_the_name_only_when_the_caller_gave_none( + trace: TraceControls, call_type: str | None, expected: Mapping[str, object] +) -> None: + assert LangfuseMapper.trace_attributes(trace, call_type) == expected diff --git a/tests/unit/integrations/test_langfuse_otel.py b/tests/unit/integrations/test_langfuse_otel.py index c7608159515..5d37a145de6 100644 --- a/tests/unit/integrations/test_langfuse_otel.py +++ b/tests/unit/integrations/test_langfuse_otel.py @@ -1,17 +1,25 @@ import base64 import json import os +from collections.abc import Mapping, Sequence +from dataclasses import dataclass from datetime import datetime, timezone from typing import Final from unittest.mock import MagicMock, patch import pytest -from opentelemetry.sdk.trace import TracerProvider +from openai.types.responses import ResponseFunctionToolCall +from opentelemetry import trace +from opentelemetry.context import Context +from opentelemetry.sdk.trace import SpanLimits, TracerProvider +from opentelemetry.trace import Span +import litellm +from litellm.constants import REDACTED_BY_LITELLM from litellm.integrations.langfuse.langfuse_otel import LangfuseOtelLogger -from litellm.integrations.opentelemetry import OpenTelemetry, OpenTelemetryConfig +from litellm.integrations.opentelemetry import LITELLM_PROXY_REQUEST_SPAN_NAME, OpenTelemetry, OpenTelemetryConfig from litellm.types.llms.openai import ResponsesAPIResponse -from litellm.types.utils import StandardCallbackDynamicParams +from litellm.types.utils import Choices, ModelResponse, StandardCallbackDynamicParams class TestLangfuseOtelIntegration: @@ -138,7 +146,7 @@ class TestLangfuseOtelIntegration: ) # safe_set_attribute(span, key, value) → positional args - mock_safe_set_attribute.assert_called_once_with( + mock_safe_set_attribute.assert_any_call( mock_span, "langfuse.environment", test_env ) @@ -386,6 +394,7 @@ class TestLangfuseOtelIntegration: actual = { call.args[1]: json.loads(call.args[2]) for call in mock_safe_set_attribute.call_args_list + if call.args[1] in expect_output } assert ( @@ -449,6 +458,7 @@ class TestLangfuseOtelIntegration: actual = { call.args[1]: json.loads(call.args[2]) for call in mock_safe_set_attribute.call_args_list + if call.args[1] in expected } assert ( actual == expected @@ -879,7 +889,11 @@ class TestLangfuseOtelResponsesAPI: Summary( text="Let me analyze this problem step by step...", type="summary_text", - ) + ), + Summary( + text="Now checking the forecast data.", + type="summary_text", + ), ], ), ResponseOutputMessage( @@ -929,22 +943,31 @@ class TestLangfuseOtelResponsesAPI: # Verify output contains reasoning and message assert isinstance(output_data, list) - assert len(output_data) == 2 + assert len(output_data) == 3 - # Verify reasoning summary assert output_data[0]["role"] == "reasoning_summary" assert ( output_data[0]["content"] == "Let me analyze this problem step by step..." ) + assert output_data[1]["role"] == "reasoning_summary" + assert output_data[1]["content"] == "Now checking the forecast data." # Verify message - assert output_data[1]["role"] == "assistant" + assert output_data[2]["role"] == "assistant" assert ( - output_data[1]["content"] + output_data[2]["content"] == "The weather in San Francisco is sunny, 20°C." ) + trace_output_calls: Final = [ + call + for call in mock_safe_set_attribute.call_args_list + if call.args[1] == LangfuseSpanAttributes.TRACE_OUTPUT.value + ] + assert len(trace_output_calls) > 0, "trace.output should be set" + assert trace_output_calls[0].args[2] == output_json + def test_responses_api_with_function_calls(self): """Test Langfuse OTEL logger with Responses API function_call output.""" from litellm.types.integrations.langfuse_otel import LangfuseSpanAttributes @@ -1054,6 +1077,381 @@ class TestLangfuseOtelResponsesAPI: assert output_data[0]["arguments"] == {} +_TRACER: Final = TracerProvider().get_tracer("langfuse-otel-tests") +TRACE_FIELD_KEYS: Final = ( + "langfuse.trace.name", + "langfuse.trace.input", + "langfuse.trace.output", + "langfuse.trace.tags", +) +CALLER: Final = trace.SpanContext( + trace_id=0x4BF92F3577B34DA6A3CE929D0E0E4736, + span_id=0x00F067AA0BA902B7, + is_remote=True, + trace_flags=trace.TraceFlags(trace.TraceFlags.SAMPLED), +) +CHAT_KWARGS: Final = { + "call_type": "acompletion", + "messages": [{"role": "user", "content": "weather?"}], +} + + +def _chat_response(content: str) -> ModelResponse: + return ModelResponse( + id="chatcmpl-derived", + model="gpt-4o", + choices=[Choices(finish_reason="stop", message={"role": "assistant", "content": content})], + ) + + +def _stamped(span: Span, kwargs: Mapping[str, object], response_obj: object = None) -> Mapping[str, object]: + LangfuseOtelLogger._set_langfuse_specific_attributes(span, kwargs, response_obj) + span.end() + return dict(getattr(span, "attributes", None) or {}) + + +def _emitted(kwargs: Mapping[str, object], response_obj: object = None) -> Mapping[str, object]: + return _stamped(_TRACER.start_span("litellm_request"), kwargs, response_obj) + + +def _trace_fields(attributes: Mapping[str, object]) -> Mapping[str, object]: + return {key: attributes[key] for key in TRACE_FIELD_KEYS if key in attributes} + + +class _PresetCache: + @staticmethod + def get_preset_cache_key_from_kwargs(**kwargs: object) -> str: # kwargs-ok: mirrors Cache's signature + return "preset-abc" + + +class TestDerivedTraceFields: + def test_chat_call_derives_trace_name_input_and_output(self) -> None: + attributes: Final = _emitted({**CHAT_KWARGS, "litellm_params": {"metadata": {}}}, _chat_response("Sunny.")) + + assert attributes["langfuse.trace.name"] == "litellm-acompletion" + assert attributes["langfuse.trace.input"] == attributes["langfuse.observation.input"] + assert attributes["langfuse.trace.output"] == attributes["langfuse.observation.output"] + assert json.loads(str(attributes["langfuse.trace.input"])) == [{"role": "user", "content": "weather?"}] + assert json.loads(str(attributes["langfuse.trace.output"])) == {"role": "assistant", "content": "Sunny."} + + def test_missing_call_type_defaults_to_completion(self) -> None: + attributes: Final = _emitted( + {"messages": [{"role": "user", "content": "hi"}], "litellm_params": {"metadata": {}}} + ) + + assert attributes["langfuse.trace.name"] == "litellm-completion" + + def test_caller_trace_name_wins_and_is_the_only_name(self) -> None: + attributes: Final = _emitted( + {"call_type": "acompletion", "litellm_params": {"metadata": {"trace_name": "caller-trace"}}} + ) + + assert attributes["langfuse.trace.name"] == "caller-trace" + + def test_existing_trace_id_keeps_caller_tags_and_leaves_derived_trace_fields_off( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setattr(litellm, "langfuse_default_tags", ["cache_hit"]) + attributes: Final = _emitted( + { + **CHAT_KWARGS, + "cache_hit": True, + "litellm_params": {"metadata": {"existing_trace_id": "abc123", "tags": ["experiment"]}}, + "standard_logging_object": {"request_tags": ["injected"]}, + }, + _chat_response("hi"), + ) + + assert _trace_fields(attributes) == {"langfuse.trace.tags": json.dumps(["experiment"])} + assert attributes["langfuse.trace.existing_id"] == "abc123" + assert "langfuse.observation.input" in attributes + assert "langfuse.observation.output" in attributes + + @pytest.mark.parametrize( + ("mask_flags", "masked_trace_keys"), + [ + ({"mask_input": True}, frozenset({"langfuse.trace.input"})), + ({"mask_output": True}, frozenset({"langfuse.trace.output"})), + ({"mask_input": True, "mask_output": True}, frozenset({"langfuse.trace.input", "langfuse.trace.output"})), + ( + {"mask_input": "true", "mask_output": "TRUE"}, + frozenset({"langfuse.trace.input", "langfuse.trace.output"}), + ), + ({"mask_input": "false", "mask_output": "False"}, frozenset()), + ({"mask_input": False, "mask_output": None}, frozenset()), + ], + ) + def test_mask_flags_redact_the_derived_trace_io( + self, mask_flags: Mapping[str, object], masked_trace_keys: frozenset[str] + ) -> None: + attributes: Final = _emitted( + { + "call_type": "acompletion", + "messages": [{"role": "user", "content": "secret question"}], + "litellm_params": {"metadata": mask_flags}, + }, + _chat_response("secret answer"), + ) + observation_of: Final = { + "langfuse.trace.input": attributes["langfuse.observation.input"], + "langfuse.trace.output": attributes["langfuse.observation.output"], + } + + assert {key: attributes.get(key) for key in observation_of} == { + key: REDACTED_BY_LITELLM if key in masked_trace_keys else observation + for key, observation in observation_of.items() + } + + def test_responses_api_output_mirrors_observation_output_items(self) -> None: + response_obj: Final = ResponsesAPIResponse( + id="response-derived", + created_at=1625247700, + output=[ + ResponseFunctionToolCall( + id="fc-1", + type="function_call", + name="get_weather", + call_id="call-1", + arguments='{"city":"sf"}', + status="completed", + ) + ], + ) + attributes: Final = _emitted({"call_type": "aresponses", "litellm_params": {"metadata": {}}}, response_obj) + + assert attributes["langfuse.trace.name"] == "litellm-aresponses" + assert attributes["langfuse.trace.output"] == attributes["langfuse.observation.output"] + assert json.loads(str(attributes["langfuse.trace.output"])) == [ + { + "id": "fc-1", + "name": "get_weather", + "call_id": "call-1", + "type": "function_call", + "arguments": {"city": "sf"}, + } + ] + + def test_tags_merge_caller_request_and_default_tags_in_order(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(litellm, "langfuse_default_tags", ["cache_hit", "user_api_key_alias"]) + attributes: Final = _emitted( + { + "call_type": "acompletion", + "cache_hit": True, + "litellm_params": {"metadata": {"tags": ["a", "b"], "user_api_key_alias": "k1"}}, + "standard_logging_object": {"request_tags": ["b", "c"]}, + } + ) + + assert json.loads(str(attributes["langfuse.trace.tags"])) == [ + "a", + "b", + "c", + "cache_hit:True", + "user_api_key_alias:k1", + ] + + @pytest.mark.parametrize( + ("metadata", "expected"), + [ + ({}, ["cache_key:preset-abc"]), + ({"hidden_params": {"cache_key": "explicit-key"}}, ["cache_key:explicit-key"]), + ], + ids=["falls_back_to_the_preset_cache_key", "prefers_hidden_params_over_the_preset"], + ) + def test_cache_key_default_tag( + self, monkeypatch: pytest.MonkeyPatch, metadata: Mapping[str, object], expected: Sequence[str] + ) -> None: + monkeypatch.setattr(litellm, "langfuse_default_tags", ["cache_key"]) + monkeypatch.setattr(litellm, "cache", _PresetCache()) + attributes: Final = _emitted({"call_type": "acompletion", "litellm_params": {"metadata": metadata}}) + + assert json.loads(str(attributes["langfuse.trace.tags"])) == expected + + @pytest.mark.parametrize( + ("caller_tags", "expected"), + [ + ("single", ["single", "req"]), + ("h1, h2", ["h1", "h2", "req"]), + ('["j1", "j2"]', ["j1", "j2", "req"]), + (["x", 1], ["x", "1", "req"]), + ("", ["req"]), + ], + ) + def test_caller_tags_are_read_the_way_langfuse_reads_a_raw_tags_value( + self, caller_tags: object, expected: Sequence[str] + ) -> None: + attributes: Final = _emitted( + { + "call_type": "acompletion", + "litellm_params": {"metadata": {"tags": caller_tags}}, + "standard_logging_object": {"request_tags": ["req"]}, + } + ) + + assert json.loads(str(attributes["langfuse.trace.tags"])) == expected + + def test_no_tags_anywhere_leaves_trace_tags_unset(self) -> None: + attributes: Final = _emitted({"call_type": "acompletion", "litellm_params": {"metadata": {}}}) + + assert "langfuse.trace.tags" not in attributes + + def test_empty_string_role_is_omitted_from_output(self) -> None: + response_obj: Final = {"id": "chatcmpl-norole", "choices": [{"message": {"role": "", "content": "hi"}}]} + attributes: Final = _emitted({"call_type": "acompletion", "litellm_params": {"metadata": {}}}, response_obj) + + assert json.loads(str(attributes["langfuse.observation.output"])) == {"content": "hi"} + + def test_no_messages_leaves_trace_input_unset_but_emits_output(self) -> None: + attributes: Final = _emitted( + {"call_type": "acompletion", "litellm_params": {"metadata": {}}}, _chat_response("hi") + ) + + assert "langfuse.trace.input" not in attributes + assert "langfuse.observation.input" not in attributes + assert attributes["langfuse.trace.output"] == attributes["langfuse.observation.output"] + + +def _proxy_span(parent: Context | None = None) -> Span: + return _TRACER.start_span(LITELLM_PROXY_REQUEST_SPAN_NAME, context=parent) + + +def _child_of(parent: Span) -> Span: + return _TRACER.start_span("litellm_request", context=trace.set_span_in_context(parent)) + + +def _carrying(parent: Span, key: str = "metadata") -> Mapping[str, object]: + return {**CHAT_KWARGS, "litellm_params": {key: {"litellm_parent_otel_span": parent}}} + + +class TestDerivedTraceFieldsOnlyOnTracesLitellmStarts: + def test_a_span_under_an_inbound_traceparent_joins_the_caller_trace(self) -> None: + span: Final = _TRACER.start_span( + "litellm_request", context=trace.set_span_in_context(trace.NonRecordingSpan(CALLER)) + ) + + attributes: Final = _stamped(span, {**CHAT_KWARGS, "litellm_params": {"metadata": {}}}, _chat_response("ok")) + + assert _trace_fields(attributes) == {} + assert "langfuse.observation.input" in attributes and "langfuse.observation.output" in attributes + + def test_the_active_app_span_litellm_writes_onto_keeps_its_trace_fields(self) -> None: + with _TRACER.start_as_current_span("checkout") as app_span: + app_span.set_attribute("langfuse.trace.name", "checkout") + + attributes: Final = _stamped( + app_span, {**CHAT_KWARGS, "litellm_params": {"metadata": {}}}, _chat_response("ok") + ) + + assert _trace_fields(attributes) == {"langfuse.trace.name": "checkout"} + + def test_a_litellm_root_span_derives_while_an_app_span_is_current(self) -> None: + with _TRACER.start_as_current_span("checkout"): + span: Final = _TRACER.start_span("litellm_request", context=Context()) + + attributes: Final = _stamped( + span, {**CHAT_KWARGS, "litellm_params": {"metadata": {}}}, _chat_response("ok") + ) + + assert attributes["langfuse.trace.name"] == "litellm-acompletion" + + def test_a_span_under_a_caller_provided_parent_span_joins_the_caller_trace(self) -> None: + caller_span: Final = _TRACER.start_span("caller-root") + + attributes: Final = _stamped(_child_of(caller_span), _carrying(caller_span), _chat_response("ok")) + + assert _trace_fields(attributes) == {} + + @pytest.mark.parametrize("key", ["metadata", "litellm_metadata"]) + def test_a_span_under_litellms_own_unparented_proxy_span_derives(self, key: str) -> None: + proxy_span: Final = _proxy_span() + + attributes: Final = _stamped(_child_of(proxy_span), _carrying(proxy_span, key), _chat_response("ok")) + + assert attributes["langfuse.trace.name"] == "litellm-acompletion" + assert attributes["langfuse.trace.input"] == attributes["langfuse.observation.input"] + assert attributes["langfuse.trace.output"] == attributes["langfuse.observation.output"] + + def test_a_span_under_a_proxy_span_with_an_inbound_traceparent_joins_the_caller_trace(self) -> None: + proxy_span: Final = _proxy_span(trace.set_span_in_context(trace.NonRecordingSpan(CALLER))) + + attributes: Final = _stamped(_child_of(proxy_span), _carrying(proxy_span), _chat_response("ok")) + + assert _trace_fields(attributes) == {} + + def test_a_span_whose_parent_is_not_the_carried_proxy_span_joins_that_parent_trace(self) -> None: + other_parent: Final = _TRACER.start_span("caller-root") + + attributes: Final = _stamped(_child_of(other_parent), _carrying(_proxy_span()), _chat_response("ok")) + + assert _trace_fields(attributes) == {} + + def test_an_inbound_traceparent_still_carries_the_caller_trace_name_and_tags(self) -> None: + span: Final = _TRACER.start_span( + "litellm_request", context=trace.set_span_in_context(trace.NonRecordingSpan(CALLER)) + ) + + attributes: Final = _stamped( + span, + {**CHAT_KWARGS, "litellm_params": {"metadata": {"trace_name": "caller-trace", "tags": ["caller"]}}}, + _chat_response("ok"), + ) + + assert _trace_fields(attributes) == { + "langfuse.trace.name": "caller-trace", + "langfuse.trace.tags": json.dumps(["caller"]), + } + + +ERROR_KEYS: Final = ( + "error.code", + "http.response.status_code", + "error.type", + "error.message", + "error.llm_provider", + "error.stack_trace", +) +SPAN_ATTRIBUTE_LIMIT: Final = 128 +_LIMITED_TRACER: Final = TracerProvider(span_limits=SpanLimits(max_span_attributes=SPAN_ATTRIBUTE_LIMIT)).get_tracer( + "langfuse-otel-limit-tests" +) + + +@dataclass(frozen=True, slots=True) +class ExportedSpan: + keys: frozenset[str] + dropped: int + + +def _exported_full_span(filler: int, *, joins_caller: bool, failed: bool) -> ExportedSpan: + span: Final = _LIMITED_TRACER.start_span( + "litellm_request", + context=trace.set_span_in_context(trace.NonRecordingSpan(CALLER)) if joins_caller else Context(), + ) + span.set_attributes({f"main.{index}": index for index in range(filler)}) + LangfuseOtelLogger._set_langfuse_specific_attributes( + span, {**CHAT_KWARGS, "litellm_params": {"metadata": {}}}, _chat_response("ok") + ) + span.set_attributes(dict.fromkeys(ERROR_KEYS, "scripted failure") if failed else {}) + span.end() + return ExportedSpan(frozenset(getattr(span, "attributes", None) or {}), getattr(span, "dropped_attributes", 0)) + + +class TestDerivedTraceFieldsAtTheSpanAttributeLimit: + @pytest.mark.parametrize("failed", [False, True]) + @pytest.mark.parametrize("filler", range(SPAN_ATTRIBUTE_LIMIT - 30, SPAN_ATTRIBUTE_LIMIT + 1)) + def test_trace_fields_never_evict_or_drop_another_attribute(self, filler: int, failed: bool) -> None: + derived: Final = _exported_full_span(filler, joins_caller=False, failed=failed) + without_trace_fields: Final = _exported_full_span(filler, joins_caller=True, failed=failed) + + assert without_trace_fields.keys <= derived.keys + assert derived.dropped == without_trace_fields.dropped + + def test_a_span_with_room_to_spare_carries_every_trace_field(self) -> None: + derived: Final = _exported_full_span(SPAN_ATTRIBUTE_LIMIT - 30, joins_caller=False, failed=False) + + assert {"langfuse.trace.name", "langfuse.trace.input", "langfuse.trace.output"} <= derived.keys + + LANGFUSE_ENV_ONLY: Final = { key: value for key, value in os.environ.items()