Add Nemotron-ASR streaming inference to Python SDK (#612)

## Add Nemotron-ASR streaming inference to Python SDK


### Description

Adds real-time audio streaming support to the Foundry Local Python SDK,
enabling live microphone-to-text transcription via ONNX Runtime GenAI's
StreamingProcessor API (Nemotron ASR).

This is the Python port of C# PR #485 with full feature parity. The
existing `AudioClient` only supports file-based transcription. This PR
introduces `LiveAudioTranscriptionSession` that accepts continuous PCM
audio chunks (e.g., from a microphone) and returns partial/final
transcription results as a synchronous generator.

### What's included

**New files**
- `src/openai/live_audio_transcription_client.py` — Streaming session
with `start()`, `append()`, `get_transcription_stream()`, `stop()`
- `src/openai/live_audio_transcription_types.py` —
`LiveAudioTranscriptionResponse` (ConversationItem-shaped),
`LiveAudioTranscriptionOptions`, `CoreErrorResponse`,
`TranscriptionContentPart`
- `test/openai/test_live_audio_transcription.py` — 22 unit tests for
deserialization, settings, state guards, streaming pipeline
- `test/openai/test_live_audio_transcription_e2e.py` — E2E test with
real native DLLs and nemotron model
- `test/openai/conftest.py` — DLL preload for E2E tests
- `samples/python/live-audio-transcription/src/app.py` — Live microphone
transcription demo

**Modified files**
- `src/openai/audio_client.py` — Added
`create_live_transcription_session()` factory method
- `src/detail/core_interop.py` — Added `StreamingRequestBuffer` struct,
`execute_command_with_binary()`, `start_audio_stream`,
`push_audio_data`, `stop_audio_stream` methods, and `_load_dll_win()`
for robust DLL loading on Windows
- `src/openai/__init__.py` — Exported new live transcription types
- `test/conftest.py` — Pre-load ORT/GenAI DLLs before brotli import to
avoid Windows DLL search conflicts

### API surface

```python
audio_client = model.get_audio_client()
session = audio_client.create_live_transcription_session()

session.settings.sample_rate = 16000
session.settings.channels = 1
session.settings.language = "en"

session.start()

# Push audio from microphone callback (thread-safe)
session.append(pcm_bytes)

# Read results as synchronous generator
for result in session.get_transcription_stream():
    print(result.content[0].text)

session.stop()
```

### C# parity

| C# API | Python API | Notes |
|---|---|---|
| `CreateLiveTranscriptionSession()` |
`create_live_transcription_session()` | ✅ |
| `StartAsync(ct)` | `start()` | Sync (matches Python SDK convention) |
| `AppendAsync(ReadOnlyMemory<byte>, ct)` | `append(bytes)` |
Thread-safe, copies data |
| `GetTranscriptionStream()` | `get_transcription_stream()` | Generator
(sync equivalent of IAsyncEnumerable) |
| `StopAsync(ct)` | `stop()` | Drains push queue, sends native stop,
surfaces final result |
| `IAsyncDisposable` | Context manager (`with`) | Idiomatic Python
equivalent |
| `LiveAudioTranscriptionOptions` | `LiveAudioTranscriptionOptions` |
Same fields: sample_rate, channels, bits_per_sample, language,
push_queue_capacity |
| `LiveAudioTranscriptionResponse` | `LiveAudioTranscriptionResponse` |
ConversationItem-shaped: content[0].text/transcript, is_final,
start_time, end_time |

### Design highlights
- **Output type alignment** — `LiveAudioTranscriptionResponse` uses the
OpenAI Realtime `ConversationItem` shape (`content[0].text/transcript`)
for forward compatibility
- **Internal push queue** — Bounded `queue.Queue` serializes audio
pushes from any thread (safe for mic callbacks) with backpressure
- **Fail-fast on errors** — Push loop terminates immediately on any
native error (no retry logic)
- **Settings freeze** — Audio format settings are snapshot-copied at
`start()` and immutable during the session
- **Buffer copy** — `append()` copies input data to avoid issues with
callers reusing buffers (e.g., PyAudio)
- **Routes through existing exports** — `start_audio_stream` and
`stop_audio_stream` route through `execute_command`; `push_audio_data`
routes through `execute_command_with_binary` — no new native entry
points required
- **DLL loading fix** — Uses `LoadLibraryExW` with
`LOAD_WITH_ALTERED_SEARCH_PATH` on Windows to prevent conflicts with
stale system-level ORT DLLs

### Verified working
- ✅ 22 unit tests passing (deserialization, settings, state guards,
streaming pipeline with mocked core)
- ✅ E2E test passing (SDK → Core.dll → onnxruntime-genai.dll →
onnxruntime.dll with nemotron model)
- ✅ Full session lifecycle: start → push synthetic PCM → stop → verify
results
- ✅ Existing tests unaffected

---------

Co-authored-by: ruiren_microsoft <ruiren@microsoft.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: 3 people

sdk/python/src/detail/core_interop.py

@@ -46,6 +46,23 @@ class RequestBuffer(ctypes.Structure):
     ]
 
 
+class StreamingRequestBuffer(ctypes.Structure):
+    """ctypes Structure matching the native ``StreamingRequestBuffer`` C struct.
+
+    Extends ``RequestBuffer`` with binary data fields for sending raw payloads
+    (e.g. PCM audio bytes) alongside JSON parameters.
+    """
+
+    _fields_ = [
+        ("Command", ctypes.c_void_p),
+        ("CommandLength", ctypes.c_int),
+        ("Data", ctypes.c_void_p),
+        ("DataLength", ctypes.c_int),
+        ("BinaryData", ctypes.c_void_p),
+        ("BinaryDataLength", ctypes.c_int),
+    ]
+
+
 class ResponseBuffer(ctypes.Structure):
     """ctypes Structure matching the native ``ResponseBuffer`` C struct."""
 
@@ -173,6 +190,16 @@ def _initialize_native_libraries() -> 'NativeBinaryPaths':
                                                       ctypes.c_void_p]  # user_data
         lib.execute_command_with_callback.restype = None
 
+        # execute_command_with_binary is required for live audio streaming.
+        # Guard with try/except until Core packages with this symbol are released.
+        try:
+            lib.execute_command_with_binary.argtypes = [ctypes.POINTER(StreamingRequestBuffer),
+                                                         ctypes.POINTER(ResponseBuffer)]
+            lib.execute_command_with_binary.restype = None
+        except AttributeError:
+            logger.debug("execute_command_with_binary not exported by Core — "
+                         "live audio streaming will not be available until Core is updated")
+
         return paths
 
     @staticmethod
@@ -295,6 +322,66 @@ def execute_command_with_callback(self, command_name: str, command_input: Option
         response = self._execute_command(command_name, command_input, callback)
         return response
 
+    def execute_command_with_binary(self, command_name: str,
+                                    command_input: Optional[InteropRequest],
+                                    binary_data: bytes) -> Response:
+        """Execute a command with both JSON parameters and a raw binary payload.
+
+        Used for operations like pushing PCM audio data alongside JSON metadata.
+
+        Args:
+            command_name: The native command name (e.g. ``"audio_stream_push"``).
+            command_input: Optional request parameters (serialized as JSON).
+            binary_data: Raw binary payload (e.g. PCM audio bytes).
+
+        Returns:
+            A ``Response`` with ``data`` on success or ``error`` on failure.
+        """
+        logger.debug("Executing command with binary: %s Input: %s BinaryLen: %d",
+                     command_name, command_input.params if command_input else None, len(binary_data))
+
+        cmd_ptr, cmd_len, cmd_buf = CoreInterop._to_c_buffer(command_name)
+        data_ptr, data_len, data_buf = CoreInterop._to_c_buffer(
+            command_input.to_json() if command_input else None
+        )
+
+        # Keep binary data alive for the duration of the native call
+        binary_buf = ctypes.create_string_buffer(binary_data)
+        binary_ptr = ctypes.cast(binary_buf, ctypes.c_void_p)
+
+        req = StreamingRequestBuffer(
+            Command=cmd_ptr, CommandLength=cmd_len,
+            Data=data_ptr, DataLength=data_len,
+            BinaryData=binary_ptr, BinaryDataLength=len(binary_data),
+        )
+        resp = ResponseBuffer()
+        lib = CoreInterop._flcore_library
+
+        lib.execute_command_with_binary(ctypes.byref(req), ctypes.byref(resp))
+
+        req = None  # Free Python reference to request
+
+        response_str = ctypes.string_at(resp.Data, resp.DataLength).decode("utf-8") if resp.Data else None
+        error_str = ctypes.string_at(resp.Error, resp.ErrorLength).decode("utf-8") if resp.Error else None
+
+        lib.free_response(resp)
+
+        return Response(data=response_str, error=error_str)
+
+    # --- Audio streaming session support ---
+
+    def start_audio_stream(self, command_input: InteropRequest) -> Response:
+        """Start a real-time audio streaming session via ``audio_stream_start``."""
+        return self.execute_command("audio_stream_start", command_input)
+
+    def push_audio_data(self, command_input: InteropRequest, audio_data: bytes) -> Response:
+        """Push a chunk of raw PCM audio data via ``audio_stream_push``."""
+        return self.execute_command_with_binary("audio_stream_push", command_input, audio_data)
+
+    def stop_audio_stream(self, command_input: InteropRequest) -> Response:
+        """Stop a real-time audio streaming session via ``audio_stream_stop``."""
+        return self.execute_command("audio_stream_stop", command_input)
+
 
 def get_cached_model_ids(core_interop: CoreInterop) -> list[str]:
     """Get the list of models that have been downloaded and are cached."""

sdk/python/src/openai/__init__.py

@@ -7,5 +7,22 @@
 from .chat_client import ChatClient, ChatClientSettings
 from .audio_client import AudioClient
 from .embedding_client import EmbeddingClient
+from .live_audio_transcription_client import LiveAudioTranscriptionSession
+from .live_audio_transcription_types import (
+    CoreErrorResponse,
+    LiveAudioTranscriptionOptions,
+    LiveAudioTranscriptionResponse,
+    TranscriptionContentPart,
+)
 
-__all__ = ["AudioClient", "ChatClient", "ChatClientSettings", "EmbeddingClient"]
+__all__ = [
+    "AudioClient",
+    "ChatClient",
+    "ChatClientSettings",
+    "CoreErrorResponse",
+    "EmbeddingClient",
+    "LiveAudioTranscriptionOptions",
+    "LiveAudioTranscriptionResponse",
+    "LiveAudioTranscriptionSession",
+    "TranscriptionContentPart",
+]

sdk/python/src/openai/audio_client.py

@@ -14,6 +14,7 @@
 
 from ..detail.core_interop import CoreInterop, InteropRequest
 from ..exception import FoundryLocalException
+from .live_audio_transcription_client import LiveAudioTranscriptionSession
 
 logger = logging.getLogger(__name__)
 
@@ -61,6 +62,25 @@ def __init__(self, model_id: str, core_interop: CoreInterop):
         self.settings = AudioSettings()
         self._core_interop = core_interop
 
+    def create_live_transcription_session(self) -> LiveAudioTranscriptionSession:
+        """Create a real-time streaming transcription session.
+
+        Audio data is pushed in as PCM chunks and transcription results are
+        returned as a synchronous generator.
+
+        Returns:
+            A streaming session that should be stopped when done.
+            Supports use as a context manager::
+
+                with audio_client.create_live_transcription_session() as session:
+                    session.settings.sample_rate = 16000
+                    session.start()
+                    session.append(pcm_bytes)
+                    for result in session.get_transcription_stream():
+                        print(result.content[0].text)
+        """
+        return LiveAudioTranscriptionSession(self.model_id, self._core_interop)
+
     @staticmethod
     def _validate_audio_file_path(audio_file_path: str) -> None:
         """Validate that the audio file path is a non-empty string."""