Merge a15309dfe8 into f2e0a5db1e

litellm/integrations/otel/mappers/openinference.py

@@ -12,6 +12,7 @@ from typing import Final
 
 from litellm.integrations.otel.mappers.base import AttributeMap, AttrValue, SpanData
 from litellm.integrations.otel.mappers.utils import (
+    MAX_MESSAGE_ATTRS_PER_SPAN,
     MAX_TOOL_DEFINITION_ATTRS_PER_SPAN,
     collect,
     drop_none,
@@ -26,6 +27,8 @@ from litellm.integrations.otel.model.payloads import (
     ToolDefinition,
 )
 
+_MAX_INDEXED_MESSAGES: Final = MAX_MESSAGE_ATTRS_PER_SPAN // 2
+
 
 class OpenInferenceMapper:
     """Emits OpenInference attributes for LLM_CALL spans.
@@ -84,27 +87,44 @@ class OpenInferenceMapper:
                 return {}
 
     def _llm_call(self, data: LLMCallSpanData) -> AttributeMap:
+        outputs: Final = output_messages(data)
+        indexed_in, indexed_out = self._indexed_split(len(data.messages_in), len(outputs))
         return {
             **collect(self._LLM_CALL_ATTRS, data),
             **collect(self._BLOB_ATTRS, data),
-            **self._messages("llm.input_messages", "input.value", data.messages_in),
-            **self._messages("llm.output_messages", "output.value", output_messages(data)),
+            **self._messages(
+                "llm.input_messages",
+                "input.value",
+                data.messages_in,
+                self._prompt_positions(len(data.messages_in), indexed_in),
+            ),
+            **self._messages("llm.output_messages", "output.value", outputs, range(indexed_out)),
             **self._tools(data),
         }
 
     @staticmethod
-    def _messages(prefix: str, value_key: str, messages: Sequence[object]) -> AttributeMap:
-        """Per-message ``{prefix}.{idx}.message.*`` keys + the ``value_key`` blob."""
+    def _indexed_split(inputs: int, outputs: int) -> tuple[int, int]:
+        """Prompt and response share one allowance; the response is reserved at least half of it."""
+        indexed_out: Final = min(outputs, max(_MAX_INDEXED_MESSAGES // 2, _MAX_INDEXED_MESSAGES - inputs))
+        return _MAX_INDEXED_MESSAGES - indexed_out, indexed_out
+
+    @staticmethod
+    def _prompt_positions(total: int, indexed: int) -> tuple[int, ...]:
+        """Prompt messages that get per-index attributes: message 0 and the most recent turns."""
+        if total <= indexed:
+            return tuple(range(total))
+        return (0, *range(total - indexed + 1, total))
+
+    @staticmethod
+    def _messages(prefix: str, value_key: str, messages: Sequence[object], positions: Sequence[int]) -> AttributeMap:
+        """``{prefix}.{idx}.message.*`` keys for the messages at ``positions`` + the ``value_key`` blob of all."""
         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 idx, (role, content) in ((idx, parsed[idx]) for idx in positions)
                 for key, value in (
-                    (
-                        f"{prefix}.{idx}.message.role",
-                        role if isinstance(role, str) else None,
-                    ),
+                    (f"{prefix}.{idx}.message.role", role if isinstance(role, str) else None),
                     (f"{prefix}.{idx}.message.content", content),
                 )
             }

litellm/integrations/otel/mappers/utils.py

@@ -32,6 +32,14 @@ core telemetry no matter how many vocabularies are configured.
 """
 
 
+MAX_MESSAGE_ATTRS_PER_SPAN: Final = DEFAULT_SPAN_ATTRIBUTE_LIMIT // 8
+"""Span-wide ceiling on per-index chat message attributes, prompt and response together.
+
+An eighth is the largest share that still fits beside the tool ceiling and the core
+of every vocabulary at once. The complete conversation still rides the JSON blobs.
+"""
+
+
 def tool_attr_budget(vocabularies: int) -> int:
     """Split the span-wide tool-definition ceiling across active vocabularies."""
     return MAX_TOOL_DEFINITION_ATTRS_PER_SPAN // max(vocabularies, 1)

tests/test_litellm/integrations/otel/test_otel_v2_emitter.py

@@ -1,6 +1,8 @@
 """Golden tests for the OTel v2 engine: span shape, kinds, semconv attributes,
 legacy dual-emit, hierarchy, error status, and idempotency. Needs the OTel SDK."""
 
+import json
+
 import pytest
 
 pytest.importorskip("opentelemetry")
@@ -18,6 +20,7 @@ from litellm.integrations.otel.plumbing import providers  # noqa: E402
 from litellm.integrations.otel.emitter import SpanEmitter  # noqa: E402
 from litellm.integrations.otel.emitter import stamp_error  # noqa: E402
 from litellm.integrations.otel.mappers.utils import (  # noqa: E402
+    MAX_MESSAGE_ATTRS_PER_SPAN,
     MAX_TOOL_DEFINITION_ATTRS_PER_SPAN,
 )
 from litellm.integrations.otel.model.payloads import (  # noqa: E402
@@ -440,3 +443,155 @@ def test_vendor_tool_definitions_are_truncated_not_dropped():
     assert a["llm.tools.0.tool.name"] == "tool_0"
     assert a["llm.tools.0.tool.json_schema"]
     assert "llm.tools.126.tool.name" not in a
+
+
+def _conversation_payload(turns, choices=1, **overrides):
+    """A ``turns``-message chat with ``choices`` response choices, content-bearing."""
+    return _payload(
+        messages=[{"role": ("user", "assistant")[i % 2], "content": f"turn {i}"} for i in range(turns)],
+        response={
+            "id": "resp_1",
+            "model": "gpt-4o-2024",
+            "choices": [
+                {"finish_reason": "stop", "message": {"role": "assistant", "content": f"reply {i}"}}
+                for i in range(choices)
+            ],
+        },
+        **overrides,
+    )
+
+
+def _conversation_span(mapper_names, payload, legacy_compat=False):
+    """The exported LLM-call span for ``payload`` with content capture on."""
+    cfg = OpenTelemetryV2Config(
+        exporter="in_memory",
+        legacy_compat=legacy_compat,
+        mapper_names=list(mapper_names),
+        capture_message_content="span_only",
+    )
+    provider, exporter = providers.in_memory_provider(cfg)
+    engine = SpanEmitter(providers.get_tracer(provider, "litellm-test"), cfg)
+    engine.emit(
+        SpanRole.LLM_CALL,
+        LLMCallSpanData.from_standard_logging_payload(payload, capture_content=True),
+    )
+    (span,) = exporter.get_finished_spans()
+    return span
+
+
+def _indexed_message_count(attributes, prefix):
+    return len({key.split(".")[2] for key in attributes if key.startswith(f"{prefix}.")})
+
+
+@pytest.mark.parametrize("turns", [60, 200])
+def test_long_conversation_does_not_evict_core_attributes(turns):
+    """Per-message OpenInference attributes must never crowd core telemetry off the span."""
+    span = _conversation_span(["genai", "openinference"], _conversation_payload(turns))
+    a = span.attributes
+
+    assert span.dropped_attributes == 0
+    assert a[GenAI.REQUEST_MODEL] == "gpt-4o"
+    assert a[GenAI.PROVIDER_NAME] == "openai"
+    assert a[GenAI.USAGE_INPUT_TOKENS] == 10
+    assert a[GenAI.USAGE_OUTPUT_TOKENS] == 5
+    assert a[GenAI.RESPONSE_FINISH_REASONS] == ("stop",)
+    assert a[f"{LiteLLM.COST_PREFIX}total"] == 0.002
+
+    assert a["llm.input_messages.0.message.content"] == "turn 0"
+    assert a["llm.output_messages.0.message.content"] == "reply 0"
+    assert a[f"llm.input_messages.{turns - 1}.message.content"] == f"turn {turns - 1}"
+    assert f"llm.input_messages.{turns // 2}.message.role" not in a
+    assert len(json.loads(a["input.value"])) == turns
+    assert len(json.loads(a["output.value"])) == 1
+    assert len(json.loads(a[GenAI.INPUT_MESSAGES])) == turns
+
+
+def test_short_conversation_keeps_every_message_indexed():
+    """Below the cap nothing is truncated in either direction."""
+    a = _conversation_span(["genai", "openinference"], _conversation_payload(4, choices=2)).attributes
+    for idx in range(4):
+        assert a[f"llm.input_messages.{idx}.message.content"] == f"turn {idx}"
+    for idx in range(2):
+        assert a[f"llm.output_messages.{idx}.message.content"] == f"reply {idx}"
+
+
+def test_indexed_prompt_keeps_opener_and_latest_turns_under_a_value_length_limit(monkeypatch):
+    """The system prompt and the live turn keep their own keys once the SDK clips ``input.value``."""
+    monkeypatch.setenv("OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT", "256")
+    payload = _conversation_payload(60)
+    payload["messages"][0] = {"role": "system", "content": "be terse"}
+    payload["messages"][-1] = {"role": "user", "content": "LATEST-TURN"}
+    a = _conversation_span(["genai", "openinference"], payload).attributes
+
+    assert len(a["input.value"]) == 256
+    assert a["llm.input_messages.0.message.role"] == "system"
+    assert a["llm.input_messages.0.message.content"] == "be terse"
+    assert a["llm.input_messages.59.message.role"] == "user"
+    assert a["llm.input_messages.59.message.content"] == "LATEST-TURN"
+    assert a["llm.output_messages.0.message.content"] == "reply 0"
+    assert [int(key.split(".")[2]) for key in a if key.endswith("message.content") and key.startswith("llm.input_")] == [
+        0,
+        *range(54, 60),
+    ]
+
+
+def test_message_cap_is_shared_across_input_and_output():
+    """One span-wide allowance covers both directions, and the response always keeps a share."""
+    long_prompt = _conversation_span(["genai", "openinference"], _conversation_payload(60, choices=1)).attributes
+    many_choices = _conversation_span(["genai", "openinference"], _conversation_payload(60, choices=20)).attributes
+
+    single_reply_indexed = _indexed_message_count(long_prompt, "llm.output_messages")
+    assert single_reply_indexed == 1
+    assert _indexed_message_count(long_prompt, "llm.input_messages") + single_reply_indexed == (
+        MAX_MESSAGE_ATTRS_PER_SPAN // 2
+    )
+
+    assert _indexed_message_count(many_choices, "llm.input_messages") > 0
+    assert _indexed_message_count(many_choices, "llm.output_messages") > single_reply_indexed
+    assert _indexed_message_count(many_choices, "llm.input_messages") + _indexed_message_count(
+        many_choices, "llm.output_messages"
+    ) == (MAX_MESSAGE_ATTRS_PER_SPAN // 2)
+
+
+def test_fully_populated_span_with_every_vocabulary_stays_within_the_attribute_limit():
+    """Every capped family maxed at once still leaves the whole core intact."""
+    payload = _conversation_payload(
+        200,
+        choices=20,
+        stream=True,
+        model_parameters={
+            **_tools_payload(127)["model_parameters"],
+            "top_p": 0.9,
+            "frequency_penalty": 0.1,
+            "presence_penalty": 0.1,
+            "seed": 7,
+            "stop": ["\n"],
+        },
+        cost_breakdown={
+            key: 0.001
+            for key in (
+                "input_cost",
+                "output_cost",
+                "cache_read_cost",
+                "cache_creation_cost",
+                "tool_usage_cost",
+                "original_cost",
+                "discount_amount",
+                "discount_percent",
+                "margin_fixed_amount",
+                "margin_percent",
+                "margin_total_amount",
+                "total_cost",
+            )
+        },
+    )
+    span = _conversation_span(["genai", "openinference", "langfuse", "weave", "langtrace"], payload, legacy_compat=True)
+    a = span.attributes
+
+    assert span.dropped_attributes == 0
+    assert a[GenAI.REQUEST_MODEL] == "gpt-4o"
+    assert a[f"{LiteLLM.COST_PREFIX}total"] == 0.002
+    assert a[LiteLLM.TOOLS_DECLARED] == 127
+    assert a["llm.input_messages.0.message.content"] == "turn 0"
+    assert a["llm.input_messages.199.message.content"] == "turn 199"
+    assert a["llm.output_messages.0.message.content"] == "reply 0"