scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-06 02:48:13 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 103

fix(otel): emit OpenInference tool calls and metadata on Arize OTel v2 spans

Browse source 
Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
yucheng 2026-09-29 09:59:59 +00:00
parent 7f95b5f361
commit f13bf491c8
9 changed files with 652 additions and 34 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
litellm/integrations/otel/logger.py
View file

@ -563,6 +563,7 @@ class OpenTelemetryV2(CustomLogger):
request_route=request_root_http_route(),
trace=call.trace,
session_id=call.session_id,
metadata_keys=tuple(self.config.baggage_metadata_keys),
)
end_time_ns: Final = to_ns(end_time)
if carrier is not None and carrier.span is not None:

88
litellm/integrations/otel/mappers/openinference.py
View file

@ -7,7 +7,7 @@ Phoenix + any other OpenInference-aware backend simultaneously.
"""
import json
from collections.abc import Callable, Mapping, Sequence
from collections.abc import Callable, Iterator, Mapping, Sequence
from itertools import accumulate, chain, groupby
from types import MappingProxyType
from typing import Final
@ -15,10 +15,12 @@ from typing import Final
from litellm.integrations.otel.mappers.base import AttributeMap, AttrValue, SpanData
from litellm.integrations.otel.mappers.utils import (
MAX_TOOL_DEFINITION_ATTRS_PER_SPAN,
MessageToolCall,
collect,
drop_none,
drop_none_pairs,
json_if,
message_content,
message_tool_calls,
output_messages,
tool_definition_attrs,
)
@ -32,6 +34,48 @@ _INPUT_MESSAGES: Final = "llm.input_messages"
_OUTPUT_MESSAGES: Final = "llm.output_messages"
_MESSAGE_FAMILIES: Final = (_INPUT_MESSAGES, _OUTPUT_MESSAGES)
_ParsedMessage = tuple[object, str | None, tuple[MessageToolCall, ...]]
def _parse_message(message: object) -> _ParsedMessage:
role: Final = message.get("role") if isinstance(message, dict) else None
return role, message_content(message), message_tool_calls(message)
def _tool_call_attribute_pairs(
prefix: str, idx: int, tool_calls: tuple[MessageToolCall, ...]
) -> Iterator[tuple[str, str | None]]:
for tool_idx, tool_call in enumerate(tool_calls):
yield f"{prefix}.{idx}.message.tool_calls.{tool_idx}.tool_call.id", tool_call.id
yield f"{prefix}.{idx}.message.tool_calls.{tool_idx}.tool_call.function.name", tool_call.name
yield f"{prefix}.{idx}.message.tool_calls.{tool_idx}.tool_call.function.arguments", tool_call.arguments
def _message_attribute_pairs(
prefix: str,
messages: Sequence[_ParsedMessage],
*,
with_tool_call_attrs: bool,
) -> Iterator[tuple[str, str | None]]:
for idx, (role, content, tool_calls) in enumerate(messages):
yield f"{prefix}.{idx}.message.role", role if isinstance(role, str) else None
yield f"{prefix}.{idx}.message.content", content
if with_tool_call_attrs:
yield from _tool_call_attribute_pairs(prefix, idx, tool_calls)
def _message_value(messages: Sequence[_ParsedMessage]) -> str:
return json.dumps(
[
{
"role": role,
"content": content,
**({"tool_calls": [tool_call.to_openai_dict() for tool_call in tool_calls]} if tool_calls else {}),
}
for role, content, tool_calls in messages
]
)
def _message_key_groups(attrs: Mapping[str, AttrValue]) -> Mapping[tuple[str, int], tuple[str, ...]]:
"""Per-index message keys in ``attrs`` grouped by ``(family, index)``."""
@ -85,7 +129,9 @@ class OpenInferenceMapper:
- ``llm.model_name`` / ``llm.provider`` / ``llm.invocation_parameters``
- ``llm.input_messages.{i}.message.role`` / ``...content``
- ``llm.output_messages.{i}.message.role`` / ``...content``
- ``llm.output_messages.{i}.message.tool_calls.{j}.tool_call.*``
- ``llm.token_count.prompt`` / ``...completion`` / ``...total``
- ``metadata`` — JSON object of allowlisted promoted request metadata
- ``input.value`` / ``output.value`` — JSON-serialized request / response
"""
@ -121,6 +167,7 @@ class OpenInferenceMapper:
"llm.invocation_parameters": lambda d: json_if(
collect(OpenInferenceMapper._INVOCATION_PARAMS, d.request_params)
),
"metadata": lambda d: json_if(dict(sorted(d.promoted_metadata.items()))),
}
def __init__(self, tool_attr_budget: int = MAX_TOOL_DEFINITION_ATTRS_PER_SPAN) -> None:
@ -137,27 +184,36 @@ class OpenInferenceMapper:
return {
**collect(self._LLM_CALL_ATTRS, data),
**collect(self._BLOB_ATTRS, data),
**self._messages(_INPUT_MESSAGES, "input.value", data.messages_in),
**self._messages(_OUTPUT_MESSAGES, "output.value", output_messages(data)),
**self._messages(
_INPUT_MESSAGES,
"input.value",
data.messages_in,
with_tool_call_attrs=False,
),
**self._messages(
_OUTPUT_MESSAGES,
"output.value",
output_messages(data),
with_tool_call_attrs=True,
),
**self._tools(data),
}
@staticmethod
def _messages(prefix: str, value_key: str, messages: Sequence[object]) -> AttributeMap:
def _messages(
prefix: str,
value_key: str,
messages: Sequence[object],
*,
with_tool_call_attrs: bool,
) -> AttributeMap:
"""``{prefix}.{idx}.message.*`` keys for every message + the ``value_key`` blob of all of them."""
parsed: Final = [(m.get("role") if isinstance(m, dict) else None, message_content(m)) for m in messages]
attrs: Final = drop_none(
{
key: value
for idx, (role, content) in enumerate(parsed)
for key, value in (
(f"{prefix}.{idx}.message.role", role if isinstance(role, str) else None),
(f"{prefix}.{idx}.message.content", content),
)
}
parsed: Final = tuple(_parse_message(message) for message in messages)
attrs: Final = drop_none_pairs(
_message_attribute_pairs(prefix, parsed, with_tool_call_attrs=with_tool_call_attrs)
)
if parsed:
attrs[value_key] = json.dumps([{"role": role, "content": content} for role, content in parsed])
attrs[value_key] = _message_value(parsed)
return attrs
def _tools(self, data: LLMCallSpanData) -> AttributeMap:

50
litellm/integrations/otel/mappers/utils.py
View file

@ -7,11 +7,28 @@ they live in one place.
import json
from collections.abc import Callable, Iterable, Mapping, Sequence
from dataclasses import dataclass
from typing import Final
from litellm.integrations.otel.mappers.base import AttributeMap, AttrValue
from litellm.integrations.otel.model.payloads import LLMCallSpanData, ToolDefinition
@dataclass(frozen=True, slots=True)
class MessageToolCall:
id: str | None
type: str
name: str | None
arguments: str | None
def to_openai_dict(self) -> dict[str, object]:
return {
"id": self.id,
"type": self.type,
"function": {"name": self.name, "arguments": self.arguments},
}
DEFAULT_SPAN_ATTRIBUTE_LIMIT: Final = 128
"""The OTel SDK's default per-span attribute count limit."""
@ -121,3 +138,36 @@ def message_content(message: object) -> str | None:
def output_messages(data: LLMCallSpanData) -> list:
"""The ``message`` payload of each response choice."""
return [c.get("message") for c in data.choices_out if isinstance(c, dict)]
def message_tool_calls(message: object) -> tuple[MessageToolCall, ...]:
if not isinstance(message, dict):
return ()
tool_calls: Final = message.get("tool_calls")
if not isinstance(tool_calls, (list, tuple)) or not tool_calls:
return ()
return tuple(tool_call for value in tool_calls if (tool_call := _message_tool_call(value)) is not None)
def _message_tool_call(value: object) -> MessageToolCall | None:
if not isinstance(value, dict):
return None
function: Final = value.get("function")
function_data: Final = function if isinstance(function, dict) else {}
raw_type: Final = value.get("type")
raw_arguments: Final = function_data.get("arguments")
arguments: Final = (
raw_arguments
if isinstance(raw_arguments, str)
else json.dumps(raw_arguments, default=str)
if raw_arguments is not None
else None
)
name: Final = function_data.get("name")
identifier: Final = value.get("id")
return MessageToolCall(
id=identifier if isinstance(identifier, str) else None,
type=raw_type if isinstance(raw_type, str) else "function",
name=name if isinstance(name, str) else None,
arguments=arguments,
)

7
litellm/integrations/otel/model/baggage.py
View file

@ -18,7 +18,7 @@ from collections.abc import Callable, Mapping
from types import MappingProxyType
from typing import Final
from litellm.integrations.otel.model.metadata import REQUESTER_METADATA_PATH, RequestIdentity
from litellm.integrations.otel.model.metadata import RequestIdentity, allowlisted_metadata
from litellm.integrations.otel.model.semconv import GenAI, LiteLLM
# Attribute key -> value extractor over (identity, request_model,
@ -92,9 +92,8 @@ def promoted_metadata(metadata: Mapping[str, str], metadata_keys: tuple[str, ...
"""Allowlisted entries of a flattened metadata mapping under ``litellm.metadata.*``."""
return MappingProxyType(
{
f"{LiteLLM.METADATA_PREFIX}{meta_key.removeprefix(REQUESTER_METADATA_PATH)}": value
for meta_key in metadata_keys
if (value := metadata.get(meta_key))
f"{LiteLLM.METADATA_PREFIX}{meta_key}": value
for meta_key, value in allowlisted_metadata(metadata, metadata_keys).items()
}
)

10
litellm/integrations/otel/model/metadata.py
View file

@ -53,6 +53,16 @@ REQUESTER_METADATA_KEY: Final = "requester_metadata"
REQUESTER_METADATA_PATH: Final = f"{REQUESTER_METADATA_KEY}."
def allowlisted_metadata(metadata: Mapping[str, str], metadata_keys: tuple[str, ...]) -> Mapping[str, str]:
return MappingProxyType(
{
meta_key.removeprefix(REQUESTER_METADATA_PATH): value
for meta_key in metadata_keys
if (value := metadata.get(meta_key))
}
)
@dataclass(frozen=True)
class RequestIdentity:
call_id: str | None = None

8
litellm/integrations/otel/model/payloads.py
View file

@ -12,7 +12,7 @@ from urllib.parse import urlsplit
from typing_extensions import ReadOnly, TypedDict
from litellm.integrations.otel.model.metadata import RequestContext, RequestIdentity
from litellm.integrations.otel.model.metadata import RequestContext, RequestIdentity, allowlisted_metadata
from litellm.integrations.otel.model.semconv import (
GenAIOperation,
GenAIOutputType,
@ -60,6 +60,8 @@ if TYPE_CHECKING:
StandardLoggingPayload,
)
_EMPTY_METADATA: Final[Mapping[str, str]] = MappingProxyType({})
# --- typed sub-structures ---------------------------------------------------- #
@ -430,6 +432,7 @@ class LLMCallSpanData:
trace: TraceControls = field(default_factory=TraceControls)
session_id: str | None = None
embedding_output: EmbeddingOutput | None = None
promoted_metadata: Mapping[str, str] = field(default_factory=lambda: _EMPTY_METADATA)
@classmethod
def from_standard_logging_payload(
@ -440,6 +443,8 @@ class LLMCallSpanData:
request_route: str | None = None,
trace: TraceControls | None = None,
session_id: str | None = None,
*,
metadata_keys: tuple[str, ...] = (),
) -> LLMCallSpanData:
params: Final = cast(Mapping[str, object], payload.get("model_parameters") or {})
# The single parse of the request's metadata — the request-vs-provider
@ -476,6 +481,7 @@ 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,
promoted_metadata=allowlisted_metadata(context.identity.metadata, metadata_keys),
is_streaming=as_bool(payload.get("stream")),
tools=_extract_tools(params),
messages_in=_dicts(payload.get("messages")) if capture_content else (),

295
tests/integration/observability/test_arize_otel_v2_openinference_spans.py Normal file
View file

@ -0,0 +1,295 @@
import json
import uuid
from collections.abc import Callable, Iterator
from contextlib import contextmanager
from pathlib import Path
from typing import Final
import yaml
from integration._support.client import Gateway, eventually
from integration._support.process import owned_proxy
from integration._support.wire import Reply, Request, Wire, wire_server
from opentelemetry.proto.collector.trace.v1.trace_service_pb2 import ExportTraceServiceRequest
def _span_attributes(request: Request) -> Iterator[dict[str, str]]:
if request.headers.get("content-type") != "application/x-protobuf":
return
batch: Final = ExportTraceServiceRequest.FromString(request.body)
for resource_spans in batch.resource_spans:
for scope_spans in resource_spans.scope_spans:
for span in scope_spans.spans:
yield {attribute.key: attribute.value.string_value for attribute in span.attributes}
def _matching_llm_spans(
requests: tuple[Request, ...],
marker: str,
marker_key: str,
) -> Iterator[dict[str, str]]:
for request in requests:
for attributes in _span_attributes(request):
if attributes.get("openinference.span.kind") == "LLM" and attributes.get(marker_key) == marker:
yield attributes
def _arize_config(tmp_path: Path) -> Path:
config: Final = yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text())
config["litellm_settings"]["callbacks"] = ["arize"]
config_path: Final = tmp_path / "arize-otel-v2.yaml"
config_path.write_text(yaml.safe_dump(config))
return config_path
def _arize_environment(destination: Wire) -> dict[str, str]:
return {
"LITELLM_OTEL_V2": "1",
"ARIZE_HTTP_ENDPOINT": destination.url + "/v1/traces",
"ARIZE_SPACE_ID": "integration-space",
"ARIZE_API_KEY": "integration-arize-key",
"OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT": "span_only",
"LITELLM_OTEL_BAGGAGE_METADATA_KEYS": "requester_metadata.trace_marker",
}
def _collector(_request: Request) -> Reply:
return Reply(body=b"", content_type="application/x-protobuf")
@contextmanager
def _arize_proxy(
gateway: Gateway,
tmp_path: Path,
upstream: Callable[[Request], Reply],
) -> Iterator[tuple[Gateway, Wire, Wire]]:
with wire_server(upstream) as provider, wire_server(_collector) as destination:
config_path: Final = _arize_config(tmp_path)
environment: Final = _arize_environment(destination)
with owned_proxy(gateway, tmp_path, environment, config=config_path) as candidate:
yield candidate, provider, destination
def _matching_span(
destination: Wire,
marker: str,
marker_key: str = "gen_ai.response.id",
) -> dict[str, str]:
batches: Final[list[Request]] = []
def matching_spans() -> tuple[dict[str, str], ...]:
batches.extend(destination.drain())
return tuple(_matching_llm_spans(tuple(batches), marker, marker_key))
return eventually(matching_spans, lambda spans: len(spans) == 1, seconds=30)[0]
def _assert_tool_call_span(span_attributes: dict[str, str], marker: str) -> None:
tool_call: Final = "llm.output_messages.0.message.tool_calls.0.tool_call."
keys: Final = (
tool_call + "id",
tool_call + "function.name",
tool_call + "function.arguments",
"metadata",
"litellm.metadata.trace_marker",
)
observed: Final = {
key: json.loads(span_attributes[key])
if key in (tool_call + "function.arguments", "metadata") and key in span_attributes
else span_attributes.get(key)
for key in keys
}
expected: Final = {
tool_call + "id": "call_" + marker,
tool_call + "function.name": "lookup_weather",
tool_call + "function.arguments": {"city": "Paris"},
"metadata": {"trace_marker": marker},
"litellm.metadata.trace_marker": marker,
}
expected_output: Final = [
{
"role": "assistant",
"content": None,
"tool_calls": [
{
"id": "call_" + marker,
"type": "function",
"function": {
"name": "lookup_weather",
"arguments": '{"city": "Paris"}',
},
}
],
}
]
actual: Final = {"attributes": observed, "output.value": json.loads(span_attributes["output.value"])}
expected_values: Final = {"attributes": expected, "output.value": expected_output}
assert actual == expected_values, f"Arize OTel v2 span values for {marker}: {actual!r}"
def test_arize_otel_v2_llm_span_carries_openinference_tool_calls_and_metadata(gateway: Gateway, tmp_path: Path) -> None:
marker: Final = "arize-otel-" + uuid.uuid4().hex
def upstream(request: Request) -> Reply:
assert request.target.endswith("/chat/completions"), request.target
body: Final = json.loads(request.body)
expected_tools: Final = [
{
"type": "function",
"function": {
"name": "lookup_weather",
"description": "Get weather",
"parameters": {
"type": "object",
"properties": {"city": {"type": "string"}},
},
},
}
]
assert body.get("messages") == [{"role": "user", "content": "weather in Paris?"}], (
f"Unexpected chat request messages: {body!r}"
)
assert body.get("tools") == expected_tools, f"Unexpected chat request tools: {body!r}"
assert body.get("tool_choice") == {
"type": "function",
"function": {"name": "lookup_weather"},
}, f"Unexpected chat request tool_choice: {body!r}"
return Reply(
body=json.dumps(
{
"id": marker,
"object": "chat.completion",
"created": 1,
"model": "gpt-4o-mini",
"choices": [
{
"index": 0,
"finish_reason": "tool_calls",
"message": {
"role": "assistant",
"content": None,
"tool_calls": [
{
"id": "call_" + marker,
"type": "function",
"function": {
"name": "lookup_weather",
"arguments": '{"city": "Paris"}',
},
}
],
},
}
],
"usage": {"prompt_tokens": 11, "completion_tokens": 4, "total_tokens": 15},
}
).encode()
)
with _arize_proxy(gateway, tmp_path, upstream) as (candidate, provider, destination):
with candidate.scenario() as scenario:
model: Final = scenario.model(model="openai/gpt-4o-mini", api_base=provider.url + "/v1")
response: Final = candidate.request(
"POST",
"/v1/chat/completions",
{
"model": model,
"messages": [{"role": "user", "content": "weather in Paris?"}],
"tools": [
{
"type": "function",
"function": {
"name": "lookup_weather",
"description": "Get weather",
"parameters": {
"type": "object",
"properties": {"city": {"type": "string"}},
},
},
}
],
"tool_choice": {"type": "function", "function": {"name": "lookup_weather"}},
"metadata": {"trace_marker": marker},
"cache": {"no-cache": True},
},
)
assert response.status_code == 200, response.text
_assert_tool_call_span(_matching_span(destination, marker), marker)
def test_arize_otel_v2_responses_span_carries_openinference_tool_calls_and_metadata(
gateway: Gateway, tmp_path: Path
) -> None:
marker: Final = "arize-otel-responses-" + uuid.uuid4().hex
def upstream(request: Request) -> Reply:
assert request.target.endswith("/responses"), request.target
body: Final = json.loads(request.body)
expected_tools: Final = [
{
"type": "function",
"name": "lookup_weather",
"description": "Get weather",
"parameters": {
"type": "object",
"properties": {"city": {"type": "string"}},
},
}
]
assert body.get("input") == "weather in Paris?", f"Unexpected Responses request input: {body!r}"
assert body.get("tools") == expected_tools, f"Unexpected Responses request tools: {body!r}"
assert body.get("tool_choice") == {
"type": "function",
"name": "lookup_weather",
}, f"Unexpected Responses request tool_choice: {body!r}"
return Reply(
body=json.dumps(
{
"id": marker,
"object": "response",
"created_at": 1,
"status": "completed",
"model": "gpt-4o-mini",
"output": [
{
"type": "function_call",
"id": "fc_" + marker,
"call_id": "call_" + marker,
"name": "lookup_weather",
"arguments": '{"city": "Paris"}',
"status": "completed",
}
],
"usage": {"input_tokens": 11, "output_tokens": 4, "total_tokens": 15},
}
).encode()
)
with _arize_proxy(gateway, tmp_path, upstream) as (candidate, provider, destination):
with candidate.scenario() as scenario:
model: Final = scenario.model(model="openai/gpt-4o-mini", api_base=provider.url + "/v1")
response: Final = candidate.request(
"POST",
"/v1/responses",
{
"model": model,
"input": "weather in Paris?",
"tools": [
{
"type": "function",
"name": "lookup_weather",
"description": "Get weather",
"parameters": {
"type": "object",
"properties": {"city": {"type": "string"}},
},
}
],
"tool_choice": {"type": "function", "name": "lookup_weather"},
"metadata": {"trace_marker": marker},
"cache": {"no-cache": True},
},
)
assert response.status_code == 200, response.text
_assert_tool_call_span(
_matching_span(destination, marker, marker_key="litellm.metadata.trace_marker"), marker
)

31
tests/unit/integrations/otel/test_otel_v2_baggage.py
View file

@ -7,22 +7,22 @@ import pytest
pytest.importorskip("opentelemetry")
from litellm.integrations.otel import ( # noqa: E402
GenAI,
HTTP,
GenAI,
LiteLLM,
OpenTelemetryV2Config,
promoted_baggage,
)
from litellm.integrations.otel.plumbing import context as ctx_mod # noqa: E402
from litellm.integrations.otel.plumbing import providers # noqa: E402
from litellm.integrations.otel.emitter import SpanEmitter # noqa: E402
from litellm.integrations.otel.model.baggage import BAGGAGE_PROMOTED_KEYS # noqa: E402
from litellm.integrations.otel.model.payloads import ( # noqa: E402
GuardrailSpanData,
LLMCallSpanData,
ServiceSpanData,
)
from litellm.integrations.otel.model.baggage import BAGGAGE_PROMOTED_KEYS # noqa: E402
from litellm.integrations.otel.model.spans import SpanRole # noqa: E402
from litellm.integrations.otel.plumbing import context as ctx_mod # noqa: E402
from litellm.integrations.otel.plumbing import providers # noqa: E402
def _payload():
@ -63,9 +63,7 @@ def test_identity_promoted_onto_every_span():
root = engine.start_span(SpanRole.PROXY_REQUEST, "POST /chat/completions", ctx)
root_ctx = ctx_mod.context_from_span(root, ctx)
engine.emit(SpanRole.LLM_CALL, data, parent_context=root_ctx)
engine.emit(
SpanRole.GUARDRAIL, GuardrailSpanData("presidio", status="success"), root_ctx
)
engine.emit(SpanRole.GUARDRAIL, GuardrailSpanData("presidio", status="success"), root_ctx)
engine.emit(SpanRole.SERVICE, ServiceSpanData("redis", call_type="set"), root_ctx)
root.end()
@ -161,9 +159,7 @@ def test_allowlisted_metadata_subkey_promoted_blob_excluded():
engine.emit(SpanRole.SERVICE, ServiceSpanData("redis", call_type="set"), ctx)
(span,) = exporter.get_finished_spans()
# allowlisted metadata sub-key is promoted
assert (
span.attributes.get(f"{LiteLLM.METADATA_PREFIX}user_api_key_org_id") == "org1"
)
assert span.attributes.get(f"{LiteLLM.METADATA_PREFIX}user_api_key_org_id") == "org1"
# non-allowlisted metadata is NOT promoted (no full-blob dumping)
assert all("private_note" not in k for k in span.attributes)
@ -208,6 +204,17 @@ def test_nested_metadata_key_promoted_under_caller_path():
assert not any(k.startswith(f"{LiteLLM.METADATA_PREFIX}requester_metadata") for k in span.attributes)
def test_llm_call_promoted_metadata_strips_requester_prefix_and_uses_allowlist():
payload = _payload()
payload["metadata"]["requester_metadata"] = {"trace_id": "trace-123"}
data = LLMCallSpanData.from_standard_logging_payload(
payload,
metadata_keys=("requester_metadata.trace_id", "user_api_key_org_id", "missing"),
)
assert data.promoted_metadata == {"trace_id": "trace-123", "user_api_key_org_id": "org1"}
assert LLMCallSpanData.from_standard_logging_payload(payload).promoted_metadata == {}
def test_http_attributes_never_promoted():
"""Even if http.* is present in baggage, the processor must not stamp it on
child spans (it belongs on the SERVER span only)."""
@ -228,9 +235,7 @@ def test_http_attributes_never_promoted():
def test_arbitrary_upstream_baggage_not_promoted():
engine, exporter = _engine_and_exporter()
ctx = ctx_mod.set_request_baggage(
{LiteLLM.TEAM_ID: "t1", "some.upstream.key": "leak"}
)
ctx = ctx_mod.set_request_baggage({LiteLLM.TEAM_ID: "t1", "some.upstream.key": "leak"})
engine.emit(SpanRole.SERVICE, ServiceSpanData("redis", call_type="set"), ctx)
(span,) = exporter.get_finished_spans()
assert span.attributes.get(LiteLLM.TEAM_ID) == "t1"

196
tests/unit/integrations/otel/test_otel_v2_vendor_mappers.py
View file

@ -7,6 +7,7 @@ backends, so one trace lights up every configured destination.
import json
from collections.abc import Mapping
from itertools import chain
from typing import Final
import pytest
@ -19,6 +20,7 @@ from litellm.integrations.otel.mappers import (
WeaveMapper,
resolve_mappers,
)
from litellm.integrations.otel.mappers.openinference import fit_indexed_messages
from litellm.integrations.otel.model.payloads import (
EmbeddingOutput,
LLMCallSpanData,
@ -29,6 +31,7 @@ from litellm.integrations.otel.model.payloads import (
ToolDefinition,
)
from litellm.integrations.otel.model.trace_controls import TraceControls
from tests.unit.integrations.otel.test_otel_v2_sources_of_truth import _responses_payload
def _llm_call(**overrides):
@ -118,6 +121,199 @@ def test_openinference_multimodal_content_text_only():
assert attrs["llm.input_messages.0.message.content"] == "hi there"
def test_openinference_output_tool_calls_preserve_calls_in_attributes_and_value():
tool_calls: Final = [
{
"id": "call_paris",
"type": "function",
"function": {"name": "lookup_weather", "arguments": '{"city": "Paris"}'},
},
{
"id": "call_search",
"function": {"name": "search", "arguments": {"q": 1}},
"index": 0,
},
"ignored",
]
data: Final = _llm_call(
choices_out=(
{
"finish_reason": "tool_calls",
"message": {"role": "assistant", "content": None, "tool_calls": tool_calls},
},
)
)
attrs: Final = OpenInferenceMapper().map(data)
assert {key: value for key, value in attrs.items() if ".tool_calls." in key} == {
"llm.output_messages.0.message.tool_calls.0.tool_call.id": "call_paris",
"llm.output_messages.0.message.tool_calls.0.tool_call.function.name": "lookup_weather",
"llm.output_messages.0.message.tool_calls.0.tool_call.function.arguments": '{"city": "Paris"}',
"llm.output_messages.0.message.tool_calls.1.tool_call.id": "call_search",
"llm.output_messages.0.message.tool_calls.1.tool_call.function.name": "search",
"llm.output_messages.0.message.tool_calls.1.tool_call.function.arguments": '{"q": 1}',
}
assert json.loads(attrs["output.value"]) == [
{
"role": "assistant",
"content": None,
"tool_calls": [
{
"id": "call_paris",
"type": "function",
"function": {"name": "lookup_weather", "arguments": '{"city": "Paris"}'},
},
{
"id": "call_search",
"type": "function",
"function": {"name": "search", "arguments": '{"q": 1}'},
},
],
}
]
def test_openinference_responses_tool_calls_are_emitted_as_output_attributes():
data: Final = LLMCallSpanData.from_standard_logging_payload(
_responses_payload(
[
{
"type": "function_call",
"call_id": "call_resp",
"name": "lookup_weather",
"arguments": '{"city": "Paris"}',
}
]
),
capture_content=True,
)
attrs: Final = OpenInferenceMapper().map(data)
tool_call: Final = "llm.output_messages.0.message.tool_calls.0.tool_call."
assert {key: value for key, value in attrs.items() if ".tool_calls." in key} == {
tool_call + "id": "call_resp",
tool_call + "function.name": "lookup_weather",
tool_call + "function.arguments": '{"city": "Paris"}',
}
assert json.loads(attrs["output.value"]) == [
{
"role": "assistant",
"content": None,
"tool_calls": [
{
"id": "call_resp",
"type": "function",
"function": {"name": "lookup_weather", "arguments": '{"city": "Paris"}'},
}
],
}
]
def test_openinference_input_tool_calls_stay_in_value_only():
data: Final = _llm_call(
messages_in=(
{
"role": "assistant",
"content": None,
"tool_calls": [
{
"id": "call_weather",
"type": "function",
"function": {"name": "lookup_weather", "arguments": '{"city": "Paris"}'},
}
],
},
)
)
attrs: Final = OpenInferenceMapper().map(data)
assert all(".tool_calls." not in key for key in attrs)
assert json.loads(attrs["input.value"]) == [
{
"role": "assistant",
"content": None,
"tool_calls": [
{
"id": "call_weather",
"type": "function",
"function": {"name": "lookup_weather", "arguments": '{"city": "Paris"}'},
}
],
}
]
def test_openinference_output_tool_calls_do_not_shed_input_roles_under_budget():
message_groups: Final = tuple(
(
{"role": "user", "content": f"Question {index}"},
{
"role": "assistant",
"content": None,
"tool_calls": [
{
"id": f"call_{index}",
"type": "function",
"function": {"name": "lookup_weather", "arguments": '{"city": "Paris"}'},
}
],
},
{"role": "tool", "tool_call_id": f"call_{index}", "content": f"Result {index}"},
)
for index in range(13)
)
messages_in: Final = tuple(chain.from_iterable(message_groups)) + ({"role": "user", "content": "Final request"},)
tools: Final = tuple(
ToolDefinition(name=name, description="Tool", parameters_json='{"type":"object"}')
for name in ("lookup_weather", "search", "get_location", "convert_units")
)
data: Final = _llm_call(
messages_in=messages_in,
tools=tools,
choices_out=(
{
"finish_reason": "tool_calls",
"message": {
"role": "assistant",
"content": None,
"tool_calls": [
{
"id": "call_output",
"type": "function",
"function": {"name": "lookup_weather", "arguments": '{"city": "Paris"}'},
}
],
},
},
),
)
attrs: Final = fit_indexed_messages(OpenInferenceMapper().map(data), 128)
tool_call: Final = "llm.output_messages.0.message.tool_calls.0.tool_call."
assert {
f"llm.input_messages.{index}.message.role": attrs.get(f"llm.input_messages.{index}.message.role")
for index in range(40)
} == {f"llm.input_messages.{index}.message.role": message["role"] for index, message in enumerate(messages_in)}
assert {key: value for key, value in attrs.items() if ".tool_calls." in key} == {
tool_call + "id": "call_output",
tool_call + "function.name": "lookup_weather",
tool_call + "function.arguments": '{"city": "Paris"}',
}
def test_openinference_plain_output_messages_keep_the_existing_value_shape():
attrs: Final = OpenInferenceMapper().map(_llm_call())
assert all(".tool_calls." not in key for key in attrs)
assert json.loads(attrs["output.value"]) == [{"role": "assistant", "content": "Sunny."}]
def test_openinference_metadata_contains_only_promoted_metadata():
attrs: Final = OpenInferenceMapper().map(
_llm_call(promoted_metadata={"trace_marker": "m", "user_api_key_alias": "k"})
)
assert json.loads(attrs["metadata"]) == {"trace_marker": "m", "user_api_key_alias": "k"}
assert "metadata" not in OpenInferenceMapper().map(_llm_call())
# --------------------------------------------------------------------------- #
# Langfuse
# --------------------------------------------------------------------------- #