feat(telemetry): full OTel GenAI semantic conventions + in-process PII stripping

livekit-agents/livekit/agents/job.py

@@ -339,7 +339,10 @@ async def _on_session_end(self) -> None:
                     report=report,
                     tagger=self._tagger,
                     http_session=http_context.http_session(),
-                    metadata=self._otel_metadata(report.options.recording_options),
+                    metadata=self._otel_metadata(
+                        report.options.recording_options,
+                        redaction_enabled=self._redaction_enabled,
+                    ),
                 )
             except Exception:
                 logger.exception("failed to upload the session report to LiveKit Cloud")
@@ -843,7 +846,7 @@ def init_recording(self, options: RecordingOptions) -> None:
             observability_url=obs_url,
             enable_traces=options["traces"],
             enable_logs=options["logs"],
-            metadata=self._otel_metadata(options),
+            metadata=self._otel_metadata(options, redaction_enabled=redaction_enabled),
         )
         # init_recording is typically called during session.start(), at which point a bunch of
         # the logs would have already been emitted. we want to capture all of the logs as it
@@ -890,7 +893,9 @@ def _on_done(task: asyncio.Task[Any], *, coro: Any = coro) -> None:
     def token_claims(self) -> Claims:
         return api.TokenVerifier().verify(self._info.token, verify_signature=False)
 
-    def _otel_metadata(self, options: RecordingOptions | None = None) -> dict[str, Any] | None:
+    def _otel_metadata(
+        self, options: RecordingOptions | None = None, *, redaction_enabled: bool = False
+    ) -> dict[str, Any] | None:
         metadata: dict[str, Any] = {}
         if (sim := self.simulation_context()) is not None:
             metadata[ATTRIBUTE_SIMULATION_ENABLED] = True
@@ -901,7 +906,10 @@ def _otel_metadata(self, options: RecordingOptions | None = None) -> dict[str, A
                 metadata[ATTRIBUTE_SIMULATION_RUN_ID] = sim.simulation_run_id
             if sim.simulation_job_id:
                 metadata[ATTRIBUTE_SIMULATION_JOB_ID] = sim.simulation_job_id
-        if options and options.get("redaction", False):
+        # stamped on every span and log so redaction can be resolved per-record, off the
+        # record itself, rather than from the ambient job context. Takes the resolved
+        # flag so project-wide redaction counts, not just the per-session option.
+        if redaction_enabled or (options and options.get("redaction", False)):
             metadata[ATTRIBUTE_REDACTION_ENABLED] = True
         return metadata or None
 

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

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 import asyncio
-import json
 import time
 from abc import ABC, abstractmethod
 from collections.abc import AsyncIterable, AsyncIterator
@@ -10,7 +9,6 @@
 from typing import Any, ClassVar, Generic, Literal, TypeVar
 
 from opentelemetry import trace
-from opentelemetry.util.types import AttributeValue
 from pydantic import BaseModel, ConfigDict, Field
 
 from livekit import rtc
@@ -20,7 +18,12 @@
 from .._exceptions import APIConnectionError, APIError, APIStatusError
 from ..log import logger
 from ..metrics import LLMMetrics
-from ..telemetry import _chat_ctx_to_otel_events, trace_types, tracer, utils as telemetry_utils
+from ..telemetry import (
+    gen_ai as gen_ai_telemetry,
+    trace_types,
+    tracer,
+    utils as telemetry_utils,
+)
 from ..types import (
     DEFAULT_API_CONNECT_OPTIONS,
     NOT_GIVEN,
@@ -246,12 +249,15 @@ def __init__(
             self._metrics_monitor_task(monitor_aiter), name="LLM._metrics_task"
         )
 
+        # tells an enclosing `llm_node` span that this call is instrumented, so it does
+        # not record the convention's attributes a second time
+        gen_ai_telemetry.mark_inference_span_recorded()
+
         async def _traceable_main_task() -> None:
             with tracer.start_as_current_span(
                 self._llm_request_span_name, end_on_exit=False
             ) as span:
-                for name, attributes in _chat_ctx_to_otel_events(self._chat_ctx):
-                    span.add_event(name, attributes)
+                self._record_genai_request(span)
                 await self._main_task()
 
         self._task = asyncio.create_task(_traceable_main_task(), name="LLM._main_task")
@@ -262,15 +268,25 @@ async def _traceable_main_task() -> None:
     @abstractmethod
     async def _run(self) -> None: ...
 
+    def _record_genai_request(self, span: trace.Span) -> None:
+        """The GenAI inference span's request side, per the OTel GenAI conventions."""
+        gen_ai_telemetry.set_request_attributes(
+            span,
+            operation=trace_types.GenAIOperationName.CHAT,
+            provider=self._llm.provider,
+            model=self._llm.model,
+            stream=True,
+            output_type=trace_types.GenAIOutputType.TEXT,
+        )
+        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),
+            tool_definitions=gen_ai_telemetry.to_tool_definitions(self._tools),
+        )
+
     async def _main_task(self) -> None:
         self._llm_request_span = trace.get_current_span()
-        self._llm_request_span.set_attributes(
-            {
-                trace_types.ATTR_GEN_AI_OPERATION_NAME: "chat",
-                trace_types.ATTR_GEN_AI_PROVIDER_NAME: self._llm.provider,
-                trace_types.ATTR_GEN_AI_REQUEST_MODEL: self._llm.model,
-            }
-        )
 
         for i in range(self._conn_options.max_retry + 1):
             try:
@@ -402,44 +418,31 @@ async def _metrics_monitor_task(self, event_aiter: AsyncIterable[ChatChunk]) ->
                 trace_types.ATTR_LLM_METRICS, metrics.model_dump_json()
             )
 
-            # set gen_ai attributes
-            self._llm_request_span.set_attributes(
-                {
-                    trace_types.ATTR_GEN_AI_OPERATION_NAME: "chat",
-                    trace_types.ATTR_GEN_AI_REQUEST_MODEL: self._llm.model,
-                    trace_types.ATTR_GEN_AI_PROVIDER_NAME: self._llm.provider,
-                    trace_types.ATTR_GEN_AI_USAGE_INPUT_TOKENS: metrics.prompt_tokens,
-                    trace_types.ATTR_GEN_AI_USAGE_CACHE_READ_INPUT_TOKENS: (
-                        metrics.prompt_cached_tokens
-                    ),
-                    trace_types.ATTR_GEN_AI_USAGE_OUTPUT_TOKENS: metrics.completion_tokens,
-                },
+            # the GenAI response side; the request side was recorded at span creation
+            gen_ai_telemetry.set_usage_attributes(self._llm_request_span, metrics)
+            finish_reason = gen_ai_telemetry.finish_reason_for(
+                function_calls=tool_calls, interrupted=metrics.cancelled
+            )
+            gen_ai_telemetry.set_response_attributes(
+                self._llm_request_span,
+                response_id=request_id or None,
+                model=self._llm.model,
+                finish_reasons=[finish_reason],
+                time_to_first_chunk=ttft if ttft >= 0 else None,
+            )
+            gen_ai_telemetry.set_content_attributes(
+                self._llm_request_span,
+                output_messages=gen_ai_telemetry.to_output_messages(
+                    text=response_content,
+                    function_calls=tool_calls,
+                    finish_reason=finish_reason,
+                ),
             )
-            if metrics.reasoning_tokens:
-                self._llm_request_span.set_attribute(
-                    trace_types.ATTR_GEN_AI_USAGE_REASONING_TOKENS, metrics.reasoning_tokens
-                )
             if completion_start_time:
                 self._llm_request_span.set_attribute(
                     trace_types.ATTR_LANGFUSE_COMPLETION_START_TIME, f'"{completion_start_time}"'
                 )
 
-            completion_event_body: dict[str, AttributeValue] = {"role": "assistant"}
-            if response_content:
-                completion_event_body["content"] = response_content
-            if tool_calls:
-                completion_event_body["tool_calls"] = [
-                    json.dumps(
-                        {
-                            "function": {"name": tool_call.name, "arguments": tool_call.arguments},
-                            "id": tool_call.call_id,
-                            "type": "function",
-                        }
-                    )
-                    for tool_call in tool_calls
-                ]
-            self._llm_request_span.add_event(trace_types.EVENT_GEN_AI_CHOICE, completion_event_body)
-
         self._llm.emit("metrics_collected", metrics)
 
     @property

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

@@ -1,6 +1,5 @@
-from . import http_server, metrics, otel_metrics, trace_types, utils
+from . import gen_ai, http_server, metrics, otel_metrics, pii, trace_types, utils
 from .traces import (
-    _chat_ctx_to_otel_events,
     _setup_cloud_tracer,
     _upload_session_report,
     set_tracer_provider,
@@ -9,6 +8,8 @@
 
 __all__ = [
     "tracer",
+    "gen_ai",
+    "pii",
     "metrics",
     "otel_metrics",
     "trace_types",
@@ -17,7 +18,6 @@
     "utils",
     "_setup_cloud_tracer",
     "_upload_session_report",
-    "_chat_ctx_to_otel_events",
 ]
 
 # Cleanup docs of unexported modules