DEV Community: VRM AI Technology

How AI Calling Agents Work: STT, LLM and TTS in One Call

VRM AI Technology — Thu, 22 Oct 2026 12:38:38 +0000

A voice AI agent looks like one thing on the phone, but a single call is three systems passing data to each other: speech-to-text (STT), a language model (LLM) and text-to-speech (TTS). Most quality problems come from how those three are connected, not from any one of them.

This post walks through the pipeline, shows a small runnable skeleton in Python, and lists the design choices that matter in production.

The three stages

STT turns the caller's audio into text. Streaming STT sends partial transcripts while the caller is still speaking, and a final transcript when they stop.
LLM reads the transcript plus conversation context and produces a reply, token by token.
TTS turns reply text into audio that is played back to the caller.

Around these sit telephony (the audio transport), turn detection (deciding when the caller has finished) and business logic (tools, CRM lookups, escalation to a human).

The naive version and why it feels slow

The simplest design runs the stages one after another: wait for the full transcript, wait for the full LLM reply, then synthesize the whole reply. Each stage waits for the previous one to finish, so the caller hears silence for the sum of all three. On a phone call that gap feels like the line has dropped.

Stream between stages

The fix is to overlap the stages. Start speaking as soon as the first sentence of the reply is complete, instead of waiting for the entire reply.

Here is a minimal skeleton. The three functions are stand-ins for real STT, LLM and TTS clients. The structure is the point:

import asyncio
import re

SENTENCE_END = re.compile(r"(?<=[.!?])\s+")


async def transcribe(audio_chunks):
    """Stand-in for a streaming STT. Yields (text, is_final)."""
    words = ["what", "are", "your", "support", "hours"]
    for i, w in enumerate(words):
        await asyncio.sleep(0.01)
        yield " ".join(words[: i + 1]), i == len(words) - 1


async def llm_stream(prompt):
    """Stand-in for a streaming LLM. Yields tokens."""
    reply = "We are open from nine to six. Monday to Saturday. Anything else?"
    for tok in reply.split(" "):
        await asyncio.sleep(0.005)
        yield tok + " "


async def synthesize(sentence):
    """Stand-in for TTS. Returns audio bytes for one sentence."""
    await asyncio.sleep(0.01)
    return sentence.encode()


async def speak(audio_out, audio):
    await audio_out.put(audio)


async def handle_turn(audio_chunks, audio_out):
    final_text = ""
    async for text, is_final in transcribe(audio_chunks):
        if is_final:
            final_text = text

    buffer = ""
    async for token in llm_stream(final_text):
        buffer += token
        parts = SENTENCE_END.split(buffer)
        # every part except the last is a complete sentence
        for sentence in parts[:-1]:
            await speak(audio_out, await synthesize(sentence))
        buffer = parts[-1]
    if buffer.strip():
        await speak(audio_out, await synthesize(buffer.strip()))

Running it prints the reply one sentence at a time:

We are open from nine to six.
Monday to Saturday.
Anything else?

The first sentence reaches TTS while the LLM is still generating the rest. In a real system the caller starts hearing audio after the first sentence, not after the whole reply.

Design choices that matter

Turn detection. Cutting the caller off is worse than a small delay. Use the STT's end-of-speech signal plus a short silence window, and tune that window against real calls.

Barge-in. Callers interrupt. When new speech arrives while the agent is talking, stop playback, cancel the pending LLM and TTS work, and start a new turn.

Context and tools. Keep the prompt small. Fetch data (order status, appointments) with tool calls, and have the agent say a short filler line while a slow tool runs.

Fallbacks. Decide in advance what happens when STT confidence is low, an API times out, or the caller asks for a human. A clear handoff beats a confident wrong answer.

Measure it. Log timestamps per stage for every turn so you can see where time goes. Without per-stage numbers, you are guessing.

Summary

A calling agent is a streaming pipeline, not three separate calls. Overlap the stages, handle interruptions, and measure each stage. The model matters, but the plumbing decides whether the call feels natural.

We build conversational voice agents at VRM AI Technology. If you are planning a Generative AI project, see our generative AI development work too.

Measuring and Reducing Turn Latency in Voice AI Agents

VRM AI Technology — Mon, 19 Oct 2026 06:55:26 +0000

In a voice agent, response latency shapes how natural the conversation feels. A text chatbot can pause for a second or two without much harm. On a phone call, the same pause is noticeable and often reads as a dropped line. This post breaks one conversational turn into stages, shows how to instrument each stage, and lists common techniques for reducing latency.

Where a turn spends its time

A turn runs from the moment the caller stops speaking to the first audio of the reply. The main stages are:

Endpointing: deciding that the caller has finished speaking.
ASR finalization: producing the final transcript after the last audio frame.
LLM time to first token (TTFT): the delay before the model starts generating.
Tool calls: external lookups such as CRM, calendar or database queries.
TTS time to first byte (TTFB): the delay before speech synthesis emits its first audio chunk.

Without streaming, total turn latency is approximately the sum of these stages. With streaming, some stages overlap (covered below). Either way, each stage needs its own measurement before you decide where to optimize.

Instrumenting a turn

A small tracer is enough to start. It stores the duration of each stage in milliseconds and reports percentiles, because averages hide the slow turns that callers actually notice.

import time
import statistics
from collections import defaultdict
from contextlib import contextmanager

MIN_SAMPLES = 20  # a p95 needs enough samples to mean anything


class TurnTracer:
    """Collects per-stage durations (in milliseconds) across many turns."""

    def __init__(self):
        self.samples = defaultdict(list)

    def add(self, stage, elapsed_ms):
        self.samples[stage].append(elapsed_ms)

    @contextmanager
    def record(self, stage):
        start = time.monotonic()
        try:
            yield
        finally:
            self.add(stage, (time.monotonic() - start) * 1000)

    def report(self):
        out = {}
        for stage, values in self.samples.items():
            if len(values) < MIN_SAMPLES:
                continue
            cuts = statistics.quantiles(values, n=20)  # 19 cut points
            out[stage] = {
                "count": len(values),
                "p50_ms": round(statistics.median(values), 1),
                "p95_ms": round(cuts[18], 1),  # 95th percentile
            }
        return out

For stages that complete in one call, such as a tool lookup, wrap them with record:

tracer = TurnTracer()

with tracer.record("tool_call"):
    result = lookup_customer(phone_number)  # your own function

For streaming stages such as LLM TTFT, stop the clock at the first token rather than at the end of the response, then keep consuming the stream:

start = time.monotonic()
first_token_seen = False

for token in llm_stream(prompt):  # your own streaming client
    if not first_token_seen:
        tracer.add("llm_ttft", (time.monotonic() - start) * 1000)
        first_token_seen = True
    # pass each token on to the next stage (for example, TTS) here

Use time.monotonic() instead of wall-clock time so system clock adjustments cannot distort durations.

Reading the numbers

Compare p95 with p50. When p50 looks fine but p95 is high, the cause is often intermittent: cold connections, long prompts, retries or network jitter. Then compare stages against each other, find the one that takes the largest share of the turn, and work on that one first.

Streaming between stages

Waiting for each stage to finish before starting the next one adds up quickly. Overlapping stages is one of the most effective ways to reduce perceived latency:

Stream ASR partial results so the final transcript is available soon after speech ends.
Stream LLM tokens and start TTS as soon as the first complete sentence or clause is ready.
Stream TTS audio to the caller in chunks instead of waiting for a finished file.

With streaming, the caller can hear the start of the reply while later parts are still being generated.

Barge-in

Callers interrupt. When voice activity is detected during playback, stop audio output, cancel any in-flight generation and return to listening. Treat the interruption path as part of your latency budget, because a slow reaction to an interruption feels as unnatural as a slow answer.

A regression test for latency

Latency often creeps up as prompts, tools and models change. One approach is to keep a fixed set of recorded test utterances, replay them in CI and fail the build when a stage exceeds a budget.

def test_turn_latency_budget():
    report = run_replay_suite()  # your own helper: replays utterances, returns tracer.report()
    assert report["llm_ttft"]["p95_ms"] < 800
    assert report["tool_call"]["p95_ms"] < 300

The thresholds above are placeholders. Set budgets from your own measurements and product requirements.

Summary

Break each turn into stages and measure every stage.
Track p50 and p95 rather than averages.
Stream between stages so they overlap.
Handle barge-in as a first-class path.
Guard your latency budgets with a replay test in CI.

Written by the team at VRM AI Technology, an AI development company building Generative AI systems, conversational voice agents and custom machine learning software.