How to Build a Voice AI Agent with LiveKit Agents or Pipecat (2026)

Oct 31, 2026 · 16 min · Ajay Kumar

This guide is for engineers who want a working voice AI agent they can talk to from a browser or a phone, built on an open-source framework rather than a black-box voice platform. By the end you will have a cascaded agent (speech-to-text, an LLM with two tools, text-to-speech) running on LiveKit Agents 1.8.10 for Python, with audio end-of-turn detection, per-turn latency logging, a LiveKit Cloud deployment and a US phone number routed to it over SIP. The last step rebuilds it on Pipecat 1.15.0. Both versions were checked against PyPI on 31 October 2026, and both files below import and start cleanly in a fresh virtual environment. You need Python 3.10 to 3.14 (Pipecat needs 3.11 or later), uv, a free LiveKit Cloud project, and for the Pipecat step, Deepgram, OpenAI and Cartesia API keys.

Step 1: Pick the architecture before the framework

There are three ways to build a voice agent: a cascaded STT, LLM and TTS pipeline, a turn-based speech-to-speech model such as OpenAI's gpt-realtime-2.1, or a full-duplex model. I explain the trade-offs and the latency arithmetic in the companion post on voice AI agents in 2026. For this guide I use cascaded, because LiveKit's own pipeline types page recommends it as the production default, and because every stage produces text you can log, test and audit.

LiveKit Agents (Apache-2.0) bundles rooms, dispatch, SIP, a turn detector, inference and hosting. Pipecat (BSD-2-Clause) is a frame pipeline over whatever transport you hand it, which suits teams that already own a telephony stack.

Here is the pipeline we are building and where each latency number comes from:

Cascaded voice agent: stages and the metric that times each one Caller audio WebRTC or SIP trunk network in/out Turn detection Silero + TurnDetector end_of_turn_delay STT deepgram/nova-3 transcription_delay LLM + tools openai/gpt-5.4-mini llm_node_ttft TTS cartesia/sonic-3 tts_node_ttfb Agent audio streams back to the caller as soon as the first TTS chunk exists e2e_latency (LiveKit): from the user stopping speaking to the agent starting to respond. Preemptive generation (on by default) starts the LLM before the end-of-turn wait finishes, so stages overlap. Pipecat: UserBotLatencyObserver times VAD stop to BotStartedSpeakingFrame, with a per-service breakdown. Budget the p95 per region, not the average.
The pipeline built in this guide. Metric names are the per-turn fields documented in LiveKit's data hooks reference and the Pipecat UserBotLatencyObserver source in pipecat-ai 1.15.0.

Step 2: Install the LiveKit CLI and create the project

The lk CLI manages keys, deployments, phone numbers and dispatch rules. The current release is v2.19.2, published 22 October 2026.

# macOS (Linux: curl -sSL https://get.livekit.io/cli | bash)
brew install livekit-cli

# Link the CLI to your LiveKit Cloud project (opens a browser)
lk cloud auth

# Create a uv project with pinned dependencies
uv init voice-agent --python 3.12 && cd voice-agent
uv add "livekit-agents==1.8.10" python-dotenv

# Write LIVEKIT_URL, LIVEKIT_API_KEY and LIVEKIT_API_SECRET to .env.local
lk app env -w

lk agent init my-agent --template agent-starter-python clones LiveKit's Python starter instead, with tests. The audio turn detector is built into the SDK (since 1.6.1), and with LiveKit Inference one LiveKit key pays for STT, LLM and TTS.

Step 3: Write the agent with STT, LLM and TTS

Create agent.py. The Agent subclass holds the instructions and tools; the AgentSession holds the models and turn handling; the AgentServer registers an entrypoint that runs once per room.

# agent.py: cascaded voice agent on LiveKit Agents 1.8.10 (Python 3.12)
import logging
import statistics
from datetime import datetime, timezone

from dotenv import load_dotenv
from livekit import agents
from livekit.agents import (
    Agent,
    AgentServer,
    AgentSession,
    ConversationItemAddedEvent,
    RunContext,
    TurnHandlingOptions,
    function_tool,
    inference,
)
from livekit.agents.llm import ChatMessage

load_dotenv(".env.local")
logger = logging.getLogger("voice-agent")


class Receptionist(Agent):
    def __init__(self) -> None:
        super().__init__(
            instructions=(
                "You are a phone receptionist for a small clinic. Keep answers to one or two "
                "short sentences. Never use lists, markdown or emojis: everything you write is spoken."
            ),
        )

    @function_tool()
    async def check_opening_hours(self, context: RunContext, day: str) -> dict:
        """Look up the clinic's opening hours for a day of the week.

        Args:
            day: Day of the week in English, for example "Monday".
        """
        hours = {"saturday": "9am to 1pm", "sunday": "closed"}
        return {"day": day, "hours": hours.get(day.lower(), "8am to 6pm")}

    @function_tool()
    async def current_time_utc(self, context: RunContext) -> str:
        """Return the current time in UTC, for questions about today or now."""
        return datetime.now(timezone.utc).strftime("%A %H:%M UTC")


server = AgentServer()


@server.rtc_session(agent_name="receptionist")  # explicit dispatch name
async def entrypoint(ctx: agents.JobContext):
    session = AgentSession(
        # LiveKit Inference model IDs: one LIVEKIT_API_KEY covers all three
        stt=inference.STT(model="deepgram/nova-3", language="en"),
        llm=inference.LLM(model="openai/gpt-5.4-mini"),
        tts=inference.TTS(model="cartesia/sonic-3", voice="f786b574-daa5-4673-aa0c-cbe3e8534c02"),
        turn_handling=TurnHandlingOptions(
            turn_detection=inference.TurnDetector(),  # audio end-of-turn model
            endpointing={"mode": "fixed", "min_delay": 0.3, "max_delay": 2.5},
        ),
    )

    e2e: list[float] = []

    @session.on("conversation_item_added")
    def _log_latency(ev: ConversationItemAddedEvent):
        if not isinstance(ev.item, ChatMessage):
            return
        m = ev.item.metrics
        if ev.item.role == "user":
            logger.info("eot_delay=%s stt_delay=%s", m.get("end_of_turn_delay"), m.get("transcription_delay"))
        elif m.get("e2e_latency") is not None:
            e2e.append(m["e2e_latency"])
            logger.info(
                "e2e=%.3fs llm_ttft=%s tts_ttfb=%s",
                m["e2e_latency"], m.get("llm_node_ttft"), m.get("tts_node_ttfb"),
            )

    async def _summary():
        if len(e2e) >= 2:
            p95 = statistics.quantiles(e2e, n=20)[-1]
            logger.info("turns=%d e2e_p50=%.3fs e2e_p95=%.3fs", len(e2e), statistics.median(e2e), p95)

    ctx.add_shutdown_callback(_summary)

    await session.start(room=ctx.room, agent=Receptionist())
    await session.generate_reply(instructions="Greet the caller and ask how you can help.")


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

The model IDs come from LiveKit's STT, LLM and TTS model tables, which also mark retirements. On the day I looked, cartesia/sonic-2 had been retired on 20 October 2026 and openai/gpt-4.1-mini was due to retire in January 2027. Treat model IDs like base images: they go end of life, so put the dates in a calendar.

To call providers directly with your own keys instead, swap the three inference.* lines for plugins (uv add "livekit-agents[deepgram,openai,cartesia]==1.8.10"):

from livekit.plugins import cartesia, deepgram, openai

session = AgentSession(
    stt=deepgram.STT(model="nova-3", language="en-US"),   # reads DEEPGRAM_API_KEY
    llm=openai.LLM(model="gpt-5.4-mini"),                 # reads OPENAI_API_KEY
    tts=cartesia.TTS(model="sonic-3"),                    # reads CARTESIA_API_KEY
    turn_handling=TurnHandlingOptions(turn_detection=inference.TurnDetector()),
)

Step 4: Configure VAD and turn detection

VAD only knows that sound stopped; it cannot tell that "I'd like to book a flight from Dublin" has more coming. LiveKit's turn detector is an audio model that listens to intonation and rhythm as well as content, and it is the default in AgentSession. It comes in two versions: v1 runs on LiveKit Inference, and v1-mini runs locally on CPU and is free anywhere. Both cover 14 languages.

The defaults you are tuning against, from the turn-taking tuning page:

Option Default With audio turn detector What it does
endpointing.min_delay 0.5 s 0.3 s Minimum wait after silence before the turn closes
endpointing.max_delay 3.0 s 2.5 s Longest wait before forcing the turn closed
interruption.mode adaptive if available, else vad same Whether "uh-huh" counts as a barge-in
interruption.min_duration 0.5 s same Speech needed to register an interruption
preemptive_generation.enabled True same Start the LLM on the final transcript, before the turn is confirmed
Silero min_silence_duration 0.55 s must be at least 0.25 s VAD silence before speech is marked as ended

Two practical rules follow. First, if you lower Silero's min_silence_duration below 0.25 s while the turn detector is on, the session raises a ValueError at start. Second, the detector commits a turn anyway if it has not answered within about a second, and by default a timeout on v1 falls back to v1-mini for the rest of the session. To pin the local model (useful when self-hosting, or to keep turn detection free and offline):

turn_handling=TurnHandlingOptions(
    turn_detection=inference.TurnDetector(version="v1-mini"),
    endpointing={"mode": "dynamic", "min_delay": 0.3, "max_delay": 2.5},  # dynamic is Python only
    interruption={"mode": "adaptive", "min_duration": 0.5},
)

If you are building a push-to-talk app, use turn_detection="manual" and commit turns yourself. If your STT does its own endpointing (AssemblyAI, Deepgram Flux), turn_detection="stt" uses that signal and keeps VAD for interruptions.

Step 5: Add tool calling

The two @function_tool() methods above are all it takes: the function name becomes the tool name and the docstring, including Args:, becomes the schema. The general patterns are in how to build an AI agent with tool calling in Python; voice adds three habits. Return small, speakable data, because a large JSON result is context the LLM reads before its first word. Make slow tools asynchronous so the agent keeps talking. And note that AgentSession defaults to max_tool_steps=3, so longer chains of dependent calls stop short.

For a tool that should not fire during the greeting, use the IGNORE_ON_ENTER flag described in the function tools docs. It only covers replies generated inside the agent's on_enter method, so move the greeting there if you rely on it:

from livekit.agents.llm.tool_context import ToolFlag

@function_tool(flags=ToolFlag.IGNORE_ON_ENTER)
async def book_appointment(self, context: RunContext, day: str, time: str) -> str:
    """Book an appointment once the caller has given a day and a time."""
    return f"Booked for {day} at {time}."

Step 6: Run the agent locally

Download the model files the local components need ahead of time, then talk to the agent in your terminal:

uv run agent.py download-files

# Terminal-only session using your mic and speakers; Ctrl+T toggles text mode
uv run agent.py console          # same as: lk agent console

# Register with LiveKit Cloud with auto-reload, then use the Agent Console in the browser
uv run agent.py dev              # same as: lk agent dev

uv run agent.py --help lists the commands. In my scratch environment it printed console, start, dev, connect and download-files, which matches the startup modes page. Console mode does not connect to a LiveKit server for media, but it still needs your LiveKit credentials in the environment because the models run on LiveKit Inference.

In dev mode the agent registers as receptionist and waits. Because agent_name is set, LiveKit uses explicit dispatch: the agent joins only rooms it is dispatched to. In the Agent Console, set the agent name to receptionist, or dispatch it from the CLI:

lk dispatch create --agent-name receptionist --room test-room

Step 7: Measure voice-to-voice latency

The handler in Step 3 already logs LiveKit's per-turn metrics, documented on the data hooks page. User messages carry end_of_turn_delay and transcription_delay; assistant messages carry llm_node_ttft, tts_node_ttfb and e2e_latency, the time from the user stopping speaking to the agent starting to respond. The shutdown callback prints p50 and p95 for the session. Two cautions: llm_node_ttft and tts_node_ttfb are empty for realtime models, and the session-level metrics_collected event is deprecated in favour of these fields and session_usage_updated.

The Voice AI and Voice Agents primer budgets a cascaded pipeline at about 1.3 seconds voice to voice, with 1,500 ms as the target. Do not trust one run on home Wi-Fi: run twenty turns from your callers' region and read the p95, then alert on it per region as I describe for TTFT in SRE for LLM apps. The per-stage fields tell you whether to change the LLM, the TTS or the endpointing delay.

Step 8: Deploy to LiveKit Cloud or self-host

From the project directory, lk agent create registers the agent, generates a Dockerfile if you do not have one, uploads the build context and deploys it:

uv lock                      # commit uv.lock; the template Dockerfile runs uv sync --locked
lk agent create              # prompts to load secrets from .env.local
lk agent update-secrets --secrets "CRM_API_TOKEN=change-me"   # rolling restart

LiveKit's build docs set the rules: a glibc base image (Alpine is not supported), a non-root user, no .env files in the image, no LIVEKIT_* variables in the Dockerfile (they are injected at runtime), a build under 10 minutes and a context under 1 GB. An abridged copy of LiveKit's uv template, with Python pinned to 3.12, looks like this:

# Abridged from LiveKit's uv Dockerfile template
FROM ghcr.io/astral-sh/uv:python3.12-bookworm-slim AS base
ENV PYTHONUNBUFFERED=1 UV_COMPILE_BYTECODE=1
# ... build stage runs: uv sync --locked ...
USER appuser
CMD ["uv", "run", "agent.py", "start"]

To self-host, run the same image with LIVEKIT_URL, LIVEKIT_API_KEY and LIVEKIT_API_SECRET set. start mode drains active calls on SIGTERM, so set a termination grace period longer than your longest call or every rolling deploy hangs up on someone. Outside LiveKit Cloud the full turn detector and adaptive interruptions have a monthly development allowance; pin v1-mini to avoid the metered dependency.

Step 9: Give the agent a phone number over SIP

LiveKit sells US numbers directly, but only inbound for now; every plan includes one free local number and the Build plan includes 50 free inbound minutes. For other countries or outbound calls, bring a SIP trunk from Twilio, Telnyx or another carrier and create an inbound trunk.

lk number search --country-code US --area-code 415
lk number purchase --numbers +14155550100

Create dispatch-rule.json so each caller gets their own room and the receptionist agent is dispatched into it:

{
  "dispatch_rule": {
    "name": "receptionist-inbound",
    "rule": { "dispatch_rule_individual": { "room_prefix": "call-" } },
    "room_config": { "agents": [{ "agent_name": "receptionist" }] }
  }
}
lk sip dispatch create dispatch-rule.json     # prints the rule ID
lk number list                                # find the phone number ID
lk number update --id <PHONE_NUMBER_ID> --sip-dispatch-rule-id <DISPATCH_RULE_ID>

Call the number. One warning from the dispatch rule docs: an individual rule names each room after the caller's phone number, and room names land in logs and traces that PII redaction does not touch, so in production route each call to a room ID you choose. For noisy phone audio, LiveKit suggests the telephony-tuned noise_cancellation.BVCTelephony() from livekit-plugins-noise-cancellation.

Step 10: Build the same agent on Pipecat 1.15.0

In Pipecat, frames flow through processors in the order you list them. Install with uv add "pipecat-ai[webrtc,silero,deepgram,openai,cartesia,runner]==1.15.0" python-dotenv and put DEEPGRAM_API_KEY, OPENAI_API_KEY and CARTESIA_API_KEY in .env. This follows the current quickstart, which uses PipelineWorker and WorkerRunner; the older PipelineTask name is a deprecated alias that 2.0 will remove.

# bot.py: the same receptionist on Pipecat 1.15.0 (Python 3.12)
import os

from dotenv import load_dotenv
from loguru import logger
from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.frames.frames import LLMRunFrame
from pipecat.observers.user_bot_latency_observer import UserBotLatencyObserver
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.worker import PipelineParams, PipelineWorker
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response_universal import (
    LLMContextAggregatorPair,
    LLMUserAggregatorParams,
)
from pipecat.runner.types import RunnerArguments
from pipecat.runner.utils import create_transport
from pipecat.services.cartesia.tts import CartesiaTTSService
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.llm_service import FunctionCallParams
from pipecat.services.openai.responses.llm import OpenAIResponsesLLMService
from pipecat.transports.base_transport import TransportParams
from pipecat.workers.runner import WorkerRunner

load_dotenv(override=True)


async def check_opening_hours(params: FunctionCallParams, day: str):
    """Look up the clinic's opening hours for a day of the week.

    Args:
        day: Day of the week in English, for example "Monday".
    """
    hours = {"saturday": "9am to 1pm", "sunday": "closed"}
    await params.result_callback({"day": day, "hours": hours.get(day.lower(), "8am to 6pm")})


async def run_bot(transport, runner_args: RunnerArguments):
    stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
    llm = OpenAIResponsesLLMService(
        api_key=os.getenv("OPENAI_API_KEY"),
        settings=OpenAIResponsesLLMService.Settings(
            model="gpt-5.4-mini",
            system_instruction="You are a clinic receptionist on a voice call. One or two short spoken sentences, no formatting.",
        ),
    )
    tts = CartesiaTTSService(
        api_key=os.getenv("CARTESIA_API_KEY"),
        settings=CartesiaTTSService.Settings(voice="86e30c1d-714b-4074-a1f2-1cb6b552fb49"),
    )

    context = LLMContext(tools=[check_opening_hours])
    user_agg, assistant_agg = LLMContextAggregatorPair(
        context,
        # Silero VAD (stop_secs 0.2) feeds the default Smart Turn v3 stop strategy
        user_params=LLMUserAggregatorParams(vad_analyzer=SileroVADAnalyzer()),
    )

    pipeline = Pipeline([
        transport.input(), stt, user_agg, llm, tts, transport.output(), assistant_agg,
    ])

    latency = UserBotLatencyObserver()

    @latency.event_handler("on_latency_measured")
    async def on_latency(observer, seconds):
        logger.info(f"voice-to-voice {seconds:.3f}s")  # VAD stop to first bot audio

    worker = PipelineWorker(
        pipeline,
        params=PipelineParams(enable_metrics=True, enable_usage_metrics=True),
        observers=[latency],
    )
    runner = WorkerRunner(handle_sigint=runner_args.handle_sigint)
    await runner.add_workers(worker)

    @worker.rtvi.event_handler("on_client_ready")
    async def on_client_ready(rtvi):
        context.add_message({"role": "developer", "content": "Greet the caller briefly."})
        await worker.queue_frames([LLMRunFrame()])

    @transport.event_handler("on_client_disconnected")
    async def on_client_disconnected(transport, client):
        logger.info("Client disconnected")
        await runner.cancel()

    await runner.run()


async def bot(runner_args: RunnerArguments):
    transport = await create_transport(
        runner_args,
        {"webrtc": lambda: TransportParams(audio_in_enabled=True, audio_out_enabled=True)},
    )
    await run_bot(transport, runner_args)


if __name__ == "__main__":
    from pipecat.runner.run import main

    main()

Run uv run bot.py and open http://localhost:7860/client. Tools are plain async functions whose docstrings become the schema (function calling docs), and turn-taking defaults to Smart Turn over Silero VAD with a 0.2 s stop_secs (speech input docs). With metrics enabled, the observer also emits on_latency_breakdown, splitting each turn into endpointing, transcription, LLM and TTS time.

For phone calls, the 1.15.0 dev runner accepts -t twilio (as well as telnyx, plivo, exotel and daily). Point a Twilio TwiML Bin's <Stream url="wss://your-host/ws" /> at the bot, as in Pipecat's Twilio WebSocket guide, and you have dial-in without a SIP trunk. Deploy with pipecat cloud deploy, or run the container yourself.

Troubleshooting

Symptom Likely cause Fix
Agent registers in dev mode but never joins the room agent_name is set, so dispatch is explicit and nothing dispatched it Set the same name in the Agent Console or run lk dispatch create --agent-name receptionist --room test-room
ValueError at session start mentioning VAD silence Silero min_silence_duration below 0.25 s with the audio turn detector Keep it at 0.25 s or more (default 0.55 s)
Agent cuts callers off mid-thought VAD-only detection or min_delay too low Use inference.TurnDetector(), raise endpointing.min_delay, add voice isolation for noisy lines
Agent stops talking every time the caller says "uh-huh" Interruption mode is vad Set interruption.mode to adaptive, or raise min_words (needs STT) or min_duration
Long silence before every reply Slow LLM, a high min_delay, or heavy tool results Read llm_node_ttft and end_of_turn_delay; pick a smaller model, try preemptive_tts, trim tool output
Phone rings but no agent answers Number not assigned to a dispatch rule, or rule has no room_config.agents lk number update --id ... --sip-dispatch-rule-id ... and check the rule JSON
Callers in the same room hear each other Direct dispatch rule with a static room_name Use an individual rule or route each call to a unique room
ValueError when disabling interruptions on a realtime model The model's server-side turn detection is on Set the model's turn_detection=None and use VAD on the session
Build fails, or failed to retrieve region info at startup Alpine base image, or a slim Node image without CA certificates Use a glibc slim image; for Node install ca-certificates
Pipecat bot answers twice after parallel tool calls group_parallel_tools disabled Leave group_parallel_tools=True (the default)

Most rows come from LiveKit's tuning symptom table and turns overview. If it works in console mode but not on the phone, suspect the 8 kHz audio first.

FAQ

Do I need separate STT, LLM and TTS API keys with LiveKit?

Not if you use LiveKit Inference: the inference.STT, inference.LLM and inference.TTS classes bill through your LiveKit project key. You need provider keys only when you use the provider plugins directly, which is also how you reach models LiveKit Inference does not list.

Can I use OpenAI's Realtime API instead of a cascaded pipeline?

Yes. Replace the three model lines with llm=inference.RealtimeModel(model="openai/gpt-realtime-2.1", voice="marin"), as LiveKit's quickstart shows. You lose the per-stage TTFT and TTFB metrics, and turn detection moves to the model unless you hand it back to the session.

Which is easier for a phone agent, LiveKit or Pipecat?

For US inbound numbers, LiveKit is fewer moving parts: buy the number, create a dispatch rule, done. Pipecat is easier when you already have Twilio, Telnyx or Plivo numbers, because it connects to their media streams over WebSocket without a SIP trunk.

How do I test a voice agent without calling it?

Run console --text to drive it by typing, and add behavioural tests (LiveKit's starter includes some; Pipecat ships evals with simulated callers). Neither replaces twenty real phone calls before launch.

When I put a voice agent in front of real callers, the first thing I check is not the model. It is the p95 of e2e_latency and end_of_turn_delay by region, the dispatch rule that decides which room a caller lands in, and whether a rolling deploy drains calls or drops them. Those are ordinary platform problems in a new costume. The official references: LiveKit Agents docs, LiveKit telephony, Pipecat docs and the Pipecat examples repo.

Related: Voice AI Agents in 2026: Realtime vs Cascaded, 1.3s Latency, $0.05 a Minute, How to Build an AI Agent With Tool Calling in Python (2026) and SRE for LLM Apps in 2026: SLOs for TTFT, Provider Outages and Evals.

I'm Ajay Kumar — I build and operate PandaStack, an open-source Firecracker microVM cloud for AI agents, and write about the infrastructure and tooling around it.

Need this kind of infrastructure work? See what I do or email hello@ajayk.sh.


Related