VideoSDK AI Agents

用于构建生产级、实时语音及多模态AI代理的开源Python框架。

VideoSDK AI Agents框架 是一个Python SDK，用于构建能够作为实时参与者加入VideoSDK房间的AI代理。它将您的代理工作进程、AI模型和用户设备连接成一条低延迟的统一流水线——自动处理音频流、发言轮次检测、打断以及媒体路由等问题，使您能够专注于代理逻辑本身。

概述

VideoSDK AI Agents 是一个Python框架，允许您构建直接参与VideoSDK会议的语音及多模态AI代理。该框架管理代理的完整生命周期——从加入会议、处理实时音频，到运行STT → LLM → TTS流水线或接入统一的实时模型，再到处理发言轮次检测、VAD、打断以及干净地结束会话等环节。

v1.0.0 引入了一个统一的Pipeline类，取代了之前的CascadingPipeline和RealtimePipeline。您可以传入任意组合的组件——STT、LLM、TTS、VAD、发言轮次检测器、虚拟形象等，框架会自动将它们串联起来，并选择最优的执行模式。基于装饰器的钩子系统（@pipeline.on(...)）让您无需继承即可在任何阶段拦截并转换数据。

🎙️ 使用级联模式的代理 使用级联模式（STT → LLM → TTS）构建AI语音代理。	⚡ 使用实时模式的代理 使用统一的实时模型（如Gemini Live）构建AI语音代理。
💻 代理文档 VideoSDK代理官方文档。	📚 SDK参考 代理框架的参考文档。

#	特性	描述
1	🎤 实时通信（音频/视频）	代理可以在会议中实时收听、发言并互动。
2	📞 SIP与电话集成	通过SIP无缝连接代理至电话系统，实现呼叫处理、路由及PSTN接入。
3	🧍 虚拟形象	可以构建或接入任何虚拟形象提供商——框架会自动处理音频路由、同步及清理等工作。
4	🤖 多模型支持	可与OpenAI、Gemini、AWS NovaSonic、Anthropic等集成。
5	🧩 级联模式	可跨不同提供商组合任意STT → LLM → TTS链条，获得完全的控制权和灵活性。
6	⚡ 实时模式	使用统一的实时模型（如OpenAI Realtime、AWS Nova Sonic、Gemini Live）以实现最低延迟。
7	🔀 混合模式	可混合使用级联和实时组件——例如自定义STT搭配实时模型，或实时模式结合自定义TTS。
8	🪝 流水线钩子	使用@pipeline.on(...)在任何阶段（STT、LLM、TTS、发言轮次）拦截并转换数据。
9	🛠️ 函数工具	通过任何外部工具或API调用扩展代理功能。
10	🌐 MCP集成	使用模型上下文协议（MCP）将代理连接到外部数据源和工具。
11	🔗 A2A协议	提供可靠的代理间路由，并基于关联性追踪请求。
12	🦜 LangChain & LangGraph	可将任何LangChain BaseChatModel或LangGraph StateGraph作为代理的LLM接入。
13	📊 可观测性	内置指标、OpenTelemetry追踪以及按组件划分的结构化日志记录。

[!IMPORTANT]

请给VideoSDK仓库加星标 ⭐️

即可第一时间获取新版本和更新通知。您的支持将帮助我们不断成长并改进VideoSDK！

---

流水线模式

所有代理都围绕一个Pipeline类构建。只需传入所需的组件，SDK便会自动选择合适的执行模式。

级联模式 — STT → LLM → TTS

您可以为每个环节自由搭配不同的服务提供商。当您需要自定义STT、特定的LLM行为或独特的TTS声音时，此模式最为适用。

async def start_session(context: JobContext):
    pipeline = Pipeline(
        stt=DeepgramSTT(),
        llm=GoogleLLM(),
        tts=CartesiaTTS(),
        vad=SileroVAD(),
        turn_detector=TurnDetector(),
    )
    session = AgentSession(agent=MyAgent(), pipeline=pipeline)
    await session.start(wait_for_participant=True, run_until_shutdown=True)

实时模式 — 统一模型实现最低延迟

在整个语音处理流程中使用单个实时模型。最适合响应延迟低于500毫秒的场景。

async def start_session(context: JobContext):
    pipeline = Pipeline(
        llm=GeminiRealtime(
            model="gemini-3.1-flash-live-preview",
            config=GeminiLiveConfig(voice="Leda", response_modalities=["AUDIO"]),
        )
    )
    session = AgentSession(agent=MyAgent(), pipeline=pipeline)
    await session.start(wait_for_participant=True, run_until_shutdown=True)

混合模式 — 自由组合

可以将外部 STT 与实时 LLM 结合使用，或者将实时模型与自定义 TTS 结合使用：

# 外部 STT → 实时 LLM
pipeline = Pipeline(stt=DeepgramSTT(), llm=OpenAIRealtime(...))

# 实时 LLM → 外部 TTS
pipeline = Pipeline(llm=OpenAIRealtime(...), tts=ElevenLabsTTS(...))

管道钩子 — 拦截任意阶段

@pipeline.on("stt")
async def clean_transcript(text: str) -> str:
    return text.strip()

@pipeline.on("llm")
async def route_llm(messages):
    if "transfer" in messages[-1].content:
        yield "正在为您转接。"  # 完全绕过 LLM

@pipeline.on("tts")
async def fix_pronunciation(text: str) -> str:
    return text.replace("VideoSDK", "Video S D K")

@pipeline.on("user_turn_start")
async def on_user_starts():
    print("用户正在讲话...")

可用的钩子点：stt · tts · llm · vision_frame · user_turn_start · user_turn_end · agent_turn_start · agent_turn_end

---

前置条件

在开始之前，请确保您已具备以下内容：

• VideoSDK 身份验证令牌（可从 app.videosdk.live 生成）
  • VideoSDK 会议 ID（可通过 创建房间 API 或 VideoSDK 控制台生成）
• Python 3.12 或更高版本
• 第三方 API 密钥：
  • 您计划使用的各项服务的 API 密钥（例如，用于 LLM/STT/TTS 的 OpenAI、用于 TTS 的 ElevenLabs、用于 Gemini 的 Google 等）。

安装

使用 UV（推荐）

UV 是一个快速的 Python 包管理器，可自动处理虚拟环境和依赖关系管理。

如果尚未安装 UV，请参阅 UV 安装指南。

• 安装核心 VideoSDK AI 代理包：

  uv add videosdk-agents
  

• 安装可选插件：

  uv add videosdk-plugins-openai
  uv add videosdk-plugins-deepgram
  

• 运行您的代理：

  uv run python main.py
  

使用 pip

• 创建并激活一个 Python 3.12 或更高版本的虚拟环境。

  macOS / Linux

  python3 -m venv venv
  source venv/bin/activate
  

  Windows

  python -m venv venv
  venv\Scripts\activate
  

• 安装核心 VideoSDK AI 代理包

  pip install videosdk-agents
  

• 安装可选插件。插件有助于集成不同的提供商以实现实时交互、STT、LLM、TTS 等功能。根据您的使用场景安装所需插件：

  # 示例：安装话轮检测插件
  pip install videosdk-plugins-turn-detector
  

  👉 支持的插件（实时交互、LLM、STT、TTS、VAD、Avatar、SIP）列于下方的【支持的库与插件】部分。

开发环境设置

要在本地设置项目，请克隆仓库并以可编辑模式安装所有包（核心 + 所有插件）：

使用 UV（推荐）：

git clone https://github.com/videosdk-live/agents.git
cd agents
uv sync
uv run python examples/cascade_basic.py

使用 pip：

git clone https://github.com/videosdk-live/agents.git
cd agents
bash setup.sh
source venv/bin/activate
python examples/cascade_basic.py

生成 VideoSDK 会议 ID

在您的 AI 代理加入会议之前，您需要先创建一个会议 ID。可以通过 VideoSDK 的创建房间 API 来生成：

使用 cURL

curl -X POST https://api.videosdk.live/v2/rooms \
  -H "Authorization: YOUR_JWT_TOKEN_HERE" \
  -H "Content-Type: application/json"

有关创建房间 API 的更多详细信息，请参阅 VideoSDK 文档。

开始使用：您的第一个代理

快速入门

现在您已经安装了必要的软件包，就可以开始构建了！

第一步：创建自定义代理

首先，让我们通过继承基础 Agent 类来创建一个自定义语音代理：

from videosdk.agents import Agent, function_tool

# 外部工具
# async def get_weather(self, latitude: str, longitude: str):

class VoiceAgent(Agent):
    def __init__(self):
        super().__init__(
            instructions="您是一位乐于助人的语音助手，能够回答问题并协助完成任务。",
             tools=[get_weather] # 您可以在该作用域之外注册任何外部工具
        )

    async def on_enter(self) -> None:
        """当代理首次加入会议时调用"""
        await self.session.say("您好！今天有什么可以帮您的吗？")
    
    async def on_exit(self) -> None:
      """当代理退出会议时调用"""
        await self.session.say("再见！")

此代码定义了一个基本的语音代理，包含：

• 自定义指令，用于定义代理的性格和能力
• 加入会议时的欢迎语
• 状态变化处理，用于跟踪代理当前的活动状态

第二步：实现函数工具

函数工具使您的代理能够执行对话之外的操作。定义工具有两种方式：

• 外部工具： 作为代理类之外的独立函数定义，并通过代理构造函数中的 tools 参数进行注册。
• 内部工具： 作为代理类内部的方法定义，并使用 @function_tool 装饰器标记。

以下是两者的示例：

import aiohttp

# 外部函数工具
@function_tool
def get_weather(latitude: str, longitude: str):
    print(f"正在获取 {latitude}, {longitude} 的天气信息")
    url = f"https://api.open-meteo.com/v1/forecast?latitude={latitude}&longitude={longitude}&current=temperature_2m"
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            if response.status == 200:
                data = await response.json()
                return {
                    "temperature": data["current"]["temperature_2m"],
                    "temperature_unit": "摄氏度",
                }
            else:
                raise Exception(
                    f"未能获取天气数据，状态码：{response.status}"
                )

class VoiceAgent(Agent):
# ... 上文代码 ...

# 内部函数工具
    @function_tool
    async def get_horoscope(self, sign: str) -> dict:
        horoscopes = {
            "Aries": "今天是你的幸运日！",
            "Taurus": "今天要专注于你的目标。",
            "Gemini": "今天的沟通将非常重要。",
        }
        return {
            "sign": sign,
            "horoscope": horoscopes.get(sign, "今天星星为你排好了队！"),
        }

• 对于可复用的独立函数，使用外部工具（通过 tools=[...] 注册）。
• 对于代理特定的逻辑，则使用内部工具作为类方法。
• 无论是外部工具还是内部工具，都必须使用 @function_tool 装饰器，以便代理能够识别并调用它们。

第3步：设置管道

使用统一的 Pipeline 类将您的代理连接到AI模型。只需传入所需的组件，SDK会处理其余部分。

实时模式（单个模型，最低延迟）：

async def start_session(context: JobContext):
    pipeline = Pipeline(
        llm=GeminiRealtime(
            model="gemini-3.1-flash-live-preview",
            config=GeminiLiveConfig(voice="Leda", response_modalities=["AUDIO"]),
        )
    )
    session = AgentSession(agent=VoiceAgent(), pipeline=pipeline)
    await session.start(wait_for_participant=True, run_until_shutdown=True)

级联模式（STT → LLM → TTS，完全由提供商控制）：

async def start_session(context: JobContext):
    pipeline = Pipeline(
        stt=DeepgramSTT(),
        llm=GoogleLLM(),
        tts=CartesiaTTS(),
        vad=SileroVAD(),
        turn_detector=TurnDetector(),
    )
    session = AgentSession(agent=VoiceAgent(), pipeline=pipeline)
    await session.start(wait_for_participant=True, run_until_shutdown=True)

第4步：组装并启动代理会话

async def start_session(context: JobContext):
    session = AgentSession(
        agent=VoiceAgent(),
        pipeline=pipeline,
    )
    await session.start(wait_for_participant=True, run_until_shutdown=True)

def make_context() -> JobContext:
    room_options = RoomOptions(
        room_id="<meeting_id>",
        name="测试代理",
        playground=True,
    )
    return JobContext(room_options=room_options)

if __name__ == "__main__":
    job = WorkerJob(entrypoint=start_session, jobctx=make_context)
    job.start()

第5步：与VideoSDK客户端应用连接

在设置好您的AI代理后，您需要一个客户端应用程序来与其连接。您可以使用任何VideoSDK快速入门示例来创建一个加入同一会议的客户端：

• JavaScript
• React
• React Native
• Android
• Flutter
• iOS
• Unity
• 物联网

在设置客户端应用程序时，请确保使用与您的AI代理相同的会议ID。

第6步：运行项目

完成设置后，您可以使用Python运行您的AI语音代理项目。请确保您的 .env 文件已正确配置，并且所有依赖项均已安装。

python main.py

[!TIP]

控制台模式 — 在没有会议室的情况下，在本地测试您的代理。 将 RoomOptions 中的 playground=True 设置为真值，然后运行 python main.py，即可直接通过终端的麦克风和扬声器进行交互。

第7步：部署

有关部署选项和指南，请查看官方文档：部署

---

VideoSDK推理

VideoSDK推理提供了一个统一的网关，用于访问STT、LLM、TTS、降噪和实时模型——无需管理各个提供商的API密钥。身份验证通过您的 VIDEOSDK_AUTH_TOKEN 进行，费用则从您的VideoSDK账户余额中扣除。

from videosdk.agents.inference import STT, LLM, TTS, Denoise, Realtime

使用VideoSDK推理的级联模式：

async def start_session(context: JobContext):
    pipeline = Pipeline(
        stt=STT.sarvam(model_id="saarika:v2.5", language="en-IN"),
        llm=LLM.google(model_id="gemini-2.5-flash"),
        tts=TTS.sarvam(model_id="bulbul:v2", speaker="anushka", language="en-IN"),
        denoise=Denoise.sanas(),
        vad=SileroVAD(),
    )
    session = AgentSession(agent=MyAgent(), pipeline=pipeline)
    await session.start(wait_for_participant=True, run_until_shutdown=True)

使用VideoSDK推理的实时模式：

async def start_session(context: JobContext):
    pipeline = Pipeline(
        llm=Realtime.gemini(
            model_id="gemini-3.1-flash-live-preview",
            voice="Puck",
            language_code="en-US",
            response_modalities=["AUDIO"],
        )
    )
    session = AgentSession(agent=MyAgent(), pipeline=pipeline)
    await session.start(wait_for_participant=True, run_until_shutdown=True)

有关按提供商计费的详细信息，请参阅推理定价。

---

支持的库和插件

该框架支持与多种类别下的各类 AI 模型和工具集成：

类别	服务
实时模型	OpenAI | Gemini | AWS Nova Sonic | Azure Voice Live
语音转文本 (STT)	OpenAI | Google | Azure AI Speech | Azure OpenAI | Sarvam AI | Deepgram | Cartesia | AssemblyAI | Navana
语言模型 (LLM)	OpenAI | Azure OpenAI | Google | Sarvam AI | Anthropic | Cerebras
文本转语音 (TTS)	OpenAI | Google | AWS Polly | Azure AI Speech | Azure OpenAI | Deepgram | Sarvam AI | ElevenLabs | Cartesia | Resemble AI | Smallest AI | Speechify | InWorld | Neuphonic | Rime AI | Hume AI | Groq | LMNT AI | Papla Media
语音活动检测 (VAD)	SileroVAD
轮次检测模型	Namo Turn Detector
虚拟化身	Simli | Anam | 自定义（实现 connect / aclose 协议）
LLM 编排	LangChain | LangGraph
降噪	RNNoise

[!TIP] 安装示例

# 安装带有特定插件的包
pip install videosdk-agents[openai,elevenlabs,silero]

# 安装单个插件
pip install videosdk-plugins-anthropic
pip install videosdk-plugins-deepgram

示例

通过以下示例，您可以了解该框架的实际应用：

核心模式示例

🎙️ 级联模式（基础） 使用 Google LLM + Deepgram STT + Cartesia TTS 的简单 STT → LLM → TTS 语音助手。	🔧 级联模式（高级） 具备 VAD、轮次检测及打断处理功能的高级级联助手。
⚡ 实时模式 使用 Gemini Live 实现最低延迟语音交互的极简实时助手。	🔀 混合模式 将级联模式与实时模式结合——自定义 STT 配合实时模型，或实时模式搭配自定义 TTS。
🧩 可组合管道 灵活的管道配置——仅转录、仅 LLM、语音+聊天、完整语音助手。	🪝 管道钩子 使用 @pipeline.on(...) 在任何阶段拦截并转换 STT、LLM 和 TTS 数据。

集成与高级功能

🌐 带有 MCP 服务器的代理 股票市场分析师代理，可通过模型上下文协议实时访问市场数据。	🤝 代理间通信（A2A） 多代理工作流：客户代理将贷款咨询转交给贷款专员代理。
🦜 LangChain 集成 在 VideoSDK 代理框架中使用 LangChain 工具和代理。	🕸️ LangGraph 集成 使用 LangGraph 状态机编排多步骤代理工作流。
🧠 记忆代理（Mem0） 利用 Mem0 实现跨会话的持久化记忆，以保持长期上下文。	👁️ 视觉代理 多模态代理，通过级联或实时管道同时处理视频帧和语音。
🔄 n8n 工作流集成 使用 Webhook 在您的代理中触发 n8n 自动化工作流。	🧑‍💼 人工介入 在对话过程中通过 Discord 或其他渠道升级到人工客服。

使用场景示例

📞 AI 电话客服代理 通过语音驱动的电话客服代理预约医院挂号。	✈️ AI WhatsApp 代理 随时随地查询并预订酒店客房。
🛒 具备知识库的代理（RAG） 基于文档知识回答问题的代理。	🎭 虚拟化身代理 展示天气预报的虚拟化身代理。
🏥 预约挂号 用于诊所预约的医疗前台接待员。	📣 公告代理 主动外呼代理，用于发布公告。
🎧 客户支持 具备升级机制和知识库的 AI 客户支持代理。	📂 更多使用场景 呼叫中心、IVR、医疗分诊、语言辅导、会议记录等。

文档

获取全面指南和 API 参考：

📄 官方文档 完整的框架文档

📝 API 参考 详细的 API 文档

📂 示例目录 更多代码示例

贡献

我们欢迎贡献！以下是您可以提供帮助的方式：

🐞 报告问题 提交关于 bug 或功能请求的问题

🔀 提交 PR 创建包含改进的拉取请求

🛠️ 开发插件 按照我们的插件开发指南进行操作

💬 加入社区 在 Discord 上与我们交流

该框架目前处于积极开发阶段，因此我们非常欢迎以新插件、新功能、bug 修复或文档改进等形式做出的贡献。

🛠️ 构建自定义插件

想要集成新的 AI 提供商吗？请查看 构建您自己的插件，了解：

• 分步插件创建指南
• 目录结构和文件要求
• STT、LLM 和 TTS 的实现示例
• 测试和提交指南

社区与支持

与 VideoSDK 保持联系：

💬 Discord 加入我们的社区

🐦 Twitter @video_sdk

▶️ YouTube VideoSDK 频道

🔗 LinkedIn VideoSDK 公司

[!TIP]

支持本项目！ ⭐️
请给仓库点个赞，加入社区，并通过提供反馈、报告 bug 或贡献插件来帮助我们改进 VideoSDK。

---

由 VideoSDK 团队用心打造

VideoSDK AI Agents 快速上手指南

VideoSDK AI Agents 是一个开源 Python 框架，用于构建生产级的实时语音和多模态 AI 智能体。它能让 AI 作为实时参与者加入视频会议房间，自动处理音频流、说话人检测、打断逻辑和媒体路由，让你专注于智能体业务逻辑。

环境准备

在开始之前，请确保满足以下要求：

• 操作系统: macOS, Linux 或 Windows
• Python 版本: Python 3.12 或更高版本
• VideoSDK 凭证:
  • VideoSDK 认证 Token (JWT)：可在 VideoSDK 控制台 生成。
  • Meeting ID (房间号)：可通过 Create Room API 或控制台创建。
• 第三方 API Key: 根据你选择的模型提供商（如 OpenAI, Google Gemini, Deepgram, ElevenLabs 等）准备相应的 API Key。

安装步骤

推荐使用 UV 包管理器，它能自动处理虚拟环境和依赖，速度更快。

方式一：使用 UV (推荐)

1. 安装 UV (如果尚未安装): 参考 UV 安装指南。

2. 初始化项目并安装核心包:

   uv init my-agent-project
   cd my-agent-project
   uv add videosdk-agents
   

3. 安装所需插件: 根据需求安装特定提供商的插件（例如 OpenAI 和 Deepgram）：

   uv add videosdk-plugins-openai
   uv add videosdk-plugins-deepgram
   

4. 运行智能体:

   uv run python main.py
   

方式二：使用 pip

1. 创建并激活虚拟环境:

   • macOS / Linux:

     python3 -m venv venv
     source venv/bin/activate
     

   • Windows:

     python -m venv venv
     venv\Scripts\activate
     

2. 安装核心包:

   pip install videosdk-agents
   

3. 安装所需插件:

   # 示例：安装 OpenAI 插件和语音活动检测插件
   pip install videosdk-plugins-openai
   pip install videosdk-plugins-turn-detector
   

基本使用

以下是构建第一个语音智能体的最小化示例。该示例展示了如何定义一个继承自 Agent 类的自定义智能体，并配置级联模式（STT → LLM → TTS）。

1. 编写代码 (main.py)

import os
from videosdk.agents import Agent, AgentSession, JobContext, Pipeline
from videosdk.plugins.openai import OpenAIRealtime # 或者使用 OpenAILLM, OpenAITTS 等组合
from videosdk.plugins.deepgram import DeepgramSTT
from videosdk.plugins.elevenlabs import ElevenLabsTTS

# 设置环境变量 (在实际项目中建议使用 .env 文件)
os.environ["VIDEOSDK_TOKEN"] = "YOUR_VIDEO_SDK_TOKEN"
os.environ["OPENAI_API_KEY"] = "YOUR_OPENAI_API_KEY"
os.environ["ELEVENLABS_API_KEY"] = "YOUR_ELEVENLABS_API_KEY"

class VoiceAgent(Agent):
    def __init__(self):
        super().__init__(
            instructions="You are a helpful voice assistant. Keep responses concise.",
            tools=[] # 可在此注册外部工具函数
        )

    async def on_enter(self) -> None:
        """当智能体加入会议时触发"""
        await self.session.say("Hi there! How can I help you today?")
    
    async def on_exit(self) -> None:
        """当智能体离开会议时触发"""
        await self.session.say("Goodbye!")

async def start_session(context: JobContext):
    # 配置 Pipeline：级联模式 (STT -> LLM -> TTS)
    pipeline = Pipeline(
        stt=DeepgramSTT(),
        llm=OpenAIRealtime(model="gpt-4o-realtime-preview"), # 或使用 GoogleLLM, AnthropicLLM 等
        tts=ElevenLabsTTS(),
    )
    
    session = AgentSession(agent=VoiceAgent(), pipeline=pipeline)
    
    # 启动会话：等待用户加入后开始，直到关闭
    await session.start(wait_for_participant=True, run_until_shutdown=True)

# 入口点 (具体启动方式取决于你的部署环境，本地测试通常配合 VideoSDK Worker)
# 此处仅为逻辑展示，实际运行需配合 VideoSDK Worker 服务

2. 获取 Meeting ID

在运行智能体前，你需要一个有效的 Meeting ID。可以使用 cURL 调用 API 生成：

curl -X POST https://api.videosdk.live/v2/rooms \
  -H "Authorization: YOUR_JWT_TOKEN_HERE" \
  -H "Content-Type: application/json"

3. 运行与测试

1. 将上述代码保存为 main.py。
2. 替换代码中的 YOUR_VIDEO_SDK_TOKEN 和其他 API Key。
3. 使用安装步骤中的命令运行脚本（例如 uv run python main.py）。
4. 智能体启动后，使用生成的 Meeting ID 在客户端（Web/Mobile/Desktop）加入房间，即可与 AI 进行实时语音对话。

提示: VideoSDK 支持多种运行模式，包括级联模式（灵活组合不同厂商的 STT/LLM/TTS）、实时模式（使用 Gemini Live 或 OpenAI Realtime 等统一模型以获得超低延迟）以及混合模式。只需更改 Pipeline 初始化时的组件即可切换。