企业官网建设流程全解析

Voice Agents Examples：用 LiveKit Agents 构建实时语音 AI Agent 的完整示例指南

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

导读

examples/voice_agents 是当前仓库中演示如何用 LiveKit Agents 框架构建实时语音 AI Agent 的示例集合，覆盖了从“开箱即用”的基础语音 Agent 到实时语音到语音模型（xAI Grok）、MCP 外部工具集成、LlamaIndex RAG 知识库问答以及 OpenTelemetry 链路追踪等完整能力。读完本文，你将掌握如何用一行代码级别的inference统一接口接入 STT/LLM/TTS 模型、如何配置抢跑生成与防误打断、如何把 Agent 接到 MCP 服务器、如何用 LlamaIndex 给语音助手插上知识库，以及如何对整场会话做 OTLP 可观测性追踪。

一、示例集总览：从入门到生产级能力

examples/voice_agents目录以basic_agent.py为入口，围绕 LiveKit Agents 框架展示了多个相互独立、可直接运行的实战示例：

这些示例都通过pyproject.toml声明依赖（livekit-agents[openai,cartesia,elevenlabs,deepgram,silero,turn-detector,mcp]>=1.6与python-dotenv），属于脚本集合而非可安装发行包，因此可以脱离仓库独立解析运行，具体见 pyproject.toml。

二、模型配置：LiveKit Inference 统一接入 STT/LLM/TTS

2.1 为什么用 LiveKit Inference

大多数示例默认使用LiveKit Inference，它提供了一套统一 API来访问 STT、LLM 和 TTS 模型，调用方无需关心底层是哪个厂商、走 HTTP 还是流式协议，只需给出一个provider/model形式的模型标识字符串：

from livekit.agents import inference session = AgentSession( stt=inference.STT("deepgram/nova-3"), llm=inference.LLM("google/gemma-4-31b-it"), # low-latency gemma, hosted on LiveKit tts=inference.TTS("cartesia/sonic-3"), )

从源码结构看，inference.STT、inference.LLM、inference.TTS分别封装在 inference/stt.py（STT类，第 409 行）、inference/llm.py（LLM类，第 226 行）、inference/tts.py（TTS类，第 231 行），它们继承自框架的stt.STT、llm.LLM、tts.TTS基类，因此可以无缝替换为任何直接使用厂商 Plugin 的写法。

2.2 实时语音到语音模型的边界

原文档特别强调了一个关键约束：实时语音到语音（voice-to-voice）模型（如 Amazon Nova Sonic、xAI Grok 等）不受 LiveKit Inference 支持，必须直接使用对应的厂商 Plugin。这是因为 voice-to-voice 模型以音频为输入输出、不走“语音转文字 → 文字生成 → 文字转语音”的级联管线，无法用统一推理 API 表达。这也解释了为什么grok/示例要直接使用livekit.plugins.xai的RealtimeModel（详见下文第五节）。

三、基础语音 Agent：basic_agent.py 逐段精读

basic_agent.py 被原文档定位为“具备多语种 STT、抢跑生成、预生成与指标采集能力的基础语音 Agent”，其实现几乎覆盖了一个生产级语音 Agent 的所有关键旋钮。

3.1 Agent 定义：指令、工具与进场行为

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: # when the agent is added to the session, it'll generate a reply # according to its instructions self.session.generate_reply(instructions="greet the user and introduce yourself")

要点：

• instructions是 Agent 的系统提示词，这里刻意要求“回复简洁、不使用 emoji / 星号 / markdown 等特殊字符”，这是语音场景的通用最佳实践——文字排版符号在 TTS 播报时会产生噪声。
• tools=[EndCallTool()]来自livekit.agents.beta，让 LLM 可以在对话中自主决定结束通话。
• on_enter在 Agent 加入会话时触发，通过generate_reply(instructions=...)让 Agent 按指令主动生成开场白。

3.2 函数工具：让 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. Ensure the user's location (city or region) is provided. ...""" logger.info(f"Looking up weather for {location}") return "sunny with a temperature of 70 degrees."

@function_tool装饰器（来自livekit.agents.llm）会把方法自动注册为 LLM 可调用的函数工具：docstring 中的参数说明会成为 LLM 的函数 schema 描述，注释还明确要求“不要向用户索要经纬度，自行估算”。示例返回的是硬编码的天气字符串，实际项目可在此处调用真实天气 API。

3.3 AgentSession：STT / LLM / TTS 与轮次控制

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", voice="9626c31c-bec5-4cca-baa8-f8ba9e84c8bc"), 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, 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, }, }, )

逐项解读（含源码佐证）：

• STT/LLM/TTS：language="multi"表示多语种识别；voice 参数指定 Cartesia 的具体音色 ID。
• 防误打断：resume_false_interruption=True表示当检测到背景噪声等误判打断时，可恢复 Agent 的播报；false_interruption_timeout=1.0是误打断的判定时间窗。在 agent_activity.py 中可见框架按interruption["false_interruption_timeout"]进行误打断判定与恢复逻辑（第 2207、2374、2406 行附近）。
• 抢跑生成：preemptive_generation={"enabled": True, "max_retries": 3}允许 LLM 在轮次结束前就提前生成回复，降低端到端延迟；max_retries控制最多重试次数。对应实现见 agent_activity.py 第 2416 行起的on_preemptive_generation，其中会检查preemptive_opts["max_retries"]并递增计数（第 2435、2438 行）。
• AEC 预热：aec_warmup_duration=3.0在 Agent 开始说话后的数秒内屏蔽打断，给客户端留出回声消除（AEC）校准时间；框架在 agent_activity.py 中以_aec_warmup_remaining/_aec_warmup_timer跟踪剩余预热时间并在预热期内丢弃输入（第 1510-1524 行）。
• TTS 文本变换：filter_emoji、filter_markdown清洗文本，text_transforms.replace(...)用发音标注替换专有名词，保证 “LiveKit” 被正确拼读。
• STT 上下文：keyterms=["LiveKit"]让 STT 在每轮用户语音中自动检测关键词并注入识别上下文，keyterm_detection.turn_interval=1控制关键词检测的触发间隔（调大可减少 LLM API 调用）。框架在 agent_activity.py 中会依据stt_context_options["forward_chat_context"]等配置把聊天上下文前向传给 STT（第 774、1209-1214 行）。

3.4 指标采集与会话生命周期

@session.on("metrics_collected") def _on_metrics_collected(ev: MetricsCollectedEvent) -> None: if ev.metrics.type == "stt_metrics": return metrics.log_metrics(ev.metrics) async def log_usage(): logger.info(f"Usage: {session.usage}") ctx.add_shutdown_callback(log_usage)

• metrics_collected事件回调中过滤掉stt_metrics（避免日志刷屏），其余指标交给metrics.log_metrics记录。
• session.usage汇总模型用量，通过ctx.add_shutdown_callback(log_usage)在会话结束时打印，用于成本核算。
• session.start()时通过room_io.RoomOptions(audio_input=...)配置房间音频输入，注释里预留了开启 Krisp Viva 语音隔离（KrispVivaFilterFrameProcessor）的位置。

最后cli.run_app(server)启动 AgentServer 并注册@server.rtc_session()入口点，这是 LiveKit Agents 标准运行方式。

四、MCP 集成：把 Agent 接到外部工具服务器

Model Context Protocol（MCP）让 LLM 应用可以标准方式连接外部工具与数据源。示例包含 MCP 服务器与 Agent 两端，二者通过 HTTP SSE 通信。

4.1 MCP 服务器端：server.py

server.py 用mcp.server.fastmcp.FastMCP定义了两个工具：

• get_weather(location)：同步工具，返回固定天气文本；
• book_flight(origin, destination, date, ctx)：异步长任务工具，模拟“搜索航空公司（约 30 秒）→ 确认预订（约 40 秒）”两阶段，期间通过ctx.report_progress(0.0, 1.0, message)上报进度，最后返回航班号、价格与确认码。

服务端以mcp.run(transport="sse")启动，默认监听 SSE 传输。

4.2 Agent 端：mcp-agent.py

mcp-agent.py 展示了如何在AgentSession中挂载 MCP 工具集：

tools=[ mcp.MCPToolset( id="mcp_toolset_1", mcp_server=mcp.MCPServerHTTP( url="http://localhost:8000/sse", # book_flight takes ~70s; bump the per-request timeout above that. client_session_timeout_seconds=120, ), # book_flight is long-running: forward progress and allow cancellation tool_options={ "book_flight": { "flags": ToolFlag.CANCELLABLE, "on_duplicate": "confirm", "report_progress": True, }, }, ) ],

关键点：

• mcp.MCPServerHTTP(url=...)通过 HTTP SSE 连接 MCP 服务器；client_session_timeout_seconds=120必须大于长任务的执行时长（示例中book_flight约 70 秒），否则请求会被超时中断。
• tool_options针对特定工具做精细控制：ToolFlag.CANCELLABLE允许用户打断该工具；on_duplicate="confirm"表示重复调用前需与用户确认；report_progress=True会把 MCP 服务端的进度上报转发给会话，让 Agent 能够“边执行边向用户播报进度”。
• Agent 的instructions明确告诉 LLM 何时调用哪个工具，并要求“自然地叙述进度、不要编造航班细节”，这保证了长任务场景下 LLM 的发言可控。

五、实时语音到语音：xAI Grok Voice Agents API

实时语音到语音模型跳过“STT→LLM→TTS”级联，直接以语音为输入输出，延迟更低。原文档指出这类模型必须使用厂商 Plugin 直连，grok/grok_voice_agent_api.py 正是这一写法的代表：

session = AgentSession( llm=xai.realtime.RealtimeModel(voice="ara"), tools=[xai.realtime.XSearch(), xai.realtime.WebSearch()], preemptive_generation=True, )

• xai.realtime.RealtimeModel(voice="ara")直接使用 xAI Grok 的实时语音模型（Plugin 直连，不走inference）。
• XSearch()与WebSearch()是 Grok 内置的搜索工具，可检索 X.com 与全网信息，因此 Agent 能回答“Elon Musk 最近发了什么 X 帖”这类需要实时信息的问题。
• preemptive_generation=True同样开启抢跑生成以降低延迟。
• 房间音频输入处预留了 Krisp 噪声消除（BVCTelephony/BVC）的接入点，可根据参与者类型（如 SIP 电话 vs 普通客户端）选择不同降噪模型。

5.1 运行 Grok 示例

按 grok/README.md 的 Quickstart，需要四步：

1. 设置环境变量：所有LIVEKIT_前缀的密钥来自 LiveKit Cloud 项目或自托管 LiveKit 服务器；XAI_API_KEY来自 xAI。

   export LIVEKIT_API_KEY=<your-livekit-api-key> export LIVEKIT_API_SECRET=<your-livekit-api-secret> export LIVEKIT_URL=<your-livekit-url> export XAI_API_KEY=<your-xai-api-key>

2. 安装依赖：

   uv add "livekit-agents[xai,silero]" livekit-plugins-noise-cancellation

3. 运行 Agent，有三种方式：

   • 控制台模式：uv run grok_voice_agent_api.py console，直接在终端里与 Grok 语音对话；
   • Agents playground：先uv run grok_voice_agent_api.py dev启动，再打开 LiveKit 托管的 agents-playground 网页，选择与 Agent 关联的 LiveKit Cloud 项目并点击连接，即可在浏览器中通话；
   • 自定义前端：同样先uv run grok_voice_agent_api.py dev，然后运行官方的 agent-starter-react 等前端模板（也提供 Android、Swift、Flutter、React Native、ESP32、Embed 等版本）。注意：前端与后端必须使用相同的LIVEKIT_环境变量才能互联。

4. 验证效果：向 Grok 提问需要实时信息的问题，例如 “What is Elon Musk's most recent X post?”，检验其内置搜索能力。

六、RAG 与知识管理：LlamaIndex 三件套

llamaindex-rag/目录用三种不同方式把 LlamaIndex 的检索能力接入语音 Agent，三者共享同一套“索引构建与持久化”代码：若*-storage目录不存在，则用SimpleDirectoryReader读取data/下的文档、构建VectorStoreIndex并persist；否则直接从磁盘load_index_from_storage加载，避免重复建索引。

6.1 方式一：chat_engine.py——用聊天引擎接管 LLM 节点

chat_engine.py 的思路是：用 LlamaIndex 的 ChatEngine 完全替代框架默认 LLM 节点。

class DummyLLM(llm.LLM): async def chat(self, *args, **kwargs): raise NotImplementedError("DummyLLM does not support chat")

• Agent 的llm传入了DummyLLM——一个故意抛异常的占位实现，注释说明“用 dummy LLM 以启用 pipeline reply”，即只占住管线位置，真正推理交给index.as_chat_engine(chat_mode=ChatMode.CONTEXT, llm="default")。
• 重写llm_node方法：从chat_ctx取出本轮用户消息，把chat_ctx中的历史消息映射为 LlamaIndex 的ChatMessage，然后await self.chat_engine.astream_chat(user_query, chat_history=...)并async for逐 token 产出回复。这样 RAG 检索与对话历史完全由 LlamaIndex 管理。

6.2 方式二：query_engine.py——查询引擎作为函数工具

query_engine.py 是最轻量的方案：LLM 仍是框架的inference.LLM("openai/gpt-4.1-mini")，只是把一个query_info函数工具注入tools=[query_info]：

@llm.function_tool async def query_info(query: str) -> str: """Get more information about a specific topic""" query_engine = index.as_query_engine(use_async=True) res = await query_engine.aquery(query) return str(res)

LLM 自主决定何时调用该工具，查询结果作为工具返回值再交给 LLM 组织成语音回答。这是“Agent 自主决定何时检索”的典型工具化 RAG 模式。

6.3 方式三：retrieval.py——检索注入系统提示词

retrieval.py 走“每轮检索、把上下文塞进 system 消息”的经典 RAG 路线，且 STT/LLM/TTS 全部换成直接 Plugin（deepgram.STT()、openai.LLM()、openai.TTS()），展示不依赖inference的写法：

retriever = self.index.as_retriever() nodes = await retriever.aretrieve(user_query) instructions = "Context that might help answer the user's question:" for node in nodes: node_content = node.get_content(metadata_mode=MetadataMode.LLM) instructions += f"\n\n{node_content}" system_msg = chat_ctx.items[0] if isinstance(system_msg, llm.ChatMessage) and system_msg.role == "system": system_msg.content.append(instructions) else: chat_ctx.items.insert(0, llm.ChatMessage(role="system", content=[instructions])) return Agent.default.llm_node(self, chat_ctx, tools, model_settings)

在llm_node中：先用aretrieve检索 Top-K 相关节点（MetadataMode.LLM保留 LLM 友好的元数据），再把拼接好的上下文追加到chat_ctx的 system 消息里，最后委托Agent.default.llm_node走框架默认 LLM 推理。代码注释也提示了“也可以改用update_instructions注入指令”的备选路径。

三种方式的取舍：chat_engine.py检索与对话能力最完整但定制最深；query_engine.py接入成本最低；retrieval.py对检索流程的控制最直观。三个示例都通过ctx.connect(auto_subscribe=AutoSubscribe.AUDIO_ONLY)只订阅音频，并在进场后用session.say("Hey, how can I help you today?", allow_interruptions=False)播报欢迎语。

七、可观测性与容错：otel_trace.py

otel_trace.py 演示了两件事：OpenTelemetry 链路追踪与多模型 Fallback 适配器。

7.1 OTLP 追踪接入

def setup_otel(metadata=None, *, endpoint=None, headers=None) -> TracerProvider: trace_provider = TracerProvider() trace_provider.add_span_processor( BatchSpanProcessor(OTLPSpanExporter(endpoint=endpoint, headers=headers)) ) set_tracer_provider(trace_provider, metadata=metadata) return trace_provider

• 通过OTLPSpanExporter以 HTTP 导出 span，兼容任何 OTLP 后端（Langfuse、Jaeger、Grafana Tempo、Honeycomb 等）。
• endpoint/headers不传时回退到标准 OTLP 环境变量：OTEL_EXPORTER_OTLP_ENDPOINT、OTEL_EXPORTER_OTLP_HEADERS。
• 代码注释给出了 Langfuse 的具体接法：endpoint 为<LANGFUSE_HOST>/api/public/otel，认证用公钥/私钥做 Base64 的Authorization: Basic头。
• 在entrypoint中调用setup_otel(metadata={"session.id": ctx.room.name})，所有 span 都会带上session.id属性（Langfuse 等后端用它做会话分组），并通过ctx.add_shutdown_callback(flush_trace)在进程退出前force_flush()落盘所有 span。

7.2 模型 Fallback 适配器

该示例同时展示了容错设计——每个模型位置都配置了一主一备：

llm=FallbackLLMAdapter(llm=[ inference.LLM("openai/gpt-4.1-mini"), inference.LLM("google/gemini-2.5-flash"), ]), stt=FallbackSTTAdapter(stt=[ inference.STT("deepgram/nova-3"), inference.STT("cartesia/ink-whisper"), ]), tts=FallbackTTSAdapter(tts=[ inference.TTS("cartesia"), inference.TTS("rime/coda", voice="astra"), ]),

主模型故障时自动切换备选模型，显著提升生产可用性。

7.3 Agent 间转接（Transfer）

示例还包含两个 Agent（Kelly与Alloy）互相转接的演示：Kelly使用 Fallback 推理管线，Alloy使用openai.realtime.RealtimeModel(voice="alloy")实时语音模型；各自定义transfer_to_alloy/transfer_to_kelly函数工具，返回另一个Agent实例即可完成会话转接——这与仓库中examples/warm-transfer的暖转接思路一脉相承。

八、实践要点小结

1. 能走inference就走inference：统一 API 一行切换厂商与模型；只有 voice-to-voice 实时模型（xAI Grok、Amazon Nova Sonic 等）必须用 Plugin 直连。
2. 语音场景的提示词纪律：要求 Agent 不用 emoji / markdown / 特殊符号，回复保持简短；配合filter_emoji、filter_markdown与发音替换变换，可显著提升 TTS 播报质量。
3. 延迟与体验旋钮：preemptive_generation抢跑生成降低首包延迟；interruption.resume_false_interruption缓解背景噪声导致的误打断；aec_warmup_duration保护回声消除校准窗口。
4. 长任务工具要“可取消 + 报进度”：MCP 场景下调大client_session_timeout_seconds，对长任务开启ToolFlag.CANCELLABLE与report_progress，并让 LLM 自然叙述进度。
5. 生产级标配：metrics_collected采集指标、session.usage核算成本、OpenTelemetry 追踪会话、Fallback 适配器兜底模型故障、add_shutdown_callback优雅收尾。

上述示例可直接在仓库的 examples/voice_agents 目录下按pyproject.toml声明的依赖运行，作为你自己实时语音 Agent 项目的起点模板。

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