ALLM.AudioStream (allm v0.6.1)

Reducers and adapters for streamed audio. Layer C, pure functions over enumerables.

collect_speech/1 folds an ALLM.SpeechEvent stream (from ALLM.stream_synthesize/3 or ALLM.stream_synthesize_input/3) into the ALLM.SpeechResponse that ALLM.synthesize/3 would return.
collect_transcription/1 folds an ALLM.TranscriptionEvent stream (from ALLM.stream_transcribe/3) into an ALLM.TranscriptionResponse.
text_deltas/1 turns a chat event stream (from ALLM.stream/3 or ALLM.stream_generate/3) into the text chunks ALLM.stream_synthesize_input/3 reads.

A failed stream is an error, not a partial response

An audio stream that fails after it opened ends with {:error, err}, and the collectors return {:error, err}. This differs from chat, where a mid-stream error folds into a response with finish_reason: :error: a speech response has no finish reason, and a half-rendered clip is not a valid response. What did arrive is described on the error's :metadata:

collect_speech/1 adds bytes_received, the count of audio bytes before the error. The bytes themselves are not kept on the error: errors encode to JSON, and raw audio is not UTF-8.
collect_transcription/1 adds committed_text, the committed segments so far, trimmed and joined with one space.

When the failure came from the input enumerable rather than the provider, metadata.cause is :input_raised (the input raised, threw or exited) or :input_crashed (a process linked inside the input crashed), and the error's :cause is a map %{kind: kind, message: text}, where kind is the atom :error, :throw or :exit and text is a string, never the raw exception or exit term. After a JSON round-trip kind reads back as a string ("error").

Speaking a chat stream

text_deltas/1 keeps only the text of :text_delta events. A chat {:error, err} event raises inside the speech adapter's input reduction, so the speech stream ends with {:error, _}, metadata.cause: :input_raised, and the chat error's reason and message in the error's :cause map. A truncated answer therefore never finishes as a successful clip. To speak whatever arrived before a chat failure, filter the chat stream before text_deltas/1.

Summary

Functions

collect_speech(events)

Fold a speech event stream into an ALLM.SpeechResponse.

collect_transcription(events)

Fold a transcription event stream into an ALLM.TranscriptionResponse.

text_deltas(chat_events)

Turn a chat event stream into the text chunks it carries. Lazy.

Functions

collect_speech(events)

@spec collect_speech(Enumerable.t(ALLM.SpeechEvent.t())) ::
  {:ok, ALLM.SpeechResponse.t()} | {:error, ALLM.Error.SpeechAdapterError.t()}

Fold a speech event stream into an ALLM.SpeechResponse.

:model, :provider, :format, :sample_rate and the audio's MIME type come from :speech_started; :usage, :id, :request_id and :metadata from :speech_completed. The audio is every :audio_delta payload concatenated, as an ALLM.Audio. :raw is nil.

An {:error, err} event returns {:error, err} with metadata.bytes_received set. A stream that ends without a terminal event, or completes without a :speech_started, returns {:error, %ALLM.Error.SpeechAdapterError{reason: :malformed_response}}. Reduction stops at the first terminal event.

Examples

iex> events = [
...>   ALLM.SpeechEvent.speech_started(%{request_id: "r1", model: "m", provider: :fake,
...>     format: :pcm, mime_type: "audio/pcm", sample_rate: 24_000}),
...>   ALLM.SpeechEvent.audio_delta("ab"),
...>   ALLM.SpeechEvent.audio_delta("c"),
...>   ALLM.SpeechEvent.speech_completed(%{request_id: "r1", id: nil,
...>     usage: %ALLM.Usage{}, metadata: %{}})
...> ]
iex> {:ok, resp} = ALLM.AudioStream.collect_speech(events)
iex> {ALLM.Audio.to_binary(resp.audio), resp.format, resp.sample_rate}
{{:ok, "abc"}, :pcm, 24000}

iex> err = ALLM.Error.SpeechAdapterError.new(:timeout)
iex> {:error, err} = ALLM.AudioStream.collect_speech([{:audio_delta, "abc"}, {:error, err}])
iex> err.metadata.bytes_received
3

collect_transcription(events)

@spec collect_transcription(Enumerable.t(ALLM.TranscriptionEvent.t())) ::
  {:ok, ALLM.TranscriptionResponse.t()}
  | {:error, ALLM.Error.TranscriptionAdapterError.t()}

Fold a transcription event stream into an ALLM.TranscriptionResponse.

:text, :language, :duration_seconds, :usage, :request_id and :metadata come from :transcription_completed, whose :text is already the adapter's join of the committed segments; :model, :provider and :id (the realtime session id) from :transcription_started. :raw is nil.

:spans is copied from :transcription_completed's optional :spans key, which an adapter writes (as the concatenation of its committed segments' spans) only when timestamps: true or logprobs: true was requested. Committed events are not folded for spans, so a completed event without the key collects to spans: nil ("not requested") and a requested stream that committed nothing collects to spans: [].

iex> span = ALLM.TranscriptSpan.new(text: "hello", kind: :word, logprob: -0.5)
iex> events = [
...>   ALLM.TranscriptionEvent.committed_transcript("hello", nil, [span]),
...>   ALLM.TranscriptionEvent.transcription_completed(%{text: "hello", language: nil,
...>     duration_seconds: 0.5, request_id: nil, usage: %ALLM.Usage{}, metadata: %{},
...>     spans: [span]})
...> ]
iex> {:ok, resp} = ALLM.AudioStream.collect_transcription(events)
iex> resp.spans == [span]
true

An {:error, err} event returns {:error, err} with metadata.committed_text set to the committed segments so far. A stream that ends without a terminal event returns {:error, %ALLM.Error.TranscriptionAdapterError{reason: :malformed_response}}.

Unlike collect_speech/1, a missing :transcription_started is not an error: the transcript itself is on the completed event, so the response is returned with :model, :provider and :id nil. (Speech needs the start event's :format and :mime_type to build its ALLM.Audio.)

Examples

iex> events = [
...>   ALLM.TranscriptionEvent.transcription_started(%{request_id: "r1", model: nil,
...>     provider: :fake, session_id: "s1"}),
...>   ALLM.TranscriptionEvent.partial_transcript("hel"),
...>   ALLM.TranscriptionEvent.committed_transcript("hello"),
...>   ALLM.TranscriptionEvent.transcription_completed(%{text: "hello", language: nil,
...>     duration_seconds: 0.5, request_id: "r1", usage: %ALLM.Usage{}, metadata: %{}})
...> ]
iex> {:ok, resp} = ALLM.AudioStream.collect_transcription(events)
iex> {resp.text, resp.id, resp.duration_seconds}
{"hello", "s1", 0.5}

text_deltas(chat_events)

@spec text_deltas(Enumerable.t(ALLM.Event.t())) :: Enumerable.t(String.t())

Turn a chat event stream into the text chunks it carries. Lazy.

Each {:text_delta, %{delta: text}} becomes text; every other event is dropped. A chat {:error, err} event raises when the stream is reduced, so passed to ALLM.stream_synthesize_input/3 it ends the speech stream with metadata.cause: :input_raised and the chat error's reason and message in the speech error's :cause map (see the moduledoc).

Examples

iex> chat = [
...>   {:message_started, %{message: ALLM.assistant("")}},
...>   {:text_delta, %{id: nil, delta: "Hel"}},
...>   {:text_delta, %{id: nil, delta: "lo."}}
...> ]
iex> chat |> ALLM.AudioStream.text_deltas() |> Enum.to_list()
["Hel", "lo."]