mirror of
https://github.com/BerriAI/litellm.git
synced 2026-11-12 08:42:18 +00:00
fix(langfuse_otel): export only caller identity under the Langfuse metadata keys
- The Langfuse metadata keys carried the whole request metadata. On OTel v2 that sent request headers, caller metadata, user email, user agent and auth metadata to a Langfuse destination that never received them on main; on v1 it doubled each span's metadata payload, so big caller metadata doubled the export size - Both paths now export only the key alias, user, end user, team, team alias and org. On v2 that is the team plus the identity among the operator's `baggage_metadata_keys`, which v2 spans already carry as `litellm.metadata.*`. Everything else stays in the bare `metadata` attribute, as on main - `output_messages` returns a list again, and the arize v2 b3 key pin is back to main's set
This commit is contained in:
yucheng
parent
c264613578
commit
c9410b2aeb
9 changed files with 291 additions and 337 deletions
|
|
@ -14,8 +14,7 @@ from litellm.integrations.langfuse.langfuse_otel_attributes import (
|
|||
)
|
||||
from litellm.integrations.opentelemetry import OpenTelemetry, OpenTelemetryConfig
|
||||
from litellm.integrations.otel.mappers.langfuse import TRACE_IDENTITY_FIELDS
|
||||
from litellm.integrations.otel.model.metadata import exported_request_metadata
|
||||
from litellm.litellm_core_utils.safe_json_dumps import safe_dumps
|
||||
from litellm.integrations.otel.model.metadata import IDENTITY_METADATA_FIELDS, exported_request_metadata
|
||||
from litellm.litellm_core_utils.safe_json_loads import safe_json_loads
|
||||
from litellm.types.integrations.langfuse_otel import (
|
||||
LangfuseSpanAttributes,
|
||||
|
|
@ -30,7 +29,7 @@ else:
|
|||
Span = Any
|
||||
|
||||
|
||||
_STANDARD_LOGGING_OBJECT_ADAPTER: Final[TypeAdapter[Mapping[str, object]]] = TypeAdapter(Mapping[str, object])
|
||||
_MAPPING_ADAPTER: Final[TypeAdapter[Mapping[object, object]]] = TypeAdapter(Mapping[object, object])
|
||||
|
||||
LANGFUSE_CLOUD_EU_ENDPOINT: Final = "https://cloud.langfuse.com/api/public/otel"
|
||||
LANGFUSE_CLOUD_US_ENDPOINT: Final = "https://us.cloud.langfuse.com/api/public/otel"
|
||||
|
|
@ -136,20 +135,23 @@ class LangfuseOtelLogger(OpenTelemetry):
|
|||
@staticmethod
|
||||
def _set_request_metadata_attributes(
|
||||
span: Span,
|
||||
kwargs: dict[str, object], # mutable-ok: required callback signature accepts kwargs as a dict
|
||||
kwargs: Mapping[object, object],
|
||||
) -> None:
|
||||
raw_standard_logging_object: Final = kwargs.get("standard_logging_object")
|
||||
if not isinstance(raw_standard_logging_object, dict):
|
||||
return
|
||||
standard_logging_object: Final = _STANDARD_LOGGING_OBJECT_ADAPTER.validate_python(raw_standard_logging_object)
|
||||
if not isinstance(standard_logging_object.get("metadata"), dict):
|
||||
raw_metadata: Final = _MAPPING_ADAPTER.validate_python(raw_standard_logging_object).get("metadata")
|
||||
if not isinstance(raw_metadata, dict):
|
||||
return
|
||||
observation_metadata: Final = exported_request_metadata(standard_logging_object)
|
||||
metadata: Final = _MAPPING_ADAPTER.validate_python(raw_metadata)
|
||||
observation_metadata: Final = exported_request_metadata(
|
||||
{field: metadata.get(field) for field in IDENTITY_METADATA_FIELDS}
|
||||
)
|
||||
if observation_metadata:
|
||||
_utils.safe_set_attribute(
|
||||
span,
|
||||
LangfuseSpanAttributes.OBSERVATION_METADATA.value,
|
||||
safe_dumps(dict(observation_metadata)),
|
||||
json.dumps(dict(observation_metadata)),
|
||||
)
|
||||
trace_prefix: Final = LangfuseSpanAttributes.TRACE_METADATA.value
|
||||
for field in TRACE_IDENTITY_FIELDS:
|
||||
|
|
@ -276,7 +278,7 @@ class LangfuseOtelLogger(OpenTelemetry):
|
|||
|
||||
metadata: Final = LangfuseOtelLogger._extract_langfuse_metadata(kwargs)
|
||||
LangfuseOtelLogger._set_metadata_attributes(span=span, metadata=metadata)
|
||||
LangfuseOtelLogger._set_request_metadata_attributes(span=span, kwargs=kwargs)
|
||||
LangfuseOtelLogger._set_request_metadata_attributes(span=span, kwargs=_MAPPING_ADAPTER.validate_python(kwargs))
|
||||
|
||||
messages: Final = kwargs.get("messages")
|
||||
if messages:
|
||||
|
|
|
|||
|
|
@ -11,12 +11,10 @@ the JSON-serialized payloads. ``trace_attributes`` maps the caller's trace contr
|
|||
"""
|
||||
|
||||
import json
|
||||
from collections.abc import Callable, Mapping
|
||||
from collections.abc import Callable
|
||||
from types import MappingProxyType
|
||||
from typing import Final
|
||||
|
||||
from pydantic import TypeAdapter
|
||||
|
||||
from litellm.integrations.otel.mappers.base import AttributeMap, AttrValue, SpanData
|
||||
from litellm.integrations.otel.mappers.utils import (
|
||||
collect,
|
||||
|
|
@ -31,8 +29,6 @@ from litellm.integrations.otel.model.payloads import (
|
|||
LLMUsage,
|
||||
)
|
||||
from litellm.integrations.otel.model.trace_controls import TraceControls
|
||||
from litellm.integrations.otel.model.utils import as_str
|
||||
from litellm.litellm_core_utils.safe_json_dumps import safe_dumps
|
||||
|
||||
LANGFUSE_OBSERVATION_INPUT: Final = "langfuse.observation.input"
|
||||
LANGFUSE_OBSERVATION_OUTPUT: Final = "langfuse.observation.output"
|
||||
|
|
@ -42,7 +38,6 @@ LANGFUSE_TRACE_SESSION_ID: Final = "session.id"
|
|||
LANGFUSE_TRACE_TAGS: Final = "langfuse.trace.tags"
|
||||
LANGFUSE_OBSERVATION_METADATA: Final = "langfuse.observation.metadata"
|
||||
LANGFUSE_TRACE_METADATA_PREFIX: Final = "langfuse.trace.metadata."
|
||||
_REQUEST_METADATA_ADAPTER: Final[TypeAdapter[Mapping[str, object]]] = TypeAdapter(Mapping[str, object])
|
||||
TRACE_IDENTITY_FIELDS: Final = (
|
||||
"user_api_key_alias",
|
||||
"user_api_key_user_id",
|
||||
|
|
@ -57,7 +52,7 @@ LANGFUSE_REQUEST_METADATA_ATTRIBUTES: Final[frozenset[str]] = frozenset(
|
|||
|
||||
|
||||
def _identity_field(name: str) -> Callable[[LLMCallSpanData], AttrValue | None]:
|
||||
return lambda d: as_str(d.request_metadata.get(name)) or None
|
||||
return lambda d: d.request_metadata.get(name)
|
||||
|
||||
|
||||
class LangfuseMapper:
|
||||
|
|
@ -94,11 +89,7 @@ class LangfuseMapper:
|
|||
|
||||
# JSON-payload attributes: each builder returns the serialized blob or None.
|
||||
_BLOB_ATTRS: dict[str, Callable[[LLMCallSpanData], AttrValue | None]] = {
|
||||
LANGFUSE_OBSERVATION_METADATA: lambda d: (
|
||||
safe_dumps({**_REQUEST_METADATA_ADAPTER.validate_python(d.request_metadata)})
|
||||
if d.request_metadata
|
||||
else None
|
||||
),
|
||||
LANGFUSE_OBSERVATION_METADATA: lambda d: json_if(dict(d.request_metadata)),
|
||||
"langfuse.observation.model.parameters": lambda d: json_if(
|
||||
collect(LangfuseMapper._MODEL_PARAMS, d.request_params)
|
||||
),
|
||||
|
|
|
|||
|
|
@ -135,9 +135,9 @@ def message_content(message: object) -> str | None:
|
|||
return None
|
||||
|
||||
|
||||
def output_messages(data: LLMCallSpanData) -> tuple[object, ...]:
|
||||
def output_messages(data: LLMCallSpanData) -> list:
|
||||
"""The ``message`` payload of each response choice."""
|
||||
return tuple(c.get("message") for c in data.choices_out if isinstance(c, dict))
|
||||
return [c.get("message") for c in data.choices_out if isinstance(c, dict)]
|
||||
|
||||
|
||||
def message_tool_calls(message: object) -> tuple[MessageToolCall, ...]:
|
||||
|
|
|
|||
|
|
@ -52,18 +52,24 @@ from litellm.constants import (
|
|||
from litellm.integrations.otel.model.semconv import resolve_operation
|
||||
from litellm.integrations.otel.model.trace_controls import TraceControls, caller_trace_controls
|
||||
from litellm.integrations.otel.model.utils import as_str, as_str_mapping, to_seconds
|
||||
from litellm.types.utils import PROMPT_QUOTING_ROUTING_DECISION_FIELDS, InternalCallOrigin
|
||||
from litellm.types.utils import InternalCallOrigin
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from litellm.types.utils import StandardLoggingPayload
|
||||
|
||||
_INTERNAL_CALL_ORIGINS: Final[frozenset[str]] = frozenset(get_args(InternalCallOrigin))
|
||||
_AUTH_ATTRIBUTE: Final = TypeAdapter(object)
|
||||
_REQUEST_METADATA_MAPPING_ADAPTER: Final[TypeAdapter[Mapping[str, object]]] = TypeAdapter(Mapping[str, object])
|
||||
_NESTED_METADATA_MAPPING_ADAPTER: Final[TypeAdapter[Mapping[object, object]]] = TypeAdapter(Mapping[object, object])
|
||||
_PROMPT_MANAGEMENT_CONTENT_FIELDS: Final[frozenset[str]] = frozenset({"prompt_variables"})
|
||||
_MCP_TOOL_CALL_CONTENT_FIELDS: Final[frozenset[str]] = frozenset({"arguments", "result"})
|
||||
_VECTOR_STORE_CONTENT_FIELDS: Final[frozenset[str]] = frozenset({"query", "vector_store_search_response"})
|
||||
_EXPORTED_METADATA_ADAPTER: Final[TypeAdapter[Mapping[str, str]]] = TypeAdapter(Mapping[str, str])
|
||||
# The caller identity the Langfuse metadata keys export; the rest of the request metadata stays in the bare
|
||||
# ``metadata`` attribute.
|
||||
IDENTITY_METADATA_FIELDS: Final = (
|
||||
"user_api_key_alias",
|
||||
"user_api_key_user_id",
|
||||
"user_api_key_end_user_id",
|
||||
"user_api_key_team_id",
|
||||
"user_api_key_team_alias",
|
||||
"user_api_key_org_id",
|
||||
)
|
||||
|
||||
REQUESTER_METADATA_KEY: Final = "requester_metadata"
|
||||
REQUESTER_METADATA_PATH: Final = f"{REQUESTER_METADATA_KEY}."
|
||||
|
|
@ -469,66 +475,21 @@ def _model_info_id(model_info: object) -> str | None:
|
|||
return None
|
||||
|
||||
|
||||
def exported_request_metadata(payload: Mapping[str, object]) -> Mapping[str, object]:
|
||||
"""The request metadata as the logging callbacks export it: ``user_api_key_*``
|
||||
dropped when ``litellm.redact_user_api_key_info`` is on, header keys nested
|
||||
under a mapping ``requester_metadata`` (any other value is exported as sent),
|
||||
``None`` values dropped."""
|
||||
from litellm.integrations.langfuse.langfuse import log_requester_metadata
|
||||
from litellm.litellm_core_utils.redact_messages import ( # pyright: ignore[reportUnknownVariableType] # helper uses bare dict
|
||||
redact_user_api_key_info as _redact_user_api_key_info,
|
||||
def exported_request_metadata(metadata: Mapping[str, object]) -> Mapping[str, str]:
|
||||
"""The non-empty string ``IDENTITY_METADATA_FIELDS`` of ``metadata``, with ``user_api_key_*`` dropped when
|
||||
``litellm.redact_user_api_key_info`` is on, as the ``langfuse`` callback does."""
|
||||
from litellm.litellm_core_utils.redact_messages import (
|
||||
redact_user_api_key_info as _redact_user_api_key_info, # pyright: ignore[reportUnknownVariableType] # helper uses bare dict
|
||||
)
|
||||
|
||||
raw_metadata: Final = payload.get("metadata")
|
||||
metadata: Final = _REQUEST_METADATA_MAPPING_ADAPTER.validate_python(
|
||||
raw_metadata if isinstance(raw_metadata, Mapping) else MappingProxyType({})
|
||||
)
|
||||
redact_metadata: Final = cast( # cast-ok: adapter validates the helper's output below
|
||||
Callable[[dict[str, object]], object],
|
||||
Callable[[dict[str, str]], object],
|
||||
_redact_user_api_key_info,
|
||||
)
|
||||
redacted: Final = _REQUEST_METADATA_MAPPING_ADAPTER.validate_python(redact_metadata(dict(metadata)))
|
||||
requester_metadata: Final = redacted.get(REQUESTER_METADATA_KEY)
|
||||
exported: Final = (
|
||||
_REQUEST_METADATA_MAPPING_ADAPTER.validate_python(log_requester_metadata(redacted))
|
||||
if requester_metadata is None or isinstance(requester_metadata, Mapping)
|
||||
else redacted
|
||||
)
|
||||
null_filtered: Final = MappingProxyType({key: value for key, value in exported.items() if value is not None})
|
||||
return _without_request_metadata_content(null_filtered)
|
||||
|
||||
|
||||
def _without_mapping_content_fields(value: object, content_fields: frozenset[str]) -> object:
|
||||
if not isinstance(value, dict):
|
||||
return value
|
||||
mapping: Final = _NESTED_METADATA_MAPPING_ADAPTER.validate_python(value)
|
||||
return {key: item for key, item in mapping.items() if key not in content_fields}
|
||||
|
||||
|
||||
def _without_vector_store_content_fields(value: object) -> object:
|
||||
if not isinstance(value, list):
|
||||
return value
|
||||
return tuple(_without_mapping_content_fields(item, _VECTOR_STORE_CONTENT_FIELDS) for item in value)
|
||||
|
||||
|
||||
def _request_metadata_value_without_content(key: str, value: object) -> object:
|
||||
match key:
|
||||
case "prompt_management_metadata":
|
||||
return _without_mapping_content_fields(value, _PROMPT_MANAGEMENT_CONTENT_FIELDS)
|
||||
case "mcp_tool_call_metadata":
|
||||
return _without_mapping_content_fields(value, _MCP_TOOL_CALL_CONTENT_FIELDS)
|
||||
case "vector_store_request_metadata":
|
||||
return _without_vector_store_content_fields(value)
|
||||
case "routing_decision":
|
||||
return _without_mapping_content_fields(value, PROMPT_QUOTING_ROUTING_DECISION_FIELDS)
|
||||
case _:
|
||||
return value
|
||||
|
||||
|
||||
def _without_request_metadata_content(metadata: Mapping[str, object]) -> Mapping[str, object]:
|
||||
return MappingProxyType(
|
||||
{key: _request_metadata_value_without_content(key, value) for key, value in metadata.items()}
|
||||
)
|
||||
present: Final = {
|
||||
field: value for field in IDENTITY_METADATA_FIELDS if isinstance(value := metadata.get(field), str) and value
|
||||
}
|
||||
return MappingProxyType(dict(_EXPORTED_METADATA_ADAPTER.validate_python(redact_metadata(present))))
|
||||
|
||||
|
||||
def _team_metadata_dict(value: object) -> Mapping[str, object] | None:
|
||||
|
|
|
|||
|
|
@ -422,7 +422,8 @@ class LLMCallSpanData:
|
|||
response_cost: float | None
|
||||
server: ServerInfo | None
|
||||
identity: RequestIdentity
|
||||
request_metadata: Mapping[str, object] = field(default_factory=dict)
|
||||
# The caller identity among the promoted metadata plus the team (see ``exported_request_metadata``).
|
||||
request_metadata: Mapping[str, str] = field(default_factory=lambda: _EMPTY_METADATA)
|
||||
is_streaming: bool | None = None
|
||||
cost: LLMCost = field(default_factory=LLMCost)
|
||||
tools: tuple[ToolDefinition, ...] = ()
|
||||
|
|
@ -468,6 +469,7 @@ class LLMCallSpanData:
|
|||
context: Final = RequestContext.from_standard_logging_payload(payload)
|
||||
metadata: Final = payload.get("metadata")
|
||||
decision: Final = metadata["routing_decision"] if metadata and "routing_decision" in metadata else None
|
||||
promoted_metadata: Final = allowlisted_metadata(context.identity.metadata, metadata_keys)
|
||||
# Normalize ``response`` to a dict once so the content/id reads below are a
|
||||
# plain ``.get`` — no repeated ``isinstance`` guards.
|
||||
raw_response: Final = payload.get("response")
|
||||
|
|
@ -498,8 +500,14 @@ class LLMCallSpanData:
|
|||
cost=LLMCost.from_breakdown(cast("Mapping[str, object] | None", payload.get("cost_breakdown"))),
|
||||
server=ServerInfo.from_api_base(context.api_base),
|
||||
identity=context.identity,
|
||||
request_metadata=exported_request_metadata(payload),
|
||||
promoted_metadata=allowlisted_metadata(context.identity.metadata, metadata_keys),
|
||||
request_metadata=exported_request_metadata(
|
||||
{
|
||||
"user_api_key_team_id": context.identity.team_id,
|
||||
"user_api_key_team_alias": context.identity.team_alias,
|
||||
**promoted_metadata,
|
||||
}
|
||||
),
|
||||
promoted_metadata=promoted_metadata,
|
||||
is_streaming=as_bool(payload.get("stream")),
|
||||
tools=_extract_tools(params),
|
||||
messages_in=_dicts(payload.get("messages")) if capture_content else (),
|
||||
|
|
|
|||
|
|
@ -88,13 +88,11 @@ _LANGFUSE_B3_KEYS: Final = frozenset(
|
|||
"langfuse.observation.cost_details",
|
||||
"langfuse.observation.id",
|
||||
"langfuse.observation.input",
|
||||
"langfuse.observation.metadata",
|
||||
"langfuse.observation.metadata.provider",
|
||||
"langfuse.observation.model.name",
|
||||
"langfuse.observation.output",
|
||||
"langfuse.observation.type",
|
||||
"langfuse.observation.usage_details",
|
||||
"langfuse.trace.metadata.user_api_key_user_id",
|
||||
"litellm.api_key.hash",
|
||||
"litellm.call_id",
|
||||
"litellm.call_type",
|
||||
|
|
|
|||
|
|
@ -27,6 +27,8 @@ _IDENTITIES: Final = (
|
|||
"user_api_key_team_id",
|
||||
"user_api_key_team_alias",
|
||||
)
|
||||
# The only request metadata the Langfuse metadata keys carry; everything else stays in the bare ``metadata``.
|
||||
_EXPORTED_FIELDS: Final = frozenset({*_IDENTITIES, "user_api_key_org_id"})
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
|
|
@ -357,19 +359,13 @@ def _scenario_identity(
|
|||
}
|
||||
|
||||
|
||||
def _assert_identity(
|
||||
attrs: Mapping[str, object],
|
||||
expected: Mapping[str, object],
|
||||
*,
|
||||
spend_logs_metadata: Mapping[str, object] | None = None,
|
||||
) -> None:
|
||||
def _assert_identity(attrs: Mapping[str, object], expected: Mapping[str, object]) -> None:
|
||||
assert "metadata" in attrs, sorted(attrs)
|
||||
assert "langfuse.observation.metadata" in attrs, sorted(attrs)
|
||||
observation: Final = json.loads(str(attrs["langfuse.observation.metadata"]))
|
||||
assert {key: observation.get(key) for key in expected} == dict(expected), observation
|
||||
assert not [key for key, value in observation.items() if value is None], observation
|
||||
if spend_logs_metadata is not None:
|
||||
assert observation["spend_logs_metadata"] == dict(spend_logs_metadata), observation
|
||||
assert set(observation) <= _EXPORTED_FIELDS, observation
|
||||
assert all(isinstance(value, str) and value for value in observation.values()), observation
|
||||
flattened: Final = {
|
||||
attribute: attrs.get(attribute)
|
||||
for attribute in (f"langfuse.trace.metadata.{field}" for field in _IDENTITIES)
|
||||
|
|
@ -411,7 +407,7 @@ def test_langfuse_otel_emits_request_metadata_under_langfuse_observation_and_tra
|
|||
assert response.json()["choices"][0]["message"]["content"] == "langfuse-echo"
|
||||
|
||||
attrs: Final = _observation_span(rig.collector, response.json()["id"])
|
||||
_assert_identity(attrs, _chat_identity_fields(identity), spend_logs_metadata={"ticket": "LIT-8283"})
|
||||
_assert_identity(attrs, _chat_identity_fields(identity))
|
||||
|
||||
|
||||
def test_langfuse_otel_metadata_redacts_user_api_key_fields_like_vanilla_langfuse(
|
||||
|
|
@ -440,11 +436,7 @@ def test_langfuse_otel_metadata_redacts_user_api_key_fields_like_vanilla_langfus
|
|||
assert response.status_code == 200, response.text
|
||||
|
||||
attrs: Final = _observation_span(rig.collector, response.json()["id"])
|
||||
assert "langfuse.observation.metadata" in attrs, sorted(attrs)
|
||||
observation: Final = json.loads(str(attrs["langfuse.observation.metadata"]))
|
||||
assert not [name for name in observation if name.startswith("user_api_key")], observation
|
||||
assert not [key_ for key_, value in observation.items() if value is None], observation
|
||||
assert observation["spend_logs_metadata"] == {"ticket": "LIT-8283"}, observation
|
||||
assert "langfuse.observation.metadata" not in attrs, attrs["langfuse.observation.metadata"]
|
||||
assert not [name for name in attrs if name.startswith("langfuse.trace.metadata.")], sorted(attrs)
|
||||
|
||||
|
||||
|
|
@ -472,7 +464,7 @@ def test_langfuse_otel_metadata_reaches_langfuse_for_openai_sdk_stream(gateway:
|
|||
assert chunks, "stream produced no chunks"
|
||||
response_id: Final = chunks[-1].id
|
||||
attrs: Final = _observation_span(rig.collector, response_id)
|
||||
_assert_identity(attrs, _chat_identity_fields(identity), spend_logs_metadata={"ticket": "LIT-8283"})
|
||||
_assert_identity(attrs, _chat_identity_fields(identity))
|
||||
|
||||
|
||||
async def test_langfuse_otel_metadata_reaches_langfuse_for_openai_sdk_async(gateway: Gateway, tmp_path: Path) -> None:
|
||||
|
|
@ -494,7 +486,7 @@ async def test_langfuse_otel_metadata_reaches_langfuse_for_openai_sdk_async(gate
|
|||
extra_body={"cache": {"no-cache": True}},
|
||||
)
|
||||
attrs: Final = _observation_span(rig.collector, response.id)
|
||||
_assert_identity(attrs, _chat_identity_fields(identity), spend_logs_metadata={"ticket": "LIT-8283"})
|
||||
_assert_identity(attrs, _chat_identity_fields(identity))
|
||||
|
||||
|
||||
def test_langfuse_otel_metadata_reaches_langfuse_for_anthropic_sdk_messages(gateway: Gateway, tmp_path: Path) -> None:
|
||||
|
|
@ -517,7 +509,7 @@ def test_langfuse_otel_metadata_reaches_langfuse_for_anthropic_sdk_messages(gate
|
|||
extra_body={"litellm_metadata": {"tags": ["lf8283"]}, "metadata": {"user_id": identity["end_user"]}},
|
||||
)
|
||||
attrs: Final = _observation_span(rig.collector, message.id)
|
||||
_assert_identity(attrs, _chat_identity_fields(identity), spend_logs_metadata={"ticket": "LIT-8283"})
|
||||
_assert_identity(attrs, _chat_identity_fields(identity))
|
||||
|
||||
|
||||
async def test_langfuse_otel_metadata_reaches_langfuse_for_anthropic_sdk_stream(
|
||||
|
|
@ -545,7 +537,7 @@ async def test_langfuse_otel_metadata_reaches_langfuse_for_anthropic_sdk_stream(
|
|||
):
|
||||
final: Final = await stream.get_final_message()
|
||||
attrs: Final = _observation_span(rig.collector, final.id)
|
||||
_assert_identity(attrs, _chat_identity_fields(identity), spend_logs_metadata={"ticket": "LIT-8283"})
|
||||
_assert_identity(attrs, _chat_identity_fields(identity))
|
||||
|
||||
|
||||
def test_langfuse_otel_metadata_reaches_langfuse_for_responses_api(gateway: Gateway, tmp_path: Path) -> None:
|
||||
|
|
@ -567,7 +559,7 @@ def test_langfuse_otel_metadata_reaches_langfuse_for_responses_api(gateway: Gate
|
|||
extra_body={"cache": {"no-cache": True}},
|
||||
)
|
||||
attrs: Final = _observation_span(rig.collector, response.id)
|
||||
_assert_identity(attrs, _chat_identity_fields(identity), spend_logs_metadata={"ticket": "LIT-8283"})
|
||||
_assert_identity(attrs, _chat_identity_fields(identity))
|
||||
|
||||
|
||||
def test_langfuse_otel_metadata_reaches_langfuse_for_responses_sse_stream(gateway: Gateway, tmp_path: Path) -> None:
|
||||
|
|
@ -598,7 +590,7 @@ def test_langfuse_otel_metadata_reaches_langfuse_for_responses_sse_stream(gatewa
|
|||
event["response_id"] for event in events if event.get("type") == "response.output_text.delta"
|
||||
)
|
||||
attrs: Final = _observation_span(rig.collector, response_id)
|
||||
_assert_identity(attrs, _chat_identity_fields(identity), spend_logs_metadata={"ticket": "LIT-8283"})
|
||||
_assert_identity(attrs, _chat_identity_fields(identity))
|
||||
|
||||
|
||||
def test_langfuse_otel_caller_trace_metadata_survives_flattened_identity(gateway: Gateway, tmp_path: Path) -> None:
|
||||
|
|
@ -623,7 +615,7 @@ def test_langfuse_otel_caller_trace_metadata_survives_flattened_identity(gateway
|
|||
assert response.status_code == 200, response.text
|
||||
attrs: Final = _observation_span(rig.collector, response.json()["id"])
|
||||
assert json.loads(str(attrs["langfuse.trace.metadata"])) == {"deploy": "blue"}, attrs
|
||||
_assert_identity(attrs, _chat_identity_fields(identity), spend_logs_metadata={"ticket": "LIT-8283"})
|
||||
_assert_identity(attrs, _chat_identity_fields(identity))
|
||||
|
||||
|
||||
def test_langfuse_otel_failure_span_carries_request_metadata(gateway: Gateway, tmp_path: Path) -> None:
|
||||
|
|
@ -646,7 +638,7 @@ def test_langfuse_otel_failure_span_carries_request_metadata(gateway: Gateway, t
|
|||
)
|
||||
assert response.status_code >= 500, response.text
|
||||
attrs: Final = _marked_span(rig.collector, marker, "langfuse.observation.")
|
||||
_assert_identity(attrs, _chat_identity_fields(identity), spend_logs_metadata={"ticket": "LIT-8283"})
|
||||
_assert_identity(attrs, _chat_identity_fields(identity))
|
||||
|
||||
|
||||
def test_langfuse_otel_teamless_key_emits_no_team_identity(gateway: Gateway, tmp_path: Path) -> None:
|
||||
|
|
@ -672,7 +664,7 @@ def test_langfuse_otel_teamless_key_emits_no_team_identity(gateway: Gateway, tmp
|
|||
observation: Final = json.loads(str(attrs["langfuse.observation.metadata"]))
|
||||
for field in ("user_api_key_team_id", "user_api_key_team_alias", "user_api_key_end_user_id"):
|
||||
assert field not in observation, observation
|
||||
assert not [key_ for key_, value in observation.items() if value is None], observation
|
||||
assert set(observation) <= _EXPORTED_FIELDS, observation
|
||||
assert {
|
||||
attribute: attrs.get(attribute)
|
||||
for attribute in (f"langfuse.trace.metadata.{field}" for field in _IDENTITIES)
|
||||
|
|
@ -706,7 +698,7 @@ def test_langfuse_otel_teamless_key_keeps_caller_trace_metadata(gateway: Gateway
|
|||
assert attrs.get("langfuse.trace.metadata.user_api_key_alias") == identity["key_alias"], attrs
|
||||
|
||||
|
||||
def test_langfuse_otel_spend_logs_metadata_roundtrips_oversized_and_nonstring_values(
|
||||
def test_langfuse_otel_oversized_caller_metadata_is_not_copied_into_the_langfuse_keys(
|
||||
gateway: Gateway, tmp_path: Path
|
||||
) -> None:
|
||||
marker: Final = "lf-bigmeta-" + uuid.uuid4().hex
|
||||
|
|
@ -716,6 +708,7 @@ def test_langfuse_otel_spend_logs_metadata_roundtrips_oversized_and_nonstring_va
|
|||
"steps": ["a", "b"],
|
||||
"nested": {"inner": {"leaf": True}},
|
||||
}
|
||||
caller_note: Final = "y" * 50000
|
||||
with (
|
||||
_langfuse_rig(gateway, tmp_path, {}, marker) as rig,
|
||||
rig.candidate.scenario() as scenario,
|
||||
|
|
@ -728,15 +721,17 @@ def test_langfuse_otel_spend_logs_metadata_roundtrips_oversized_and_nonstring_va
|
|||
"model": identity["model"],
|
||||
"messages": [{"role": "user", "content": marker}],
|
||||
"user": identity["end_user"],
|
||||
"metadata": {"caller_note": caller_note},
|
||||
"cache": {"no-cache": True},
|
||||
},
|
||||
key=key,
|
||||
)
|
||||
assert response.status_code == 200, response.text
|
||||
attrs: Final = _observation_span(rig.collector, response.json()["id"])
|
||||
observation: Final = json.loads(str(attrs["langfuse.observation.metadata"]))
|
||||
assert observation["spend_logs_metadata"] == dict(spend_logs_metadata), observation
|
||||
_assert_identity(attrs, _chat_identity_fields(identity))
|
||||
assert len(str(attrs["langfuse.observation.metadata"])) < 1000, len(str(attrs["langfuse.observation.metadata"]))
|
||||
assert caller_note in str(attrs["metadata"]), "the bare metadata attribute keeps the caller metadata"
|
||||
assert spend_logs_metadata["blob"] in str(attrs["metadata"]), "the bare metadata attribute keeps it"
|
||||
|
||||
|
||||
def test_langfuse_otel_unauthenticated_request_leaks_no_identity(gateway: Gateway, tmp_path: Path) -> None:
|
||||
|
|
@ -1041,8 +1036,7 @@ def test_langfuse_otel_v2_mapper_emits_request_metadata_and_identity(gateway: Ga
|
|||
observation: Final = json.loads(str(attrs["langfuse.observation.metadata"]))
|
||||
expected: Final = _chat_identity_fields(identity)
|
||||
assert {key: observation.get(key) for key in expected} == dict(expected), observation
|
||||
assert observation["spend_logs_metadata"] == {"ticket": "LIT-8283"}, observation
|
||||
assert not [key_ for key_, value in observation.items() if value is None], observation
|
||||
assert set(observation) <= _EXPORTED_FIELDS, observation
|
||||
flattened: Final = {
|
||||
attribute: attrs.get(attribute)
|
||||
for attribute in (f"langfuse.trace.metadata.{field}" for field in _IDENTITIES)
|
||||
|
|
@ -1086,11 +1080,7 @@ def test_langfuse_otel_v2_mapper_redacts_user_api_key_fields(gateway: Gateway, t
|
|||
seconds=25,
|
||||
)
|
||||
attrs: Final = next(s for s in observed if s.get("gen_ai.response.id") == response.json()["id"])
|
||||
assert "langfuse.observation.metadata" in attrs, sorted(attrs)
|
||||
observation: Final = json.loads(str(attrs["langfuse.observation.metadata"]))
|
||||
assert not [name for name in observation if name.startswith("user_api_key")], observation
|
||||
assert not [key_ for key_, value in observation.items() if value is None], observation
|
||||
assert observation["spend_logs_metadata"] == {"ticket": "LIT-8283"}, observation
|
||||
assert "langfuse.observation.metadata" not in attrs, attrs["langfuse.observation.metadata"]
|
||||
assert not [name for name in attrs if name.startswith("langfuse.trace.metadata.user_api_key_")], sorted(attrs)
|
||||
|
||||
|
||||
|
|
@ -1129,7 +1119,7 @@ def test_langfuse_otel_v2_mapper_teamless_key_emits_no_team_identity(gateway: Ga
|
|||
observation: Final = json.loads(str(attrs["langfuse.observation.metadata"]))
|
||||
for field in ("user_api_key_team_id", "user_api_key_team_alias", "user_api_key_end_user_id"):
|
||||
assert field not in observation, observation
|
||||
assert not [key_ for key_, value in observation.items() if value is None], observation
|
||||
assert set(observation) <= _EXPORTED_FIELDS, observation
|
||||
assert {
|
||||
attribute: attrs.get(attribute)
|
||||
for attribute in (f"langfuse.trace.metadata.{field}" for field in _IDENTITIES)
|
||||
|
|
@ -1185,7 +1175,7 @@ def test_otel_v2_scalar_requester_metadata_keeps_the_llm_span(
|
|||
assert attrs.get("gen_ai.usage.input_tokens") == 3, attrs
|
||||
if callback == "langfuse_otel":
|
||||
observation: Final = json.loads(str(attrs["langfuse.observation.metadata"]))
|
||||
assert observation["requester_metadata"] == 1, observation
|
||||
assert "requester_metadata" not in observation, observation
|
||||
assert observation["user_api_key_alias"] == identity["key_alias"], observation
|
||||
|
||||
|
||||
|
|
@ -1205,7 +1195,7 @@ def test_langfuse_otel_scalar_requester_metadata_keeps_the_observation_input(gat
|
|||
attrs: Final = _observation_span(rig.collector, response.json()["id"])
|
||||
assert marker in str(attrs.get("langfuse.observation.input")), sorted(attrs)
|
||||
observation: Final = json.loads(str(attrs["langfuse.observation.metadata"]))
|
||||
assert observation["requester_metadata"] == 1, observation
|
||||
assert "requester_metadata" not in observation, observation
|
||||
assert observation["user_api_key_alias"] == identity["key_alias"], observation
|
||||
|
||||
|
||||
|
|
@ -1249,6 +1239,7 @@ def test_otel_v2_collector_beside_langfuse_gets_no_langfuse_request_metadata(
|
|||
observation: Final = json.loads(str(langfuse["langfuse.observation.metadata"]))
|
||||
assert observation["user_api_key_alias"] == identity["key_alias"], observation
|
||||
assert langfuse["langfuse.trace.metadata.user_api_key_alias"] == identity["key_alias"], langfuse
|
||||
assert probe not in json.dumps(langfuse, default=str), langfuse
|
||||
|
||||
collector: Final = _v2_llm_span(generic, response.json()["id"])
|
||||
assert "langfuse.observation.metadata" not in collector, sorted(collector)
|
||||
|
|
@ -1260,7 +1251,71 @@ def test_otel_v2_collector_beside_langfuse_gets_no_langfuse_request_metadata(
|
|||
assert collector.get("gen_ai.usage.input_tokens") == 3, collector
|
||||
|
||||
|
||||
_PROMPT_QUOTING_ROUTING_FIELDS: Final = ("signals", "matched_keyword", "escalation_keyword", "classifier_crux")
|
||||
@pytest.mark.parametrize("route", ["/v1/chat/completions", "/v1/messages", "/v1/responses"])
|
||||
def test_otel_v2_langfuse_gets_no_request_headers_or_caller_metadata(
|
||||
route: str, gateway: Gateway, tmp_path: Path
|
||||
) -> None:
|
||||
marker: Final = "lf-v2-probe-" + uuid.uuid4().hex
|
||||
probe: Final = "lf-probe-" + uuid.uuid4().hex
|
||||
with (
|
||||
_langfuse_rig(gateway, tmp_path, {}, marker, overrides={"LITELLM_OTEL_V2": "1"}) as rig,
|
||||
rig.candidate.scenario() as scenario,
|
||||
):
|
||||
key, identity = _scenario_identity(scenario, rig.provider)
|
||||
model: Final = (
|
||||
scenario.model(model="anthropic/claude-sonnet-4-6", api_base=rig.provider.url)
|
||||
if route == "/v1/messages"
|
||||
else identity["model"]
|
||||
)
|
||||
body: Final = {
|
||||
"/v1/chat/completions": {
|
||||
"model": model,
|
||||
"messages": [{"role": "user", "content": marker}],
|
||||
"metadata": {"caller_note": probe},
|
||||
"cache": {"no-cache": True},
|
||||
},
|
||||
"/v1/messages": {
|
||||
"model": model,
|
||||
"max_tokens": 16,
|
||||
"messages": [{"role": "user", "content": marker}],
|
||||
"litellm_metadata": {"caller_note": probe},
|
||||
},
|
||||
"/v1/responses": {"model": model, "input": marker, "litellm_metadata": {"caller_note": probe}},
|
||||
}[route]
|
||||
watch: Final = _watched_spans(rig.collector)
|
||||
response: Final = rig.candidate.request("POST", route, body, key=key, headers={"x-lit8283-probe": probe})
|
||||
assert response.status_code == 200, response.text
|
||||
settle: Final = rig.candidate.request(
|
||||
"POST",
|
||||
"/v1/chat/completions",
|
||||
{
|
||||
"model": identity["model"],
|
||||
"messages": [{"role": "user", "content": marker}],
|
||||
"cache": {"no-cache": True},
|
||||
},
|
||||
key=key,
|
||||
)
|
||||
assert settle.status_code == 200, settle.text
|
||||
|
||||
def llm_span(spans: Sequence[Mapping[str, object]], response_id: str) -> Mapping[str, object] | None:
|
||||
return next(
|
||||
(span for span in spans if _response_id(span.get("gen_ai.response.id")) == _response_id(response_id)),
|
||||
None,
|
||||
)
|
||||
|
||||
exported: Final = eventually(
|
||||
watch,
|
||||
lambda spans: (
|
||||
llm_span(spans, response.json()["id"]) is not None and llm_span(spans, settle.json()["id"]) is not None
|
||||
),
|
||||
seconds=25,
|
||||
)
|
||||
attrs: Final = llm_span(exported, response.json()["id"])
|
||||
assert attrs is not None
|
||||
observation: Final = json.loads(str(attrs["langfuse.observation.metadata"]))
|
||||
assert observation["user_api_key_alias"] == identity["key_alias"], observation
|
||||
assert set(observation) <= _EXPORTED_FIELDS, observation
|
||||
assert not [span for span in exported if probe in json.dumps(span, default=str)], exported
|
||||
|
||||
|
||||
def _keyword_router(router: str, keyword: str) -> Callable[[str], Sequence[Mapping[str, object]]]:
|
||||
|
|
@ -1331,9 +1386,13 @@ def test_langfuse_otel_metadata_omits_prompt_quoting_routing_fields(
|
|||
if otel_v2
|
||||
else _observation_span(rig.collector, response.json()["id"])
|
||||
)
|
||||
routing: Final = json.loads(str(attrs["langfuse.observation.metadata"]))["routing_decision"]
|
||||
assert routing["tier"] == "REASONING", routing
|
||||
assert routing["router_model_name"] == router, routing
|
||||
assert not [field for field in _PROMPT_QUOTING_ROUTING_FIELDS if field in routing], routing
|
||||
observation: Final = json.loads(str(attrs["langfuse.observation.metadata"]))
|
||||
assert set(observation) <= _EXPORTED_FIELDS, observation
|
||||
langfuse_metadata: Final = {
|
||||
name: value
|
||||
for name, value in attrs.items()
|
||||
if name == "langfuse.observation.metadata" or name.startswith("langfuse.trace.metadata.")
|
||||
}
|
||||
assert keyword not in json.dumps(langfuse_metadata, default=str), langfuse_metadata
|
||||
if otel_v2:
|
||||
assert keyword not in json.dumps(attrs, default=str), attrs
|
||||
|
|
|
|||
|
|
@ -21,6 +21,7 @@ from litellm.integrations.otel.mappers import (
|
|||
resolve_mappers,
|
||||
)
|
||||
from litellm.integrations.otel.mappers.openinference import fit_indexed_messages
|
||||
from litellm.integrations.otel.model.baggage import DEFAULT_BAGGAGE_METADATA_KEYS
|
||||
from litellm.integrations.otel.model.payloads import (
|
||||
EmbeddingOutput,
|
||||
LLMCallSpanData,
|
||||
|
|
@ -777,33 +778,40 @@ def test_resolve_mappers_rejects_unknown_name():
|
|||
resolve_mappers(["genai", "nope"])
|
||||
|
||||
|
||||
def test_langfuse_mapper_emits_request_metadata_and_identity():
|
||||
def _langfuse_attrs(
|
||||
metadata: Mapping[str, object], metadata_keys: tuple[str, ...] = DEFAULT_BAGGAGE_METADATA_KEYS
|
||||
) -> Mapping[str, object]:
|
||||
payload: Final[dict[str, object]] = {
|
||||
"call_type": "acompletion",
|
||||
"custom_llm_provider": "openai",
|
||||
"model": "gpt-4o",
|
||||
"metadata": {
|
||||
"metadata": metadata,
|
||||
"response": {},
|
||||
}
|
||||
return LangfuseMapper().map(LLMCallSpanData.from_standard_logging_payload(payload, metadata_keys=metadata_keys))
|
||||
|
||||
|
||||
def test_langfuse_mapper_emits_request_metadata_and_identity():
|
||||
attrs: Final = _langfuse_attrs(
|
||||
{
|
||||
"user_api_key_alias": "prod-key",
|
||||
"user_api_key_user_id": "user-1",
|
||||
"user_api_key_end_user_id": "end-1",
|
||||
"user_api_key_team_id": "team-9",
|
||||
"user_api_key_team_alias": "team nine",
|
||||
"user_api_key_max_budget": None,
|
||||
"requester_ip_address": "10.0.0.1",
|
||||
"spend_logs_metadata": {"ticket": "LIT-8283"},
|
||||
"requester_metadata": {"headers": {"x-tenant": "acme"}},
|
||||
},
|
||||
"response": {},
|
||||
}
|
||||
attrs: Final = LangfuseMapper().map(LLMCallSpanData.from_standard_logging_payload(payload))
|
||||
}
|
||||
)
|
||||
|
||||
assert json.loads(attrs["langfuse.observation.metadata"]) == {
|
||||
"user_api_key_alias": "prod-key",
|
||||
"user_api_key_user_id": "user-1",
|
||||
"user_api_key_end_user_id": "end-1",
|
||||
"user_api_key_team_id": "team-9",
|
||||
"user_api_key_team_alias": "team nine",
|
||||
"spend_logs_metadata": {"ticket": "LIT-8283"},
|
||||
"requester_metadata": {"headers": {"x-tenant": "acme"}},
|
||||
"user_api_key_user_id": "user-1",
|
||||
"user_api_key_alias": "prod-key",
|
||||
"user_api_key_end_user_id": "end-1",
|
||||
}
|
||||
assert attrs["langfuse.trace.metadata.user_api_key_alias"] == "prod-key"
|
||||
assert attrs["langfuse.trace.metadata.user_api_key_user_id"] == "user-1"
|
||||
|
|
@ -815,69 +823,69 @@ def test_langfuse_mapper_emits_request_metadata_and_identity():
|
|||
assert attrs["langfuse.observation.metadata.provider"] == "openai"
|
||||
|
||||
|
||||
def test_langfuse_mapper_omits_prompt_mcp_and_vector_store_content():
|
||||
payload: Final[dict[str, object]] = {
|
||||
"call_type": "acompletion",
|
||||
"custom_llm_provider": "openai",
|
||||
"model": "gpt-4o",
|
||||
"metadata": {
|
||||
"prompt_management_metadata": {
|
||||
"prompt_id": "prompt-1",
|
||||
"prompt_variables": {"input": "prompt-variables-secret"},
|
||||
"prompt_integration": "prompt-service",
|
||||
},
|
||||
"mcp_tool_call_metadata": {
|
||||
"name": "lookup_weather",
|
||||
"mcp_server_name": "weather-server",
|
||||
"arguments": {"location": "mcp-arguments-secret"},
|
||||
"result": {"forecast": "mcp-result-secret"},
|
||||
},
|
||||
"vector_store_request_metadata": [
|
||||
{
|
||||
"vector_store_id": "vs-1",
|
||||
"custom_llm_provider": "openai",
|
||||
"query": "vector-query-secret",
|
||||
"vector_store_search_response": {"text": "vector-response-secret"},
|
||||
"start_time": 1.0,
|
||||
"end_time": 2.0,
|
||||
}
|
||||
],
|
||||
},
|
||||
"response": {},
|
||||
}
|
||||
attrs: Final = LangfuseMapper().map(LLMCallSpanData.from_standard_logging_payload(payload))
|
||||
observation_metadata: Final = attrs["langfuse.observation.metadata"]
|
||||
serialized: Final = json.dumps(observation_metadata)
|
||||
parsed: Final = json.loads(observation_metadata)
|
||||
def test_langfuse_mapper_exports_only_the_promoted_metadata_and_the_team():
|
||||
attrs: Final = _langfuse_attrs(
|
||||
{
|
||||
"user_api_key_alias": "prod-key",
|
||||
"user_api_key_team_id": "team-9",
|
||||
"requester_ip_address": "10.0.0.1",
|
||||
"user_api_key_user_email": "caller-email-secret",
|
||||
"user_api_key_auth_metadata": {"note": "auth-metadata-secret"},
|
||||
"user_agent": "user-agent-secret",
|
||||
"headers": {"x-tenant-secret": "header-secret"},
|
||||
"requester_custom_headers": {"x-tenant-secret": "custom-header-secret"},
|
||||
"requester_metadata": {"note": "requester-metadata-secret"},
|
||||
"spend_logs_metadata": {"note": "spend-logs-secret"},
|
||||
"prompt_management_metadata": {"prompt_variables": {"input": "prompt-variables-secret"}},
|
||||
"mcp_tool_call_metadata": {"arguments": {"location": "mcp-arguments-secret"}},
|
||||
"vector_store_request_metadata": [{"query": "vector-query-secret"}],
|
||||
"routing_decision": {"matched_keyword": "routing-prompt-secret"},
|
||||
}
|
||||
)
|
||||
|
||||
assert "prompt-variables-secret" not in serialized
|
||||
assert "mcp-arguments-secret" not in serialized
|
||||
assert "mcp-result-secret" not in serialized
|
||||
assert "vector-query-secret" not in serialized
|
||||
assert "vector-response-secret" not in serialized
|
||||
assert parsed["prompt_management_metadata"]["prompt_id"] == "prompt-1"
|
||||
assert parsed["prompt_management_metadata"]["prompt_integration"] == "prompt-service"
|
||||
assert parsed["mcp_tool_call_metadata"]["name"] == "lookup_weather"
|
||||
assert parsed["mcp_tool_call_metadata"]["mcp_server_name"] == "weather-server"
|
||||
assert parsed["vector_store_request_metadata"][0]["vector_store_id"] == "vs-1"
|
||||
assert parsed["vector_store_request_metadata"][0]["custom_llm_provider"] == "openai"
|
||||
assert parsed["vector_store_request_metadata"][0]["start_time"] == 1.0
|
||||
assert parsed["vector_store_request_metadata"][0]["end_time"] == 2.0
|
||||
assert "secret" not in json.dumps(attrs)
|
||||
assert json.loads(attrs["langfuse.observation.metadata"]) == {
|
||||
"user_api_key_team_id": "team-9",
|
||||
"user_api_key_alias": "prod-key",
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("metadata_keys", "expected"),
|
||||
[
|
||||
((), {"user_api_key_team_id": "team-9"}),
|
||||
(("user_api_key_user_id",), {"user_api_key_team_id": "team-9", "user_api_key_user_id": "user-1"}),
|
||||
(("requester_metadata.trace_id", "requester_ip_address"), {"user_api_key_team_id": "team-9"}),
|
||||
],
|
||||
)
|
||||
def test_langfuse_mapper_follows_the_operator_metadata_allowlist(
|
||||
metadata_keys: tuple[str, ...], expected: Mapping[str, str]
|
||||
) -> None:
|
||||
attrs: Final = _langfuse_attrs(
|
||||
{
|
||||
"user_api_key_alias": "prod-key",
|
||||
"user_api_key_user_id": "user-1",
|
||||
"user_api_key_team_id": "team-9",
|
||||
"requester_ip_address": "10.0.0.1",
|
||||
"requester_metadata": {"trace_id": "trace-7", "note": "requester-metadata-secret"},
|
||||
},
|
||||
metadata_keys=metadata_keys,
|
||||
)
|
||||
|
||||
assert json.loads(attrs["langfuse.observation.metadata"]) == expected
|
||||
assert {key for key in attrs if key.startswith("langfuse.trace.metadata.user_api_key_")} == {
|
||||
f"langfuse.trace.metadata.{key}" for key in expected if key.startswith("user_api_key_")
|
||||
}
|
||||
|
||||
|
||||
def test_langfuse_mapper_skips_empty_identity_values():
|
||||
payload: Final[dict[str, object]] = {
|
||||
"call_type": "acompletion",
|
||||
"custom_llm_provider": "openai",
|
||||
"model": "gpt-4o",
|
||||
"metadata": {
|
||||
attrs: Final = _langfuse_attrs(
|
||||
{
|
||||
"user_api_key_alias": None,
|
||||
"user_api_key_user_id": "",
|
||||
"user_api_key_end_user_id": "end-1",
|
||||
},
|
||||
"response": {},
|
||||
}
|
||||
attrs: Final = LangfuseMapper().map(LLMCallSpanData.from_standard_logging_payload(payload))
|
||||
}
|
||||
)
|
||||
|
||||
assert "langfuse.trace.metadata.user_api_key_alias" not in attrs
|
||||
assert "langfuse.trace.metadata.user_api_key_user_id" not in attrs
|
||||
|
|
@ -888,37 +896,23 @@ def test_langfuse_mapper_redacts_user_api_key_fields(monkeypatch: pytest.MonkeyP
|
|||
import litellm
|
||||
|
||||
monkeypatch.setattr(litellm, "redact_user_api_key_info", True)
|
||||
payload: Final[dict[str, object]] = {
|
||||
"call_type": "acompletion",
|
||||
"custom_llm_provider": "openai",
|
||||
"model": "gpt-4o",
|
||||
"metadata": {
|
||||
attrs: Final = _langfuse_attrs(
|
||||
{
|
||||
"user_api_key_alias": "prod-key",
|
||||
"user_api_key_team_id": "team-9",
|
||||
"spend_logs_metadata": {"ticket": "LIT-8283"},
|
||||
},
|
||||
"response": {},
|
||||
}
|
||||
attrs: Final = LangfuseMapper().map(LLMCallSpanData.from_standard_logging_payload(payload))
|
||||
"requester_ip_address": "10.0.0.1",
|
||||
}
|
||||
)
|
||||
|
||||
exported: Final = json.loads(attrs["langfuse.observation.metadata"])
|
||||
assert not [key for key in exported if key.startswith("user_api_key")]
|
||||
assert exported["spend_logs_metadata"] == {"ticket": "LIT-8283"}
|
||||
assert "langfuse.observation.metadata" not in attrs
|
||||
assert not [key for key in attrs if key.startswith("langfuse.trace.metadata.user_api_key_")]
|
||||
assert attrs["langfuse.trace.metadata.team_id"] == "team-9"
|
||||
|
||||
|
||||
def test_langfuse_mapper_omits_observation_metadata_without_payload_metadata():
|
||||
payload: Final[dict[str, object]] = {
|
||||
"call_type": "acompletion",
|
||||
"custom_llm_provider": "openai",
|
||||
"model": "gpt-4o",
|
||||
"metadata": {},
|
||||
"response": {},
|
||||
}
|
||||
attrs: Final = LangfuseMapper().map(LLMCallSpanData.from_standard_logging_payload(payload))
|
||||
attrs: Final = _langfuse_attrs({})
|
||||
|
||||
assert json.loads(attrs["langfuse.observation.metadata"]) == {"requester_metadata": {}}
|
||||
assert "langfuse.observation.metadata" not in attrs
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
|
|
@ -933,43 +927,6 @@ def test_langfuse_mapper_omits_observation_metadata_without_payload_metadata():
|
|||
],
|
||||
)
|
||||
def test_span_data_builds_from_request_metadata_of_unexpected_shape(unexpected_shape: dict[str, object]) -> None:
|
||||
payload: Final[dict[str, object]] = {
|
||||
"call_type": "acompletion",
|
||||
"custom_llm_provider": "openai",
|
||||
"model": "gpt-4o",
|
||||
"metadata": {**unexpected_shape, "spend_logs_metadata": {"ticket": "LIT-8283"}},
|
||||
"response": {},
|
||||
}
|
||||
attrs: Final = LangfuseMapper().map(LLMCallSpanData.from_standard_logging_payload(payload))
|
||||
attrs: Final = _langfuse_attrs({**unexpected_shape, "user_api_key_alias": "prod-key"})
|
||||
|
||||
assert json.loads(attrs["langfuse.observation.metadata"])["spend_logs_metadata"] == {"ticket": "LIT-8283"}
|
||||
|
||||
|
||||
def test_langfuse_mapper_omits_prompt_quoting_routing_fields():
|
||||
payload: Final[dict[str, object]] = {
|
||||
"call_type": "acompletion",
|
||||
"custom_llm_provider": "openai",
|
||||
"model": "gpt-4o",
|
||||
"metadata": {
|
||||
"routing_decision": {
|
||||
"router_model_name": "smart-router",
|
||||
"router_type": "complexity",
|
||||
"cause": "literal_keyword_match",
|
||||
"tier": "REASONING",
|
||||
"signals": ["keyword (routing-prompt-secret)"],
|
||||
"matched_keyword": "routing-prompt-secret",
|
||||
"escalation_keyword": "routing-prompt-secret",
|
||||
"classifier_crux": "routing-prompt-secret",
|
||||
}
|
||||
},
|
||||
"response": {},
|
||||
}
|
||||
attrs: Final = LangfuseMapper().map(LLMCallSpanData.from_standard_logging_payload(payload))
|
||||
|
||||
assert "routing-prompt-secret" not in json.dumps(attrs)
|
||||
assert json.loads(attrs["langfuse.observation.metadata"])["routing_decision"] == {
|
||||
"router_model_name": "smart-router",
|
||||
"router_type": "complexity",
|
||||
"cause": "literal_keyword_match",
|
||||
"tier": "REASONING",
|
||||
}
|
||||
assert json.loads(attrs["langfuse.observation.metadata"]) == {"user_api_key_alias": "prod-key"}
|
||||
|
|
|
|||
|
|
@ -496,6 +496,7 @@ class TestLangfuseOtelIntegration:
|
|||
"user_api_key_end_user_id": "end-user-1",
|
||||
"user_api_key_team_id": "team-1",
|
||||
"user_api_key_team_alias": "team-a",
|
||||
"user_api_key_org_id": "org-1",
|
||||
"spend_logs_metadata": {"env": "prod"},
|
||||
"requester_ip_address": "10.0.0.1",
|
||||
"requester_metadata": {"ticket": "LIT-8283"},
|
||||
|
|
@ -510,7 +511,14 @@ class TestLangfuseOtelIntegration:
|
|||
|
||||
actual: Final = {call.args[1]: call.args[2] for call in mock_safe_set_attribute.call_args_list}
|
||||
|
||||
assert json.loads(actual["langfuse.observation.metadata"]) == {**request_metadata}
|
||||
assert json.loads(actual["langfuse.observation.metadata"]) == {
|
||||
"user_api_key_alias": "prod-key",
|
||||
"user_api_key_user_id": "user-1",
|
||||
"user_api_key_end_user_id": "end-user-1",
|
||||
"user_api_key_team_id": "team-1",
|
||||
"user_api_key_team_alias": "team-a",
|
||||
"user_api_key_org_id": "org-1",
|
||||
}
|
||||
assert {key: value for key, value in actual.items() if key.startswith("langfuse.trace.metadata.")} == {
|
||||
"langfuse.trace.metadata.user_api_key_alias": "prod-key",
|
||||
"langfuse.trace.metadata.user_api_key_user_id": "user-1",
|
||||
|
|
@ -542,36 +550,24 @@ class TestLangfuseOtelIntegration:
|
|||
|
||||
actual: Final = {call.args[1]: call.args[2] for call in mock_safe_set_attribute.call_args_list}
|
||||
|
||||
assert json.loads(actual["langfuse.observation.metadata"]) == {
|
||||
"spend_logs_metadata": {"env": "prod"},
|
||||
"requester_ip_address": "10.0.0.1",
|
||||
"requester_metadata": {"ticket": "LIT-8283"},
|
||||
}
|
||||
assert "langfuse.observation.metadata" not in actual
|
||||
assert not [key for key in actual if key.startswith("langfuse.trace.metadata.")]
|
||||
|
||||
def test_request_metadata_omits_prompt_mcp_and_vector_store_content(self) -> None:
|
||||
def test_request_metadata_beyond_the_identity_fields_stays_off_the_langfuse_metadata_keys(self) -> None:
|
||||
request_metadata: Final = {
|
||||
"prompt_management_metadata": {
|
||||
"prompt_id": "prompt-1",
|
||||
"prompt_variables": {"input": "prompt-variables-secret"},
|
||||
"prompt_integration": "prompt-service",
|
||||
},
|
||||
"mcp_tool_call_metadata": {
|
||||
"name": "lookup_weather",
|
||||
"mcp_server_name": "weather-server",
|
||||
"arguments": {"location": "mcp-arguments-secret"},
|
||||
"result": {"forecast": "mcp-result-secret"},
|
||||
},
|
||||
"vector_store_request_metadata": [
|
||||
{
|
||||
"vector_store_id": "vs-1",
|
||||
"custom_llm_provider": "openai",
|
||||
"query": "vector-query-secret",
|
||||
"vector_store_search_response": {"text": "vector-response-secret"},
|
||||
"start_time": 1.0,
|
||||
"end_time": 2.0,
|
||||
}
|
||||
],
|
||||
"user_api_key_alias": "prod-key",
|
||||
"requester_ip_address": "10.0.0.1",
|
||||
"user_api_key_user_email": "caller-email-secret",
|
||||
"user_api_key_auth_metadata": {"note": "auth-metadata-secret"},
|
||||
"user_agent": "user-agent-secret",
|
||||
"headers": {"x-tenant-secret": "header-secret"},
|
||||
"requester_custom_headers": {"x-tenant-secret": "custom-header-secret"},
|
||||
"requester_metadata": {"note": "requester-metadata-secret"},
|
||||
"spend_logs_metadata": {"note": "spend-logs-secret"},
|
||||
"prompt_management_metadata": {"prompt_variables": {"input": "prompt-variables-secret"}},
|
||||
"mcp_tool_call_metadata": {"arguments": {"location": "mcp-arguments-secret"}},
|
||||
"vector_store_request_metadata": [{"query": "vector-query-secret"}],
|
||||
"routing_decision": {"matched_keyword": "routing-prompt-secret"},
|
||||
}
|
||||
kwargs: Final[dict[str, object]] = {
|
||||
"litellm_params": {"metadata": {}},
|
||||
|
|
@ -582,25 +578,18 @@ class TestLangfuseOtelIntegration:
|
|||
LangfuseOtelLogger._set_langfuse_specific_attributes(MagicMock(), kwargs, None)
|
||||
|
||||
actual: Final = {call.args[1]: call.args[2] for call in mock_safe_set_attribute.call_args_list}
|
||||
observation_metadata: Final = actual["langfuse.observation.metadata"]
|
||||
serialized: Final = json.dumps(observation_metadata)
|
||||
parsed: Final = json.loads(observation_metadata)
|
||||
langfuse_metadata: Final = {
|
||||
key: value
|
||||
for key, value in actual.items()
|
||||
if key == "langfuse.observation.metadata" or key.startswith("langfuse.trace.metadata.")
|
||||
}
|
||||
|
||||
assert "prompt-variables-secret" not in serialized
|
||||
assert "mcp-arguments-secret" not in serialized
|
||||
assert "mcp-result-secret" not in serialized
|
||||
assert "vector-query-secret" not in serialized
|
||||
assert "vector-response-secret" not in serialized
|
||||
assert parsed["prompt_management_metadata"]["prompt_id"] == "prompt-1"
|
||||
assert parsed["prompt_management_metadata"]["prompt_integration"] == "prompt-service"
|
||||
assert parsed["mcp_tool_call_metadata"]["name"] == "lookup_weather"
|
||||
assert parsed["mcp_tool_call_metadata"]["mcp_server_name"] == "weather-server"
|
||||
assert parsed["vector_store_request_metadata"][0]["vector_store_id"] == "vs-1"
|
||||
assert parsed["vector_store_request_metadata"][0]["custom_llm_provider"] == "openai"
|
||||
assert parsed["vector_store_request_metadata"][0]["start_time"] == 1.0
|
||||
assert parsed["vector_store_request_metadata"][0]["end_time"] == 2.0
|
||||
assert langfuse_metadata == {
|
||||
"langfuse.observation.metadata": json.dumps({"user_api_key_alias": "prod-key"}),
|
||||
"langfuse.trace.metadata.user_api_key_alias": "prod-key",
|
||||
}
|
||||
|
||||
def test_request_metadata_drops_null_fields_and_skips_empty_trace_identities(self) -> None:
|
||||
def test_request_metadata_drops_empty_fields_and_skips_empty_trace_identities(self) -> None:
|
||||
request_metadata: Final = {
|
||||
"user_api_key_alias": "prod-key",
|
||||
"user_api_key_user_id": "user-1",
|
||||
|
|
@ -624,15 +613,28 @@ class TestLangfuseOtelIntegration:
|
|||
assert json.loads(actual["langfuse.observation.metadata"]) == {
|
||||
"user_api_key_alias": "prod-key",
|
||||
"user_api_key_user_id": "user-1",
|
||||
"user_api_key_team_id": "",
|
||||
"spend_logs_metadata": {"env": "prod"},
|
||||
"requester_metadata": {},
|
||||
}
|
||||
assert {key: value for key, value in actual.items() if key.startswith("langfuse.trace.metadata.")} == {
|
||||
"langfuse.trace.metadata.user_api_key_alias": "prod-key",
|
||||
"langfuse.trace.metadata.user_api_key_user_id": "user-1",
|
||||
}
|
||||
|
||||
def test_request_metadata_keys_are_absent_without_identity_fields(self) -> None:
|
||||
kwargs: Final[dict[str, object]] = {
|
||||
"litellm_params": {"metadata": {}},
|
||||
"standard_logging_object": {
|
||||
"metadata": {"spend_logs_metadata": {"env": "prod"}, "requester_ip_address": "10.0.0.1"}
|
||||
},
|
||||
}
|
||||
|
||||
with patch("litellm.integrations.arize._utils.safe_set_attribute") as mock_safe_set_attribute:
|
||||
LangfuseOtelLogger._set_langfuse_specific_attributes(MagicMock(), kwargs, None)
|
||||
|
||||
actual: Final = {call.args[1]: call.args[2] for call in mock_safe_set_attribute.call_args_list}
|
||||
|
||||
assert "langfuse.observation.metadata" not in actual
|
||||
assert not [key for key in actual if key.startswith("langfuse.trace.metadata.")]
|
||||
|
||||
def test_request_metadata_keys_are_absent_without_standard_logging_object(self) -> None:
|
||||
kwargs: Final = {"litellm_params": {"metadata": {"trace_metadata": {"k": "v"}}}}
|
||||
|
||||
|
|
@ -653,14 +655,17 @@ class TestLangfuseOtelIntegration:
|
|||
{"requester_metadata": 1},
|
||||
{"requester_metadata": True},
|
||||
{"requester_metadata": 1.5},
|
||||
{1: "one"},
|
||||
{"user_api_key_user_id": 5},
|
||||
{"user_api_key_team_id": {"id": "team-1"}},
|
||||
],
|
||||
)
|
||||
def test_request_metadata_of_unexpected_shape_keeps_the_observation_output(
|
||||
self, unexpected_shape: dict[str, object]
|
||||
self, unexpected_shape: dict[object, object]
|
||||
) -> None:
|
||||
kwargs: Final[dict[str, object]] = {
|
||||
"litellm_params": {"metadata": {}},
|
||||
"standard_logging_object": {"metadata": {**unexpected_shape, "spend_logs_metadata": {"env": "prod"}}},
|
||||
"standard_logging_object": {"metadata": {**unexpected_shape, "user_api_key_alias": "prod-key"}},
|
||||
}
|
||||
response_obj: Final = {"choices": [{"message": {"role": "assistant", "content": "hello"}}]}
|
||||
|
||||
|
|
@ -669,36 +674,9 @@ class TestLangfuseOtelIntegration:
|
|||
|
||||
actual: Final = {call.args[1]: call.args[2] for call in mock_safe_set_attribute.call_args_list}
|
||||
|
||||
assert json.loads(actual["langfuse.observation.metadata"])["spend_logs_metadata"] == {"env": "prod"}
|
||||
assert json.loads(actual["langfuse.observation.metadata"]) == {"user_api_key_alias": "prod-key"}
|
||||
assert json.loads(actual["langfuse.observation.output"]) == {"role": "assistant", "content": "hello"}
|
||||
|
||||
def test_observation_metadata_omits_prompt_quoting_routing_fields(self) -> None:
|
||||
kwargs: Final[dict[str, object]] = {
|
||||
"litellm_params": {"metadata": {}},
|
||||
"standard_logging_object": {
|
||||
"metadata": {
|
||||
"routing_decision": {
|
||||
"router_type": "complexity",
|
||||
"tier": "REASONING",
|
||||
"signals": ["keyword (routing-prompt-secret)"],
|
||||
"matched_keyword": "routing-prompt-secret",
|
||||
"escalation_keyword": "routing-prompt-secret",
|
||||
"classifier_crux": "routing-prompt-secret",
|
||||
}
|
||||
}
|
||||
},
|
||||
}
|
||||
|
||||
with patch("litellm.integrations.arize._utils.safe_set_attribute") as mock_safe_set_attribute:
|
||||
LangfuseOtelLogger._set_langfuse_specific_attributes(MagicMock(), kwargs, None)
|
||||
|
||||
actual: Final = {call.args[1]: call.args[2] for call in mock_safe_set_attribute.call_args_list}
|
||||
|
||||
assert json.loads(actual["langfuse.observation.metadata"])["routing_decision"] == {
|
||||
"router_type": "complexity",
|
||||
"tier": "REASONING",
|
||||
}
|
||||
|
||||
|
||||
class TestLangfuseOtelKeyDynamicConfig:
|
||||
"""Key/team-scoped Langfuse credentials must define the full export target
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue