Streaming text-to-speech contract: audio arrives as it is synthesized.
Layer B — runtime. A speech adapter opts into streaming by implementing
this behaviour in addition to ALLM.SpeechAdapter, in the same module.
There is no separate engine slot: the engine's :speech_adapter is
checked with function_exported?/3, the same way the chat :adapter is
checked for ALLM.StreamAdapter.stream/2. A slot adapter that does not
export the callback cannot stream.
Two callbacks, one per input shape:
stream_synthesize/2— the whole text is known up front, the audio streams back.stream_synthesize_input/3(optional) — the text itself arrives as an enumerable of chunks, for example the text deltas of a chat stream, and audio streams back while text is still arriving.
Each returns {:ok, enumerable} of ALLM.SpeechEvent values. The
enumerable obeys the grammar stated in ALLM.SpeechEvent.
Minimum impl skeleton
defmodule MyStreamingSpeech do
@behaviour ALLM.SpeechAdapter
@behaviour ALLM.SpeechStreamAdapter
alias ALLM.Error.SpeechAdapterError
alias ALLM.SpeechEvent
@impl ALLM.SpeechAdapter
def synthesize(request, opts), do: MyProvider.synthesize(request, opts)
@impl ALLM.SpeechStreamAdapter
def stream_synthesize(%ALLM.SpeechRequest{input: ""}, _opts) do
# Invariant 2: gates run before ALLM.Keys.fetch!/2 and before the
# enumerable is returned.
{:error, SpeechAdapterError.new(:invalid_request)}
end
def stream_synthesize(%ALLM.SpeechRequest{} = request, opts) do
key = ALLM.Keys.fetch!(:my_provider, opts)
stream =
Stream.resource(
fn -> MyProvider.open(request, key) end,
fn conn -> MyProvider.next_events(conn) end,
# Invariant 4: release the transport, drain its messages.
fn conn -> MyProvider.close(conn) end
)
{:ok, stream}
end
endInvariants
- The synchronous return is exactly
{:ok, enumerable}or{:error, %ALLM.Error.SpeechAdapterError{}}. The one documented exception isALLM.Keys.fetch!/2, which raises%ALLM.Error.EngineError{reason: :missing_key}by design. - Lazy. No I/O happens until the enumerable is reduced. Pre-flight
gates run synchronously and return
{:error, _}beforeALLM.Keys.fetch!/2: forstream_synthesize/2the empty-input gate, then per-adapter gates; forstream_synthesize_input/3the request shape checks ofALLM.Validate.speech_request(request, input: :streamed), then per-adapter gates. A keyless environment therefore observes the rejection, not a missing-key raise. - The enumerable obeys the
ALLM.SpeechEventgrammar: one:speech_started, one or more:audio_delta, one:speech_completed; or a failure ending in one{:error, _}. Nothing follows a terminal event. A stream that reaches its end with zero audio bytes, including astream_synthesize_input/3input that yields no non-empty chunk, ends with:invalid_requestandmetadata.cause: :empty_input. This differs fromALLM.SpeechAdapter.synthesize/2, where a successful response with an empty body is:malformed_response(ALLM.SpeechAdapterinvariant 2). - Cleanup: halt-safe. A consumer halt (
Enum.take/2) releases the transport (the HTTP request is cancelled, or the WebSocket is closed and the input pump stopped) within 500 ms. The after function stops the pump withALLM.Providers.Support.InputPump.stop/2and drains the transport's messages. On a WebSocket or input-pump path no stream-owned message is left in the consumer's mailbox. On an HTTP path the drain is best-effort: messages already queued at the halt are removed, but the HTTP client cancels asynchronously, so one late message from the cancelled request can still arrive afterwards. The input's own resources are released by process exit, not by its after functions. opts[:stream_timeout](milliseconds of silence, default 60,000) is honoured. The timer resets on every transport message and every input-pump message, so a slow input does not time out a socket whose server is waiting for text, and a silent server does not time out while input still flows. Expiry ends the stream with{:error, %ALLM.Error.SpeechAdapterError{reason: :timeout}}.opts[:request_id]appears on:speech_startedand:speech_completed.request.metadataappears unchanged on:speech_completed's:metadata.- Input chunks (
stream_synthesize_input/3). A non-binary or non-UTF-8 element ends the stream with:invalid_requestandmetadata.cause: :invalid_input_chunk. An empty string element is skipped. An input that raises, throws or exits ends the stream with:invalid_requestandmetadata.cause: :input_raised; an input whose reducing process dies from an exit signal ends it withmetadata.cause: :input_crashed. In both cases the error's:causefield is a%{kind: kind, message: message}map (kindis:error,:throwor:exit,messagea string), never the raw exception or exit term, and the consuming process is never killed. - Concatenating every
:audio_deltapayload gives non-empty audio, and the:mime_typeon:speech_startedbeginsaudio/. - Ordering of I/O. Key resolution runs synchronously, after the gates and before the enumerable is returned. A transport that must be opened before input flows (a WebSocket) is opened in the enumerable's start function, and the input is only reduced after the connection is established, so a failed connection never consumes the input.
Input is reduced in another process
An adapter reduces the input of stream_synthesize_input/3 through
ALLM.Providers.Support.InputPump, so the input runs in a helper
process. An input that depends on the caller's mailbox or process
dictionary must be relayed.
Backpressure
Transport messages arrive in the consumer's mailbox at the provider's
generation rate, as on the chat streaming path; a slow consumer grows
its mailbox. On the input side the pump's credit window (default 8,
adapter_opts[:input_window]) bounds how much unsent input waits in
the consumer's mailbox.
Summary
Types
One chunk of text fed to stream_synthesize_input/3.
Callbacks
Stream the synthesis of request.input.
Stream the synthesis of text that itself arrives as an enumerable of
chunks. request.input is ignored.
Types
@type text_chunk() :: String.t()
One chunk of text fed to stream_synthesize_input/3.
Callbacks
@callback stream_synthesize( ALLM.SpeechRequest.t(), keyword() ) :: {:ok, Enumerable.t(ALLM.SpeechEvent.t())} | {:error, ALLM.Error.SpeechAdapterError.t()}
Stream the synthesis of request.input.
Returns {:ok, enumerable} of ALLM.SpeechEvent values, or
{:error, %ALLM.Error.SpeechAdapterError{}} from a pre-flight gate.
@callback stream_synthesize_input( ALLM.SpeechRequest.t(), Enumerable.t(text_chunk()), keyword() ) :: {:ok, Enumerable.t(ALLM.SpeechEvent.t())} | {:error, ALLM.Error.SpeechAdapterError.t()}
Stream the synthesis of text that itself arrives as an enumerable of
chunks. request.input is ignored.
Optional. Returns {:ok, enumerable} of ALLM.SpeechEvent values, or
{:error, %ALLM.Error.SpeechAdapterError{}} from a pre-flight gate.