# `ALLM.AudioStream`
[🔗](https://github.com/cykod/ALLM/blob/v0.6.1/lib/allm/audio_stream.ex#L1)

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`.

# `collect_speech`

```elixir
@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`

```elixir
@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`

```elixir
@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."]

---

*Consult [api-reference.md](api-reference.md) for complete listing*