diff --git a/examples/tracing/azure-speech/azure_speech_tracing.ipynb b/examples/tracing/azure-speech/azure_speech_tracing.ipynb
new file mode 100644
index 00000000..9eb196d9
--- /dev/null
+++ b/examples/tracing/azure-speech/azure_speech_tracing.ipynb
@@ -0,0 +1,186 @@
+{
+ "cells": [
+ {
+ "cell_type": "markdown",
+ "metadata": {},
+ "source": [
+ "[](https://colab.research.google.com/github/openlayer-ai/openlayer-python/blob/main/examples/tracing/azure-speech/azure_speech_tracing.ipynb)\n",
+ "\n",
+ "\n",
+ "# Azure AI Speech tracing quickstart\n",
+ "\n",
+ "This notebook illustrates how to monitor Azure AI Speech (speech-to-text, speech translation and text-to-speech) with Openlayer.\n",
+ "\n",
+ "The Speech SDK and the Openlayer tracer both run in your backend: the Azure key stays in your server-side configuration and is never sent to Openlayer. Only non-secret settings (region, language, voice, custom endpoint ID, output format), the recognized text or synthesized-audio stats, latency and failure details are traced. Do not ship either the Azure key or the Openlayer API key in browser or mobile code."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "!pip install openlayer azure-cognitiveservices-speech"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {},
+ "source": [
+ "## 1. Set the environment variables"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "import os\n",
+ "\n",
+ "# Azure AI Speech env variables (load these from your secret store in production)\n",
+ "os.environ[\"AZURE_SPEECH_KEY\"] = \"YOUR_AZURE_SPEECH_KEY_HERE\"\n",
+ "os.environ[\"AZURE_SPEECH_REGION\"] = \"YOUR_AZURE_SPEECH_REGION_HERE\"\n",
+ "\n",
+ "# Openlayer env variables\n",
+ "os.environ[\"OPENLAYER_API_KEY\"] = \"YOUR_OPENLAYER_API_KEY_HERE\"\n",
+ "os.environ[\"OPENLAYER_INFERENCE_PIPELINE_ID\"] = \"YOUR_OPENLAYER_INFERENCE_PIPELINE_ID_HERE\""
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {},
+ "source": [
+ "## 2. Initialize Openlayer and create the Speech clients\n",
+ "\n",
+ "`init()` auto-instruments every `SpeechRecognizer`, `TranslationRecognizer` and `SpeechSynthesizer` created afterwards. You can also trace a single client explicitly with `trace_azure_speech(client)`.\n",
+ "\n",
+ "Audio is **not** sent to Openlayer by default. Set `attachment_upload_enabled=True` only if your security and privacy requirements allow storing audio in Openlayer; synthesized audio and any input audio you pass as `openlayer_audio` are then uploaded alongside the trace."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "import azure.cognitiveservices.speech as speechsdk\n",
+ "\n",
+ "from openlayer.lib import init\n",
+ "\n",
+ "init(\n",
+ " attachment_upload_enabled=False, # set to True to upload audio to Openlayer\n",
+ ")\n",
+ "\n",
+ "speech_config = speechsdk.SpeechConfig(\n",
+ " subscription=os.environ[\"AZURE_SPEECH_KEY\"],\n",
+ " region=os.environ[\"AZURE_SPEECH_REGION\"],\n",
+ ")\n",
+ "speech_config.speech_recognition_language = \"en-US\"\n",
+ "speech_config.speech_synthesis_voice_name = \"en-US-JennyNeural\""
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {},
+ "source": [
+ "## 3. Use your traced clients normally\n",
+ "\n",
+ "### Text-to-speech"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "synthesizer = speechsdk.SpeechSynthesizer( # auto-traced by Openlayer\n",
+ " speech_config=speech_config,\n",
+ " audio_config=speechsdk.audio.AudioOutputConfig(filename=\"greeting.wav\"),\n",
+ ")\n",
+ "\n",
+ "result = synthesizer.speak_text(\"Hello! Thanks for calling. How can I help you today?\")\n",
+ "result.reason"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {},
+ "source": [
+ "### Speech-to-text\n",
+ "\n",
+ "The Speech SDK does not expose the audio behind an `AudioConfig`, so pass the same file as `openlayer_audio` if you want it attached to the trace (only used when attachment uploads are enabled)."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "recognizer = speechsdk.SpeechRecognizer( # auto-traced by Openlayer\n",
+ " speech_config=speech_config,\n",
+ " audio_config=speechsdk.audio.AudioConfig(filename=\"greeting.wav\"),\n",
+ ")\n",
+ "\n",
+ "result = recognizer.recognize_once(openlayer_audio=\"greeting.wav\")\n",
+ "result.reason, result.text"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {},
+ "source": [
+ "### Grouping calls into one trace\n",
+ "\n",
+ "Wrap a conversational turn in `@trace` to see recognition and synthesis as steps of the same trace."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "from openlayer.lib import trace\n",
+ "\n",
+ "\n",
+ "@trace()\n",
+ "def voice_turn(audio_path: str) -> str:\n",
+ " recognizer = speechsdk.SpeechRecognizer(\n",
+ " speech_config=speech_config,\n",
+ " audio_config=speechsdk.audio.AudioConfig(filename=audio_path),\n",
+ " )\n",
+ " heard = recognizer.recognize_once(openlayer_audio=audio_path).text\n",
+ "\n",
+ " reply = f\"You said: {heard}\"\n",
+ " speechsdk.SpeechSynthesizer(speech_config=speech_config, audio_config=None).speak_text(reply)\n",
+ " return reply\n",
+ "\n",
+ "\n",
+ "voice_turn(\"greeting.wav\")"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {},
+ "source": [
+ "That's it! The Speech calls are published to Openlayer and you can start creating tests around them."
+ ]
+ }
+ ],
+ "metadata": {
+ "kernelspec": {
+ "display_name": "Python 3",
+ "language": "python",
+ "name": "python3"
+ },
+ "language_info": {
+ "name": "python",
+ "version": "3.12"
+ }
+ },
+ "nbformat": 4,
+ "nbformat_minor": 5
+}
\ No newline at end of file
diff --git a/src/openlayer/lib/__init__.py b/src/openlayer/lib/__init__.py
index 55ede7e3..a7477dfd 100644
--- a/src/openlayer/lib/__init__.py
+++ b/src/openlayer/lib/__init__.py
@@ -16,6 +16,7 @@
"trace_async",
"trace_bedrock",
"trace_azure_content_understanding",
+ "trace_azure_speech",
"trace_oci_genai",
"trace_oci", # Alias for backward compatibility
"trace_litellm",
@@ -167,6 +168,22 @@ def trace_azure_content_understanding(client):
return azure_content_understanding_tracer.trace_azure_content_understanding(client)
+def trace_azure_speech(client):
+ """Trace Azure AI Speech recognition, translation and synthesis calls."""
+ # pylint: disable=import-outside-toplevel
+ try:
+ import azure.cognitiveservices.speech # noqa: F401
+ except ImportError:
+ raise ImportError(
+ "azure-cognitiveservices-speech is required for Azure Speech tracing. "
+ "Install with: pip install azure-cognitiveservices-speech"
+ )
+
+ from .integrations import azure_speech_tracer
+
+ return azure_speech_tracer.trace_azure_speech(client)
+
+
def trace_oci_genai(client, estimate_tokens: bool = True):
"""Trace OCI GenAI chat completions.
diff --git a/src/openlayer/lib/integrations/_auto.py b/src/openlayer/lib/integrations/_auto.py
index 81ef2f02..733c0d60 100644
--- a/src/openlayer/lib/integrations/_auto.py
+++ b/src/openlayer/lib/integrations/_auto.py
@@ -169,6 +169,12 @@ def _do_patch() -> None:
_patch_via("azure_content_understanding_tracer", "_patch_acu"),
_patch_via("azure_content_understanding_tracer", "_unpatch_acu"),
),
+ IntegrationSpec(
+ "azure_speech",
+ "azure.cognitiveservices.speech",
+ _patch_via("azure_speech_tracer", "_patch_azure_speech"),
+ _patch_via("azure_speech_tracer", "_unpatch_azure_speech"),
+ ),
IntegrationSpec(
"litellm",
"litellm",
diff --git a/src/openlayer/lib/integrations/azure_speech_tracer.py b/src/openlayer/lib/integrations/azure_speech_tracer.py
new file mode 100644
index 00000000..b5cf2b2d
--- /dev/null
+++ b/src/openlayer/lib/integrations/azure_speech_tracer.py
@@ -0,0 +1,670 @@
+"""Module with methods used to trace the Azure AI Speech SDK.
+
+Traces speech-to-text (``SpeechRecognizer``), speech translation
+(``TranslationRecognizer``) and text-to-speech (``SpeechSynthesizer``) calls made
+with ``azure-cognitiveservices-speech``.
+
+The wrapper runs wherever the Speech SDK runs (the customer's backend), so the
+Azure credential never leaves it: only an allowlist of non-secret configuration
+(region, language, voice, custom endpoint ID, output format) is recorded. Audio
+is attached only when ``attachment_upload_enabled`` is on in the tracer
+configuration, and it is uploaded separately rather than inlined into the trace.
+"""
+
+import contextvars
+import logging
+import mimetypes
+import re
+import struct
+import time
+from functools import wraps
+from pathlib import Path
+from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional, Sequence, Union
+from urllib.parse import quote, unquote, urlsplit, urlunsplit, parse_qsl
+
+try:
+ import azure.cognitiveservices.speech as speechsdk
+
+ HAVE_AZURE_SPEECH = True
+except ImportError:
+ HAVE_AZURE_SPEECH = False
+
+if TYPE_CHECKING:
+ import azure.cognitiveservices.speech as speechsdk
+
+from ..tracing import tracer
+from ..tracing.attachments import Attachment
+
+logger = logging.getLogger(__name__)
+
+# No space: cost lookup matches `provider` against an llm-costs slug exactly.
+PROVIDER = "Azure_Speech"
+
+# Speech SDK offsets and durations are expressed in 100-nanosecond ticks.
+_TICKS_PER_MS = 10_000
+
+_RECOGNIZE_METHODS = ("recognize_once", "recognize_once_async")
+_SYNTHESIZE_METHODS = {
+ "speak_text": "text",
+ "speak_text_async": "text",
+ "speak_ssml": "ssml",
+ "speak_ssml_async": "ssml",
+}
+
+_warned_audio_dropped = False
+
+# True while a traced sync method runs; see ``_wrap_method``.
+_in_traced_sync_call: "contextvars.ContextVar[bool]" = contextvars.ContextVar(
+ "openlayer_azure_speech_in_traced_sync_call", default=False
+)
+
+
+def trace_azure_speech(client: Any) -> Any:
+ """Patch an Azure Speech recognizer or synthesizer to trace its calls.
+
+ Supported clients and methods:
+
+ - ``SpeechRecognizer`` / ``TranslationRecognizer``: ``recognize_once`` and
+ ``recognize_once_async``.
+ - ``SpeechSynthesizer``: ``speak_text``, ``speak_ssml`` and their ``_async``
+ variants.
+
+ The following information is collected for each call:
+
+ - start_time, end_time and latency (for ``_async`` methods, until
+ ``future.get()`` returns).
+ - inputs: the recognition language (and translation target languages), or
+ the text/SSML that was synthesized.
+ - output: the recognized text (plus translations), or the synthesized audio's
+ duration and size.
+ - model: the custom endpoint ID or ``speech-to-text`` for recognition; the
+ voice name or ``text-to-speech`` for synthesis.
+ - model_parameters: region, language, voice, endpoint ID and output format.
+ The subscription key and authorization token are never recorded.
+ - metadata: result ID, result reason, offset/duration, and no-match or
+ cancellation details (the Speech SDK reports failures as canceled results
+ rather than raising).
+
+ Two extra keyword arguments are accepted by every traced method and are never
+ forwarded to the Speech SDK:
+
+ - ``inference_id``: sets the ID of the traced step.
+ - ``openlayer_audio`` (recognition only): the audio being recognized, as a
+ file path, raw bytes, or an ``Attachment``. The Speech SDK does not expose
+ the audio behind an ``AudioConfig``, so input audio is captured only when
+ passed explicitly.
+
+ Audio (``openlayer_audio`` and synthesized output) is attached only when
+ ``attachment_upload_enabled=True`` is set via ``openlayer.lib.init()`` or
+ ``configure()``; otherwise it is dropped and never leaves the process.
+
+ Parameters
+ ----------
+ client : SpeechRecognizer | TranslationRecognizer | SpeechSynthesizer
+ The Azure Speech client to patch.
+
+ Returns
+ -------
+ SpeechRecognizer | TranslationRecognizer | SpeechSynthesizer
+ The patched client.
+ """
+ if not HAVE_AZURE_SPEECH:
+ raise ImportError(
+ "azure-cognitiveservices-speech library is not installed. "
+ "Please install it with: pip install azure-cognitiveservices-speech"
+ )
+
+ if getattr(client, "_openlayer_patched", False) is True:
+ return client
+
+ if isinstance(client, (speechsdk.SpeechRecognizer, speechsdk.translation.TranslationRecognizer)):
+ for method in _RECOGNIZE_METHODS:
+ _wrap_method(client, method, _recognition_tracer(client))
+ elif isinstance(client, speechsdk.SpeechSynthesizer):
+ for method, input_key in _SYNTHESIZE_METHODS.items():
+ _wrap_method(client, method, _synthesis_tracer(client, input_key))
+ else:
+ raise ValueError(
+ "Invalid client. Please provide a SpeechRecognizer, TranslationRecognizer "
+ "or SpeechSynthesizer from azure.cognitiveservices.speech."
+ )
+
+ client._openlayer_patched = True
+ return client
+
+
+def _patch_azure_speech() -> None:
+ """Patch the Speech SDK client classes' ``__init__`` so every newly-constructed
+ recognizer and synthesizer is auto-traced. Idempotent."""
+ if not HAVE_AZURE_SPEECH:
+ return
+ # pylint: disable=import-outside-toplevel
+ from ._auto import _patch_class_init
+
+ for cls in _client_classes():
+ _patch_class_init(cls, trace_azure_speech)
+
+
+def _unpatch_azure_speech() -> None:
+ if not HAVE_AZURE_SPEECH:
+ return
+ # pylint: disable=import-outside-toplevel
+ from ._auto import _unpatch_class_init
+
+ for cls in _client_classes():
+ _unpatch_class_init(cls)
+
+
+def _client_classes() -> List[type]:
+ return [
+ speechsdk.SpeechRecognizer,
+ speechsdk.translation.TranslationRecognizer,
+ speechsdk.SpeechSynthesizer,
+ ]
+
+
+# ----------------------------- Wrapping ----------------------------- #
+
+
+def _wrap_method(
+ client: Any,
+ method_name: str,
+ make_step: Callable[[Any, tuple, Dict[str, Any], Any, float, float, Optional[str]], None],
+) -> None:
+ """Wrap ``client.`` so a step is traced when its result is ready.
+
+ ``_async`` methods return a ``ResultFuture``; the step is traced when
+ ``future.get()`` returns, the same way the Content Understanding tracer hooks
+ ``poller.result``.
+
+ The SDK implements each sync method as ``self._async(...).get()``, so
+ a traced sync call would also run through the traced async method. The sync
+ wrapper sets ``_in_traced_sync_call`` so the inner async call passes through
+ and the call is traced once.
+ """
+ original = getattr(client, method_name)
+ is_async = method_name.endswith("_async")
+
+ @wraps(original)
+ def traced(*args: Any, **kwargs: Any) -> Any:
+ if is_async and _in_traced_sync_call.get():
+ return original(*args, **kwargs)
+
+ inference_id = kwargs.pop("inference_id", None)
+ audio = kwargs.pop("openlayer_audio", None)
+ start_time = time.time()
+
+ if not is_async:
+ token = _in_traced_sync_call.set(True)
+ try:
+ result = original(*args, **kwargs)
+ finally:
+ _in_traced_sync_call.reset(token)
+ _safe_trace(make_step, result, args, kwargs, audio, start_time, inference_id)
+ return result
+
+ future = original(*args, **kwargs)
+ original_get = future.get
+
+ @wraps(original_get)
+ def traced_get() -> Any:
+ result = original_get()
+ _safe_trace(make_step, result, args, kwargs, audio, start_time, inference_id)
+ return result
+
+ future.get = traced_get
+ return future
+
+ setattr(client, method_name, traced)
+
+
+def _safe_trace(
+ make_step: Callable[..., None],
+ result: Any,
+ args: tuple,
+ kwargs: Dict[str, Any],
+ audio: Any,
+ start_time: float,
+ inference_id: Optional[str],
+) -> None:
+ try:
+ make_step(result, args, kwargs, audio, start_time, time.time(), inference_id)
+ # pylint: disable=broad-except
+ except Exception as e:
+ logger.error("Failed to trace the Azure Speech call with Openlayer. %s", e)
+
+
+def _recognition_tracer(client: Any) -> Callable[..., None]:
+ is_translation = isinstance(client, speechsdk.translation.TranslationRecognizer)
+
+ def make_step(
+ result: Any,
+ _args: tuple,
+ _kwargs: Dict[str, Any],
+ audio: Any,
+ start_time: float,
+ end_time: float,
+ inference_id: Optional[str],
+ ) -> None:
+ config = get_model_parameters(client)
+ secrets = _collect_secrets(client)
+ inputs: Dict[str, Any] = {"language": config.get("language")}
+ if is_translation:
+ inputs["targetLanguages"] = list(getattr(client, "target_languages", None) or [])
+ # A bare attachment (not an AudioContent item) directly under the value is
+ # the shape the Openlayer UI renders as an audio player.
+ audio_attachment = _audio_input_attachment(audio)
+ if audio_attachment is not None:
+ inputs["audio"] = audio_attachment
+
+ text = getattr(result, "text", None)
+ translations = getattr(result, "translations", None)
+ output: Union[str, Dict[str, Any], None] = text
+ if is_translation:
+ output = {"text": text, "translations": dict(translations or {})}
+
+ add_to_trace(
+ **create_trace_args(
+ name="Azure Speech Translation" if is_translation else "Azure Speech Recognition",
+ start_time=start_time,
+ end_time=end_time,
+ inputs=inputs,
+ output=output,
+ model=config.get("endpoint_id") or "speech-to-text",
+ model_parameters=config,
+ raw_output=_redact_secrets(getattr(result, "json", None) or "", secrets) or None,
+ metadata=get_result_metadata(result, secrets),
+ id=inference_id,
+ )
+ )
+
+ return make_step
+
+
+def _synthesis_tracer(client: Any, input_key: str) -> Callable[..., None]:
+ def make_step(
+ result: Any,
+ args: tuple,
+ kwargs: Dict[str, Any],
+ _audio: Any,
+ start_time: float,
+ end_time: float,
+ inference_id: Optional[str],
+ ) -> None:
+ config = get_model_parameters(client)
+ text = args[0] if args else kwargs.get(input_key)
+
+ audio_data: bytes = getattr(result, "audio_data", None) or b""
+ output: Dict[str, Any] = {
+ "audioDurationMs": _timedelta_ms(getattr(result, "audio_duration", None)),
+ "audioSizeBytes": len(audio_data),
+ }
+ if audio_data and _audio_upload_enabled():
+ audio_bytes, media_type, extension, audio_metadata = _describe_synthesis_audio(
+ audio_data, config.get("output_format")
+ )
+ attachment = Attachment.from_bytes(audio_bytes, name=f"synthesis.{extension}", media_type=media_type)
+ attachment.metadata.update(audio_metadata)
+ output["audio"] = attachment
+
+ add_to_trace(
+ **create_trace_args(
+ name="Azure Speech Synthesis",
+ start_time=start_time,
+ end_time=end_time,
+ inputs={input_key: text},
+ output=output,
+ model=config.get("voice") or "text-to-speech",
+ model_parameters=config,
+ # The result carries the audio bytes; never copy it into the trace.
+ raw_output=None,
+ metadata=get_result_metadata(result, _collect_secrets(client)),
+ id=inference_id,
+ )
+ )
+
+ return make_step
+
+
+# ----------------------------- Parsing ----------------------------- #
+
+
+def get_model_parameters(client: Any) -> Dict[str, Any]:
+ """Read an allowlist of non-secret settings from the client's properties.
+
+ The property collection also holds the subscription key and authorization
+ token, so it must never be dumped wholesale.
+ """
+ property_ids = {
+ "region": speechsdk.PropertyId.SpeechServiceConnection_Region,
+ "endpoint_id": speechsdk.PropertyId.SpeechServiceConnection_EndpointId,
+ "language": speechsdk.PropertyId.SpeechServiceConnection_RecoLanguage,
+ "recognition_mode": speechsdk.PropertyId.SpeechServiceConnection_RecoMode,
+ "voice": speechsdk.PropertyId.SpeechServiceConnection_SynthVoice,
+ "synthesis_language": speechsdk.PropertyId.SpeechServiceConnection_SynthLanguage,
+ "output_format": speechsdk.PropertyId.SpeechServiceConnection_SynthOutputFormat,
+ }
+ properties = getattr(client, "properties", None)
+ if properties is None:
+ return {}
+
+ parameters: Dict[str, Any] = {}
+ for name, property_id in property_ids.items():
+ try:
+ value = properties.get_property(property_id)
+ # pylint: disable=broad-except
+ except Exception:
+ continue
+ if value:
+ parameters[name] = value
+ return parameters
+
+
+def get_result_metadata(result: Any, secrets: Sequence[str] = ()) -> Dict[str, Any]:
+ """Extract the result ID, reason, timings and failure details.
+
+ ``errorDetails`` is passed through ``_redact_secrets``: the Speech SDK can
+ embed the endpoint URL (and any credential in its query string) in it.
+ """
+ metadata: Dict[str, Any] = {}
+
+ result_id = getattr(result, "result_id", None)
+ if result_id:
+ metadata["resultId"] = result_id
+
+ reason = getattr(result, "reason", None)
+ if reason is not None:
+ metadata["reason"] = _enum_name(reason)
+
+ offset = getattr(result, "offset", None)
+ if isinstance(offset, int):
+ metadata["offsetMs"] = offset / _TICKS_PER_MS
+ duration = getattr(result, "duration", None)
+ if isinstance(duration, int):
+ metadata["durationMs"] = duration / _TICKS_PER_MS
+
+ if reason is not None and _enum_name(reason) == "NoMatch":
+ no_match = getattr(result, "no_match_details", None)
+ if no_match is not None and getattr(no_match, "reason", None) is not None:
+ metadata["noMatchReason"] = _enum_name(no_match.reason)
+
+ if reason is not None and _enum_name(reason) == "Canceled":
+ cancellation = getattr(result, "cancellation_details", None)
+ if cancellation is not None:
+ # Recognition results expose ``code``; synthesis results ``error_code``.
+ code = getattr(cancellation, "error_code", None) or getattr(cancellation, "code", None)
+ metadata["cancellation"] = {
+ "reason": _enum_name(getattr(cancellation, "reason", None)),
+ "errorCode": _enum_name(code),
+ "errorDetails": _redact_secrets(getattr(cancellation, "error_details", None) or "", secrets) or None,
+ }
+
+ return metadata
+
+
+def _enum_name(value: Any) -> Optional[str]:
+ if value is None:
+ return None
+ return getattr(value, "name", None) or str(value)
+
+
+def _timedelta_ms(value: Any) -> Optional[float]:
+ if value is None:
+ return None
+ try:
+ return float(value.total_seconds()) * 1000
+ except AttributeError:
+ return None
+
+
+# ----------------------------- Audio ----------------------------- #
+
+
+def _audio_upload_enabled() -> bool:
+ return bool(tracer._resolve("attachment_upload_enabled")) # pylint: disable=protected-access
+
+
+def _audio_input_attachment(audio: Any) -> Optional[Attachment]:
+ """Build an ``Attachment`` for explicitly-passed input audio.
+
+ Returns None (and warns once) when attachment uploads are disabled, so audio
+ never reaches Openlayer unless the tracer is configured to upload it.
+ """
+ global _warned_audio_dropped # pylint: disable=global-statement
+
+ if audio is None:
+ return None
+ if not _audio_upload_enabled():
+ if not _warned_audio_dropped:
+ logger.warning(
+ "Openlayer: `openlayer_audio` was passed but attachment uploads are "
+ "disabled, so the audio is not attached. Enable them with "
+ "openlayer.lib.init(attachment_upload_enabled=True)."
+ )
+ _warned_audio_dropped = True
+ return None
+
+ if isinstance(audio, Attachment):
+ return audio
+ if isinstance(audio, (bytes, bytearray)):
+ return Attachment.from_bytes(bytes(audio), name="audio.wav", media_type="audio/wav")
+ if isinstance(audio, (str, Path)):
+ # Read the bytes instead of using Attachment.from_file(), which would
+ # record the absolute local path in the trace.
+ path = Path(audio).expanduser()
+ media_type = mimetypes.guess_type(str(path))[0] or "audio/wav"
+ return Attachment.from_bytes(path.read_bytes(), name=path.name, media_type=media_type)
+
+ logger.warning("Openlayer: unsupported `openlayer_audio` type %s; audio not attached.", type(audio).__name__)
+ return None
+
+
+# The WAVE format tags for headerless output that can be wrapped in a WAV container.
+_WAVE_FORMAT_TAGS = {"pcm": 1, "alaw": 6, "mulaw": 7}
+
+
+def _sniff_container(data: bytes) -> Optional[tuple]:
+ """(media type, extension) from the audio's own header, if recognizable.
+
+ No bare MPEG frame-sync check: headerless PCM samples can start with 0xFFEx.
+ """
+ if data[:4] == b"RIFF" and data[8:12] == b"WAVE":
+ return "audio/wav", "wav"
+ if data[:4] == b"OggS":
+ return "audio/ogg", "ogg"
+ if data[:4] == b"\x1a\x45\xdf\xa3":
+ return "audio/webm", "webm"
+ if data[:9] == b"#!AMR-WB\n":
+ return "audio/amr-wb", "amr"
+ if data[:3] == b"ID3":
+ return "audio/mpeg", "mp3"
+ return None
+
+
+def _parse_raw_format(normalized: str) -> Optional[tuple]:
+ """(sample rate Hz, bits per sample) from e.g. ``raw22050hz16bitmonopcm``."""
+ khz = re.search(r"(\d+)khz", normalized)
+ hz = re.search(r"(\d+)hz", normalized)
+ bits = re.search(r"(\d+)bit", normalized)
+ sample_rate = int(khz.group(1)) * 1000 if khz else int(hz.group(1)) if hz else None
+ if sample_rate is None or bits is None:
+ return None
+ return sample_rate, int(bits.group(1))
+
+
+def _wrap_in_wav(samples: bytes, format_tag: int, sample_rate: int, bits_per_sample: int, channels: int = 1) -> bytes:
+ """Wrap headerless samples in a WAV (RIFF) container."""
+ block_align = channels * bits_per_sample // 8
+ header = b"RIFF" + struct.pack(" tuple:
+ """Type the synthesized audio accurately: ``(bytes, media_type, extension, metadata)``.
+
+ The container is detected from the audio's own header (WAV, Ogg, WebM,
+ AMR-WB, ID3-tagged MP3); untagged MP3 is recognized by the format name.
+ Headerless PCM, mu-law and A-law (``raw-*`` formats) are wrapped in a WAV
+ container so they play, using the sample rate and bit depth in the format
+ name. Anything else (raw Opus frames, TrueSilk, Siren, G.722, unknown) is
+ labeled ``application/octet-stream`` rather than guessed.
+
+ ``output_format`` is the Python SDK's kebab-case name
+ (``raw-8khz-8bit-mono-mulaw``); the JS SDK's enum names normalize the same.
+ """
+ normalized = (output_format or "").lower().replace("-", "").replace("_", "")
+ metadata: Dict[str, Any] = {"outputFormat": output_format} if output_format else {}
+
+ container = _sniff_container(data)
+ if container:
+ return data, container[0], container[1], metadata
+ if normalized.endswith("mp3"):
+ # Azure's MP3 output is a bare frame stream (no ID3 tag).
+ return data, "audio/mpeg", "mp3", metadata
+
+ if normalized.startswith("raw"):
+ encoding = next((name for name in ("pcm", "mulaw", "alaw") if normalized.endswith(name)), None)
+ raw = _parse_raw_format(normalized)
+ if encoding and raw:
+ sample_rate, bits = raw
+ wav = _wrap_in_wav(data, _WAVE_FORMAT_TAGS[encoding], sample_rate, bits)
+ metadata.update({"encoding": encoding, "sampleRateHz": sample_rate, "wrappedInWav": True})
+ return wav, "audio/wav", "wav", metadata
+
+ return data, "application/octet-stream", "bin", metadata
+
+
+# ----------------------------- Redaction ----------------------------- #
+
+_REDACTED = "[REDACTED]"
+_SECRET_PROPERTY_NAMES = (
+ "SpeechServiceConnection_Key",
+ "SpeechServiceAuthorization_Token",
+ "SpeechServiceConnection_ProxyPassword",
+)
+_URL_PROPERTY_NAMES = ("SpeechServiceConnection_Endpoint", "SpeechServiceConnection_Host")
+# Values shorter than this are not scrubbed verbatim (too likely to collide with ordinary text).
+_MIN_SECRET_LENGTH = 6
+
+_URL_PATTERN = re.compile(r"\b(?:wss?|https?)://[^\s'\"<>`]+", re.IGNORECASE)
+_BEARER_PATTERN = re.compile(r"\b(Bearer|Basic)\s+[A-Za-z0-9\-._~+/]+=*", re.IGNORECASE)
+_SENSITIVE_PAIR_PATTERN = re.compile(
+ r"\b((?:ocp-apim-)?subscription[-_]?key|api[-_]?key|access[-_]?token|auth(?:orization)?|token|sig|signature"
+ r"|password|secret)(\s*[=:]\s*)(?!Bearer\b|Basic\b|\[REDACTED\])[^\s&,;'\")\]}>]+",
+ re.IGNORECASE,
+)
+
+
+def _collect_secrets(client: Any) -> List[str]:
+ """Secret values configured on a Speech client: the subscription key, auth
+ token and proxy password, plus every query value and userinfo credential in
+ its custom endpoint/host URL. Used only to scrub text; never recorded."""
+ properties = getattr(client, "properties", None)
+ if properties is None:
+ return []
+
+ def read(name: str) -> str:
+ try:
+ return properties.get_property(getattr(speechsdk.PropertyId, name)) or ""
+ # pylint: disable=broad-except
+ except Exception:
+ return ""
+
+ secrets = [read(name) for name in _SECRET_PROPERTY_NAMES]
+ for name in _URL_PROPERTY_NAMES:
+ raw = read(name)
+ if not raw:
+ continue
+ try:
+ parts = urlsplit(raw)
+ secrets.extend(value for _, value in parse_qsl(parts.query, keep_blank_values=True))
+ secrets.extend(unquote(part) for part in (parts.username or "", parts.password or ""))
+ except ValueError:
+ continue
+ return sorted({secret for secret in secrets if len(secret) >= _MIN_SECRET_LENGTH}, key=len, reverse=True)
+
+
+def _redact_url(match: "re.Match[str]") -> str:
+ url = match.group(0)
+ try:
+ parts = urlsplit(url)
+ netloc = parts.netloc
+ if "@" in netloc:
+ userinfo, host = netloc.rsplit("@", 1)
+ netloc = f"{_REDACTED}:{_REDACTED}@{host}" if ":" in userinfo else f"{_REDACTED}@{host}"
+ keys = list(dict.fromkeys(key for key, _ in parse_qsl(parts.query, keep_blank_values=True)))
+ query = "&".join(f"{quote(key)}={_REDACTED}" for key in keys)
+ return urlunsplit((parts.scheme, netloc, parts.path, query, ""))
+ except ValueError:
+ cut = min((i for i in (url.find("?"), url.find("#")) if i != -1), default=-1)
+ return url if cut == -1 else f"{url[:cut]}?{_REDACTED}"
+
+
+def _redact_secrets(text: Optional[str], secrets: Sequence[str] = ()) -> str:
+ """Remove credentials from free text the Speech SDK produces (error details).
+
+ The SDK can embed the full endpoint URL in connection errors, so a custom
+ endpoint carrying a token or signature in its query string would otherwise
+ be published. URLs keep scheme, host, path and parameter names (values and
+ userinfo are replaced); ``Bearer``/``Basic`` tokens and ``key=``/``token=``/
+ ``sig=``-style pairs are replaced; and every value in ``secrets`` (see
+ ``_collect_secrets``) is replaced wherever it appears, raw or URL-encoded.
+ """
+ if not text:
+ return ""
+ result = _URL_PATTERN.sub(_redact_url, text)
+ result = _BEARER_PATTERN.sub(lambda m: f"{m.group(1)} {_REDACTED}", result)
+ result = _SENSITIVE_PAIR_PATTERN.sub(lambda m: f"{m.group(1)}{m.group(2)}{_REDACTED}", result)
+ for secret in sorted(secrets, key=len, reverse=True):
+ if len(secret) < _MIN_SECRET_LENGTH:
+ continue
+ result = result.replace(secret, _REDACTED)
+ encoded = quote(secret, safe="")
+ if encoded != secret:
+ result = result.replace(encoded, _REDACTED)
+ return result
+
+
+# ----------------------------- Trace ----------------------------- #
+
+
+def create_trace_args(
+ name: str,
+ start_time: float,
+ end_time: float,
+ inputs: Dict[str, Any],
+ output: Any,
+ model: str,
+ model_parameters: Optional[Dict[str, Any]] = None,
+ raw_output: Optional[str] = None,
+ metadata: Optional[Dict[str, Any]] = None,
+ id: Optional[str] = None,
+) -> Dict[str, Any]:
+ """Returns a dictionary with the trace arguments."""
+ trace_args = {
+ "name": name,
+ "start_time": start_time,
+ "end_time": end_time,
+ "latency": (end_time - start_time) * 1000,
+ "inputs": inputs,
+ "output": output,
+ "tokens": 0,
+ "prompt_tokens": 0,
+ "completion_tokens": 0,
+ "model": model,
+ "model_parameters": model_parameters,
+ "raw_output": raw_output,
+ "metadata": metadata if metadata else {},
+ }
+ if id:
+ trace_args["id"] = id
+ return trace_args
+
+
+def add_to_trace(**kwargs: Any) -> None:
+ """Add an Azure Speech step to the trace."""
+ tracer.add_chat_completion_step_to_trace(**kwargs, provider=PROVIDER)
diff --git a/src/openlayer/lib/tracing/tracer.py b/src/openlayer/lib/tracing/tracer.py
index 4f89b3a8..171ccbd4 100644
--- a/src/openlayer/lib/tracing/tracer.py
+++ b/src/openlayer/lib/tracing/tracer.py
@@ -223,8 +223,9 @@ def init(
logged) instead of accumulating in memory. Defaults to 100.
auto_instrument: When truthy (default ``True``), detects every installed
supported LLM SDK (openai, anthropic, mistral, groq, gemini, oci,
- azure_content_understanding, litellm, portkey, google_adk) and patches
- it so newly-constructed clients are auto-traced. Pass ``False`` to skip
+ azure_content_understanding, azure_speech, litellm, portkey,
+ google_adk) and patches it so newly-constructed clients are
+ auto-traced. Pass ``False`` to skip
patching, or a list of names (e.g. ``["openai", "anthropic"]``) to
patch only that subset. This is a procedural argument — it is NOT
persisted across calls, and ``False`` does not unpatch (use
diff --git a/tests/test_azure_speech_integration.py b/tests/test_azure_speech_integration.py
new file mode 100644
index 00000000..36c4bebd
--- /dev/null
+++ b/tests/test_azure_speech_integration.py
@@ -0,0 +1,762 @@
+"""Tests for the Azure AI Speech (``azure-cognitiveservices-speech``) tracer.
+
+No network calls. Real ``SpeechRecognizer`` / ``SpeechSynthesizer`` objects are
+constructed with a fake key (the constructors don't connect), and their
+recognize/speak methods are replaced with stubs that return fake results *before*
+tracing, so the tracer wraps the stub. Steps are asserted by patching the tracer
+module's ``add_to_trace``, matching tests/test_google_genai_integration.py; the
+credential-leak test runs the real step path and serializes the whole trace.
+"""
+
+# The Speech SDK isn't installed in the lint env, and pytest fixtures hide autouse
+# functions from static analysis.
+# pyright: reportMissingImports=false, reportUnknownMemberType=false, reportUnknownVariableType=false, reportUnknownArgumentType=false, reportUnknownParameterType=false, reportMissingParameterType=false, reportUnusedFunction=false
+# pyright: reportMissingTypeStubs=false, reportAttributeAccessIssue=false, reportCallIssue=false
+
+import json
+from types import SimpleNamespace
+from typing import Any, Dict, List, Optional
+from datetime import timedelta
+from urllib.parse import quote
+from unittest.mock import patch
+
+import pytest
+
+speechsdk = pytest.importorskip("azure.cognitiveservices.speech")
+
+from openlayer.lib.integrations import azure_speech_tracer as ast
+from openlayer.lib.tracing.attachments import Attachment
+
+FAKE_KEY = "FAKE-AZURE-SPEECH-KEY-0123456789"
+# A minimal RIFF/WAVE header, as the SDK's default output format produces.
+RIFF_WAV = b"RIFF\x24\x00\x00\x00WAVEfmt fake-audio"
+
+
+# ------------------------------- fixtures ------------------------------- #
+@pytest.fixture(autouse=True)
+def _disable_publish(monkeypatch: pytest.MonkeyPatch) -> None:
+ """Keep every tracer publish path off."""
+ monkeypatch.setenv("OPENLAYER_DISABLE_PUBLISH", "true")
+ monkeypatch.setenv("OPENLAYER_API_KEY", "fake")
+
+ from openlayer.lib.tracing import tracer as _tracer
+
+ monkeypatch.setattr(_tracer, "_publish", False, raising=False)
+
+
+@pytest.fixture(autouse=True)
+def _reset_class_patches():
+ """Undo any class-level ``__init__`` patch so the idempotency marker doesn't
+ leak between tests."""
+ yield
+ ast._unpatch_azure_speech()
+
+
+# ------------------------------- helpers ------------------------------- #
+def _speech_config(**overrides: Any) -> Any:
+ config = speechsdk.SpeechConfig(subscription=FAKE_KEY, region="eastus")
+ for key, value in overrides.items():
+ setattr(config, key, value)
+ return config
+
+
+def _push_stream_audio() -> Any:
+ return speechsdk.audio.AudioConfig(stream=speechsdk.audio.PushAudioInputStream())
+
+
+def _make_recognizer(language: str = "en-US") -> Any:
+ return speechsdk.SpeechRecognizer(
+ speech_config=_speech_config(), audio_config=_push_stream_audio(), language=language
+ )
+
+
+def _make_synthesizer(voice: str = "en-US-JennyNeural") -> Any:
+ return speechsdk.SpeechSynthesizer(
+ speech_config=_speech_config(speech_synthesis_voice_name=voice), audio_config=None
+ )
+
+
+def _recognition_result(
+ text: str = "Hello world.",
+ reason: Any = None,
+ offset: int = 5_000_000,
+ duration: int = 12_300_000,
+ no_match_details: Any = None,
+ cancellation_details: Any = None,
+ translations: Optional[Dict[str, str]] = None,
+) -> SimpleNamespace:
+ result = SimpleNamespace(
+ text=text,
+ reason=reason if reason is not None else speechsdk.ResultReason.RecognizedSpeech,
+ result_id="res-123",
+ offset=offset,
+ duration=duration,
+ json=json.dumps({"DisplayText": text, "RecognitionStatus": "Success"}),
+ no_match_details=no_match_details,
+ cancellation_details=cancellation_details,
+ )
+ if translations is not None:
+ result.translations = translations
+ return result
+
+
+def _synthesis_result(
+ audio: bytes = b"RIFF....WAVEfmt fake-audio",
+ reason: Any = None,
+ cancellation_details: Any = None,
+) -> SimpleNamespace:
+ return SimpleNamespace(
+ audio_data=audio,
+ audio_duration=timedelta(milliseconds=1500),
+ reason=reason if reason is not None else speechsdk.ResultReason.SynthesizingAudioCompleted,
+ result_id="syn-456",
+ cancellation_details=cancellation_details,
+ )
+
+
+class _Future:
+ """Stand-in for ``speechsdk.ResultFuture``."""
+
+ def __init__(self, result: Any) -> None:
+ self._result = result
+
+ def get(self) -> Any:
+ return self._result
+
+
+def _stub(client: Any, method: str, result: Any) -> List[Dict[str, Any]]:
+ """Replace ``client.`` with a stub; returns the recorded calls."""
+ calls: List[Dict[str, Any]] = []
+
+ def _impl(*args: Any, **kwargs: Any) -> Any:
+ calls.append({"args": args, "kwargs": kwargs})
+ return result
+
+ setattr(client, method, _impl)
+ return calls
+
+
+# ------------------------------- dependency handling ------------------------------- #
+class TestDependency:
+ def test_module_exposes_availability_flag(self) -> None:
+ assert ast.HAVE_AZURE_SPEECH is True
+
+ def test_raises_helpful_import_error_without_sdk(self) -> None:
+ with patch.object(ast, "HAVE_AZURE_SPEECH", False):
+ with pytest.raises(ImportError) as exc_info:
+ ast.trace_azure_speech(object())
+ assert "pip install azure-cognitiveservices-speech" in str(exc_info.value)
+
+ def test_rejects_unsupported_objects(self) -> None:
+ with pytest.raises(ValueError):
+ ast.trace_azure_speech(object())
+
+
+# ------------------------------- recognition ------------------------------- #
+class TestRecognition:
+ def test_recognize_once_emits_step(self) -> None:
+ recognizer = _make_recognizer(language="pt-BR")
+ _stub(recognizer, "recognize_once", _recognition_result("Olá mundo."))
+ ast.trace_azure_speech(recognizer)
+
+ with patch.object(ast, "add_to_trace") as mock_add:
+ result = recognizer.recognize_once()
+
+ assert result.text == "Olá mundo."
+ mock_add.assert_called_once()
+ kwargs = mock_add.call_args.kwargs
+ assert kwargs["name"] == "Azure Speech Recognition"
+ assert kwargs["output"] == "Olá mundo."
+ assert kwargs["inputs"] == {"language": "pt-BR"}
+ assert kwargs["model"] == "speech-to-text"
+ assert kwargs["model_parameters"]["region"] == "eastus"
+ assert kwargs["model_parameters"]["language"] == "pt-BR"
+ assert kwargs["latency"] >= 0
+ assert kwargs["start_time"] <= kwargs["end_time"]
+ metadata = kwargs["metadata"]
+ assert metadata["reason"] == "RecognizedSpeech"
+ assert metadata["resultId"] == "res-123"
+ # 100-ns ticks -> ms
+ assert metadata["offsetMs"] == 500.0
+ assert metadata["durationMs"] == 1230.0
+ assert "RecognitionStatus" in kwargs["raw_output"]
+
+ def test_custom_endpoint_id_is_the_model(self) -> None:
+ config = _speech_config()
+ config.endpoint_id = "custom-model-abc"
+ recognizer = speechsdk.SpeechRecognizer(speech_config=config, audio_config=_push_stream_audio())
+ _stub(recognizer, "recognize_once", _recognition_result())
+ ast.trace_azure_speech(recognizer)
+
+ with patch.object(ast, "add_to_trace") as mock_add:
+ recognizer.recognize_once()
+
+ assert mock_add.call_args.kwargs["model"] == "custom-model-abc"
+
+ def test_no_match_is_recorded(self) -> None:
+ recognizer = _make_recognizer()
+ no_match = SimpleNamespace(reason=speechsdk.NoMatchReason.InitialSilenceTimeout)
+ _stub(
+ recognizer,
+ "recognize_once",
+ _recognition_result(text="", reason=speechsdk.ResultReason.NoMatch, no_match_details=no_match),
+ )
+ ast.trace_azure_speech(recognizer)
+
+ with patch.object(ast, "add_to_trace") as mock_add:
+ recognizer.recognize_once()
+
+ metadata = mock_add.call_args.kwargs["metadata"]
+ assert metadata["reason"] == "NoMatch"
+ assert metadata["noMatchReason"] == "InitialSilenceTimeout"
+
+ def test_cancellation_is_recorded_as_error(self) -> None:
+ recognizer = _make_recognizer()
+ cancellation = SimpleNamespace(
+ reason=speechsdk.CancellationReason.Error,
+ code=speechsdk.CancellationErrorCode.AuthenticationFailure,
+ error_details="WebSocket upgrade failed: Authentication error (401).",
+ )
+ _stub(
+ recognizer,
+ "recognize_once",
+ _recognition_result(text="", reason=speechsdk.ResultReason.Canceled, cancellation_details=cancellation),
+ )
+ ast.trace_azure_speech(recognizer)
+
+ with patch.object(ast, "add_to_trace") as mock_add:
+ recognizer.recognize_once()
+
+ metadata = mock_add.call_args.kwargs["metadata"]
+ assert metadata["reason"] == "Canceled"
+ assert metadata["cancellation"] == {
+ "reason": "Error",
+ "errorCode": "AuthenticationFailure",
+ "errorDetails": "WebSocket upgrade failed: Authentication error (401).",
+ }
+
+ def test_recognize_once_async_traces_on_get(self) -> None:
+ recognizer = _make_recognizer()
+ _stub(recognizer, "recognize_once_async", _Future(_recognition_result("async text")))
+ ast.trace_azure_speech(recognizer)
+
+ with patch.object(ast, "add_to_trace") as mock_add:
+ future = recognizer.recognize_once_async()
+ mock_add.assert_not_called()
+ result = future.get()
+
+ assert result.text == "async text"
+ mock_add.assert_called_once()
+ assert mock_add.call_args.kwargs["output"] == "async text"
+
+ def test_translation_output_includes_translations(self) -> None:
+ config = speechsdk.translation.SpeechTranslationConfig(subscription=FAKE_KEY, region="eastus")
+ config.speech_recognition_language = "en-US"
+ config.add_target_language("de")
+ recognizer = speechsdk.translation.TranslationRecognizer(
+ translation_config=config, audio_config=_push_stream_audio()
+ )
+ _stub(
+ recognizer,
+ "recognize_once",
+ _recognition_result(
+ "Hello.", reason=speechsdk.ResultReason.TranslatedSpeech, translations={"de": "Hallo."}
+ ),
+ )
+ ast.trace_azure_speech(recognizer)
+
+ with patch.object(ast, "add_to_trace") as mock_add:
+ recognizer.recognize_once()
+
+ kwargs = mock_add.call_args.kwargs
+ assert kwargs["name"] == "Azure Speech Translation"
+ assert kwargs["output"] == {"text": "Hello.", "translations": {"de": "Hallo."}}
+ assert kwargs["inputs"]["language"] == "en-US"
+ assert kwargs["inputs"]["targetLanguages"] == ["de"]
+
+ def test_openlayer_kwargs_never_reach_the_sdk(self) -> None:
+ recognizer = _make_recognizer()
+ calls = _stub(recognizer, "recognize_once", _recognition_result())
+ ast.trace_azure_speech(recognizer)
+
+ with patch.object(ast, "add_to_trace") as mock_add:
+ recognizer.recognize_once(inference_id="abc-123", openlayer_audio=b"raw-audio")
+
+ assert calls[0]["kwargs"] == {}
+ assert mock_add.call_args.kwargs["id"] == "abc-123"
+
+ def test_provider_has_no_spaces(self) -> None:
+ """Cost lookup matches provider against a slug exactly; spaces never match."""
+ assert " " not in ast.PROVIDER
+
+ recognizer = _make_recognizer()
+ _stub(recognizer, "recognize_once", _recognition_result())
+ ast.trace_azure_speech(recognizer)
+
+ from openlayer.lib.tracing import tracer as _tracer
+
+ with patch.object(_tracer, "add_chat_completion_step_to_trace") as mock_step:
+ recognizer.recognize_once()
+
+ assert mock_step.call_args.kwargs["provider"] == "Azure_Speech"
+
+
+# ------------------------------- synthesis ------------------------------- #
+class TestSynthesis:
+ @pytest.mark.parametrize("method,input_key", [("speak_text", "text"), ("speak_ssml", "ssml")])
+ def test_speak_emits_step(self, method: str, input_key: str) -> None:
+ synthesizer = _make_synthesizer()
+ _stub(synthesizer, method, _synthesis_result())
+ ast.trace_azure_speech(synthesizer)
+
+ with patch.object(ast, "add_to_trace") as mock_add:
+ result = getattr(synthesizer, method)("Hi there")
+
+ assert result.result_id == "syn-456"
+ kwargs = mock_add.call_args.kwargs
+ assert kwargs["name"] == "Azure Speech Synthesis"
+ assert kwargs["inputs"] == {input_key: "Hi there"}
+ assert kwargs["model"] == "en-US-JennyNeural"
+ assert kwargs["model_parameters"]["voice"] == "en-US-JennyNeural"
+ assert kwargs["output"] == {"audioDurationMs": 1500.0, "audioSizeBytes": 26}
+ assert kwargs["metadata"]["reason"] == "SynthesizingAudioCompleted"
+ # audio bytes must never land in raw_output
+ assert kwargs["raw_output"] is None
+
+ def test_speak_text_accepts_keyword_text(self) -> None:
+ synthesizer = _make_synthesizer()
+ _stub(synthesizer, "speak_text", _synthesis_result())
+ ast.trace_azure_speech(synthesizer)
+
+ with patch.object(ast, "add_to_trace") as mock_add:
+ synthesizer.speak_text(text="keyword text")
+
+ assert mock_add.call_args.kwargs["inputs"] == {"text": "keyword text"}
+
+ def test_speak_text_async_traces_on_get(self) -> None:
+ synthesizer = _make_synthesizer()
+ _stub(synthesizer, "speak_text_async", _Future(_synthesis_result()))
+ ast.trace_azure_speech(synthesizer)
+
+ with patch.object(ast, "add_to_trace") as mock_add:
+ future = synthesizer.speak_text_async("Hi")
+ mock_add.assert_not_called()
+ future.get()
+
+ mock_add.assert_called_once()
+ assert mock_add.call_args.kwargs["inputs"] == {"text": "Hi"}
+
+ def test_synthesis_cancellation_is_recorded(self) -> None:
+ synthesizer = _make_synthesizer()
+ cancellation = SimpleNamespace(
+ reason=speechsdk.CancellationReason.Error,
+ error_code=speechsdk.CancellationErrorCode.ConnectionFailure,
+ error_details="Connection failed.",
+ )
+ _stub(
+ synthesizer,
+ "speak_text",
+ _synthesis_result(audio=b"", reason=speechsdk.ResultReason.Canceled, cancellation_details=cancellation),
+ )
+ ast.trace_azure_speech(synthesizer)
+
+ with patch.object(ast, "add_to_trace") as mock_add:
+ synthesizer.speak_text("Hi")
+
+ assert mock_add.call_args.kwargs["metadata"]["cancellation"] == {
+ "reason": "Error",
+ "errorCode": "ConnectionFailure",
+ "errorDetails": "Connection failed.",
+ }
+
+
+# ------------------------------- audio capture ------------------------------- #
+class TestAudioCapture:
+ def test_no_audio_attached_when_uploads_disabled(self) -> None:
+ recognizer = _make_recognizer()
+ _stub(recognizer, "recognize_once", _recognition_result())
+ synthesizer = _make_synthesizer()
+ _stub(synthesizer, "speak_text", _synthesis_result())
+ ast.trace_azure_speech(recognizer)
+ ast.trace_azure_speech(synthesizer)
+
+ with patch.object(ast, "_audio_upload_enabled", return_value=False), patch.object(
+ ast, "add_to_trace"
+ ) as mock_add:
+ recognizer.recognize_once(openlayer_audio=b"raw-audio")
+ synthesizer.speak_text("Hi")
+
+ recognition, synthesis = (call.kwargs for call in mock_add.call_args_list)
+ assert "audio" not in recognition["inputs"]
+ assert "audio" not in synthesis["output"]
+
+ def test_input_audio_attached_when_uploads_enabled(self, tmp_path: Any) -> None:
+ wav = tmp_path / "caller.wav"
+ wav.write_bytes(b"RIFF fake wav")
+ recognizer = _make_recognizer()
+ _stub(recognizer, "recognize_once", _recognition_result())
+ ast.trace_azure_speech(recognizer)
+
+ with patch.object(ast, "_audio_upload_enabled", return_value=True), patch.object(
+ ast, "add_to_trace"
+ ) as mock_add:
+ recognizer.recognize_once(openlayer_audio=str(wav))
+
+ audio = mock_add.call_args.kwargs["inputs"]["audio"]
+ assert isinstance(audio, Attachment)
+ assert audio.name == "caller.wav"
+ assert audio.media_type in ("audio/wav", "audio/x-wav")
+ # Read as bytes: the local path must not be recorded in the trace.
+ assert audio.file_path is None
+ assert audio.get_bytes() == b"RIFF fake wav"
+
+ def test_output_audio_attached_when_uploads_enabled(self) -> None:
+ synthesizer = _make_synthesizer()
+ _stub(synthesizer, "speak_text", _synthesis_result(audio=RIFF_WAV))
+ ast.trace_azure_speech(synthesizer)
+
+ with patch.object(ast, "_audio_upload_enabled", return_value=True), patch.object(
+ ast, "add_to_trace"
+ ) as mock_add:
+ synthesizer.speak_text("Hi")
+
+ audio = mock_add.call_args.kwargs["output"]["audio"]
+ assert isinstance(audio, Attachment)
+ assert audio.media_type == "audio/wav"
+ assert audio.get_bytes() == RIFF_WAV
+ # Never inlined into the trace JSON; the uploader sends it separately.
+ assert audio.data_base64 is None
+
+ def test_audio_shape_is_renderable_by_the_frontend(self) -> None:
+ """Mirror of the frontend's ``parseMultimodalContent`` (lib/util/multimodal.ts):
+ it renders a value that IS an attachment ({storageUri, mediaType}) or an
+ object whose direct values are attachments. A typed ``{type, attachment}``
+ item nested inside a dict is NOT found, so audio must be a bare attachment
+ directly under the input/output value."""
+ from openlayer.lib.tracing import tracer as _tracer
+ from openlayer.lib.tracing.attachment_uploader import find_attachments
+
+ def is_attachment_like(value: Any) -> bool:
+ return isinstance(value, dict) and isinstance(value.get("storageUri"), str)
+
+ def renders_audio(value: Any) -> bool:
+ if is_attachment_like(value):
+ return True
+ return isinstance(value, dict) and any(
+ is_attachment_like(v) and str(v.get("mediaType", "")).startswith("audio/") for v in value.values()
+ )
+
+ recognizer = _make_recognizer()
+ _stub(recognizer, "recognize_once", _recognition_result())
+ synthesizer = _make_synthesizer()
+ _stub(synthesizer, "speak_text", _synthesis_result())
+ ast.trace_azure_speech(recognizer)
+ ast.trace_azure_speech(synthesizer)
+
+ with patch.object(ast, "_audio_upload_enabled", return_value=True):
+ with _tracer.create_step(name="voice turn") as root:
+ recognizer.recognize_once(openlayer_audio=b"RIFF in")
+ synthesizer.speak_text("Hi")
+
+ recognition, synthesis = root.steps
+ # The uploader must still find both attachments ...
+ for step in (recognition, synthesis):
+ found = find_attachments(step.inputs) + find_attachments(step.output)
+ assert len(found) == 1
+ found[0].storage_uri = "s3://bucket/attachments/x.wav" # simulate upload
+
+ # ... and the serialized shape must be what the frontend parses.
+ recognition_dict, synthesis_dict = recognition.to_dict(), synthesis.to_dict()
+ assert renders_audio(recognition_dict["inputs"]["audio"]) # inputs render per key
+ assert renders_audio(synthesis_dict["output"]) # output renders as one value
+
+ def test_audio_upload_setting_is_read_from_tracer_config(self) -> None:
+ from openlayer.lib.tracing import tracer as _tracer
+
+ with patch.object(_tracer, "_resolve", return_value=True) as mock_resolve:
+ assert ast._audio_upload_enabled() is True
+ mock_resolve.assert_called_with("attachment_upload_enabled")
+
+
+# ------------------------------- credential safety ------------------------------- #
+class TestAudioEncoding:
+ """Synthesized audio is labeled by its actual encoding (review finding)."""
+
+ @pytest.mark.parametrize(
+ "output_format,data,media_type,extension",
+ [
+ ("", RIFF_WAV, "audio/wav", "wav"),
+ ("audio-24khz-48kbitrate-mono-mp3", b"\xff\xf3\x44\xc4\x00", "audio/mpeg", "mp3"),
+ ("audio-24khz-48kbitrate-mono-mp3", b"ID3\x04rest", "audio/mpeg", "mp3"),
+ ("ogg-24khz-16bit-mono-opus", b"OggS\x00\x02", "audio/ogg", "ogg"),
+ ("webm-24khz-16bit-mono-opus", b"\x1a\x45\xdf\xa3\x01", "audio/webm", "webm"),
+ ("amr-wb-16000hz", b"#!AMR-WB\n\x04", "audio/amr-wb", "amr"),
+ ("amr-wb-16000hz", RIFF_WAV, "audio/wav", "wav"), # the service actually returns RIFF here
+ ("raw-24khz-16bit-mono-truesilk", b"\x0e\x00\xa7\x54", "application/octet-stream", "bin"),
+ ("audio-24khz-16bit-48kbps-mono-opus", b"\x68\x0b\xeb", "application/octet-stream", "bin"),
+ ("g722-16khz-64kbps", b"\x7a\xde", "application/octet-stream", "bin"),
+ ("", b"\x01\x02\x03", "application/octet-stream", "bin"), # unknown: not guessed as WAV
+ ],
+ )
+ def test_labels_by_actual_encoding(self, output_format: str, data: bytes, media_type: str, extension: str) -> None:
+ _, got_type, got_ext, _ = ast._describe_synthesis_audio(data, output_format)
+ assert (got_type, got_ext) == (media_type, extension)
+
+ @pytest.mark.parametrize(
+ "output_format,format_tag,sample_rate,bits,encoding",
+ [
+ ("raw-8khz-8bit-mono-mulaw", 7, 8000, 8, "mulaw"),
+ ("raw-8khz-8bit-mono-alaw", 6, 8000, 8, "alaw"),
+ ("raw-22050hz-16bit-mono-pcm", 1, 22050, 16, "pcm"),
+ ("raw-48khz-16bit-mono-pcm", 1, 48000, 16, "pcm"),
+ ],
+ )
+ def test_wraps_headerless_audio_in_wav(
+ self, output_format: str, format_tag: int, sample_rate: int, bits: int, encoding: str
+ ) -> None:
+ import struct
+
+ samples = b"\x01\x02\x03\x04"
+ data, media_type, _, metadata = ast._describe_synthesis_audio(samples, output_format)
+ assert media_type == "audio/wav"
+ assert data[:4] == b"RIFF" and data[8:16] == b"WAVEfmt " and data[36:40] == b"data"
+ assert struct.unpack(" None:
+ import wave
+
+ data, _, _, _ = ast._describe_synthesis_audio(b"\x00\x01" * 1600, "raw-16khz-16bit-mono-pcm")
+ path = tmp_path / "out.wav"
+ path.write_bytes(data)
+ with wave.open(str(path)) as wav:
+ assert (wav.getframerate(), wav.getsampwidth(), wav.getnchannels(), wav.getnframes()) == (16000, 2, 1, 1600)
+
+ def test_raw_pcm_that_starts_like_an_mpeg_frame_is_not_mp3(self) -> None:
+ # Real raw PCM from the service starts with bytes like ff ff fe ff.
+ _, media_type, _, metadata = ast._describe_synthesis_audio(b"\xff\xff\xfe\xff", "raw-16khz-16bit-mono-pcm")
+ assert media_type == "audio/wav"
+ assert metadata["wrappedInWav"] is True
+
+ def test_synthesis_step_carries_the_wrapped_audio(self) -> None:
+ config = _speech_config()
+ config.set_speech_synthesis_output_format(speechsdk.SpeechSynthesisOutputFormat.Raw8Khz8BitMonoMULaw)
+ synthesizer = speechsdk.SpeechSynthesizer(speech_config=config, audio_config=None)
+ _stub(synthesizer, "speak_text", _synthesis_result(audio=b"\xff\x7f\x80\x00"))
+ ast.trace_azure_speech(synthesizer)
+
+ with patch.object(ast, "_audio_upload_enabled", return_value=True), patch.object(
+ ast, "add_to_trace"
+ ) as mock_add:
+ synthesizer.speak_text("Hi")
+
+ audio = mock_add.call_args.kwargs["output"]["audio"]
+ assert (audio.name, audio.media_type) == ("synthesis.wav", "audio/wav")
+ assert audio.get_bytes()[:4] == b"RIFF"
+ assert audio.metadata["encoding"] == "mulaw"
+
+
+class TestErrorRedaction:
+ """The Speech SDK embeds the endpoint URL (with its query string) in connection
+ errors, so credentials in a custom endpoint must be redacted (review finding)."""
+
+ ENDPOINT_TOKEN = "SECRET-ENDPOINT-TOKEN"
+
+ def test_real_sdk_connection_failure_is_redacted(self) -> None:
+ """Offline: the SDK fails to connect to a refused local port and reports the full URL."""
+ from openlayer.lib.tracing import tracer as _tracer
+
+ config = speechsdk.SpeechConfig(
+ subscription=FAKE_KEY,
+ endpoint=f"ws://127.0.0.1:1/speech/recognition/v1?token={self.ENDPOINT_TOKEN}&sig=SIGSECRET99",
+ )
+ stream = speechsdk.audio.PushAudioInputStream()
+ stream.write(b"\x00" * 3200)
+ stream.close()
+ recognizer = speechsdk.SpeechRecognizer(
+ speech_config=config, audio_config=speechsdk.audio.AudioConfig(stream=stream), language="en-US"
+ )
+ ast.trace_azure_speech(recognizer)
+
+ with _tracer.create_step(name="turn") as root:
+ result = recognizer.recognize_once()
+
+ # The raw SDK text really does contain the token (the caller still gets it).
+ assert self.ENDPOINT_TOKEN in result.cancellation_details.error_details
+ serialized = json.dumps(root.to_dict())
+ assert self.ENDPOINT_TOKEN not in serialized
+ assert "SIGSECRET99" not in serialized
+ assert FAKE_KEY not in serialized
+ details = root.steps[0].metadata["cancellation"]["errorDetails"]
+ assert "token=[REDACTED]" in details
+ assert "Connection failed" in details
+
+ def test_js_style_unable_to_contact_server_message(self) -> None:
+ text = "Unable to contact server. StatusCode: 1006, wss://h.example.com/v1?token=abc123def&x=1 Reason: 401"
+ assert ast._redact_secrets(text) == (
+ "Unable to contact server. StatusCode: 1006, wss://h.example.com/v1?token=[REDACTED]&x=[REDACTED] Reason: 401"
+ )
+
+ def test_bearer_pairs_userinfo_and_known_secrets(self) -> None:
+ assert ast._redact_secrets("Authorization: Bearer eyJ.a.b") == "Authorization: Bearer [REDACTED]"
+ assert ast._redact_secrets("failed (subscription-key=abcdef123, sig: ZZZZZZ)") == (
+ "failed (subscription-key=[REDACTED], sig: [REDACTED])"
+ )
+ assert (
+ ast._redact_secrets("wss://user:pa55word@h.example.com/p") == "wss://[REDACTED]:[REDACTED]@h.example.com/p"
+ )
+ secret = "k3y/with+special=chars"
+ assert ast._redact_secrets(f"{secret} and {quote(secret, safe='')}", [secret]) == "[REDACTED] and [REDACTED]"
+
+ def test_ordinary_error_text_is_untouched(self) -> None:
+ text = "Unsupported voice xx-XX-NoSuchVoiceNeural. websocket error code: 1007"
+ assert ast._redact_secrets(text, ["short"]) == text
+
+ def test_collect_secrets_reads_key_and_endpoint_query(self) -> None:
+ config = speechsdk.SpeechConfig(
+ subscription=FAKE_KEY, endpoint="wss://h.example.com/p?token=endpoint-token-1&x=1"
+ )
+ recognizer = speechsdk.SpeechRecognizer(speech_config=config, audio_config=_push_stream_audio())
+ secrets = ast._collect_secrets(recognizer)
+ assert FAKE_KEY in secrets and "endpoint-token-1" in secrets
+ assert "1" not in secrets
+
+
+class TestCredentialSafety:
+ def test_key_never_appears_in_serialized_trace(self) -> None:
+ """Runs the real step path (no add_to_trace mock) and serializes the trace."""
+ from openlayer.lib.tracing import tracer as _tracer
+
+ recognizer = _make_recognizer()
+ _stub(recognizer, "recognize_once", _recognition_result())
+ synthesizer = _make_synthesizer()
+ _stub(synthesizer, "speak_ssml", _synthesis_result())
+ ast.trace_azure_speech(recognizer)
+ ast.trace_azure_speech(synthesizer)
+
+ with _tracer.create_step(name="voice turn") as root:
+ recognizer.recognize_once()
+ synthesizer.speak_ssml("Hi")
+
+ serialized = json.dumps(root.to_dict())
+ assert len(root.steps) == 2
+ assert FAKE_KEY not in serialized
+
+
+# ------------------------------- sync -> async delegation ------------------------------- #
+class TestSyncDelegatesToAsync:
+ """The SDK implements ``recognize_once()`` as ``self.recognize_once_async().get()``
+ (and ``speak_*`` likewise), so a sync call runs through BOTH wrapped methods.
+ Only the async method is stubbed here; the real sync method delegates to it."""
+
+ def test_recognize_once_emits_a_single_step(self) -> None:
+ recognizer = _make_recognizer()
+ _stub(recognizer, "recognize_once_async", _Future(_recognition_result("once")))
+ ast.trace_azure_speech(recognizer)
+
+ with patch.object(ast, "add_to_trace") as mock_add:
+ result = recognizer.recognize_once(inference_id="abc-123")
+
+ assert result.text == "once"
+ assert mock_add.call_count == 1, "sync call must not also trace the inner async call"
+ assert mock_add.call_args.kwargs["id"] == "abc-123"
+
+ @pytest.mark.parametrize("method,input_key", [("speak_text", "text"), ("speak_ssml", "ssml")])
+ def test_speak_emits_a_single_step(self, method: str, input_key: str) -> None:
+ synthesizer = _make_synthesizer()
+ _stub(synthesizer, f"{method}_async", _Future(_synthesis_result()))
+ ast.trace_azure_speech(synthesizer)
+
+ with patch.object(ast, "add_to_trace") as mock_add:
+ getattr(synthesizer, method)("Hi")
+
+ assert mock_add.call_count == 1
+ assert mock_add.call_args.kwargs["inputs"] == {input_key: "Hi"}
+
+ def test_async_call_after_sync_call_is_still_traced(self) -> None:
+ recognizer = _make_recognizer()
+ _stub(recognizer, "recognize_once_async", _Future(_recognition_result()))
+ ast.trace_azure_speech(recognizer)
+
+ with patch.object(ast, "add_to_trace") as mock_add:
+ recognizer.recognize_once()
+ recognizer.recognize_once_async().get()
+
+ assert mock_add.call_count == 2
+
+
+# ------------------------------- robustness ------------------------------- #
+class TestRobustness:
+ def test_tracing_failure_does_not_break_the_call(self) -> None:
+ recognizer = _make_recognizer()
+ _stub(recognizer, "recognize_once", _recognition_result("still works"))
+ ast.trace_azure_speech(recognizer)
+
+ with patch.object(ast, "add_to_trace", side_effect=RuntimeError("boom")):
+ result = recognizer.recognize_once()
+
+ assert result.text == "still works"
+
+ def test_sdk_exception_propagates(self) -> None:
+ recognizer = _make_recognizer()
+
+ def _raise() -> Any:
+ raise RuntimeError("sdk failure")
+
+ recognizer.recognize_once = _raise
+ ast.trace_azure_speech(recognizer)
+
+ with patch.object(ast, "add_to_trace") as mock_add, pytest.raises(RuntimeError, match="sdk failure"):
+ recognizer.recognize_once()
+ mock_add.assert_not_called()
+
+
+# ------------------------------- idempotency & auto-instrument ------------------------------- #
+class TestIdempotency:
+ def test_double_patch_wraps_once(self) -> None:
+ recognizer = _make_recognizer()
+ _stub(recognizer, "recognize_once", _recognition_result())
+
+ ast.trace_azure_speech(recognizer)
+ first = recognizer.recognize_once
+ assert ast.trace_azure_speech(recognizer) is recognizer
+ assert recognizer.recognize_once is first
+
+ with patch.object(ast, "add_to_trace") as mock_add:
+ recognizer.recognize_once()
+ assert mock_add.call_count == 1
+
+ def test_auto_instrument_patches_new_clients(self) -> None:
+ ast._patch_azure_speech()
+ recognizer = _make_recognizer()
+ synthesizer = _make_synthesizer()
+
+ assert getattr(recognizer, "_openlayer_patched", False) is True
+ assert getattr(synthesizer, "_openlayer_patched", False) is True
+
+ def test_unpatch_restores_constructors(self) -> None:
+ ast._patch_azure_speech()
+ ast._unpatch_azure_speech()
+ recognizer = _make_recognizer()
+ assert getattr(recognizer, "_openlayer_patched", False) is False
+
+ def test_registered_for_auto_instrument(self) -> None:
+ from openlayer.lib.integrations._auto import _REGISTRY_BY_NAME
+
+ spec = _REGISTRY_BY_NAME["azure_speech"]
+ assert spec.probe == "azure.cognitiveservices.speech"
+
+ def test_public_alias(self) -> None:
+ from openlayer.lib import trace_azure_speech
+
+ recognizer = _make_recognizer()
+ assert trace_azure_speech(recognizer) is recognizer
+ assert getattr(recognizer, "_openlayer_patched", False) is True
diff --git a/tests/test_integration_conditional_imports.py b/tests/test_integration_conditional_imports.py
index adb5bf9d..848e0017 100644
--- a/tests/test_integration_conditional_imports.py
+++ b/tests/test_integration_conditional_imports.py
@@ -35,6 +35,7 @@
"langchain_callback": ["langchain", "langchain_core", "langchain_community"],
"litellm_tracer": ["litellm"],
"portkey_tracer": ["portkey_ai"],
+ "azure_speech_tracer": ["azure"],
}
# Expected patterns for integration modules