LiveKit Agents
LiveKit Agents is a powerful framework for building realtime voice AI agents that interact over WebRTC. It provides high-level abstractions for managing LiveKit rooms, participants, audio/video streams, and integrating with various AI models (LLMs, STT, TTS) through a flexible plugin architecture. The current version is 1.5.2, with frequent minor releases delivering new features, bug fixes, and updated plugin support.
Warnings
- breaking Major versions (e.g., 1.5.0) often introduce significant features and internal refactoring. Specifically, 1.5.0 added 'Adaptive Interruption Handling' (enabled by default), and later 1.5.2 updated MistralAI SDK to v2. These changes may subtly alter agent behavior or require code adjustments when using specific plugins.
- gotcha The `livekit-agents` library provides the core framework, but actual AI functionalities (LLM, STT, TTS) are provided by separate `livekit-plugins-*` packages (e.g., `livekit-plugins-openai`). These must be installed independently based on your chosen AI providers.
- gotcha All LiveKit Agents require connection details for a LiveKit server. `LIVEKIT_URL`, `LIVEKIT_API_KEY`, and `LIVEKIT_API_SECRET` must be correctly configured as environment variables or explicitly passed to `cli.run_agent`.
- gotcha The library has strict Python version requirements, currently `>=3.10, <3.15`. Using unsupported Python versions (e.g., 3.9 or 3.15+) will lead to installation failures or unexpected runtime errors.
- gotcha LiveKit Agents is built entirely on `asyncio`. Blocking operations or incorrect `async/await` patterns within your agent logic can lead to performance bottlenecks, unresponsive agents, or deadlocks in the event loop.
Install
-
pip install livekit-agents
Imports
- Agent
from livekit.agents import Agent
- JobContext
from livekit.agents import JobContext
- cli
from livekit.agents import cli
Quickstart
import asyncio
import os
from livekit.agents import cli, JobContext, Agent
class EchoAgent(Agent):
def __init__(self):
super().__init__()
async def _on_participant_connected(self, ctx: JobContext, participant_sid: str):
print(f"Participant {participant_sid} connected! Listening for chat messages.")
# Access the first speaker node (representing the room or the primary speaker)
# and listen for incoming chat messages.
async for msg in ctx.initial_speaker_node.chat.iter_messages():
print(f"Received chat message from {msg.participant_identity}: {msg.message}")
# Echo the message back into the room
await ctx.initial_speaker_node.chat.send_message(
f"Echo from agent ({ctx.agent_id}): {msg.message}"
)
async def entrypoint(ctx: JobContext):
agent = EchoAgent()
agent.on("participant_connected", agent._on_participant_connected)
print(f"Agent {ctx.agent_id} ready in room {ctx.room.name}. Waiting for participants...")
if __name__ == "__main__":
# Ensure LiveKit server connection details are available.
# For local development, set these as environment variables or pass them directly:
# LIVEKIT_URL="ws://localhost:7880" LIVEKIT_API_KEY="devkey" LIVEKIT_API_SECRET="secret"
livekit_url = os.environ.get('LIVEKIT_URL', '')
livekit_api_key = os.environ.get('LIVEKIT_API_KEY', '')
livekit_api_secret = os.environ.get('LIVEKIT_API_SECRET', '')
if not all([livekit_url, livekit_api_key, livekit_api_secret]):
print("Error: LiveKit environment variables (LIVEKIT_URL, LIVEKIT_API_KEY, LIVEKIT_API_SECRET) are not set.")
print("Please set them or pass them to cli.run_agent.")
exit(1)
cli.run_agent(
entrypoint,
agent_name="EchoAgent",
livekit_url=livekit_url,
livekit_api_key=livekit_api_key,
livekit_api_secret=livekit_api_secret
)