feat(llmobs): instrument openai realtime api for audio

ddtrace/contrib/internal/openai/__init__.py

@@ -57,4 +57,12 @@
    The service name reported by default for OpenAI requests.
 
    Alternatively, set this option with the ``DD_OPENAI_SERVICE`` environment variable.
+
+
+.. py:data:: DD_OPENAI_REALTIME_ENABLED
+
+   Enables instrumentation of the OpenAI Realtime API. Set to ``false`` to disable Realtime
+   instrumentation while keeping the rest of the OpenAI integration enabled.
+
+   Default: ``true``
 """  # noqa: E501

ddtrace/contrib/internal/openai/patch.py

@@ -7,6 +7,7 @@
 
 from ddtrace import config
 from ddtrace.contrib.internal.openai import _endpoint_hooks
+from ddtrace.contrib.internal.openai import _realtime
 from ddtrace.contrib.trace_utils import unwrap
 from ddtrace.contrib.trace_utils import wrap
 from ddtrace.internal import core
@@ -141,6 +142,8 @@ def patch():
             if deep_getattr(openai, async_method) is not None:
                 wrap(openai, async_method, _patched_endpoint_async(endpoint_hook))
 
+    _realtime.patch_realtime()
+
     openai.__datadog_patch = True
 
     # Notify AI Guard (and any other plugin) that wrapping is complete so they
@@ -187,6 +190,8 @@ def unpatch():
             if async_resource is not None and hasattr(async_resource, method_name):
                 unwrap(async_resource, method_name)
 
+    _realtime.unpatch_realtime()
+
     delattr(openai, "_datadog_integration")
 
 

ddtrace/internal/settings/_supported_configurations.py

@@ -345,6 +345,7 @@
         "DD_MYSQL_SERVICE",
         "DD_MYSQL_TRACE_FETCH_METHODS",
         "DD_OPENAI_AGENTS_SERVICE",
+        "DD_OPENAI_REALTIME_ENABLED",
         "DD_OPENAI_SERVICE",
         "DD_ORIGIN_DETECTION_ENABLED",
         "DD_PARTITION",

ddtrace/llmobs/_integrations/audio_utils.py

@@ -0,0 +1,211 @@
+"""Audio helpers shared by the LLM Observability integrations.
+
+Building ``AudioPart``s, mapping provider audio formats to MIME types, and turning raw audio
+(PCM16, G.711) into playable WAV within the per-span-event size budget. Kept separate from the
+larger ``utils.py`` since the audio surface has grown (chat audio, realtime, telephony).
+"""
+
+import base64
+import io
+import struct
+from typing import Any
+from typing import Optional
+from typing import Union
+import wave
+
+from ddtrace.internal.logger import get_logger
+from ddtrace.llmobs._utils import _get_attr
+from ddtrace.llmobs.types import AudioPart
+
+
+logger = get_logger(__name__)
+
+
+def format_audio_part(data: Union[bytes, str], mime_type: str) -> AudioPart:
+    """Build an ``AudioPart`` from raw audio bytes (base64-encoded) or an existing base64 string."""
+    content = base64.b64encode(data).decode("utf-8") if isinstance(data, bytes) else data
+    return AudioPart(mime_type=mime_type, content=content)
+
+
+# OpenAI audio ``format`` values that don't map to ``audio/<format>``.
+_OPENAI_AUDIO_MIME_TYPES = {
+    "mp3": "audio/mpeg",
+}
+
+
+def audio_mime_type_from_format(fmt: str) -> str:
+    """Map an OpenAI audio ``format`` (e.g. "wav", "mp3") to a MIME type."""
+    fmt = (fmt or "").strip().lower()
+    return _OPENAI_AUDIO_MIME_TYPES.get(fmt, "audio/{}".format(fmt) if fmt else "audio/wav")
+
+
+# Raw audio formats the UI cannot render as a player. For these we keep the transcript as the
+# message content and skip the inline audio_part (raw bytes would only bloat the payload).
+_NON_RENDERABLE_AUDIO_MIME_TYPES = frozenset(
+    {
+        "audio/pcm",
+        "audio/pcm16",
+        "audio/l16",
+        "audio/pcmu",
+        "audio/pcma",
+        "audio/g711_ulaw",
+        "audio/g711_alaw",
+        "audio/basic",
+    }
+)
+
+# Budget for inline audio measured on the *base64-encoded* size (what actually rides the span
+# event), so the guard reflects the real payload after encoding. Kept below the 5 MB per-span-event
+# limit with headroom for the rest of the event (oversize events have their whole I/O dropped
+# backend-side). 4 MiB encoded ≈ 3 MiB of raw audio.
+LLMOBS_AUDIO_INLINE_MAX_BYTES = 4 * 1024 * 1024
+
+
+def _base64_encoded_len(num_bytes: int) -> int:
+    """Length of ``num_bytes`` after standard base64 encoding (4 chars per 3 bytes, padded)."""
+    return ((num_bytes + 2) // 3) * 4
+
+
+# OpenAI Realtime audio ``format`` values (legacy string form) that don't map to ``audio/<format>``.
+_REALTIME_AUDIO_FORMAT_MIME_TYPES = {
+    "pcm16": "audio/pcm",
+    "pcm": "audio/pcm",
+    "g711_ulaw": "audio/pcmu",
+    "g711_alaw": "audio/pcma",
+}
+
+
+def realtime_audio_format_to_mime(fmt: Any) -> str:
+    """Map an OpenAI Realtime audio format to a MIME type.
+
+    Handles both the legacy string form (e.g. "pcm16", "g711_ulaw") and the newer
+    discriminated-union object whose ``type`` is already a MIME type (e.g. "audio/pcm").
+    """
+    if fmt is None:
+        return ""
+    type_attr = _get_attr(fmt, "type", None)
+    if type_attr:
+        fmt = type_attr
+    normalized = str(fmt).strip().lower()
+    if not normalized:
+        return ""
+    if normalized.startswith("audio/"):
+        return normalized
+    return _REALTIME_AUDIO_FORMAT_MIME_TYPES.get(normalized, "audio/{}".format(normalized))
+
+
+def is_renderable_audio_mime(mime_type: str) -> bool:
+    """Whether a MIME type can be rendered as an audio player in the UI (raw PCM cannot)."""
+    return bool(mime_type) and mime_type.strip().lower() not in _NON_RENDERABLE_AUDIO_MIME_TYPES
+
+
+def concat_base64_audio(b64_chunks: "list[str]") -> bytes:
+    """Decode and concatenate a list of base64 audio chunks into raw bytes.
+
+    Chunks must be decoded before concatenation: directly joining base64 strings is invalid
+    unless each chunk is aligned on a 3-byte boundary.
+    """
+    buf = bytearray()
+    for chunk in b64_chunks:
+        if not chunk:
+            continue
+        try:
+            buf.extend(base64.b64decode(chunk))
+        except (ValueError, TypeError):
+            continue
+    return bytes(buf)
+
+
+# Raw little-endian PCM16 mime types. These aren't renderable on their own, but can be losslessly
+# wrapped in a WAV container (just a header) to produce a playable ``audio/wav`` part.
+_PCM16_MIME_TYPES = frozenset({"audio/pcm", "audio/pcm16", "audio/l16"})
+
+
+def is_pcm16_audio_mime(mime_type: str) -> bool:
+    return bool(mime_type) and mime_type.strip().lower() in _PCM16_MIME_TYPES
+
+
+def pcm16_to_wav(pcm_bytes: bytes, sample_rate: int = 24000, channels: int = 1) -> bytes:
+    """Wrap raw little-endian PCM16 audio in a WAV container.
+
+    This is lossless and cheap (it only prepends a header), and turns raw PCM — which the UI can't
+    render — into a playable ``audio/wav`` payload.
+    """
+    buf = io.BytesIO()
+    with wave.open(buf, "wb") as wav_file:
+        wav_file.setnchannels(channels)
+        wav_file.setsampwidth(2)  # PCM16 = 2 bytes/sample
+        wav_file.setframerate(sample_rate)
+        wav_file.writeframes(pcm_bytes)
+    return buf.getvalue()
+
+
+# G.711 telephony audio (8kHz, 8-bit companded). Realtime uses these for phone-call integrations
+# (Twilio/SIP). We decode to PCM16 so it can be WAV-wrapped and played in the UI; the stdlib `wave`
+# module can't emit G.711-tagged WAV, and `audioop` (ulaw2lin/alaw2lin) was removed in Python 3.13.
+_G711_MIME_TO_VARIANT = {
+    "audio/pcmu": "ulaw",
+    "audio/g711_ulaw": "ulaw",
+    "audio/pcma": "alaw",
+    "audio/g711_alaw": "alaw",
+}
+G711_SAMPLE_RATE = 8000  # G.711 is always 8kHz mono.
+
+
+def g711_variant(mime_type: str) -> Optional[str]:
+    """Return "ulaw"/"alaw" for a G.711 MIME type, else ``None``."""
+    return _G711_MIME_TO_VARIANT.get((mime_type or "").strip().lower())
+
+
+def _ulaw_decode_sample(byte: int) -> int:
+    """Decode one G.711 μ-law byte to a signed 16-bit linear PCM sample (CCITT G.711)."""
+    byte = ~byte & 0xFF
+    sample = ((byte & 0x0F) << 3) + 0x84
+    sample <<= (byte & 0x70) >> 4
+    return (0x84 - sample) if (byte & 0x80) else (sample - 0x84)
+
+
+def _alaw_decode_sample(byte: int) -> int:
+    """Decode one G.711 A-law byte to a signed 16-bit linear PCM sample (CCITT G.711)."""
+    byte ^= 0x55
+    sample = (byte & 0x0F) << 4
+    seg = (byte & 0x70) >> 4
+    if seg == 0:
+        sample += 8
+    elif seg == 1:
+        sample += 0x108
+    else:
+        sample = (sample + 0x108) << (seg - 1)
+    return sample if (byte & 0x80) else -sample
+
+
+# Precompute 256-entry decode tables (the input domain is a single byte).
+_ULAW_TABLE = [_ulaw_decode_sample(i) for i in range(256)]
+_ALAW_TABLE = [_alaw_decode_sample(i) for i in range(256)]
+
+
+def g711_to_pcm16(data: bytes, variant: str) -> bytes:
+    """Decode G.711 ("ulaw"/"alaw") bytes to raw little-endian PCM16."""
+    table = _ALAW_TABLE if variant == "alaw" else _ULAW_TABLE
+    return b"".join(struct.pack("<h", table[b]) for b in data)
+
+
+def format_audio_part_with_guard(
+    audio_bytes: bytes, mime_type: str, max_bytes: int = LLMOBS_AUDIO_INLINE_MAX_BYTES
+) -> Optional[AudioPart]:
+    """Build a playable ``AudioPart`` only for renderable formats within the size budget.
+
+    Returns ``None`` for non-renderable formats (e.g. raw PCM) or oversize audio; callers should
+    fall back to the transcript as the message content in that case.
+    """
+    if not audio_bytes or not is_renderable_audio_mime(mime_type):
+        return None
+    # Compare the *encoded* size: format_audio_part base64-encodes the bytes (~4/3 expansion), and
+    # that encoded content is what counts against the per-span-event limit.
+    encoded_len = _base64_encoded_len(len(audio_bytes))
+    if encoded_len > max_bytes:
+        logger.debug(
+            "Audio (%d encoded bytes) exceeds inline budget %d; omitting inline audio content", encoded_len, max_bytes
+        )
+        return None
+    return format_audio_part(audio_bytes, mime_type)

ddtrace/llmobs/_integrations/base.py

@@ -76,7 +76,7 @@ def trace(self, operation_id: str, submit_to_llmobs: bool = False, **kwargs) ->
             child_of=parent_context,
             resource=operation_id,
             span_type=span_type,
-            activate=True,
+            activate=kwargs.get("activate", True),
         )
         service = int_service(None, self.integration_config) or ""
         set_service_and_source(span, service, self.integration_config)

ddtrace/llmobs/_integrations/openai.py

@@ -51,6 +51,7 @@ def trace(self, operation_id: str, submit_to_llmobs: bool = False, **kwargs: dic
             "createResponse",
             "parseChatCompletion",
             "parseResponse",
+            "createRealtimeResponse",
         )
         if operation_id in traced_operations:
             submit_to_llmobs = True
@@ -67,6 +68,15 @@ def _set_base_span_tags(self, span: Span, **kwargs) -> None:
             client = "Deepseek"
         span._set_attribute("openai.request.provider", client)
 
+    def _get_model_provider(self, span) -> str:
+        if self._is_provider(span, "azure"):
+            return "azure_openai"
+        elif self._is_provider(span, "openai"):
+            return "openai"
+        elif self._is_provider(span, "deepseek"):
+            return "deepseek"
+        return UNKNOWN_MODEL_PROVIDER
+
     def _is_provider(self, span, provider):
         """Check if the traced operation is from the given provider."""
         base_url = None
@@ -97,13 +107,7 @@ def _llmobs_set_tags(
         )
         model_name = span.get_tag("openai.response.model") or span.get_tag("openai.request.model") or "unknown_model"
 
-        model_provider = UNKNOWN_MODEL_PROVIDER
-        if self._is_provider(span, "azure"):
-            model_provider = "azure_openai"
-        elif self._is_provider(span, "openai"):
-            model_provider = "openai"
-        elif self._is_provider(span, "deepseek"):
-            model_provider = "deepseek"
+        model_provider = self._get_model_provider(span)
 
         metrics = self._extract_llmobs_metrics_tags(span, response, span_kind, kwargs)
         provider = span.get_tag("openai.request.provider") or "OpenAI"
@@ -171,6 +175,42 @@ def _llmobs_set_tags_from_tool(span: Span, kwargs: dict[str, Any], response: Any
             metadata={"tool_id": tool_id},
         )
 
+    def _llmobs_set_tags_from_realtime_response(
+        self,
+        span: Span,
+        model_name: Optional[str],
+        input_messages: list[Any],
+        output_messages: list[Any],
+        metadata: Optional[dict[str, Any]],
+        metrics: Optional[dict[str, Any]],
+        session_id: Optional[str] = None,
+    ) -> None:
+        """Tag a per-turn Realtime span (llm kind) built by the realtime state machine.
+
+        Each turn is its own trace; ``session_id`` groups all turns of one connection into a single
+        conversation in the UI (there is no parent session span).
+        """
+        provider = span.get_tag("openai.request.provider") or "OpenAI"
+        model_provider = self._get_model_provider(span)
+        _annotate_llmobs_span_data(
+            span,
+            name="{}.{}".format(provider, span.resource) if span.resource else None,
+            kind="llm",
+            model_name=model_name or "unknown_model",
+            model_provider=model_provider,
+            input_messages=input_messages or None,
+            output_messages=output_messages or None,
+            metadata=metadata or {},
+            metrics=metrics or None,
+            session_id=session_id,
+        )
+        # Mirror the base llmobs_set_tags path: also stamp the LLMObs->APM shadow token metrics so
+        # realtime spans are consistent with every other OpenAI span for APM-only users.
+        try:
+            self._apply_shadow_metrics(span, metrics, "llm", model_name=model_name, model_provider=model_provider)
+        except Exception:
+            log.debug("Error applying shadow metrics for realtime span %s", span, exc_info=True)
+
     def _set_apm_shadow_tags(self, span, args, kwargs, response=None, operation=""):
         span_kind = (
             "workflow"
@@ -181,13 +221,7 @@ def _set_apm_shadow_tags(self, span, args, kwargs, response=None, operation=""):
         )
         metrics = self._extract_llmobs_metrics_tags(span, response, span_kind, kwargs)
         model_name = span.get_tag("openai.response.model") or span.get_tag("openai.request.model")
-        model_provider = UNKNOWN_MODEL_PROVIDER
-        if self._is_provider(span, "azure"):
-            model_provider = "azure_openai"
-        elif self._is_provider(span, "openai"):
-            model_provider = "openai"
-        elif self._is_provider(span, "deepseek"):
-            model_provider = "deepseek"
+        model_provider = self._get_model_provider(span)
         self._apply_shadow_metrics(
             span,
             metrics,