LiveKit Agents

LiveKit Agents 框架

LiveKit Agents 是用于构建实时、可编程服务器端参与者的框架，面向多模态语音代理场景，使代理能够"看见、听见、理解"并与用户进行对话。

核心特性

• 灵活集成：提供完善的 STT、LLM、TTS、Realtime API 生态，可根据业务需求自由组合。
• 内置任务调度：通过 Dispatch API 将终端用户匹配到合适的代理。
• WebRTC 客户端：借助 LiveKit 开源 SDK 生态，覆盖各主流平台。
• 电话集成：与 LiveKit SIP 通信栈无缝对接，支持拨打与接听电话。
• 数据交换：通过 RPC 与 Data API 与客户端双向传递数据。
• 语义轮次检测：基于 Transformer 模型判断用户发言结束，减少抢话。
• MCP 支持：一行代码即可集成 MCP 服务器提供的工具。
• 内置测试框架：支持编写断言与 LLM 评判（judge）以验证代理行为。
• 开源：整套链路可自托管，包括广泛使用的 WebRTC 媒体服务器 LiveKit Server。

安装

pip install "livekit-agents[openai,deepgram,cartesia]"

核心概念

• Agent：带有指令的 LLM 应用单元。
• AgentSession：管理代理与终端用户交互的容器。
• entrypoint：会话入口，类似于 Web 服务器中的请求处理函数。
• AgentServer：主进程，负责任务调度并为每个会话启动代理。

基础用法：简单语音代理

from livekit.agents import (
    Agent, AgentServer, AgentSession,
    JobContext, RunContext, cli,
    function_tool, inference,
)

@function_tool
async def lookup_weather(context: RunContext, location: str):
    """查询天气信息。"""
    return {"weather": "sunny", "temperature": 70}

server = AgentServer()

@server.rtc_session()
async def entrypoint(ctx: JobContext):
    # AgentSession 支持 STT + LLM + TTS 组合或单一 Realtime API
    session = AgentSession(
        vad=inference.VAD(),
        stt=inference.STT("deepgram/nova-3", language="multi"),
        llm=inference.LLM("google/gemma-4-31b-it"),
        tts=inference.TTS("cartesia/sonic-3",
                          voice="9626c31c-bec5-4cca-baa8-f8ba9e84c8bc"),
    )

    agent = Agent(
        instructions="You are a friendly voice assistant built by LiveKit.",
        tools=[lookup_weather],
    )

    await session.start(agent=agent, room=ctx.room)
    await session.generate_reply(instructions="问候用户并询问他们的一天")

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

运行需要配置环境变量：LIVEKIT_URL、LIVEKIT_API_KEY、LIVEKIT_API_SECRET。

多代理交接

通过让某个 function_tool 返回新的 Agent 实例即可实现交接，并在新 Agent 中切换至 Realtime Model：

class IntroAgent(Agent):
    async def on_enter(self):
        self.session.generate_reply(instructions="问候并收集信息")

    @function_tool
    async def information_gathered(self, context, name: str, location: str):
        """用户提交姓名与所在地后交接至 StoryAgent。"""
        context.userdata.name = name
        context.userdata.location = location
        return StoryAgent(name, location), "开始讲故事！"

class StoryAgent(Agent):
    def __init__(self, name, location):
        super().__init__(
            instructions=f"你是一名讲故事的人，用户名叫 {name}，来自 {location}",
            llm=openai.realtime.RealtimeModel(voice="echo"),
        )

测试

内置 pytest 集成，支持事件断言与 LLM 评判：

@pytest.mark.asyncio
async def test_no_availability():
    async with AgentSession(llm=google.LLM()) as sess:
        await sess.start(MyAgent())
        result = await sess.run(user_input="你好，我需要下单。")
        result.expect.skip_next_event_if(type="message", role="assistant")
        result.expect.next_event().is_function_call(name="start_order")
        result.expect.next_event().is_function_call_output()
        await (
            result.expect.next_event()
            .is_message(role="assistant")
            .judge(llm, intent="应询问用户想点什么")
        )

运行模式

# 终端模式：本地音频输入输出，快速验证
python myagent.py console

# 开发模式：热重载，连接 LiveKit Cloud 或自托管服务
python myagent.py dev

# 生产模式：面向生产优化
python myagent.py start

开发与贡献

项目使用 uv 管理依赖，开发安装：

uv sync --all-extras --dev
uv run ruff format
uv run ruff check --fix
uv run pytest --unit

代码使用 ruff 格式化与检查；使用 pdoc 生成本地文档。

许可证

Agents 框架遵循 Apache-2.0 协议；LiveKit 轮次检测模型遵循 LiveKit Model License。