• October 24, 2026
• 11 min read

LiveKit — Agents Framework¶

How a LiveKit voice agent is built and run: the Agent and AgentSession classes, the voice pipeline (STT, LLM, TTS, VAD, turn detection), realtime models, and the worker side (AgentServer, jobs, dispatch, prewarm). Tools and handoffs are in 03; production operations are in 04.

Packages and Plugins¶

uv add "livekit-agents[openai,silero,turn-detector]"   # core + plugins you use
uv add "livekit-agents[deepgram,cartesia,elevenlabs,google,anthropic,groq]"
uv run python -m livekit.agents download-files          # fetch model weights (turn detector) at build time

Each provider is a livekit-plugins-<name> package imported as livekit.plugins.<name>; the extras of livekit-agents install them. from livekit.agents import inference exposes the LiveKit Inference gateway (inference.STT, inference.LLM, inference.TTS, model strings such as "openai/gpt-4o"), which needs LiveKit Cloud credentials. On your own infrastructure use the provider plugins; inference.VAD and inference.TurnDetector(version="v1-mini") run locally.

Agent and AgentSession¶

An Agent holds instructions, tools and hooks. An AgentSession owns the conversation runtime: it connects room audio and text to STT, LLM, TTS, VAD and turn detection, keeps the chat history, and runs tools. One session runs one agent at a time; handoffs swap the agent.

from livekit.agents import Agent, AgentSession, function_tool, RunContext


class Concierge(Agent):
    def __init__(self) -> None:
        super().__init__(instructions="You are a hotel concierge. Keep replies under two sentences.")

    async def on_enter(self) -> None:                     # runs when this agent becomes active
        self.session.generate_reply(instructions="Welcome the guest and ask what they need.")

    @function_tool()
    async def opening_hours(self, context: RunContext, place: str) -> str:
        """Return opening hours.

        Args:
            place: The place, for example "pool" or "spa".
        """
        return f"The {place} is open 7am to 9pm."

Session methods: await session.start(agent=..., room=..., room_options=...), session.generate_reply(user_input=..., instructions=..., tool_choice=..., allow_interruptions=...), session.say(text, allow_interruptions=..., add_to_chat_ctx=...), session.interrupt(), session.update_agent(agent), session.shutdown(drain=True), await session.aclose(), session.history (chat context), session.usage, session.current_agent.

Agent accepts the same stt/llm/tts/vad/turn_handling arguments to override the session defaults for that agent, plus chat_ctx, tools, id. Hooks: on_enter, on_exit, on_user_turn_completed, and the pipeline nodes stt_node, llm_node, tts_node, transcription_node, realtime_audio_output_node.

Room I/O¶

session.start(room=ctx.room, room_options=room_io.RoomOptions(...)) links the session to one participant (the first STANDARD or SIP participant, or participant_identity).

A text-only agent sets audio_input=False, audio_output=False and has no stt/tts; a message sent to lk.chat is answered on lk.transcription (see 03).

Turn Handling¶

turn_handling=TurnHandlingOptions(...) replaces the old min_endpointing_delay, allow_interruptions, turn_detection, ... keyword arguments (they still work and warn).

from livekit.agents import AgentSession, TurnHandlingOptions, inference

session = AgentSession(
    # stt=..., llm=..., tts=..., vad=...
    turn_handling=TurnHandlingOptions(
        turn_detection=inference.TurnDetector(version="v1-mini"),   # local model; or "vad" / "stt" / "manual"
        endpointing={"mode": "fixed", "min_delay": 0.5, "max_delay": 3.0},
        interruption={"mode": "vad", "min_duration": 0.5, "min_words": 0},
        preemptive_generation={"enabled": True, "preemptive_tts": False},
    ),
)

Turn detection decides when the user is done. VAD alone ends a turn after silence (min_delay); a turn-detector model reads the speech and extends the wait when the sentence is unfinished. Preemptive generation starts the LLM call on the interim transcript, before the turn is confirmed, and discards it if the user keeps talking: lower latency, more tokens. Turn on preemptive_tts only when latency matters more than cost.

Turn detector options:

The default session VAD (inference.VAD) is a local Silero model. To tune it use silero.VAD.load(min_silence_duration=0.55, activation_threshold=0.5, min_speech_duration=0.05, prefix_padding_duration=0.5); the values shown are the defaults.

Realtime (Speech-to-Speech) Models¶

from livekit.agents import AgentSession
from livekit.plugins import openai

session = AgentSession(llm=openai.realtime.RealtimeModel(voice="marin"))                            # audio in, audio out
hybrid = AgentSession(llm=openai.realtime.RealtimeModel(modalities=["text"]), tts=openai.TTS())     # model understands speech, your TTS speaks

Realtime models bring their own server-side turn detection (semantic or server VAD with TurnDetection(type="semantic_vad", ...)). To use LiveKit's turn detector with one, disable the built-in detection and add a separate STT.

Startup Modes¶

python agent.py start                    # production: INFO logs (JSON), drains on SIGTERM, health server on :8081
lk agent dev agent.py                    # development: debug logs, reload on file changes, random dev deployment id
lk agent console agent.py --text         # talk to the agent in the terminal (no room, no server)
python agent.py connect --room my-room   # run one job in an existing room
python -m livekit.agents download-files  # fetch plugin model files

lk agent dev runs the project in the current directory (a pyproject.toml or requirements.txt must exist) and registers the worker under a deployment id. By default that is a random dev-xxxxxxxx, so dispatch to it with --deployment <id> (CLI, API, or RoomAgentDispatch.deployment in a token), or start it with lk agent dev --deployment production to receive default dispatch. python agent.py start registers with an empty deployment and receives default dispatch.

Agent Server¶

from livekit import agents
from livekit.agents import AgentServer, JobContext, JobProcess, JobRequest

server = AgentServer(
    num_idle_processes=2,          # warm job processes kept ready
    load_threshold=0.7,            # above this the server stops accepting jobs
    drain_timeout=1800,            # seconds to wait for jobs on SIGTERM
    prometheus_port=9464,          # lk_agents_* metrics
)
server.setup_fnc = prewarm         # runs once per job process before any job (see Prewarm)

@server.rtc_session(agent_name="support", on_request=on_request, on_session_end=on_session_end)
async def entrypoint(ctx: JobContext) -> None: ...

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

@server.rtc_session registers exactly one entrypoint. cli.run_app accepts an AgentServer or the older WorkerOptions(entrypoint_fnc=..., prewarm_fnc=..., agent_name=...).

@server.rtc_session(...) arguments: agent_name (turns on explicit dispatch), type (ServerType.ROOM, one job per room, or PUBLISHER, one per publisher), on_request, on_session_end.

Jobs¶

A job is one agent assignment: a child process runs your entrypoint with a JobContext.

async def on_request(req: JobRequest) -> None:
    if req.room.name.startswith("blocked-"):
        await req.reject()
    else:
        await req.accept(name="Support bot", identity=f"agent-{req.room.name}", attributes={"role": "support"})


async def on_session_end(ctx: JobContext) -> None:
    report = ctx.make_session_report()                 # chat_history, events, model usage, options
    save(report.to_dict())

By default the room closes when the last non-agent participant leaves (after departure_timeout, default 20 s), the agent is disconnected and the job shuts down. If an agent drops unexpectedly, the server dispatches a new one to the room after about 15 seconds.

Dispatch¶

from livekit import api

async with api.LiveKitAPI() as lk:                                         # LIVEKIT_URL / KEY / SECRET from the environment
    d = await lk.agent_dispatch.create_dispatch(
        api.CreateAgentDispatchRequest(agent_name="support", room="support-1", metadata='{"user_id":"12345"}')
    )
    print(await lk.agent_dispatch.list_dispatch("support-1"))
    await lk.agent_dispatch.delete_dispatch(d.id, "support-1")

lk dispatch create --agent-name support --room support-1 --metadata '{"user_id":"12345"}'
lk token create --identity caller --room support-1 --agent support --job-metadata '{"user_id":"12345"}' --join

A token can carry the dispatch (with_room_config(api.RoomConfiguration(agents=[api.RoomAgentDispatch(agent_name="support", metadata=...)])), see 01); it applies only when the token creates the room. SIP dispatch rules carry room_config.agents (see 04). The server assigns a dispatch in under 150 ms; the first available worker below its load threshold accepts, otherwise the job goes to another worker.

Prewarm¶

from livekit.plugins import silero

def prewarm(proc: JobProcess) -> None:
    proc.userdata["vad"] = silero.VAD.load()

server.setup_fnc = prewarm

@server.rtc_session(agent_name="support")
async def entrypoint(ctx: JobContext) -> None:
    session = AgentSession(vad=ctx.proc.userdata["vad"])       # model already in memory

Load models in setup_fnc, not in the entrypoint; a job process starts from the warm pool, so the first user turn does not wait for model loading.

Names from Older Tutorials¶

Quick Rules¶

1. One AgentSession per job, created in the entrypoint; put reusable objects (VAD, LLM clients) in setup_fnc.
2. Set agent_name and dispatch explicitly; remember the deployment id when you use lk agent dev.
3. Pin turn detection (inference.TurnDetector(version="v1-mini"), "vad" or "stt") and interruption mode: "vad" on your own infrastructure.
4. Tune with TurnHandlingOptions: lower min_delay for speed, raise it to stop cut-offs; keep preemptive_tts off until latency demands it.
5. Use text-only sessions (audio_input=False, audio_output=False) for chat clients and for tests.
6. Handle on_request when you must reject jobs (capacity, tenancy) and use on_session_end for reports.
7. Give workers time to drain: drain_timeout and the orchestrator's grace period must exceed your longest call.

See also¶

• Digital Garden: Knowledge Base
• LiveKit — Open-Source WebRTC Server and Agents Framework
• LiveKit — Server, Rooms & Tokens
• Pipecat — Open-Source Framework for Voice Agents
• Agentic AI Architecture