企业官网建设流程全解析

LiveKit Agents for Python 实战：用 AgentSession 构建生产级实时语音 AI Agent

【免费下载链接】agentsA framework for building realtime voice AI agents 🤖🎙️📹项目地址: https://gitcode.com/GitHub_Trending/agen/agents

LiveKit Agents 是 LiveKit 官方推出的 Python 实时多模态与语音 AI Agent 框架，定位是"面向生产环境的实时 AI Agent 构建框架"（livekit-agents/README.md）。本文以该文档为核心，结合仓库内 AgentSession 运行时源码、Agent 抽象类、OpenAI Realtime 插件 以及 examples 示例集，从零讲解如何用十几行代码写出一个可实时对话的语音 Agent，并深入拆解AgentSession、Agent、JobContext、WorkerOptions等核心概念的底层实现。读完本文，你将掌握该框架的最小可运行范式、完整配置项、函数工具与实时模型接入方式，并能基于仓库示例搭建自己的语音助手。

LiveKit Agents 概览：一个端到端的实时语音运行时

livekit-agents是仓库中独立打包的 Python 库（pyproject.toml 中name = "livekit-agents"，描述为 "A powerful framework for building realtime voice AI agents"）。它解决的问题非常具体：把 WebRTC 房间内的音频/视频流、语音识别（STT）、语音合成（TTS）、大语言模型（LLM）、语音活动检测（VAD）、打断处理、工具调用等复杂环节，编排成一个开箱即用的实时 Agent 会话。

从源码结构看（livekit/agents 目录），框架主要分为几层：

• voice 层：核心运行时，包含AgentSession（会话编排）、Agent（Agent 定义）、AgentTask、room_io（房间音视频输入输出）、turn（轮次与端点检测）、amd（答录机检测）、ivr等；
• 能力抽象层：llm（含ChatContext、ToolContext、RealtimeModel抽象）、stt、tts、vad、tokenize（分词器）等，均以可插拔接口形式存在；
• 进程与调度层：worker.py（AgentServer、WorkerOptions）、ipc（多进程 IPC 与监督）、inference（模型推理执行器）；
• 可观测性：telemetry（OpenTelemetry trace/log/metrics）、metrics、observability.py。

所有核心能力通过 livekit/agents/init.py 对外导出，顶层即可from livekit import agents使用。

最小可运行示例：从 README 出发

README 给出了一个完整可运行的最小示例，这是理解整个框架的最佳入口。逐行拆解如下：

from dotenv import load_dotenv from livekit import agents from livekit.agents import AgentSession, Agent, RoomInputOptions from livekit.plugins import openai load_dotenv() async def entrypoint(ctx: agents.JobContext): await ctx.connect() session = AgentSession( llm=openai.realtime.RealtimeModel( voice="coral" ) ) await session.start( room=ctx.room, agent=Agent(instructions="You are a helpful voice AI assistant.") ) await session.generate_reply( instructions="Greet the user and offer your assistance." ) if __name__ == "__main__": agents.cli.run_app(agents.WorkerOptions(entrypoint_fnc=entrypoint))

1.entrypoint与JobContext

async def entrypoint(ctx: agents.JobContext):

这是每个 Agent 的入口函数，接收一个JobContext。JobContext封装了一次"任务"的完整上下文：包括连接到的room（ctx.room）、当前 job 信息、进程上下文（ctx.proc）、LiveKit API 客户端（ctx.api()）、以及各类生命周期回调。从 job.py 的源码看，JobContext还提供add_shutdown_callback（会话结束回调）、wait_for_participant（等待用户入会）、add_sip_participant（接入 SIP 电话）等能力。

await ctx.connect()负责让 Agent 以参与者身份加入 LiveKit 房间——这是后续一切音视频流转发的前提。

2. 创建AgentSession

session = AgentSession( llm=openai.realtime.RealtimeModel(voice="coral") )

AgentSession是框架的核心运行时。在这里只传了llm，使用的是 OpenAI Realtime 模型（端到端语音模型，语音输入输出由模型直接完成，无需单独 STT/TTS 管线）。voice="coral"指定了模型使用的音色。README 这段代码之所以只配置llm，正是因为 Realtime 模型自带语音能力，这也展示了框架"按需组合"的设计。

3. 启动会话：session.start

await session.start( room=ctx.room, agent=Agent(instructions="You are a helpful voice AI assistant.") )

AgentSession.start()将 Agent 挂载到会话并绑定房间。Agent(instructions=...)定义了 Agent 的系统提示词。从 agent_session.py 的源码实现看，start内部会做一系列初始化：创建默认的RoomIO（房间音频输入输出）、解析录制选项、设置 primary session、配置可观测性等。方法签名中的room_options、session_host、record、capture_run等参数分别控制房间 I/O 配置、是否允许远程会话驱动、是否录制（音频/转录/追踪/日志）以及是否捕获运行结果（RunResult）。

4. 主动开场：generate_reply

await session.generate_reply( instructions="Greet the user and offer your assistance." )

generate_reply是让 Agent "主动说话"的入口，非常适合开场白或被动场景下的主动服务。它会生成一次语音回复，instructions参数提供本次回复的额外指令。注意在最新源码中它返回的是一个SpeechHandle（agent_session.py），await该句柄会等待回复播放完成，且句柄本身不会抛出异常——需通过handle.exception()检查失败原因。

5. Worker 与 CLI 入口

if __name__ == "__main__": agents.cli.run_app(agents.WorkerOptions(entrypoint_fnc=entrypoint))

WorkerOptions描述了 Worker 的启动参数：entrypoint_fnc指定入口函数，还支持load_threshold（负载阈值）、num_idle_processes（空闲进程数）、drain_timeout、api_key/api_secret/ws_url等（见 worker.py 中WorkerOptions.__init__的完整签名）。agents.cli.run_app(...)则启动整个应用：连接 LiveKit 服务器、注册 Worker、按需接收任务并拉起子进程执行entrypoint。

AgentSession 核心参数全解

AgentSession.__init__（agent_session.py）提供了非常丰富的配置项，按功能分组如下：

能力组件（不传时使用默认/推理配置）：

轮次与打断：

• turn_handling：轮次处理配置（TurnHandlingOptions），可细分子项interruption（允许打断、打断检测模式adaptive/vad、误打断恢复resume_false_interruption等）、endpointing（端点检测延迟）、preemptive_generation（在用户说完之前预生成回复，max_retries控制重试次数）；
• aec_warmup_duration：Agent 开始说话后的一段时间内屏蔽打断（秒），用于让客户端完成回声消除（AEC）校准，examples中常用3.0；
• user_away_timeout：用户与 Agent 都静默超过该时长后，将用户状态标记为 "away"，默认15.0秒，设None关闭。

工具调用：

• tools：全局工具列表，会话内所有 Agent 共享；
• tool_handling：工具处理配置（含异步工具的进度提示模板等）；
• max_tool_steps：单次 LLM 轮次内最多连续工具调用次数，默认3；
• mcp_servers：MCP 服务器列表，自动把 MCP 工具暴露给 Agent（需安装mcp可选依赖）。

文本与转录：

• use_tts_aligned_transcript：是否使用 TTS 对齐转录作为转录节点输入（需 TTS 支持对齐能力）；
• tts_text_transforms：TTS 输入文本变换，内置"filter_markdown"、"filter_emoji"，也可用text_transforms.replace({...})自定义发音替换；
• stt_context_options：对话感知的 STT 上下文（keyterms关键词、keyterm_detection自动关键词检测、forward_chat_context前向传递对话上下文）；
• transcription_timeout：用户说话但迟迟无最终转录时，超时后触发user_transcription_timeout事件。

其他：

• expressive：表达模式（让 LLM 通过内联标签控制情绪、语速、非语言音效，由 TTS 渲染）；
• ivr_detection：检测 Agent 是否在与 IVR 电话系统交互；
• userdata：任意类型的会话级用户数据；
• video_sampler：视频采样器（多模态场景，默认在用户说话时约 1fps、静默时 0.3fps 采样）；
• conn_options：STT/LLM/TTS 的通用连接选项（重试、超时）；
• loop：绑定的事件循环，默认取当前事件循环。

Agent 与函数工具

Agent（agent.py）是 Agent 行为的最小单元。构造参数包括：

• instructions：系统提示词（必填）；
• chat_ctx：初始对话上下文（ChatContext）；
• tools：该 Agent 私有的工具列表；
• stt/vad/llm/tts：可覆盖会话级配置（支持模型字符串，会自动通过inference解析为对应实例）；
• turn_handling/tool_handling：轮次与工具处理覆盖项；
• expressive：表达模式覆盖；
• mcp_servers：MCP 工具服务器。

Agent还支持通过@function_tool装饰器把类方法自动注册为 LLM 可调用的工具（tool_context.py 中的function_tool），并支持on_enter等生命周期钩子。以下来自 examples/voice_agents/basic_agent.py：

class MyAgent(Agent): def __init__(self) -> None: super().__init__( instructions="Your name is Kelly, built by LiveKit. ...", tools=[EndCallTool()], ) async def on_enter(self) -> None: # 进入会话后自动生成开场白 self.session.generate_reply(instructions="greet the user and introduce yourself") # 所有带 @function_tool 的方法都会在该 Agent 激活时传给 LLM @function_tool async def lookup_weather( self, context: RunContext, location: str, latitude: str, longitude: str ) -> str: """Called when the user asks for weather related information...""" return "sunny with a temperature of 70 degrees."

多 Agent 场景下，每个Agent可有独立 instructions、tools 与 chat_ctx，通过AgentTask/AgentHandoff机制在会话内切换，从而实现"前台客服 → 后台专员"这类交接流。

使用 OpenAI Realtime 端到端语音模型

README 示例使用的openai.realtime.RealtimeModel来自仓库的 livekit-plugins-openai 插件包。其构造参数（realtime_model.py）非常丰富：

• model：模型名，默认"gpt-realtime"；
• voice：音色，默认"marin"（README 示例传了"coral"）；
• modalities：启用的模态，如["text", "audio"]；
• input_audio_transcription：用户语音的转录配置；
• input_audio_noise_reduction：输入降噪；
• turn_detection：服务端轮次检测配置；
• tool_choice：工具选择策略；
• speed：播放速度倍率；
• truncation/reasoning：截断与推理配置（如RealtimeReasoning(effort="low")）；
• api_key：OpenAI API Key，缺省从OPENAI_API_KEY环境变量读取；
• base_url：可指向兼容端点；
• azure_deployment/entra_token：传入任一 Azure 参数即切换为 Azure OpenAI Realtime 模式；
• max_session_duration：连接复用上限（秒），到期自动回收连接；
• conn_options：重试与连接选项。

由于 Realtime 模型本身就是"语音进、语音出"的端到端模型，README 示例中无需再配置 STT 与 TTS——这正是该框架支持两种架构的体现：传统流水线（STT → LLM → TTS）与Realtime 端到端模型，两者都可无缝接入AgentSession。

完整实战：一个配置齐全的语音 Agent

把上面的知识串起来，参考 basic_agent.py 构建一个生产风格示例：

import logging from dotenv import load_dotenv from livekit.agents import ( Agent, AgentServer, AgentSession, JobContext, RunContext, TurnHandlingOptions, cli, inference, metrics, room_io, text_transforms, ) from livekit.agents.beta import EndCallTool from livekit.agents.llm import function_tool logger = logging.getLogger("basic-agent") load_dotenv() class MyAgent(Agent): def __init__(self) -> None: super().__init__( instructions=( "Your name is Kelly, built by LiveKit. You would interact " "with users via voice. Keep responses concise. No emojis." ), tools=[EndCallTool()], ) async def on_enter(self) -> None: self.session.generate_reply(instructions="greet the user and introduce yourself") @function_tool async def lookup_weather(self, context: RunContext, location: str) -> str: """Called when the user asks for weather related information.""" return "sunny with a temperature of 70 degrees." server = AgentServer() @server.rtc_session() async def entrypoint(ctx: JobContext) -> None: ctx.log_context_fields = {"room": ctx.room.name} session: AgentSession = AgentSession( stt=inference.STT("deepgram/nova-3", language="multi"), # 耳朵 llm=inference.LLM("openai/gpt-4.1-mini"), # 大脑 tts=inference.TTS("cartesia/sonic-3"), # 声音 turn_handling=TurnHandlingOptions( interruption={ "resume_false_interruption": True, # 误打断后恢复语音 "false_interruption_timeout": 1.0, }, preemptive_generation={"enabled": True, "max_retries": 3}, ), aec_warmup_duration=3.0, # 开场屏蔽打断，校准 AEC tts_text_transforms=[ "filter_emoji", "filter_markdown", text_transforms.replace({"LiveKit": "<<ˈ|l|aɪ|v|k|ɪ|t>>"}), ], stt_context_options={ "keyterms": ["LiveKit"], "keyterm_detection": {"enabled": True, "turn_interval": 1}, }, ) @session.on("metrics_collected") def _on_metrics_collected(ev) -> None: metrics.log_metrics(ev.metrics) ctx.add_shutdown_callback(lambda: logger.info(f"Usage: {session.usage}")) await session.start( agent=MyAgent(), room=ctx.room, room_options=room_io.RoomOptions( audio_input=room_io.AudioInputOptions(), # 可挂降噪 Filter 等 ), ) if __name__ == "__main__": cli.run_app(server)

注意@server.rtc_session()与 README 中WorkerOptions(entrypoint_fnc=...)是两种等价写法：前者是AgentServer提供的装饰器风格（worker.py 中rtc_session），还支持on_request、on_session_end等回调；后者是面向单入口的简化写法。

环境配置与运行

运行示例需要（依据 examples/README.md）：

1. LiveKit 服务：一个 LiveKit Cloud 项目或本地 LiveKit 服务器；
2. Python 版本：>=3.10（pyproject.toml 中requires-python = ">=3.10,<3.15"）；
3. uv包管理器（仓库使用 uv 管理，见 uv.lock）。

在项目根目录（或 examples 目录）创建.env：

LIVEKIT_URL="wss://your-project.livekit.cloud" LIVEKIT_API_KEY="your_api_key" LIVEKIT_API_SECRET="your_api_secret" # 使用插件直连（如 RealtimeModel）时还需： OPENAI_API_KEY="sk-..."

安装依赖并启动：

uv sync # 或 pip install -e livekit-agents uv run python examples/voice_agents/basic_agent.py start --url "$LIVEKIT_URL" \ --api-key "$LIVEKIT_API_KEY" --api-secret "$LIVEKIT_API_SECRET"

CLI 还提供开发模式：python agent.py dev（启用文件变更自动重载，见 cli.py 中dev命令）、console（终端控制台直连调试，支持--text纯文本模式与--record录制）、simulate_job（离线模拟任务）等子命令。

模型与插件生态

AgentSession的能力组件全部是可插拔接口，仓库通过LiveKit Inference与插件包两种方式提供模型：

• LiveKit Inference：统一模型访问 API，一条字符串指定提供商与模型（examples/README.md）：

from livekit.agents import inference session = AgentSession( stt=inference.STT("deepgram/nova-3"), llm=inference.LLM("google/gemma-4-31b-it"), # 低延迟，托管于 LiveKit tts=inference.TTS("cartesia/sonic-3"), )

• 插件包：livekit-plugins-*系列（livekit-plugins 目录下 90+ 个包），覆盖 OpenAI、Anthropic、Google Gemini、Azure、AWS Bedrock、Deepgram、Cartesia、ElevenLabs、Silero、xAI、Meta 等。在 pyproject.toml 中以可选依赖形式声明（如openai = ["livekit-plugins-openai>=1.8.0"]、mcp = ["mcp>=1.24.0,<2"]），按需安装即可。

需要说明的是，Realtime 端到端模型（如openai.realtime.RealtimeModel）不经由 LiveKit Inference，必须直接使用插件（见 examples/README.md 中的说明）。

从源码理解框架设计

如果继续深入阅读源码，会发现几个值得关注的工程设计：

• 多进程 Worker 模型：AgentServer（worker.py）通过ipc模块维护空闲进程池（num_idle_processes），每个 Job 在独立子进程中执行entrypoint，通过 Unix socket 做 IPC，并内置 ping/pong 健康检查、内存监控（job_memory_warn_mb/job_memory_limit_mb）与崩溃监督（supervised_proc.py）。这是"生产级"的关键保障；
• 会话编排：AgentSession.start内部会先配置可观测性（录制、追踪、日志），再挂载RoomIO，并处理 primary/secondary 会话关系（agent_session.py）；
• 可观测性：基于 OpenTelemetry 的 trace/log/metrics 管线（telemetry），支持 PII 脱敏、会话报告上传、Prometheus 指标暴露，配合metrics.log_metrics即可把每轮延迟与 token 用量打进日志。

测试与验证

仓库为框架配备了规模可观的测试集（tests 目录，含 200+ 测试文件）。与本文主题直接相关的有：

• test_agent_session.py：AgentSession启动、事件、录制与会话生命周期；
• test_agent_update_options.py：会话运行中动态更新配置；
• test_agent_task_close_race.py：AgentTask 关闭竞态；
• test_tools.py 与 test_tool_proxy.py：函数工具注册与代理；
• test_plugin_openai_realtime_reasoning.py：OpenAI Realtime 推理配置。

examples/homepage/tests下还有面向业务 Agent 的单元与评估（evals）测试范式，可作为自己项目测试的参考。

总结

LiveKit Agents 的核心心智模型可以浓缩为三句话：AgentSession是运行时，Agent是行为，WorkerOptions是部署入口。通过 livekit-agents/README.md 的最小示例，你可以在十分钟内跑通一个实时语音 Agent；通过 AgentSession 配置 与 examples，你可以把它扩展为带函数工具、多 Agent 交接、电话 IVR、可观测性完备的生产级服务。深入 livekit/agents 源码与 tests 测试，则能获得对实时语音系统编排细节最准确的理解。

【免费下载链接】agentsA framework for building realtime voice AI agents 🤖🎙️📹项目地址: https://gitcode.com/GitHub_Trending/agen/agents