feat: record LLM input as deltas on trace spans

livekit-agents/livekit/agents/llm/chat_context.py

@@ -14,6 +14,8 @@
 
 from __future__ import annotations
 
+import hashlib
+import json
 import textwrap
 import time
 from collections.abc import Generator, Sequence
@@ -183,7 +185,8 @@ class ImageContent(BaseModel):
 
     id: str = Field(default_factory=lambda: utils.shortuuid("img_"))
     """
-    Unique identifier for the image
+    Unique identifier for the image. Use a new id when replacing inline image data
+    or a video frame; input-delta telemetry uses this id instead of hashing media bytes.
     """
 
     type: Literal["image_content"] = Field(default="image_content")
@@ -307,7 +310,50 @@ class MetricsReport(TypedDict, total=False):
     stt_metadata: MetricsMetadata
 
 
-class ChatMessage(BaseModel):
+class _ChatItemBase(BaseModel):
+    """Shared by every :data:`ChatItem` type; adds no fields."""
+
+    def _essential_fields(self) -> tuple[Any, ...]:
+        """What identifies this item's content beyond its id and type: what
+        :meth:`ChatContext.is_equivalent` compares and :meth:`_fingerprint` hashes.
+        Timestamps, metrics and other metadata are left out."""
+        return ()
+
+    def _fingerprint(self) -> bytes:
+        """A short digest of :meth:`_essential_fields`, to tell whether an item changed
+        without keeping a copy of it."""
+        payload = json.dumps(
+            [type(self).__name__, *self._essential_fields()],
+            ensure_ascii=False,
+            default=_fingerprint_default,
+        )
+        return hashlib.blake2b(payload.encode(), digest_size=8).digest()
+
+
+def _fingerprint_default(value: Any) -> Any:
+    # Avoid hashing inline media payloads. Their object identity detects replacement
+    # within a running session; callers should also give replacements a new image id.
+    if isinstance(value, ImageContent):
+        image_source = (
+            value.image
+            if isinstance(value.image, str) and not value.image.startswith("data:")
+            else id(value.image)
+        )
+        return [
+            "image",
+            value.id,
+            image_source,
+            value.inference_width,
+            value.inference_height,
+            value.inference_detail,
+            value.mime_type,
+        ]
+    if isinstance(value, AudioContent):
+        return ["audio", value.transcript]
+    return str(value)
+
+
+class ChatMessage(_ChatItemBase):
     id: str = Field(default_factory=lambda: utils.shortuuid("item_"))
     type: Literal["message"] = "message"
     role: ChatRole
@@ -349,11 +395,14 @@ def raw_text_content(self) -> str | None:
             return None
         return "\n".join(text_parts)
 
+    def _essential_fields(self) -> tuple[Any, ...]:
+        return (self.role, self.interrupted, self.content)
+
 
 ChatContent: TypeAlias = ImageContent | AudioContent | str
 
 
-class FunctionCall(BaseModel):
+class FunctionCall(_ChatItemBase):
     id: str = Field(default_factory=lambda: utils.shortuuid("item_"))
     type: Literal["function_call"] = "function_call"
     call_id: str
@@ -368,8 +417,11 @@ class FunctionCall(BaseModel):
     should be grouped together (e.g., parallel tool calls from a single API response),
     set this to a shared value. If not set, falls back to using id for grouping."""
 
+    def _essential_fields(self) -> tuple[Any, ...]:
+        return (self.name, self.call_id, self.arguments)
+
 
-class FunctionCallOutput(BaseModel):
+class FunctionCallOutput(_ChatItemBase):
     id: str = Field(default_factory=lambda: utils.shortuuid("item_"))
     type: Literal["function_call_output"] = Field(default="function_call_output")
     name: str = Field(default="")
@@ -384,16 +436,19 @@ class FunctionCallOutput(BaseModel):
     Realtime models can also use it to schedule their response.
     """
 
+    def _essential_fields(self) -> tuple[Any, ...]:
+        return (self.name, self.call_id, self.output, self.is_error)
 
-class AgentHandoff(BaseModel):
+
+class AgentHandoff(_ChatItemBase):
     id: str = Field(default_factory=lambda: utils.shortuuid("item_"))
     type: Literal["agent_handoff"] = Field(default="agent_handoff")
     old_agent_id: str | None = None
     new_agent_id: str
     created_at: float = Field(default_factory=time.time)
 
 
-class AgentConfigUpdate(BaseModel):
+class AgentConfigUpdate(_ChatItemBase):
     id: str = Field(default_factory=lambda: utils.shortuuid("item_"))
     type: Literal["agent_config_update"] = Field(default="agent_config_update")
 
@@ -944,36 +999,19 @@ def is_equivalent(self, other: ChatContext) -> bool:
           - Function calls: compares `name`, `call_id`, and `arguments`.
           - Function call outputs: compares `name`, `call_id`, `output`, and `is_error`.
 
-        Does not consider timestamps or other metadata.
+        Does not consider timestamps or other metadata. Each item type declares its fields
+        in ``_essential_fields``, which item fingerprints hash as well.
         """
         if self is other:
             return True
 
         if len(self.items) != len(other.items):
             return False
 
-        for a, b in zip(self.items, other.items, strict=False):
-            if a.id != b.id or a.type != b.type:
-                return False
-
-            if a.type == "message" and b.type == "message":
-                if a.role != b.role or a.interrupted != b.interrupted or a.content != b.content:
-                    return False
-
-            elif a.type == "function_call" and b.type == "function_call":
-                if a.name != b.name or a.call_id != b.call_id or a.arguments != b.arguments:
-                    return False
-
-            elif a.type == "function_call_output" and b.type == "function_call_output":
-                if (
-                    a.name != b.name
-                    or a.call_id != b.call_id
-                    or a.output != b.output
-                    or a.is_error != b.is_error
-                ):
-                    return False
-
-        return True
+        return all(
+            a.id == b.id and a.type == b.type and a._essential_fields() == b._essential_fields()
+            for a, b in zip(self.items, other.items, strict=True)
+        )
 
 
 class _ReadOnlyChatContext(ChatContext):

livekit-agents/livekit/agents/llm/llm.py

@@ -18,7 +18,7 @@
 from .._exceptions import APIConnectionError, APIError, APIStatusError
 from ..log import logger
 from ..metrics import LLMMetrics
-from ..telemetry import gen_ai as gen_ai_telemetry, trace_types, tracer
+from ..telemetry import gen_ai as gen_ai_telemetry, input_delta, trace_types, tracer
 from ..types import (
     DEFAULT_API_CONNECT_OPTIONS,
     NOT_GIVEN,
@@ -292,12 +292,21 @@ def _record_genai_request(self, span: trace.Span) -> None:
             output_type=trace_types.GenAIOutputType.TEXT,
         )
         if self._record_content:
+            if self._genai_operation_name is None and input_delta.active():
+                # a delegating span (fallback) would only repeat the input its provider
+                # span records, and take that span's place as the next delta's base
+                gen_ai_telemetry.set_content_attributes(
+                    span, tool_definitions=gen_ai_telemetry.to_tool_definitions(self._tools)
+                )
+                return
+            delta = input_delta.compute(input_delta.LLM_REQUEST, self._chat_ctx, span)
             gen_ai_telemetry.set_content_attributes(
                 span,
-                system_instructions=gen_ai_telemetry.to_system_instructions(self._chat_ctx),
-                input_messages=gen_ai_telemetry.to_input_messages(self._chat_ctx),
+                system_instructions=delta.system_instructions(),
+                input_messages=delta.input_messages(),
                 tool_definitions=gen_ai_telemetry.to_tool_definitions(self._tools),
             )
+            input_delta.set_attributes(span, delta)
 
     async def _main_task(self) -> None:
         self._llm_request_span = trace.get_current_span()

livekit-agents/livekit/agents/telemetry/__init__.py

@@ -1,6 +1,7 @@
 from . import (
     gen_ai,
     http_server,
+    input_delta,
     loop_monitor,
     metrics,
     otel_metrics,
@@ -27,6 +28,7 @@
     "otel_metrics",
     "trace_types",
     "http_server",
+    "input_delta",
     "loop_monitor",
     "session_context",
     "rpc",

livekit-agents/livekit/agents/telemetry/gen_ai.py

@@ -86,6 +86,26 @@ def mark_inference_span_recorded() -> None:
         on_created()
 
 
+def _is_system_message(item: ChatItem) -> bool:
+    return item.type == "message" and item.role in ("system", "developer")
+
+
+def _split_instructions(items: Sequence[ChatItem]) -> tuple[list[ChatItem], list[ChatItem]]:
+    """The agent's instructions are its instructions message. Every other system message
+    (``generate_reply(instructions=...)``, the expressive guide, one added through
+    ``update_chat_ctx``) is sent in place by chat-completion providers, so it belongs to
+    the conversation. A context without that message (a standalone ``llm.chat()``) takes
+    the system message it starts with, if any, as its instructions."""
+    from ..voice.generation import INSTRUCTIONS_MESSAGE_ID
+
+    for i, item in enumerate(items):
+        if item.id == INSTRUCTIONS_MESSAGE_ID and _is_system_message(item):
+            return [item], [*items[:i], *items[i + 1 :]]
+    if items and _is_system_message(items[0]):
+        return [items[0]], list(items[1:])
+    return [], list(items)
+
+
 def _text_part(content: str) -> dict[str, Any]:
     return {"type": "text", "content": content}
 
@@ -151,52 +171,67 @@ def _maybe_json(raw: str) -> Any:
 
 
 def to_system_instructions(chat_ctx: ChatContext) -> list[dict[str, Any]]:
-    """LiveKit carries an agent's instructions as ``system``/``developer`` messages in
-    the chat context, but they originate from ``Agent(instructions=...)`` rather than
-    from the conversation, so they are reported as instructions rather than history."""
+    """LiveKit carries an agent's instructions as the ``system``/``developer`` messages the
+    chat context starts with, but they originate from ``Agent(instructions=...)`` rather
+    than from the conversation, so they are reported as instructions rather than history."""
+    return _instruction_parts(_split_instructions(chat_ctx.items)[0])
+
+
+def to_input_messages(chat_ctx: ChatContext) -> list[dict[str, Any]]:
+    """History in the order it was sent. The leading ``system``/``developer`` messages go
+    to ``gen_ai.system_instructions`` instead; a later one stays in place as a ``system``
+    message. Non-conversational items (agent handoffs, config updates) are skipped."""
+    return _conversation_messages(_split_instructions(chat_ctx.items)[1])
+
+
+def _instruction_parts(items: Sequence[ChatItem]) -> list[dict[str, Any]]:
     parts: list[dict[str, Any]] = []
-    for item in chat_ctx.items:
-        if item.type == "message" and item.role in ("system", "developer"):
-            if (text := item.raw_text_content) is not None:
-                parts.append(_text_part(text))
+    for item in items:
+        if item.type == "message" and (text := item.raw_text_content) is not None:
+            parts.append(_text_part(text))
     return parts
 
 
-def to_input_messages(chat_ctx: ChatContext) -> list[dict[str, Any]]:
-    """History in the order it was sent. ``system``/``developer`` messages go to
-    ``gen_ai.system_instructions`` instead, and non-conversational items (agent
-    handoffs, config updates) are skipped."""
+def _message_role(item: ChatItem) -> str | None:
+    if item.type == "message":
+        return "system" if item.role == "developer" else item.role
+    if item.type == "function_call":
+        return "assistant"
+    if item.type == "function_call_output":
+        return "tool"
+    return None
+
+
+def _conversation_messages(items: Sequence[ChatItem]) -> list[dict[str, Any]]:
     messages: list[dict[str, Any]] = []
-    for item in chat_ctx.items:
-        role: str
-        if item.type == "message":
-            if item.role in ("system", "developer"):
-                continue
-            role = item.role
-        elif item.type == "function_call":
-            role = "assistant"
-        elif item.type == "function_call_output":
-            role = "tool"
-        else:
+    for item, placement in zip(items, _message_layout(items), strict=True):
+        if placement == "skipped":
             continue
-
         parts = _message_parts(item)
-        if not parts:
-            continue
-
-        # consecutive tool calls from one assistant turn belong to a single message
-        if (
-            messages
-            and messages[-1]["role"] == role == "assistant"
-            and item.type == "function_call"
-        ):
+        if placement == "merged":
             messages[-1]["parts"].extend(parts)
-            continue
-
-        messages.append({"role": role, "parts": parts})
+        else:
+            messages.append({"role": _message_role(item), "parts": parts})
     return messages
 
 
+def _message_layout(items: Sequence[ChatItem]) -> list[str]:
+    """How each item lands in ``gen_ai.input.messages``: a "new" message, "merged" into the
+    previous one (consecutive tool calls of one assistant turn), or "skipped"."""
+    layout: list[str] = []
+    last_role: str | None = None
+    for item in items:
+        role = _message_role(item)
+        if role is None or not _message_parts(item):
+            layout.append("skipped")
+        elif last_role == role == "assistant" and item.type == "function_call":
+            layout.append("merged")
+        else:
+            layout.append("new")
+            last_role = role
+    return layout
+
+
 def to_output_messages(
     *,
     text: str | None = None,