Crate voicepipe 

voicepipe

A pipeline for live, turn-taking voice calls with a language model, in Rust.

Audio streams in. voicepipe decides when the speaker’s turn is over, transcribes it, runs the host’s turn, and speaks the reply back a sentence at a time while it is still being written. It stops the reply the moment the speaker talks over it, and keeps only what was heard.

Speech-to-text, text-to-speech and the model are traits the host implements, so the same pipeline runs over local models or hosted APIs. voicepipe owns what is common to every voice agent: turn-taking, barge-in, ordering and timing.

What the host provides

Trait	The host’s part
Stt	One clip (a 16 kHz mono PCM16 WAV) to its words. transcribe_after takes one chunk of a turn still being spoken, plus the words before it, for a transcriber that accepts a prompt.
Tts	One piece of the reply to one audio clip, in any container.
Llm	The turn: a TurnRequest in, text deltas and tool rounds out through a sink, and an outcome (Completed, Interrupted or AwaitingApproval). It honours a CancellationToken, and truncate_reply cuts the recorded reply back to what was heard.
Approvals	Whether an approval card is waiting, and what to do with an utterance heard while one waits.
Vad, TurnDetector	Optional. EarshotVad (default feature, pure Rust) and EnergyVad ship, with Endpointer as the turn detector.

The transport is two channels, Inbound (audio frames and JSON control frames) and Outbound (events and audio clips). Bridging them to a WebSocket takes a few lines of host code, and the host authenticates the connection before it bridges.

What the pipeline does

• Turn-taking. Onset, a barge-in bar that a backchannel “mm” does not clear, pauses, end of turn, and pre-roll so the first syllable is not clipped.
• Streaming transcription. The turn is cut into chunks at the speaker’s pauses and transcribed while they talk on, so the transcript is ready about when the turn ends. Whole-clip transcription is the fallback, and a mode of its own.
• Sentence-streamed speech. The reply is cut into sentences (the first at its first clause), synthesized ahead of playback and sent in order. A filler line covers a tool round that comes before any text.
• Barge-in. Voice over the reply cancels the turn, tells the client to drop its queued audio, and asks the host to record only what had started playing, as the client reports it or as estimated from clip lengths.
• Metrics. Each response reports where its time went, from the end of speech to the first audio.

The approval invariant

A turn may stop at an approval card: a change the host wants a person to confirm. Nothing a person says resolves one. While Approvals::pending reports a card, an utterance is answered with a reminder, handed to Approvals::held_utterance, and never reaches the Llm. TurnRequest::Resolve is built only from an action.resolve control frame, which is a button press. No audio or transcript path can construct one, and the crate’s tests assert that a spoken “yes” leaves the card waiting.

Example

A complete host with stand-in providers. A real one calls its speech services and model where these return fixed values, and forwards a socket’s frames instead of a synthetic tone.

use std::sync::Arc;

use tokio::sync::mpsc;
use tokio_util::sync::CancellationToken;
use voicepipe::audio::wav_from_pcm16;
use voicepipe::{
    run, Approvals, BoxFuture, CallConfig, CallServerEvent, Clip, Endpointer, EnergyVad,
    EnergyVadConfig, Inbound, Llm, LlmEvent, LlmSink, Outbound, ProviderError, Providers, Stt,
    Tts, TurnOutcome, TurnRequest,
};

struct Ears;
impl Stt for Ears {
    fn transcribe(&self, _wav: Vec<u8>) -> BoxFuture<'_, Result<String, ProviderError>> {
        Box::pin(async { Ok("what time is it".to_string()) })
    }
}

struct Voice;
impl Tts for Voice {
    fn synthesize(&self, _text: String) -> BoxFuture<'_, Result<Clip, ProviderError>> {
        let wav = wav_from_pcm16(&[0; 8_000], 16_000);
        Box::pin(async move {
            Ok(Clip { content_type: "audio/wav".into(), bytes: wav.into() })
        })
    }
}

struct Echo;
impl Llm for Echo {
    fn turn(
        &self,
        request: TurnRequest,
        sink: LlmSink,
        _cancel: CancellationToken,
    ) -> BoxFuture<'_, Result<TurnOutcome, ProviderError>> {
        Box::pin(async move {
            let reply = match request {
                TurnRequest::Utterance(text) => format!("You said: {text}. It is noon."),
                TurnRequest::Resolve { .. } => "Done.".to_string(),
            };
            sink(LlmEvent::TextDelta(reply.clone()));
            Ok(TurnOutcome::Completed { reply: Some(reply) })
        })
    }

    fn truncate_reply(&self, _heard: String) -> BoxFuture<'_, bool> {
        Box::pin(async { true })
    }
}

struct NoCards;
impl Approvals for NoCards {
    fn pending(&self) -> BoxFuture<'_, Option<serde_json::Value>> {
        Box::pin(async { None })
    }
    fn held_utterance(&self, _text: String) -> BoxFuture<'_, ()> {
        Box::pin(async {})
    }
}

#[tokio::main]
async fn main() {
    let config = CallConfig::default();
    let vad = EnergyVad::new(EnergyVadConfig::default());
    let detector = Box::new(Endpointer::new(config.endpoint_config(), vad));
    let providers = Providers {
        stt: Arc::new(Ears),
        tts: Arc::new(Voice),
        llm: Arc::new(Echo),
        approvals: Arc::new(NoCards),
        observer: None,
    };
    let (to_call, inbound) = mpsc::channel(256);
    let (outbound, mut from_call) = mpsc::unbounded_channel();
    tokio::spawn(run("call-1".into(), config, providers, detector, inbound, outbound));

    // Room noise, a second of "speech" (a 220 Hz tone), then a silence long
    // enough to end the turn, in 20 ms frames of little-endian PCM16.
    let samples = (0..48_000).map(|i: i32| {
        let voiced = (8_000..24_000).contains(&i);
        let tone = (i as f32 * 220.0 * std::f32::consts::TAU / 16_000.0).sin() * 9_000.0;
        if voiced { tone as i16 } else { (i * 7_919 % 41 - 20) as i16 }
    });
    let bytes: Vec<u8> = samples.flat_map(i16::to_le_bytes).collect();
    for frame in bytes.chunks(640) {
        to_call.send(Inbound::Audio(frame.to_vec().into())).await.unwrap();
    }

    while let Some(message) = from_call.recv().await {
        match message {
            Outbound::Event(event) => {
                println!("{}", serde_json::to_string(&event).unwrap());
                if matches!(event, CallServerEvent::ResponseDone { .. }) {
                    break;
                }
            }
            Outbound::Audio(clip) => println!("({} bytes of audio)", clip.len()),
        }
    }
}

The wire protocol

One duplex connection per call. The client streams its microphone as binary frames of little-endian PCM16, mono, 16 kHz, and sends JSON control frames as text: session.update, response.cancel, output_audio.started / output_audio.idle (playback reports, which make “what was heard” exact), action.resolve (the approval button) and ping. The server answers with JSON events named, where the meaning is the same, after the OpenAI Realtime API’s: session.created (which carries the protocol version), call.state, input_audio_buffer.speech_started, transcription captions, response.created, response.text.delta, response.audio.start followed by one binary frame holding that piece’s clip, response.done with metrics, output_audio.clear, conversation.item.truncated, and approval.pending / approval.resolved. The types are in the protocol module.

Status

Pre-1.0. While the version is 0.x, a minor release may break the API or the wire protocol and a patch release only fixes. The minimum supported Rust version is 1.88.

voicepipe was extracted from the call path of Vogt, which is its first host; its design notes are in docs/DESIGN.md.

Licence

Licensed under either of Apache License, Version 2.0 or MIT license at your option.

Unless you explicitly state otherwise, any contribution intentionally submitted for inclusion in this crate by you, as defined in the Apache-2.0 license, shall be dual licensed as above, without any additional terms or conditions.

Modules

PCM/WAV framing (audio), voice activity detection (vad), turn-taking (turn), cutting a turn into chunks to transcribe while it is spoken (chunk), cutting a streamed reply into speakable pieces (text), the wire protocol (protocol) and the call itself (pipeline).

Re-exports

pub use audio::SAMPLE_RATE;

pub use chunk::ChunkConfig;

pub use chunk::Chunker;

pub use pipeline::run;

pub use pipeline::Approvals;

pub use pipeline::BoxFuture;

pub use pipeline::CallConfig;

pub use pipeline::Clip;

pub use pipeline::Inbound;

pub use pipeline::Llm;

pub use pipeline::LlmEvent;

pub use pipeline::LlmSink;

pub use pipeline::Observer;

pub use pipeline::Outbound;

pub use pipeline::ProviderError;

pub use pipeline::Providers;

pub use pipeline::ResponseReport;

pub use pipeline::Stt;

pub use pipeline::SttMode;

pub use pipeline::Tts;

pub use pipeline::TurnOutcome;

pub use pipeline::TurnRequest;

pub use pipeline::MAX_AUDIO_FRAME_BYTES;

pub use protocol::CallClientEvent;

pub use protocol::CallMetrics;

pub use protocol::CallResponseStatus;

pub use protocol::CallServerEvent;

pub use protocol::CallState;

pub use protocol::PROTOCOL_VERSION;

pub use text::speakable;

pub use text::SentenceChunker;

pub use turn::EndpointConfig;

pub use turn::EndpointEvent;

pub use turn::Endpointer;

pub use turn::TurnDetector;

pub use vad::EarshotVad;

pub use vad::EnergyVad;

pub use vad::EnergyVadConfig;

pub use vad::Vad;

Modules

audio

PCM16 framing and WAV encoding for 16 kHz mono call audio.

chunk

Streaming transcription by chunks, for transcribers that only take whole clips.

pipeline

The call pipeline: turn-taking, transcription, the host’s turn spoken as it streams, barge-in, and the approval invariant.

protocol

The call’s wire protocol.

text

Text side of a call: cutting a streamed reply into the pieces that are spoken one at a time, and making each piece sayable.

turn

Turn-taking: per-frame speech decisions in, conversation events out.

vad

Voice activity detection: the Vad trait and two detectors.