LiveKit Agents

Agent框架旨在构建运行在服务器上的实时、可编程参与者。使用它来创建能够看、听和理解的对话式多模态语音代理。

该框架包含用于常见工作流程的插件，如语音活动检测和语音转文本。

Agents与云端或自托管的LiveKit服务器无缝集成，将作业队列和调度责任转移给它。这消除了对额外队列基础设施的需求。在本地机器上开发的Agent代码在部署到生产服务器时可以扩展以支持数千个并发会话。

该SDK目前处于开发者预览阶段。在此期间，您可能会遇到错误，且API可能会发生变化。

我们欢迎并感谢任何反馈或贡献。您可以在这里创建问题或在LiveKit社区Slack与我们进行实时聊天。

文档和指南

• 概述
• 快速入门
• 使用插件
• 部署agents

[!注意] 0.7.x和0.8.x版本之间存在破坏性API更改。请参阅0.8迁移指南以获取更改的详细概述。

示例

• 语音助手：具有STT、LLM和TTS功能的语音助手。演示
• 视频发布：演示将RGB帧发布到LiveKit房间
• STT：将参与者的音频转录为文本的agent
• TTS：将合成语音发布到LiveKit房间的agent

安装

要安装核心Agents库：

pip install livekit-agents

Agents包含一组预构建的插件，使组合agents变得更容易。这些插件涵盖了常见任务，如将语音转换为文本或反之，以及在生成式AI模型上运行推理。您可以按以下方式安装插件：

pip install livekit-plugins-deepgram

目前可用的插件如下：

概念

• Agent：定义可编程服务器端参与者工作流程的函数。这是您的应用程序代码。
• Worker：负责管理与LiveKit服务器作业队列的容器进程。每个worker能够同时运行多个agents。
• Plugin：执行特定任务（如语音转文本）的库类，来自特定提供商。一个agent可以组合多个插件以执行更复杂的任务。

运行agent

该框架提供了一个CLI接口来运行您的agent。首先，您需要设置以下环境变量：

• LIVEKIT_URL
• LIVEKIT_API_KEY
• LIVEKIT_API_SECRET

启动worker

这将启动worker并等待用户连接到您的LiveKit服务器：

python my_agent.py start

要在开发模式下运行worker（支持热代码重载），您可以使用dev命令：

python my_agent.py dev

使用playground作为您的agent UI

为了简化构建和测试agent的过程，我们开发了一个名为"playground"的多功能Web前端。您可以使用或修改这个应用程序以满足您的特定需求。它还可以作为完全自定义agent应用程序的起点。

• 托管的playground
• 源代码
• Playground文档

加入特定房间

要加入已经活跃的LiveKit房间，您可以使用connect命令：

python my_agent.py connect --room <my-room>

运行agent时会发生什么？

当您按照上述步骤运行agent时，会启动一个worker，它会打开与LiveKit服务器实例的经过身份验证的WebSocket连接（由LIVEKIT_URL定义并使用访问令牌进行身份验证）。

此时实际上没有agent在运行。相反，worker正在等待LiveKit服务器给它分配工作。

当创建一个房间时，服务器会通知其中一个已注册的worker有新工作。被通知的worker可以决定是否接受。如果worker接受工作，它将实例化您的agent作为参与者并让它加入房间，在那里它可以开始订阅音视频轨道。一个worker可以同时管理多个agent实例。

如果被通知的worker拒绝工作或在预定的超时期限内没有接受，服务器将把工作请求路由到另一个可用的worker。

当我对worker发送SIGTERM信号时会发生什么？

编排系统是为生产用例设计的。与典型的Web服务器不同，agent是一个有状态程序，因此在活跃会话进行时不终止worker很重要。

当对worker调用SIGTERM时，worker将向LiveKit服务器发出信号，表示它不再需要额外的工作。它还会自动拒绝在服务器接收到信号之前到达的任何新工作请求。只要它还在管理连接到房间的agents，worker就会保持活跃状态。

下载模型文件

某些插件在使用前需要下载模型文件。要下载agent所需的所有模型，请执行以下命令：

python my_agent.py download-files

如果您正在开发自定义插件，可以通过在Plugin类中实现download_files方法来集成此功能：

class MyPlugin(Plugin):
    def __init__(self):
        super().__init__(__name__, __version__)

    def download_files(self):
        _ = torch.hub.load(
            repo_or_dir="my-repo",
            model="my-model",
        )