Step 2 Landed — Akka.NET Streams Now Carry the Whole Voice Path in AgentZero Lite

Series: AgentZero Lite — Part 7. The previous post (Part 6 — Step 1 Embedding an Open vLLM, Step 2 Akka Streams) laid out the roadmap. This one reports back: Step 2 is in the trunk, opt-in behind a feature flag, with eleven TestKit cases passing and the legacy batch path still primary until live testing validates the stream path.

TL;DR for the people who scroll

• The voice subsystem (microphone → STT → reactor → LLM → TTS → speaker) is now built around two Akka.NET RunnableGraphs owned by a single VoiceStreamActor at /user/stage/voice.

• INPUT graph turns 50 ms PCM frames into transcripts. OUTPUT graph turns LLM token streams into progressive TTS playback. Both share a KillSwitch so the actor can tear them down atomically — BargeIn, CancelInflight, EndTurn, DeviceLost are all Tell-only messages that end up in the same teardown path.

• STT and TTS parallelism are SmallestMailboxPool routers, not fused SelectAsync(parallelism: N) constants. Operations knob, not a code constant.

• Akka.NET 1.5 doesn't expose StatefulSelectMany on FlowOperations (it's on SourceOperations only). Custom GraphStage<FlowShape<...>> is the canonical pattern for stateful flows in this codebase, and TestKit's SourceProbe / SinkProbe hook into it cleanly.

• Opt-in via VoiceSettings.UseStreamPipeline (default false). The legacy OnVoiceUtteranceEnded → SendCurrentInput batch path stays primary until live testing on real devices.

Why streams instead of "another async helper"

UtteranceEnded event → byte[] PCM → await STT → set txtInput.Text →
SendCurrentInput → reactor finishes → await TTS → Play(byte[])

Topology — two graphs, one actor

flowchart TD
    Actor["<b>VoiceStreamActor</b><br/>/user/stage/voice<br/>• owns SharedKillSwitch<br/>• Tell-only protocol<br/>• holds materialized handles"]

    subgraph INPUT ["INPUT graph (mic → reactor)"]
        direction TB
        I1["Source.Queue&lt;MicFrame&gt;<br/>(64, OverflowStrategy.DropHead)"]
        I2["KillSwitches.Single"]
        I3["<b>VoiceSegmenterStage</b><br/>VAD + 1 s pre-roll + utterance FSM"]
        I4[".Async()"]
        I5["SelectAsync(p, sttPool.Ask)"]
        I6["Sink.ActorRefWithAck(Self)"]
        I1 --> I2 --> I3 --> I4 --> I5 --> I6
    end

    subgraph OUTPUT ["OUTPUT graph (LLM tokens → speaker)"]
        direction TB
        O1["Source.Queue&lt;string&gt;<br/>(64, OverflowStrategy.Backpressure)"]
        O2["KillSwitches.Single"]
        O3["<b>SentenceChunkerStage</b><br/>.?!\\n + min/max chunk len"]
        O4["TtsTextCleaner.StripMarkdown"]
        O5["SelectAsync(p, ttsPool.Ask)"]
        O6["Sink.ForEach(playback.Enqueue)"]
        O1 --> O2 --> O3 --> O4 --> O5 --> O6
    end

    Actor -- "owns + materializes" --> I1
    Actor -- "owns + materializes" --> O1

    classDef actor fill:#1f2937,stroke:#0ea5e9,color:#e5e7eb,stroke-width:2px
    classDef stage fill:#0b3d2e,stroke:#10b981,color:#d1fae5
    class Actor actor
    class I3,O3 stage

P0 — the boring but load-bearing part

P1 — INPUT graph: where Akka.NET diverges from the JVM docs

public override void OnPush()
{
    var frame = Grab(_stage.In);

    // Pre-roll ring (always — even outside an utterance — so when VAD
    // trips the prior PreRollSeconds of audio is available).
    _preRoll.Enqueue(frame.Pcm16k);
    _preRollBytes += frame.Pcm16k.Length;
    while (_preRollBytes > _maxPreRollBytes && _preRoll.Count > 0)
        _preRollBytes -= _preRoll.Dequeue().Length;

    var above = frame.Rms >= _stage._config.VadThreshold;
    if (above)
    {
        _utteranceSilenceFrames = 0;
        if (!_inUtterance)
        {
            // Utterance starts — seed buffer with pre-roll.
            _inUtterance = true;
            _buffer = new List<byte>(...);
            foreach (var chunk in _preRoll) _buffer.AddRange(chunk);
            _startedAt = DateTimeOffset.UtcNow;
        }
        else _buffer!.AddRange(frame.Pcm16k);
        Pull(_stage.In);
        return;
    }
    if (_inUtterance)
    {
        _buffer!.AddRange(frame.Pcm16k);
        if (++_utteranceSilenceFrames >= _stage._config.UtteranceHangoverFrames)
        {
            _inUtterance = false;
            var pcm = _buffer.ToArray();
            _buffer = null;
            Push(_stage.Out, new PcmSegment(pcm, pcm.Length / 32_000.0, _startedAt));
            return;
        }
        Pull(_stage.In);
        return;
    }
    Pull(_stage.In);
}

.SelectAsync(parallelism, async (PcmSegment seg) =>
{
    var reply = await sttPool.Ask<TranscribeReply>(
        new TranscribeRequest(seg, language),
        TimeSpan.FromSeconds(120));
    return new VoiceTranscriptReady(reply.Transcript, reply.DurationSeconds);
})

1. Operations knob. VoiceSettings.StreamSttParallelism controls the pool size. Bumping it later doesn't require recompiling.

2. One STT engine per worker. Each worker actor lazily constructs its ISpeechToText and pays the cold-start cost (Whisper.net "small" loads ~487 MB) on its first segment. The router naturally spreads load across the warmed workers.

P2 — OUTPUT graph: progressive TTS without a custom playback engine

_ = Task.Run(async () =>
{
    try
    {
        await foreach (var token in cmd.TokenStream)
        {
            var result = await queue.OfferAsync(token);
            if (result is QueueOfferResult.QueueClosed) return;
            if (result is QueueOfferResult.Failure f) { _log.Error(f.Cause, "..."); return; }
        }
        queue.Complete();
    }
    catch (OperationCanceledException) { try { queue.Complete(); } catch { } }
    catch (Exception ex) { _log.Error(ex, "..."); try { queue.Fail(ex); } catch { } }
});

P3 — actor-mediated control plane

• Decider handles "this stage threw" but doesn't speak the language of "the user just talked over the AI."

• RestartFlow.WithBackoff restarts the entire flow on failure — heavy when the failure is a single transient 5xx and the rest of the stage state (loaded STT model, open socket) is still good.

sequenceDiagram
    autonumber
    participant Mic as Mic capture
    participant Actor as VoiceStreamActor
    participant Out as OUTPUT graph
    participant Spk as NAudioPlaybackQueue

    Out->>Spk: enqueue TTS chunk
    Spk-->>Actor: PlaybackStarted
    Note over Actor: _outputActive = true<br/>_consecutiveLoudFrames = 0

    Mic->>Actor: MicFrame (loud, +1)
    Mic->>Actor: MicFrame (loud, +1)
    Mic->>Actor: MicFrame (loud, +1)
    Mic->>Actor: MicFrame (loud, +1) — threshold hit

    Actor->>Actor: Self.Tell(BargeIn)
    Actor->>Out: KillSwitch.Shutdown()
    Actor->>Out: Source.Queue&lt;string&gt;.Complete()
    Actor->>Spk: Stop() (clears queue)
    Spk-->>Actor: PlaybackStopped

    Note over Actor: _outputActive = false<br/>INPUT graph keeps running —<br/>user's interrupting utterance<br/>flows through normally

Five non-obvious things I'd tell anyone wiring this on Akka.NET

1. StatefulSelectMany is not on FlowOperations. It's SourceOperations only in 1.5. Stateful flows are custom GraphStage<FlowShape<>>. Don't waste an hour searching for the operator.

2. Source.ActorRef does not support OverflowStrategy.Backpressure. For real-time devices use Source.Queue<T>(N, OverflowStrategy.DropHead) — the audio thread cannot slow down, so dropping the oldest frame on overflow preserves recency. Trying to backpressure the soundcard is the wrong direction.

3. Sink.ActorRef has no backpressure. Use Sink.ActorRefWithAck(actor, init, ack, complete, fail). The actor must Sender.Tell(ack) on the init message and after every element; the sink drives the demand. Easy to miss until your actor mailbox is buried under mic frames.

4. The materializer's actors share the default dispatcher. A stream stage that pokes the WPF UI directly will block the dispatcher pool. Marshal back to the dispatcher only at the terminal sink, never mid-graph.

5. Per-call retry beats RestartFlow.WithBackoff for voice. A transient 5xx on TTS shouldn't tear down a loaded model and a warm socket. A 3-attempt exponential retry inside the worker actor is cheaper and matches the user's perceived latency budget (5xx ↔ recovery happens in well under one utterance).

What's still on the table

• Reactor IAsyncEnumerable<string> token stream. Right now HandleReactorResult fires SpeakText(r.FinalMessage) after the reactor finishes; the OUTPUT graph still chunks per sentence and overlaps TTS with playback, but the LLM has already ended its turn. Wiring the reactor's per-token feed directly into the graph is the next step — and the place where the seconds-of-latency win is.

• BargeInDetector as a child actor. Currently inline. The policy (4 consecutive loud frames at threshold) is good enough for a single user, single device. Voice-class detection or named-call recognition belongs in its own actor.

• BufferedWaveProvider-based gap-less playback. Sequential WaveOutEvent is the correct sequential ordering but leaves an audible seam between TTS chunks.

• Settings UI. The UseStreamPipeline and StreamSttParallelism knobs are JSON-only today.

Closing

• psmon

TECH LINKS

• 𝕏 @webnori

• 📘 Akka Labs (Facebook group)

• 📝 More writings — webnori wiki