LiveKit Agents for Python 实战用 AgentSession 构建生产级实时语音 AI Agent【免费下载链接】agentsA framework for building realtime voice AI agents ️项目地址: https://gitcode.com/GitHub_Trending/agen/agentsLiveKit 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会话编排、AgentAgent 定义、AgentTask、room_io房间音视频输入输出、turn轮次与端点检测、amd答录机检测、ivr等能力抽象层llm含ChatContext、ToolContext、RealtimeModel抽象、stt、tts、vad、tokenize分词器等均以可插拔接口形式存在进程与调度层worker.pyAgentServer、WorkerOptions、ipc多进程 IPC 与监督、inference模型推理执行器可观测性telemetryOpenTelemetry 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( llmopenai.realtime.RealtimeModel( voicecoral ) ) await session.start( roomctx.room, agentAgent(instructionsYou are a helpful voice AI assistant.) ) await session.generate_reply( instructionsGreet the user and offer your assistance. ) if __name__ __main__: agents.cli.run_app(agents.WorkerOptions(entrypoint_fncentrypoint))1.entrypoint与JobContextasync def entrypoint(ctx: agents.JobContext):这是每个 Agent 的入口函数接收一个JobContext。JobContext封装了一次任务的完整上下文包括连接到的roomctx.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. 创建AgentSessionsession AgentSession( llmopenai.realtime.RealtimeModel(voicecoral) )AgentSession是框架的核心运行时。在这里只传了llm使用的是 OpenAI Realtime 模型端到端语音模型语音输入输出由模型直接完成无需单独 STT/TTS 管线。voicecoral指定了模型使用的音色。README 这段代码之所以只配置llm正是因为 Realtime 模型自带语音能力这也展示了框架按需组合的设计。3. 启动会话session.startawait session.start( roomctx.room, agentAgent(instructionsYou 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_replyawait session.generate_reply( instructionsGreet the user and offer your assistance. )generate_reply是让 Agent 主动说话的入口非常适合开场白或被动场景下的主动服务。它会生成一次语音回复instructions参数提供本次回复的额外指令。注意在最新源码中它返回的是一个SpeechHandleagent_session.pyawait该句柄会等待回复播放完成且句柄本身不会抛出异常——需通过handle.exception()检查失败原因。5. Worker 与 CLI 入口if __name__ __main__: agents.cli.run_app(agents.WorkerOptions(entrypoint_fncentrypoint))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提供了非常丰富的配置项按功能分组如下能力组件不传时使用默认/推理配置参数作用说明stt语音转文字Agent 的耳朵传stt.STT实例或模型字符串例如inference.STT(deepgram/nova-3)vad语音活动检测默认使用内置 silero VADinference.VAD(modelsilero)传vadNone可关闭llm大语言模型Agent 的大脑llm.LLM/RealtimeModel/DuplexModel或模型字符串tts文字转语音Agent 的声音tts.TTS实例或模型字符串轮次与打断turn_handling轮次处理配置TurnHandlingOptions可细分子项interruption允许打断、打断检测模式adaptive/vad、误打断恢复resume_false_interruption等、endpointing端点检测延迟、preemptive_generation在用户说完之前预生成回复max_retries控制重试次数aec_warmup_durationAgent 开始说话后的一段时间内屏蔽打断秒用于让客户端完成回声消除AEC校准examples中常用3.0user_away_timeout用户与 Agent 都静默超过该时长后将用户状态标记为 away默认15.0秒设None关闭。工具调用tools全局工具列表会话内所有 Agent 共享tool_handling工具处理配置含异步工具的进度提示模板等max_tool_steps单次 LLM 轮次内最多连续工具调用次数默认3mcp_serversMCP 服务器列表自动把 MCP 工具暴露给 Agent需安装mcp可选依赖。文本与转录use_tts_aligned_transcript是否使用 TTS 对齐转录作为转录节点输入需 TTS 支持对齐能力tts_text_transformsTTS 输入文本变换内置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_optionsSTT/LLM/TTS 的通用连接选项重试、超时loop绑定的事件循环默认取当前事件循环。Agent 与函数工具Agentagent.py是 Agent 行为的最小单元。构造参数包括instructions系统提示词必填chat_ctx初始对话上下文ChatContexttools该 Agent 私有的工具列表stt/vad/llm/tts可覆盖会话级配置支持模型字符串会自动通过inference解析为对应实例turn_handling/tool_handling轮次与工具处理覆盖项expressive表达模式覆盖mcp_serversMCP 工具服务器。Agent还支持通过function_tool装饰器把类方法自动注册为 LLM 可调用的工具tool_context.py 中的function_tool并支持on_enter等生命周期钩子。以下来自 examples/voice_agents/basic_agent.pyclass MyAgent(Agent): def __init__(self) - None: super().__init__( instructionsYour name is Kelly, built by LiveKit. ..., tools[EndCallTool()], ) async def on_enter(self) - None: # 进入会话后自动生成开场白 self.session.generate_reply(instructionsgreet 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-realtimevoice音色默认marinREADME 示例传了coralmodalities启用的模态如[text, audio]input_audio_transcription用户语音的转录配置input_audio_noise_reduction输入降噪turn_detection服务端轮次检测配置tool_choice工具选择策略speed播放速度倍率truncation/reasoning截断与推理配置如RealtimeReasoning(effortlow)api_keyOpenAI 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(instructionsgreet 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( sttinference.STT(deepgram/nova-3, languagemulti), # 耳朵 llminference.LLM(openai/gpt-4.1-mini), # 大脑 ttsinference.TTS(cartesia/sonic-3), # 声音 turn_handlingTurnHandlingOptions( interruption{ resume_false_interruption: True, # 误打断后恢复语音 false_interruption_timeout: 1.0, }, preemptive_generation{enabled: True, max_retries: 3}, ), aec_warmup_duration3.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(fUsage: {session.usage})) await session.start( agentMyAgent(), roomctx.room, room_optionsroom_io.RoomOptions( audio_inputroom_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.mdLiveKit 服务一个 LiveKit Cloud 项目或本地 LiveKit 服务器Python 版本3.10pyproject.toml 中requires-python 3.10,3.15uv包管理器仓库使用 uv 管理见 uv.lock。在项目根目录或 examples 目录创建.envLIVEKIT_URLwss://your-project.livekit.cloud LIVEKIT_API_KEYyour_api_key LIVEKIT_API_SECRETyour_api_secret # 使用插件直连如 RealtimeModel时还需 OPENAI_API_KEYsk-...安装依赖并启动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_SECRETCLI 还提供开发模式python agent.py dev启用文件变更自动重载见 cli.py 中dev命令、console终端控制台直连调试支持--text纯文本模式与--record录制、simulate_job离线模拟任务等子命令。模型与插件生态AgentSession的能力组件全部是可插拔接口仓库通过LiveKit Inference与插件包两种方式提供模型LiveKit Inference统一模型访问 API一条字符串指定提供商与模型examples/README.mdfrom livekit.agents import inference session AgentSession( sttinference.STT(deepgram/nova-3), llminference.LLM(google/gemma-4-31b-it), # 低延迟托管于 LiveKit ttsinference.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-openai1.8.0]、mcp [mcp1.24.0,2]按需安装即可。需要说明的是Realtime 端到端模型如openai.realtime.RealtimeModel不经由 LiveKit Inference必须直接使用插件见 examples/README.md 中的说明。从源码理解框架设计如果继续深入阅读源码会发现几个值得关注的工程设计多进程 Worker 模型AgentServerworker.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.pyAgentSession启动、事件、录制与会话生命周期test_agent_update_options.py会话运行中动态更新配置test_agent_task_close_race.pyAgentTask 关闭竞态test_tools.py 与 test_tool_proxy.py函数工具注册与代理test_plugin_openai_realtime_reasoning.pyOpenAI 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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 LiveKit Agents for Python 实战：用 AgentSession 构建生产级实时语音 AI Agent
LiveKit Agents for Python 实战用 AgentSession 构建生产级实时语音 AI Agent【免费下载链接】agentsA framework for building realtime voice AI agents ️项目地址: https://gitcode.com/GitHub_Trending/agen/agentsLiveKit 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会话编排、AgentAgent 定义、AgentTask、room_io房间音视频输入输出、turn轮次与端点检测、amd答录机检测、ivr等能力抽象层llm含ChatContext、ToolContext、RealtimeModel抽象、stt、tts、vad、tokenize分词器等均以可插拔接口形式存在进程与调度层worker.pyAgentServer、WorkerOptions、ipc多进程 IPC 与监督、inference模型推理执行器可观测性telemetryOpenTelemetry 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( llmopenai.realtime.RealtimeModel( voicecoral ) ) await session.start( roomctx.room, agentAgent(instructionsYou are a helpful voice AI assistant.) ) await session.generate_reply( instructionsGreet the user and offer your assistance. ) if __name__ __main__: agents.cli.run_app(agents.WorkerOptions(entrypoint_fncentrypoint))1.entrypoint与JobContextasync def entrypoint(ctx: agents.JobContext):这是每个 Agent 的入口函数接收一个JobContext。JobContext封装了一次任务的完整上下文包括连接到的roomctx.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. 创建AgentSessionsession AgentSession( llmopenai.realtime.RealtimeModel(voicecoral) )AgentSession是框架的核心运行时。在这里只传了llm使用的是 OpenAI Realtime 模型端到端语音模型语音输入输出由模型直接完成无需单独 STT/TTS 管线。voicecoral指定了模型使用的音色。README 这段代码之所以只配置llm正是因为 Realtime 模型自带语音能力这也展示了框架按需组合的设计。3. 启动会话session.startawait session.start( roomctx.room, agentAgent(instructionsYou 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_replyawait session.generate_reply( instructionsGreet the user and offer your assistance. )generate_reply是让 Agent 主动说话的入口非常适合开场白或被动场景下的主动服务。它会生成一次语音回复instructions参数提供本次回复的额外指令。注意在最新源码中它返回的是一个SpeechHandleagent_session.pyawait该句柄会等待回复播放完成且句柄本身不会抛出异常——需通过handle.exception()检查失败原因。5. Worker 与 CLI 入口if __name__ __main__: agents.cli.run_app(agents.WorkerOptions(entrypoint_fncentrypoint))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提供了非常丰富的配置项按功能分组如下能力组件不传时使用默认/推理配置参数作用说明stt语音转文字Agent 的耳朵传stt.STT实例或模型字符串例如inference.STT(deepgram/nova-3)vad语音活动检测默认使用内置 silero VADinference.VAD(modelsilero)传vadNone可关闭llm大语言模型Agent 的大脑llm.LLM/RealtimeModel/DuplexModel或模型字符串tts文字转语音Agent 的声音tts.TTS实例或模型字符串轮次与打断turn_handling轮次处理配置TurnHandlingOptions可细分子项interruption允许打断、打断检测模式adaptive/vad、误打断恢复resume_false_interruption等、endpointing端点检测延迟、preemptive_generation在用户说完之前预生成回复max_retries控制重试次数aec_warmup_durationAgent 开始说话后的一段时间内屏蔽打断秒用于让客户端完成回声消除AEC校准examples中常用3.0user_away_timeout用户与 Agent 都静默超过该时长后将用户状态标记为 away默认15.0秒设None关闭。工具调用tools全局工具列表会话内所有 Agent 共享tool_handling工具处理配置含异步工具的进度提示模板等max_tool_steps单次 LLM 轮次内最多连续工具调用次数默认3mcp_serversMCP 服务器列表自动把 MCP 工具暴露给 Agent需安装mcp可选依赖。文本与转录use_tts_aligned_transcript是否使用 TTS 对齐转录作为转录节点输入需 TTS 支持对齐能力tts_text_transformsTTS 输入文本变换内置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_optionsSTT/LLM/TTS 的通用连接选项重试、超时loop绑定的事件循环默认取当前事件循环。Agent 与函数工具Agentagent.py是 Agent 行为的最小单元。构造参数包括instructions系统提示词必填chat_ctx初始对话上下文ChatContexttools该 Agent 私有的工具列表stt/vad/llm/tts可覆盖会话级配置支持模型字符串会自动通过inference解析为对应实例turn_handling/tool_handling轮次与工具处理覆盖项expressive表达模式覆盖mcp_serversMCP 工具服务器。Agent还支持通过function_tool装饰器把类方法自动注册为 LLM 可调用的工具tool_context.py 中的function_tool并支持on_enter等生命周期钩子。以下来自 examples/voice_agents/basic_agent.pyclass MyAgent(Agent): def __init__(self) - None: super().__init__( instructionsYour name is Kelly, built by LiveKit. ..., tools[EndCallTool()], ) async def on_enter(self) - None: # 进入会话后自动生成开场白 self.session.generate_reply(instructionsgreet 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-realtimevoice音色默认marinREADME 示例传了coralmodalities启用的模态如[text, audio]input_audio_transcription用户语音的转录配置input_audio_noise_reduction输入降噪turn_detection服务端轮次检测配置tool_choice工具选择策略speed播放速度倍率truncation/reasoning截断与推理配置如RealtimeReasoning(effortlow)api_keyOpenAI 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(instructionsgreet 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( sttinference.STT(deepgram/nova-3, languagemulti), # 耳朵 llminference.LLM(openai/gpt-4.1-mini), # 大脑 ttsinference.TTS(cartesia/sonic-3), # 声音 turn_handlingTurnHandlingOptions( interruption{ resume_false_interruption: True, # 误打断后恢复语音 false_interruption_timeout: 1.0, }, preemptive_generation{enabled: True, max_retries: 3}, ), aec_warmup_duration3.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(fUsage: {session.usage})) await session.start( agentMyAgent(), roomctx.room, room_optionsroom_io.RoomOptions( audio_inputroom_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.mdLiveKit 服务一个 LiveKit Cloud 项目或本地 LiveKit 服务器Python 版本3.10pyproject.toml 中requires-python 3.10,3.15uv包管理器仓库使用 uv 管理见 uv.lock。在项目根目录或 examples 目录创建.envLIVEKIT_URLwss://your-project.livekit.cloud LIVEKIT_API_KEYyour_api_key LIVEKIT_API_SECRETyour_api_secret # 使用插件直连如 RealtimeModel时还需 OPENAI_API_KEYsk-...安装依赖并启动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_SECRETCLI 还提供开发模式python agent.py dev启用文件变更自动重载见 cli.py 中dev命令、console终端控制台直连调试支持--text纯文本模式与--record录制、simulate_job离线模拟任务等子命令。模型与插件生态AgentSession的能力组件全部是可插拔接口仓库通过LiveKit Inference与插件包两种方式提供模型LiveKit Inference统一模型访问 API一条字符串指定提供商与模型examples/README.mdfrom livekit.agents import inference session AgentSession( sttinference.STT(deepgram/nova-3), llminference.LLM(google/gemma-4-31b-it), # 低延迟托管于 LiveKit ttsinference.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-openai1.8.0]、mcp [mcp1.24.0,2]按需安装即可。需要说明的是Realtime 端到端模型如openai.realtime.RealtimeModel不经由 LiveKit Inference必须直接使用插件见 examples/README.md 中的说明。从源码理解框架设计如果继续深入阅读源码会发现几个值得关注的工程设计多进程 Worker 模型AgentServerworker.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.pyAgentSession启动、事件、录制与会话生命周期test_agent_update_options.py会话运行中动态更新配置test_agent_task_close_race.pyAgentTask 关闭竞态test_tools.py 与 test_tool_proxy.py函数工具注册与代理test_plugin_openai_realtime_reasoning.pyOpenAI 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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考