How speech models power voice assistants and call centers

Speech models voice assistants rely on two distinct pipelines: speech-to-text (STT) for ingestion and text-to-speech (TTS) for output. Each pipeline has different latency budgets, accuracy requirements, and failure modes. This guide walks through the architecture decisions, model selection criteria, and integration patterns that separate production systems from demos.

Choose your deployment model

The first decision is not which model — it is where the model runs. Three deployment patterns dominate:

Cloud APIs (OpenAI Whisper API, Google Cloud Speech, Azure Speech, ElevenLabs, Deepgram) offer the lowest operational burden. You send audio, you get text or audio back. Latency is network-bound: 200–800 ms round-trip for STT, 300–1000 ms for TTS. Pricing is per-minute or per-character. Use these when team size is small, traffic is bursty, or you need multilingual support without maintaining model infrastructure.

Self-hosted on GPU (Whisper.cpp, faster-whisper, NeMo, Coqui TTS, Piper, Bark) gives you control over latency, cost at scale, and data residency. A single A10G can serve ~50 concurrent STT streams with faster-whisper large-v3 at 2× real-time factor. TTS is lighter: Piper runs real-time on CPU for single-speaker voices. Choose this when you have steady volume (>10k minutes/day), strict PII requirements, or need custom vocabulary adaptation.

Hybrid — cloud for overflow and rare languages, self-hosted for baseline traffic — is the most common production pattern. Route based on language detection, confidence thresholds, or capacity signals.

# routing logic example
async def route_stt(audio: bytes, lang: str) -> str:
    if lang in SELF_HOSTED_LANGS and not capacity_exceeded():
        return await self_hosted_stt(audio, lang)
    return await cloud_stt(audio, lang)

Pitfall: do not underestimate the operational cost of self-hosting. Model updates, CUDA driver compatibility, queue management, and autoscaling policies consume engineering time. Budget 0.5–1 FTE per model family in steady state.

Design the audio pipeline

Raw audio from clients arrives in varying sample rates, codecs, and channel configurations. Normalize at the edge.

import numpy as np
import soundfile as sf
import io

TARGET_SR = 16000  # Whisper, most STT models expect 16 kHz mono

def normalize_audio(raw: bytes, src_sr: int, channels: int) -> np.ndarray:
    # decode whatever the client sent
    audio, _ = sf.read(io.BytesIO(raw), dtype="float32")
    # downmix to mono
    if audio.ndim > 1:
        audio = audio.mean(axis=1)
    # resample if needed
    if src_sr != TARGET_SR:
        import librosa
        audio = librosa.resample(audio, orig_sr=src_sr, target_sr=TARGET_SR)
    return audio

Voice activity detection (VAD) sits before STT. Silero VAD (ONNX, ~1 MB) runs on CPU with negligible latency and filters silence, reducing compute spend by 30–60% in typical call center traffic.

# silero vad usage
import torch

model, utils = torch.hub.load(repo_or_dir="snakers4/silero-vad", model="silero_vad", force_reload=False)
(get_speech_timestamps, _, read_audio, _, _) = utils

def vad_segments(audio: np.ndarray, sr: int = 16000) -> list[tuple[float, float]]:
    tensor = torch.from_numpy(audio).unsqueeze(0)
    timestamps = get_speech_timestamps(tensor, model, sampling_rate=sr)
    return [(ts["start"] / sr, ts["end"] / sr) for ts in timestamps]

Tradeoff: aggressive VAD cuts cost but risks clipping soft speech. Tune threshold (default 0.5) and min_silence_duration_ms per domain. Call centers tolerate false negatives better than voice assistants.

Handle streaming vs. batch

Voice assistants need streaming (partial results as user speaks). Call centers often process recorded calls in batch. The model interface differs.

Streaming STT requires a WebSocket or gRPC connection with chunked audio. Whisper does not natively stream; you segment with VAD and send windows. Deepgram, Gladia, and AssemblyAI offer true streaming endpoints. For self-hosted, use faster-whisper with vad_filter=True on rolling buffers, or integrate whisper.cpp with its streaming branch.

# pseudo-streaming with faster-whisper
from faster_whisper import WhisperModel

model = WhisperModel("large-v3", device="cuda", compute_type="float16")

async def stream_transcribe(audio_chunks: AsyncIterator[bytes]) -> AsyncIterator[str]:
    buffer = bytearray()
    async for chunk in audio_chunks:
        buffer.extend(chunk)
        # process every 2 seconds of audio
        if len(buffer) >= 2 * TARGET_SR * 2:  # 16-bit mono
            segments, _ = model.transcribe(np.frombuffer(buffer, dtype=np.int16).astype(np.float32) / 32768.0,
                                           vad_filter=True, language="en")
            for seg in segments:
                yield seg.text
            buffer.clear()

Batch STT is simpler: send the full file, get a transcript with timestamps. Use this for call recording analysis, voicemail transcription, and compliance archives.

TTS streaming matters for perceived latency. Chunked synthesis (sending first audio bytes before full generation completes) reduces time-to-first-byte from ~800 ms to ~150 ms for long utterances. ElevenLabs, PlayHT, and self-hosted Piper support streaming. Buffer 1–2 sentences client-side to avoid choppy playback.

Manage vocabulary and accuracy

Generic models fail on domain terms: drug names, product SKUs, alphanumeric codes. Three levers exist:

Prompting / context — Whisper accepts a prompt parameter (up to 224 tokens) that biases decoding. Pass the last N words of conversation or a domain word list.

DOMAIN_PROMPT = "The patient was prescribed metformin lisinopril atorvastatin. Appointment scheduled with Dr. Chen."

segments, _ = model.transcribe(audio, initial_prompt=DOMAIN_PROMPT, language="en")

Custom vocabulary / phrase boost — Cloud APIs (Deepgram, Google, Azure) let you boost specific phrases with weight. Self-hosted Whisper cannot do this natively; you need a weighted finite-state transducer (WFST) shallow fusion layer or switch to a model that supports it (NeMo Conformer-CTC with WFST).
Fine-tuning / LoRA — For high-volume domains, fine-tune Whisper or a Conformer-CTC model on 50–200 hours of in-domain audio. LoRA adapters (rank 32–64) train in hours on a single GPU and preserve multilingual capability. Expect 10–20% relative WER reduction on domain terms.

Pitfall: prompting helps but cannot fix systematic errors (homophones, acronyms). Measure WER on a held-out eval set per domain before investing in fine-tuning.

Build for failure modes

Speech pipelines fail in predictable ways. Design for each:

Low confidence — STT returns no_speech_prob or average token logprob. Reject or escalate when confidence < threshold.

def should_escalate(segments, threshold=-0.8):
    avg_logprob = np.mean([seg.avg_logprob for seg in segments])
    return avg_logprob < threshold

Speaker diarization errors — Call centers need speaker labels. PyAnnote.audio 3.1 runs diarization as a separate pass. Run it async; do not block STT. Merge diarization segments with STT words by timestamp overlap.

Background noise / music — VAD helps. For music-heavy environments (hold music, retail), add a music classifier (e.g., audionet or yamnet) and route those segments to a noise-robust model or human review.

TTS pronunciation errors — Proper nouns, abbreviations, and numbers trip TTS. Pre-process text with a normalization pipeline:

import re

ABBREVIATIONS = {
    "dr.": "doctor",
    "mr.": "mister",
    "mrs.": "missus",
    "etc.": "etcetera",
    "api": "A P I",
    "sdk": "S D K",
}

def normalize_tts_text(text: str) -> str:
    # expand abbreviations
    for abbr, expansion in ABBREVIATIONS.items():
        text = re.sub(rf"\b{re.escape(abbr)}\b", expansion, text, flags=re.IGNORECASE)
    # spell out numbers in certain contexts
    text = re.sub(r"\b(\d{4,})\b", lambda m: " ".join(m.group(1)), text)  # credit cards, IDs
    return text

Test TTS output with a pronunciation eval set. Automate: synthesize, run through STT, compare to expected text.

Monitor what matters

Dashboards should show:

End-to-end latency (P50, P95, P99) from audio ingress to first token / first audio byte
Concurrent streams vs. capacity
WER / CER on sampled traffic (human-annotated weekly)
Fallback rate (cloud → self-hosted or primary → backup provider)
Cost per minute by model and provider

Alert on P99 latency > 2× target, fallback rate > 5%, or WER regression > 2% absolute.

# example prometheus alerts
- alert: STTLatencyHigh
  expr: histogram_quantile(0.99, rate(stt_latency_seconds_bucket[5m])) > 3.0
  for: 5m
  labels:
    severity: critical
  annotations:
    summary: "STT P99 latency > 3s"

- alert: STTFallbackRateHigh
  expr: rate(stt_fallback_total[5m]) / rate(stt_requests_total[5m]) > 0.05
  for: 2m
  labels:
    severity: warning

Integrate with the conversation layer

Speech models voice assistants sit behind an orchestration layer that manages turn-taking, barge-in, and context. The orchestration layer decides when to invoke STT, when to cancel TTS (user interrupted), and how to handle partial results.

class VoiceOrchestrator:
    def __init__(self, stt, tts, llm):
        self.stt = stt
        self.tts = tts
        self.llm = llm
        self.tts_task = None

    async def on_audio_chunk(self, chunk: bytes):
        # feed streaming STT
        partial = await self.stt.feed(chunk)
        if partial.is_final:
            # cancel any in-flight TTS (barge-in)
            if self.tts_task:
                self.tts_task.cancel()
            # get LLM response
            response = await self.llm.generate(partial.text)
            # stream TTS
            self.tts_task = asyncio.create_task(self.tts.stream(response))

    async def on_silence(self, duration_ms: int):
        if duration_ms > 800:  # endpointing threshold
            await self.stt.finalize()

Key parameters: endpointing silence threshold (600–1000 ms), barge-in cancellation latency (< 100 ms), and partial result emission cadence (200–500 ms). Tune per use case: voice assistants favor lower thresholds; call centers favor stability.

Plan for multilingual

If you support >3 languages, do not run separate models per language. Whisper large-v3 and SeamlessM4T handle 100+ languages in one model. Detect language on the first 1–2 seconds of audio, then pin the language for the session to avoid mid-conversation switches.

def detect_language(audio: np.ndarray) -> str:
    # use first 3 seconds
    sample = audio[:3 * TARGET_SR]
    segments, info = model.transcribe(sample, language=None, task="transcribe")
    return info.language  # iso code

For TTS, choose a multilingual model (XTTS v2, Bark, SeamlessM4T) or maintain per-language voices. XTTS v2 clones a reference speaker across languages but adds ~200 ms latency. Pre-generate common prompts in each language if latency is critical.

Summary checklist

Before shipping:

Speech models voice assistants and call centers succeed when the audio pipeline, model selection, and failure handling are engineered as a system — not when you chase the latest benchmark. Start with cloud APIs, measure your actual traffic patterns, then migrate the hot path to self-hosted only when the numbers justify it.