Build a Voice Agent in 2026: LiveKit Pipeline, Tool Calls and a Latency Budget

1. The Problem: Voice Is a Pipeline With a Clock

A voice turn crosses at least five stages: the network, voice activity detection, speech-to-text, the LLM and text-to-speech. The user hears silence for the sum of whatever you did not overlap. LiveKit’s docs say a conversation feels natural when end-to-end latency stays under one second, and Pipecat’s intro says its pipelines typically land between 500 and 800 ms. Both are vendor statements, not guarantees, and neither knows your STT, your model or your region.

The fix is structural, not a faster model: stream at every stage, decide turn boundaries with something smarter than a silence timer, and treat interruption as a first-class event. That is what the frameworks below give you, and why I would not hand-roll this loop.

For the speech-to-speech alternative and its price per minute, see my Grok Voice 3.0 post. This one is the pipeline route, built by hand.

2. Minimum Concepts Before Touching Code

Six ideas explain nearly every knob you will touch.

3. LiveKit Agents vs Pipecat: What I Verified

Both are open-source Python frameworks for exactly this pipeline, and both are good. The table lists only facts I read in the official docs, source tags or package indexes on October 14, 2026.

I use LiveKit for the worked example for one reason: its quickstart is a complete single file, so I could check every call against the 1.8.7 source tag. Pipecat’s docs quickstart shows fragments, and its docs and quickstart repository currently use different class names (see the note below the table).

4. Step 1: Scaffold the Agent and Run the Baseline

Everything in this step comes from LiveKit’s Voice AI quickstart. I have not run it against a live project: I copied it from the docs and checked the imports and signatures against the livekit-agents 1.8.7 tag on GitHub.

# 1) LiveKit CLI (pick your OS), then link your LiveKit Cloud project
winget install LiveKit.LiveKitCLI             # Windows
brew install livekit-cli                      # macOS
curl -sSL https://get.livekit.io/cli | bash   # Linux
lk cloud auth

# 2) Scaffold the official Python starter and run it
lk agent init my-agent --template agent-starter-python
cd my-agent
uv sync
lk agent console   # talk to it in your terminal (add --text for text only)
lk agent dev       # connect it to LiveKit and use the Agent Console

from dotenv import load_dotenv

from livekit import agents
from livekit.agents import AgentServer, AgentSession, Agent, inference, room_io, TurnHandlingOptions
from livekit.plugins import ai_coustics

load_dotenv(".env.local")


class Assistant(Agent):
    def __init__(self) -> None:
        super().__init__(
            instructions="""You are a helpful voice AI assistant.
            You eagerly assist users with their questions by providing information from your extensive knowledge.
            Your responses are concise, to the point, and without any complex formatting or punctuation including emojis, asterisks, or other symbols.
            You are curious, friendly, and have a sense of humor.""",
        )

server = AgentServer()

@server.rtc_session(agent_name="my-agent")
async def my_agent(ctx: agents.JobContext):
    session = AgentSession(
        stt=inference.STT(model="assemblyai/universal-3-6-pro", language="en"),
        llm=inference.LLM(model="google/gemma-4-31b-it"),
        tts=inference.TTS(
            model="fishaudio/s2.1-pro",
            voice="fa4c9eb3dccc4806b382b40d61c6b10a",
        ),
        turn_handling=TurnHandlingOptions(
            turn_detection=inference.TurnDetector(),
        ),
    )

    await session.start(
        room=ctx.room,
        agent=Assistant(),
        room_options=room_io.RoomOptions(
            audio_input=room_io.AudioInputOptions(
                noise_cancellation=ai_coustics.audio_enhancement(model=ai_coustics.EnhancerModel.QUAIL_VF_S),
            ),
        ),
    )

    await session.generate_reply(
        instructions="Greet the user and offer your assistance."
    )


if __name__ == "__main__":
    agents.cli.run_app(server)

Read the file as a pipeline. inference.STT, inference.LLM and inference.TTS are the three streaming stages, served by LiveKit Inference so you need no provider keys but do need a LiveKit Cloud project. inference.TurnDetector() is the end-of-turn model, and the session provides the VAD it needs. The instructions tell the model to answer briefly with no emojis or asterisks, because the output is spoken; the session also strips markdown and emoji by default in 1.8.7.

5. Step 2: Turn-Taking, VAD and Barge-In

Turn-taking is where a voice agent feels human or robotic, and in 1.8.x it lives in one object: TurnHandlingOptions. This is the shape of LiveKit’s recommended starting config, trimmed to the parts that matter.

from livekit.agents import AgentSession, TurnHandlingOptions, inference

session = AgentSession(
    stt=inference.STT(model="assemblyai/universal-3-6-pro", language="en"),
    llm=inference.LLM(model="google/gemma-4-31b-it"),
    tts=inference.TTS(model="fishaudio/s2.1-pro", voice="fa4c9eb3dccc4806b382b40d61c6b10a"),
    turn_handling=TurnHandlingOptions(
        # audio model that predicts end of turn on top of VAD
        turn_detection=inference.TurnDetector(),
        # wait after detected silence before closing the turn
        # (0.3 / 2.5 are the documented defaults with the audio turn detector)
        endpointing={"mode": "fixed", "min_delay": 0.3, "max_delay": 2.5},
        interruption={
            "mode": "adaptive",   # default on LiveKit Cloud; use "vad" when self-hosting
            "min_duration": 0.5,  # seconds of speech that count as a barge-in
            "min_words": 0,       # raise it (needs STT) to ignore "okay", "uh-huh"
            "resume_false_interruption": True,
            "false_interruption_timeout": 2.0,
        },
        # on by default; preemptive TTS trades wasted compute for lower latency
        preemptive_generation={"preemptive_tts": False},
    ),
)

# uv add "livekit-agents[silero,deepgram,openai,cartesia]~=1.8"
# env: DEEPGRAM_API_KEY, OPENAI_API_KEY, CARTESIA_API_KEY
from livekit.agents import AgentSession, TurnHandlingOptions, inference
from livekit.plugins import cartesia, deepgram, openai, silero

session = AgentSession(
    vad=silero.VAD.load(),
    stt=deepgram.STT(model="nova-3", language="en"),
    llm=openai.responses.LLM(model="gpt-4.1"),
    tts=cartesia.TTS(model="sonic-3", voice="f786b574-daa5-4673-aa0c-cbe3e8534c02"),
    turn_handling=TurnHandlingOptions(
        turn_detection=inference.TurnDetector(version="v1-mini"),  # local CPU
    ),
)

Three defaults worth knowing, all from LiveKit’s docs: with the audio turn detector the session closes a turn after 0.3 s minimum and 2.5 s maximum, against 0.5 s and 3.0 s without it; preemptive generation is on and preemptive TTS is off; and adaptive interruption handling, which filters backchannel with an audio model, is a LiveKit Cloud feature.

6. Step 3: Function Calling Mid-Call Without Dead Air

A tool call in a voice agent is silence the user can hear. LiveKit gives you three behaviours, and picking the right one per tool is most of the job: a short read-only lookup you cancel if the user barges in, a write that must never end half done, and a slow job that keeps the agent talking.

import asyncio

from livekit.agents import Agent, RunContext, function_tool
from livekit.agents.llm import ToolError


class SupportAgent(Agent):
    def __init__(self) -> None:
        super().__init__(instructions="You are a phone support agent. Short sentences, no formatting.")

    # 1) Short read-only lookup: stop the work if the user talks over the agent
    @function_tool
    async def get_order_status(self, order_id: str, run_ctx: RunContext) -> str | None:
        """Look up the status of an order when the user asks about it.

        Args:
            order_id: The ID of the order to look up.
        """
        lookup = asyncio.ensure_future(self._lookup_order(order_id))  # your backend call
        await run_ctx.speech_handle.wait_if_not_interrupted([lookup])
        if run_ctx.speech_handle.interrupted:
            lookup.cancel()  # Python does not cancel it for you
            return None      # empty result, no reply
        return lookup.result()

    # 2) Write: a barge-in must not leave it half done
    @function_tool
    async def cancel_order(self, order_id: str, run_ctx: RunContext) -> str:
        """Cancel an order. Call only after the user confirmed out loud.

        Args:
            order_id: The ID of the order to cancel.
        """
        run_ctx.disallow_interruptions()
        ok = await self._cancel_in_backend(order_id)  # your backend call, with a timeout
        if not ok:
            raise ToolError("The order could not be cancelled. Offer to transfer the caller to a person.")
        return f"Order {order_id} is cancelled."

    # 3) Slow job: acknowledge now, fill the silence, answer when done
    @function_tool()
    async def export_statement(self, ctx: RunContext, month: str) -> str:
        """Generate a statement and email it. Takes up to a minute.

        Args:
            month: Month to export, YYYY-MM.
        """
        await ctx.update(f"Generating the statement for {month}. This can take a minute.")
        async with ctx.with_filler("Still working on it, hang on a sec.", delay=5):
            await self._build_statement(month)  # your slow backend call
        return f"The statement for {month} was sent to the email on file."

7. Step 4: Measure Every Turn Before You Tune

LiveKit attaches a metrics report to every chat message. The fields map one to one onto the budget table, so you can see which stage ate the turn.

from livekit.agents import ConversationItemAddedEvent
from livekit.agents.llm import ChatMessage


@session.on("conversation_item_added")
def on_item(ev: ConversationItemAddedEvent):
    if not isinstance(ev.item, ChatMessage):
        return
    m = ev.item.metrics  # per-turn MetricsReport
    if ev.item.role == "user":
        print("stt:", m.get("transcription_delay"), "eot:", m.get("end_of_turn_delay"))
    elif ev.item.role == "assistant":
        print(
            "llm ttft:", m.get("llm_node_ttft"),
            "tts ttfb:", m.get("tts_node_ttfb"),
            "e2e:", m.get("e2e_latency"),
        )

Log these per turn, then look at p50 and p95 per stage over a few dozen real calls from your target region. llm_node_ttft and tts_node_ttfb are only filled in pipeline mode. The 1.8.7 release notes added the same end-to-end latency for realtime agents when local VAD is configured. Change one knob at a time: LiveKit’s own tuning page warns that preemptive generation does not always reduce latency.

8. Phone Calls: The Hand-Off at a Conceptual Level

The agent code does not change when the caller is on a phone: a SIP caller is another participant in a room. What changes is the front door and what you do when the agent should stop.

9. Failure Modes: Echo, Noisy Rooms, Long Tool Calls

Demos hide these three. Production calls find them in the first week.

10. Common Mistakes (and the Boring Fixes)

Most voice-agent bugs I read about fall into these buckets. The left column is what holds up; the right column is what pages you.

Conclusion

You now have the full loop: a streaming STT → LLM → TTS pipeline on LiveKit Agents, turn-taking you can reason about, tools that behave under interruption, a phone path in outline and a metric for every stage. Pipecat would give you the same shape with different names and a different transport story, and the comparison table is there so you can choose without a benchmark war.

This week: scaffold the quickstart, add one read-only tool with interruption handling, log the four per-turn metrics for 20 calls and fill in your own budget table. Then decide whether the pipeline or the speech-to-speech model from my previous post wins on your traffic.

Sources