mstar.api_server.openai.serving_realtime

mstar.api_server.openai.serving_realtime#

/v1/realtime (transcription intent): streaming speech-to-text over a WebSocket.

A subset of OpenAI’s Realtime transcription session, enough for a client that streams microphone audio and wants partial transcripts:

Client events

transcription_session.update {"session": {"input_audio_transcription": {"model", "language", "prompt"}, "input_audio_format", "turn_detection", "mstar": {"chunk_seconds", "unfixed_chunks", "unfixed_tokens"}}} input_audio_buffer.append {"audio": <base64 PCM16 mono>} input_audio_buffer.commit end of the utterance: flush and finish input_audio_buffer.clear drop buffered audio and start over

input_audio_format is "pcm16" (16 kHz, the default) or OpenAI’s {"type": "audio/pcm", "rate": 24000}; 24 kHz audio is resampled to the model’s 16 kHz as it arrives, so partial audio_seconds stay on the client’s timeline. The rate changes only with an empty buffer.

turn_detection null (the default) leaves ending an utterance to the client’s commit. {"type": "server_vad", "threshold", "prefix_padding_ms", "silence_duration_ms"} lets the server end it (realtime_vad), adding input_audio_buffer.speech_started / .speech_stopped events with audio_start_ms / audio_end_ms on the session’s audio timeline; a turn that reaches the item cap is committed there instead of refused.

chunk_seconds below MIN_CHUNK_SECONDS is raised to it and must not pass the item cap, and the unfixed counts are non-negative integers; numeric strings count as numbers. language and prompt are strings (or null), language at most MAX_LANGUAGE_CHARS characters. Any other value is an error and the update changes nothing.

A failed step is an error too. A request error (a 4xx from the engine, such as a model refusing the language) quotes its message, with server paths reduced to their base names and cut to MAX_ERROR_CHARS; any other failure says only that the step failed, with an error id that the server log carries the details under.

Server events

transcription_session.created / .updated conversation.item.input_audio_transcription.delta append-only text mstar.transcription.partial the whole current hypothesis, unstable tail included conversation.item.input_audio_transcription.completed {"transcript"} error

The decoding follows the Qwen3-ASR SDK’s streaming algorithm: every chunk_seconds of new audio, the audio heard so far is transcribed again as one engine request whose assistant turn is prefilled with the previous hypothesis minus its last unfixed_tokens tokens (once unfixed_chunks chunks are in), so the tail can still be revised while the head is fixed. The adapter decides how that prefix reaches its model (realtime_step_request). Each step is an ordinary request, so sessions batch on the engine like any other traffic.

Since every step decodes the whole item again, an item’s work grows with the square of its length. An item is capped at MAX_ITEM_SECONDS (or the adapter’s max_audio_seconds if shorter): an append past the cap is an error and the client commits or clears. An append that completes several chunks at once is decoded as one step.

Functions

pcm16_to_float(payload[, carry])

Base64 PCM16 (little-endian, mono) -> (float32 in [-1, 1), the odd trailing byte).

stable_prefix(raw_text, tokenizer, ...)

raw_text minus its last unfixed_tokens tokens, backed off further when the cut lands inside a multi-byte character.

Classes

RealtimeSession(session_id, request[, ...])

One WebSocket's transcription state (the SDK's ASRStreamingState).

RealtimeTranscription(api, model_name, ...)

Drives one WebSocket session against the API server.

Exceptions

StepError(message, status)

The engine's error for a step, with the HTTP status it carried: a 4xx is the request's fault (a model's ValueError), anything else the server's.
class mstar.api_server.openai.serving_realtime.RealtimeSession(session_id, request, chunk_seconds=2.0, unfixed_chunks=2, unfixed_tokens=5, max_item_seconds=300.0, pending=<factory>, pending_samples=0, heard=<factory>, chunk_id=0, raw_hypothesis='', emitted='', item_id=<factory>, failures=('', {}), odd_byte=b'', input_rate=16000, resampler=None, short=0, turn_detection=None, segmenter=None, stream_samples=0, clock_base=0, clock_in=0)[source]#

Bases: object

One WebSocket’s transcription state (the SDK’s ASRStreamingState).

Parameters:
add(audio)[source]#

Add SAMPLE_RATE audio to the item.

Parameters:

audio (ndarray)

Return type:

None

append(audio)[source]#
Parameters:

audio (ndarray)

Return type:

None

audio_so_far()[source]#
Return type:

ndarray

chunk_id: int = 0#
property chunk_samples: int#
chunk_seconds: float = 2.0#
clear()[source]#

Start a new item; the stream state (resampler, segmenter, odd byte) goes on.

Return type:

None

clock_base: int = 0#
clock_in: int = 0#
drain_resampler()[source]#

What the resampler still holds, as if silence followed; it starts over.

Return type:

ndarray

emitted: str = ''#
failures: tuple[str, dict] = ('', {})#
flush()[source]#
Return type:

ndarray | None

heard: list[ndarray]#
input_rate: int = 16000#
item_id: str#
item_samples(extra=0)[source]#

SAMPLE_RATE samples in the item once extra more input samples arrive, rounded once over the whole item (not per append), so the cap does not depend on how the audio was split.

Parameters:

extra (int)

Return type:

int

max_item_seconds: float = 300.0#
odd_byte: bytes = b''#
pending: list[ndarray]#
pending_samples: int = 0#
raw_hypothesis: str = ''#
realign()[source]#

After the resampler starts over (a commit or a clear), put the timeline back on the input clock: each start-over rounds its output up by less than a sample, which would add up over many of them.

Return type:

None

request: TranscriptionRequest#
resample(audio)[source]#
Parameters:

audio (ndarray)

Return type:

ndarray

resampler: StreamResampler | None = None#
reset_stream()[source]#

A clear: drop what the resampler and the segmenter hold, keeping the timeline (the dropped audio still counts).

Return type:

None

segment(audio)[source]#

Resample and segment audio: the segmenter’s operations.

Parameters:

audio (ndarray)

Return type:

list[tuple]

segmenter: TurnSegmenter | None = None#
session_id: str#
set_input_rate(rate)[source]#
Parameters:

rate (int)

Return type:

None

short: int = 0#
stream_samples: int = 0#
take_chunk()[source]#

Every full chunk of pending audio as one array, or None until enough arrived. Output the resampler still holds for its lookahead counts toward the chunk, so a chunk of 24 kHz input is decoded on the append that completes it, not the one after; the chunk is then short by those samples (short), which go with the next one.

Return type:

ndarray | None

turn_detection: ServerVAD | None = None#
unfixed_chunks: int = 2#
unfixed_tokens: int = 5#
class mstar.api_server.openai.serving_realtime.RealtimeTranscription(api, model_name, adapter, websocket)[source]#

Bases: object

Drives one WebSocket session against the API server.

Parameters:

  • model_name (str)

  • websocket (WebSocket)

async apply(ops)[source]#

Carry out segmenter operations in order (see TurnSegmenter). Audio steps once, after the last operation, so one long append is one step, as without turn detection; a turn’s commit decodes the rest.

Parameters:

ops (list[tuple])

Return type:

None

async commit()[source]#

Finish the item: decode what is left, report it, start a new one.

Return type:

None

current_text()[source]#
Return type:

str

async emit(text, hold)[source]#

Send the part of text past what was already sent, minus the last hold characters. A revision that reaches into sent text is not resent — deltas are append-only — the partial event carries it.

Parameters:
Return type:

None

async handle(event)[source]#
Parameters:

event (dict)

Return type:

None

async run()[source]#
Return type:

None

async send(event)[source]#
Parameters:

event (dict)

Return type:

None

async step(chunk, final=False)[source]#

Hear chunk, re-decode everything so far from the stable prefix, then send the newly firm text and the whole hypothesis.

Parameters:

  • chunk (ndarray)

  • final (bool)

Return type:

None

update(session)[source]#

Apply a session update, all or nothing.

Parameters:

session (dict)

Return type:

None

exception mstar.api_server.openai.serving_realtime.StepError(message, status)[source]#

Bases: RuntimeError

The engine’s error for a step, with the HTTP status it carried: a 4xx is the request’s fault (a model’s ValueError), anything else the server’s.

Parameters:
mstar.api_server.openai.serving_realtime.pcm16_to_float(payload, carry=b'')[source]#

Base64 PCM16 (little-endian, mono) -> (float32 in [-1, 1), the odd trailing byte). carry is the previous append’s odd byte, so a sample split across two appends is joined instead of shifting every sample after it. Strict decoding: a malformed payload is an error, not silence.

Parameters:
Return type:

tuple[ndarray, bytes]

mstar.api_server.openai.serving_realtime.stable_prefix(raw_text, tokenizer, unfixed_tokens)[source]#

raw_text minus its last unfixed_tokens tokens, backed off further when the cut lands inside a multi-byte character.

Parameters:

  • raw_text (str)

  • unfixed_tokens (int)

Return type:

str