Streaming live call audio to your own ML pipeline — a reference architecture

Most voice-AI deployments stop at "call lands on an AI agent." The AI vendor talks to the caller, the vendor sends you a transcript and some events after the call, and that's your window into what happened. For most products that's fine.

For some products — real-time sentiment detection, custom speech recognition, agent-assist, live keyword alerts, per-industry entity extraction, QA scoring during the call — you need the raw audio frames, delivered live, with latency low enough to act on them while the caller is still talking.

Voylo's <Stream>verb opens a WebSocket to your endpoint and shovels raw PCM frames into it. You consume, process, push back whatever you want. This post walks through the architecture, the wire format, and a working Python consumer. One of our customers in the mining- and-remote-ops space (anonymised) uses this exact pattern to run their own custom-vocabulary transcription stack against a language model their R&D team maintains — nothing vendor- hosted, full control over the pipeline, millisecond latency.

Architecture

Four moving pieces:

• Voylo call leg — the inbound (or outbound) call that lands on your VoyloML app.
• Stream verb — a VoyloML tag that bridges the live audio to a WebSocket of your choosing. Blocks like<Dial> does; returns when the WS closes, the call ends, or the gateway times out.
• Your WebSocket server— anywhere you can run one. We'll show Python; the wire format is vendor-neutral.
• Your ML pipeline — whatever consumes frames. Can be in-process, can be a Kafka queue to a GPU cluster, up to you.

The VoyloML

<?xml version="1.0" encoding="UTF-8"?>
<Response>
  <!-- Play a short greeting while the stream connects (optional) -->
  <Play>https://your-cdn/prompts/connecting.wav</Play>

  <!-- Open the WebSocket, start streaming live audio -->
  <Stream url="wss://your-ml.example.com/call-audio"
          name="customer-support-{{CallSid}}"
          format="pcm16/16000"
          action="#after">
    <!-- Optional metadata — appears in the start frame -->
    <Param name="caller"    value="{{From}}"/>
    <Param name="customer"  value="{{customerId}}"/>
    <Param name="language"  value="{{language}}"/>
  </Stream>

  <Flow id="after">
    <Play>https://your-cdn/prompts/thanks.wav</Play>
    <Hangup/>
  </Flow>
</Response>

Attributes worth knowing:

• url — your WebSocket endpoint. Must be WSS (TLS) in production — Voylo rejects plain WS from the edge.
• format — pcm16/16000 (16-bit linear PCM, 16kHz) is the Voylo default and what we recommend. Lossless, 16-bit words, well-aligned with what every modern ASR expects. Also supports pcm16/8000 for narrow-band savings and mulaw/8000 if your downstream prefers G.711 directly.
• name — a label echoed in the start frame; useful for correlating with your call-ID. Supports {{var}} substitution.
• <Param> children — arbitrary key-value pairs sent in the start frame as customParameters. Think of them like SIP headers for your streaming consumer — identify the caller, pass session IDs, surface routing hints.

Wire format

The protocol is Twilio Media Streams-compatible — if you have an existing consumer written for Twilio Media Streams, swap the WebSocket URL and it works. The frames you'll receive:

// 1. On connect
{"event": "connected", "protocol": "Call", "version": "1.0.0"}

// 2. When streaming begins
{
  "event": "start",
  "sequenceNumber": "1",
  "start": {
    "streamSid": "MZ...",
    "callSid":   "call_abc123",
    "mediaFormat": {"encoding": "audio/l16", "sampleRate": 16000, "channels": 1},
    "customParameters": {"caller": "+971...", "customer": "cust_xyz"}
  }
}

// 3. Media frames (dozens per second — ~20ms each by default)
{
  "event": "media",
  "sequenceNumber": "42",
  "media": {
    "track": "inbound",             // or "outbound" for the AI/callee side
    "chunk": "17",
    "timestamp": "340",             // ms since start
    "payload": "base64-encoded-pcm" // decode this
  }
}

// 4. Marks you push into the stream (optional)
{"event": "mark", "mark": {"name": "cue-1"}}

// 5. On end
{"event": "stop", "stop": {"callSid": "call_abc123", "accountSid": "..."}}

Push direction (your server → Voylo): three frame types.

• media with base64-encoded PCM — gets played INTO the live call. This is how you drive synthesized audio INTO the conversation: TTS from your stack, custom prompts, hold music.
• mark— a named label that round-trips. Useful for "when has my sent audio finished playing?" — you push a mark after a chunk, Voylo echoes it back when it finishes queuing to the caller.
• clear — flush any queued audio from your server waiting to be played. Useful for barge-in: user starts talking, you cut your TTS mid-sentence.

Minimum viable consumer in Python

Here's a 50-line consumer that logs every call, dumps the inbound audio to a per-call wav file, and prints transient energy as a liveness check. Takes about 15 minutes to extend into a real ASR / sentiment pipeline — swap wav_file.writeframes for your asr.feed_frames of choice.

import asyncio, base64, json, wave
import websockets

async def handler(ws):
    wav = None
    stream_sid = None
    try:
        async for msg in ws:
            m = json.loads(msg)
            evt = m.get("event")

            if evt == "connected":
                print("[conn] protocol:", m.get("protocol"))

            elif evt == "start":
                stream_sid = m["start"]["streamSid"]
                fmt = m["start"]["mediaFormat"]
                print(f"[start] {stream_sid} {fmt}")
                print(f"[start] custom:", m["start"].get("customParameters", {}))
                wav = wave.open(f"/tmp/{stream_sid}.wav", "wb")
                wav.setnchannels(fmt["channels"])
                wav.setsampwidth(2)                  # pcm16 = 2 bytes
                wav.setframerate(fmt["sampleRate"])

            elif evt == "media":
                pcm = base64.b64decode(m["media"]["payload"])
                if wav and m["media"]["track"] == "inbound":
                    wav.writeframes(pcm)
                # Hook your real pipeline in here — e.g. feed `pcm` into
                # a running ASR session, push to a sentiment classifier,
                # kick off a keyword alert, etc.

            elif evt == "stop":
                print(f"[stop]  {stream_sid}")
                break
    finally:
        if wav:
            wav.close()

async def main():
    async with websockets.serve(handler, "0.0.0.0", 8765):
        await asyncio.Future()  # run forever

if __name__ == "__main__":
    asyncio.run(main())

How this gets real

The 50-line consumer is a toy. What production looks like:

1. Frames go to a bus, not a file. Push every inbound PCM chunk into Kafka / Pub/Sub / Redis Streams keyed by streamSid. Downstream workers consume. This decouples the WS endpoint's uptime from your inference workers' uptime.
2. ASR per stream, not per frame.Whisper / Deepgram's streaming / your in-house CTC model — each stream opens one ASR session that lives for the call. Frames feed it; partial transcripts come out; final transcript on stop.
3. Fan out to as many models as you want.One inbound frame goes to: ASR, speaker diarisation, emotion / sentiment, keyword alerts, PII scrubber. Independent workers; one slow consumer doesn't slow the others.
4. Push results back into the call.Agent-assist UI telling a human agent what to say next doesn't need to push audio back — it needs to push text to the agent's screen. But if you want the AI to interrupt and play a specific response ("I detected the customer saying cancellation keywords — play the retention offer prompt"), you push media frames back and the audio enters the live call.

Latency reality

With 20ms frame cadence, your WebSocket in the same region as Voylo, and PCM16 at 16kHz:

• Frame-to-your-server:< 50ms end-to-end typical; < 100ms worst-case for cross-region.
• Your-ML-decision-to-caller-ear:depends on your inference stack. A streaming ASR + classifier typically adds 100-300ms. TTS back into the call adds the TTS vendor's latency (ElevenLabs streaming is ~150ms first-token).
• Round-trip for interrupt-the-caller:~400-800ms end-to-end. Fast enough for "AI reacts to what you just said", not fast enough for simultaneous crosstalk interruption. The vendors who do sub-200ms crosstalk (OpenAI Realtime, ElevenLabs Agents) are co-locating their inference with the voice edge; you can match them if you run inference on the same box as your WS consumer.

What this is good for

• Custom-vocabulary ASR. If your industry has jargon that general ASRs mangle (medical codes, mining equipment names, regional dialect vocabulary) — this is the cleanest way to run your own.
• In-house compliance and PII handling. Audio never leaves your infrastructure (except for the Voylo call path itself). Important for regulated industries.
• Multi-model fan-out. You want ASR AND emotion AND keyword alerts AND a custom entity extractor, all on the same call. A vendor-hosted agent gives you their fixed capability set; raw frames give you orthogonal processing.
• Agent-assist / supervisor tools.Real-time transcript on a human agent's screen, suggested responses, escalation triggers.
• Quality analytics. Call scoring per minute as the call progresses; alert supervisors when a call is trending toward dissatisfied.

What this is NOT good for

• Turnkey AI agents.If what you want is "AI picks up and talks to the caller", use Vapi / Retell / ElevenLabs / OpenAI Realtime via an inbound trunk. Streaming is one abstraction level lower; you're the agent now.
• Recording-only use cases. If you just want to store the audio, enable record="always" on <Dial> or on the application. You get a .wav file after the call ends; much simpler, no WS infrastructure.

The one-line summary

If your voice product needs more than a vendor-hosted AI agent — custom models, in-house transcription, real-time analytics — the <Stream> verb gives you raw audio frames over a WebSocket and lets you push synthesized audio back. One verb in your app, a WebSocket server in your language of choice, full control.