去年在做一个语音交互 Demo 时卡住我的并不是大模型本身而是如何把 ASR 语音识别、LLM 推理、TTS 语音合成这三段能力稳定地串成一条链路。网上资料大多是单点讲解讲 ASR 的只讲 ASR讲 Agent 的只讲工具调用很少有人把“语音进 → 文本理解 → 语音出”的完整闭环讲清楚。这篇文章是我把整个链路跑通后的工程记录基于一套可运行的 VoiceAgent 项目拆解级联式三明治架构的原理、代码实现和排错思路。无论你是刚入门语音智能体的新手还是想把现有 Agent 项目升级成语音交互的老手都能直接复用这套方案。1. 背景与核心概念1.1 什么是 VoiceAgentVoiceAgent 是一个以语音为主要交互入口的智能体系统。用户说话系统识别语音大模型理解意图并生成回复系统再把回复转成语音播放出来。整个过程模拟了人与人之间的自然对话也是当前智能语音助手、智能客服、AI 陪伴、智能硬件交互等场景的核心技术底座。与纯文本 Agent 相比VoiceAgent 多出了音频采集、端点检测、语音识别和语音合成这几个环节。这不仅仅是模块变多了还带来一系列实际问题录音什么时候开始、什么时候结束识别结果有错别字怎么处理用户说话很快系统能不能流式响应TTS 合成的声音是否自然整个链路的延迟能不能控制在可接受范围内。这些问题的答案都体现在系统架构设计上。1.2 什么是级联式三明治架构级联式三明治架构是本文 VoiceAgent 项目的核心设计思路。它的结构可以概括为“三层面包 内部级联”。把整个 VoiceAgent 的完整调用链想象成一个三明治上层面包上行感知层。负责把麦克风采集到的原始音频转成文本内部按照 VAD语音活动检测→ ASR语音识别的顺序级联执行。中间夹心认知决策层。负责接收文本结合对话历史、系统提示词和外部工具生成最合适的回复文本。这里是大模型和智能体的核心区域。下层面包下行表达层。负责把回复文本转成语音并播放内部按照 TTS语音合成→ 音频播放的顺序级联执行。三个层次之间只通过标准接口传递数据上行层输出文本字符串中间层输出文本字符串下行层接收文本字符串。每一层内部又可以继续拆成多个细粒度组件这就是“级联”的含义。层与层之间是松耦合的任何一层都可以独立替换不影响其他层。1.3 为什么需要这种架构最初尝试把所有功能写到一个大循环里结果就是改一个识别参数要动全局代码排查问题也不知道从哪下手。级联式三明治架构能解决几个核心痛点。第一关注点分离。每一层只关心自己的职责。VAD 不需要知道大模型用了什么提示词TTS 也不需要知道工具调用是怎么实现的。第二组件可替换。今天用 faster-whisper 做 ASR明天想换更轻量的模型只需要保证新模型实现了相同的transcribe()接口即可完全不影响上层逻辑。第三延迟可控。语音交互对时延很敏感。三层架构可以把耗时分散在各个阶段的处理中例如 VAD 在用户还没说完话时就已经开始检测端点ASR 可以做到边说边识别TTS 可以在大模型生成完整回复后再合成也可以设计成流式合成。第四便于排错。如果用户反馈“识别不准”问题大概率在上行感知层如果回答内容不对问题大概率在认知决策层如果声音卡顿、破音问题大概率在下行表达层。分层后排查范围迅速缩小。2. 环境准备与项目结构2.1 开发环境说明本文示例项目使用 Python 3.10 及以上版本在 Windows、macOS、Linux 上均可运行。语音识别部分使用 faster-whisper它基于 CTranslate2 推理框架CPU 上也能跑不需要独立安装 PyTorch 全家桶TTS 部分使用 edge-tts微软 Edge 的在线语音合成服务语音活动检测使用 webrtcvad这是 WebRTC 的 VAD 模块封装的 Python 库。大模型部分通过 OpenAI 兼容的 API 协议调用支持接入云端大模型服务也支持接入本地部署的 vLLM 或 Ollama 服务。版本需要根据你的项目实际情况调整本文示例以常见环境为例重点演示配置思路。2.2 项目目录结构先来看完整的项目目录结构voiceagent/ ├── config/ │ ├── settings.yaml │ └── __init__.py ├── core/ │ ├── audio/ │ │ ├── __init__.py │ │ └── vad.py │ ├── asr/ │ │ ├── __init__.py │ │ └── whisper_asr.py │ ├── llm/ │ │ ├── __init__.py │ │ └── agent.py │ ├── tts/ │ │ ├── __init__.py │ │ └── edge_tts_engine.py │ └── pipeline/ │ ├── __init__.py │ └── sandwich_pipeline.py ├── main.py └── requirements.txt每个模块的职责很清晰config/settings.yaml项目配置文件集中管理音频参数、ASR 参数、大模型参数和 TTS 参数。core/audio/vad.py音频采集与语音活动检测负责判断用户开始说话和结束说话。core/asr/whisper_asr.py语音识别把 PCM 音频转成文本。core/llm/agent.py大模型智能体支持多轮对话和工具调用。core/tts/edge_tts_engine.py语音合成把文本转为 MP3 音频。core/pipeline/sandwich_pipeline.py级联式流水线把上述模块按三明治架构组装起来。main.py程序入口。2.3 创建虚拟环境与安装依赖推荐先创建虚拟环境避免污染系统 Python 环境。mkdir voiceagent cd voiceagent python -m venv venv # Windows venv\Scripts\activate # macOS / Linux source venv/bin/activate创建requirements.txt内容如下faster-whisper1.0.0 edge-tts6.1.0 webrtcvad2.0.10 openai1.30.0 PyYAML6.0 sounddevice0.4.6 numpy1.24.0 pygame2.5.0安装依赖pip install -r requirements.txt这里简单解释几个依赖的用途。faster-whisper 负责 ASR它比原始 Whisper 在 CPU 上的推理速度快很多edge-tts 是纯 Python 的在线 TTS 库不需要本地模型权重sounddevice 负责读取麦克风音频pygame 负责播放 TTS 生成的 MP3 文件。3. 级联式三明治架构分层拆解3.1 上行感知层音频采集与 VAD上行感知层解决的核心问题是“从麦克风流中找出用户说话的有效片段”。麦克风源源不断地产生音频数据里面既有语音也有静音和环境噪声。如果把这堆数据全部送给大模型不仅浪费算力还会让识别结果变得混乱。VAD 的作用就是在音频流中标记哪些片段包含人声哪些是静音。VAD 的工程实现有两个关键参数帧长度和静音阈值。WebRTC VAD 要求每次检测的音频帧长度必须是 10ms、20ms 或 30ms 的 16kHz 16bit 单声道 PCM 数据。在本文项目中我们把帧长度设置为 30ms采样率 16kHz因此每帧包含 480 个采样点对应的字节数是 960 字节。静音阈值表示连续多少帧没有检测到人声就认为一句话已经结束了这个值直接决定了用户体验阈值太短容易把句子从中间切断阈值太长又会让交互变得拖沓。本文项目使用 1.2 秒的连续静音作为结束条件。也就是说用户说完一句话后停顿超过 1.2 秒系统就认为这句话说完了开始进入识别流程。这是经过多次测试比较合适的值既不会频繁切断长句也不会让用户在停顿后等待太久。3.2 上行感知层ASR 语音识别VAD 完成后我们得到一段干净的 16kHz 单声道 PCM 音频。接下来要交给 ASR 模块转成文本。本文使用 faster-whisper 作为识别引擎。faster-whisper 支持多种模型大小从 tiny、base、small 到 medium、large。模型越大识别准确率越高但推理速度越慢内存占用也越大。在实际项目中如果运行在 CPU 上建议从small模型起步如果机器性能较强可以尝试medium如果对延迟非常敏感可以使用base甚至tiny。faster-whisper 的另一个重要参数是compute_type。CPU 推理时使用int8量化可以显著提升速度同时识别准确率损失很小。如果使用 GPU 推理float16通常是更好的选择。这里需要注意faster-whisper 要求输入音频的采样率是 16kHzVAD 阶段输出的音频正好满足这个条件所以在三明治架构中上行感知层内部的两个级联组件之间数据类型是天然匹配的。3.3 认知决策层大模型驱动的智能体认知决策层是整个三明治的“夹心”也是 VoiceAgent 的“大脑”。这一层接收上行层传输过来的用户文本结合对话历史、系统提示词和外部工具最终生成回复文本。在本文项目中认知决策层被设计成一个支持工具调用的智能体。所谓工具调用是指大模型在回答时如果发现自己需要实时信息或执行特定操作可以生成一个结构化的工具调用请求。例如用户问“现在几点了”模型可以调用get_current_time工具用户问“北京天气怎么样”模型可以调用天气查询工具。工具执行的结果会再次返回给模型模型基于真实结果组织最终回复。这里要理解一个关键点大模型本身不直接执行工具它只是“决定”需要调用什么工具、传什么参数。真正执行工具的代码是外部函数执行结果再拼接到对话上下文里。这种设计让智能体的能力边界可以无限扩展接入订单查询、日程管理、设备控制、数据库查询等任何外部系统只要以工具的形式注册给智能体即可。3.4 下行表达层TTS 语音合成认知决策层输出的是文本字符串。下行表达层负责把这串文本变成声音。本文使用 edge-tts 实现语音合成。edge-tts 是 Python 语言实现的微软 Edge 在线语音合成客户端支持多种中文发音人例如zh-CN-XiaoxiaoNeural是女声zh-CN-YunxiNeural是男声。开发者可以根据产品调性选择不同的音色。edge-tts 的使用方式非常简单创建一个Communicate对象调用save()方法就可以保存音频文件。但需要注意它是异步接口所以外层需要用asyncio.run()来驱动。下行表达层内部也可以做级联。本文项目在 TTS 合成完 MP3 文件后使用 pygame 播放音频。在实际产品中下行层还可以进一步拆分成文本润色、韵律预测、流式音频块合成等多个子模块形成更深层的级联结构。4. 完整实战从 0 到 1 搭建 VoiceAgent4.1 编写配置文件在config/settings.yaml中集中管理所有参数audio: sample_rate: 16000 frame_ms: 30 silence_seconds: 1.2 max_seconds: 15 asr: model_size: small device: cpu compute_type: int8 language: zh llm: api_key: ${LLM_API_KEY} base_url: https://api.openai.com/v1 model: gpt-4o-mini system_prompt: | 你是一个友好的语音助手请用简洁自然的中文回答用户的问题。 回答尽量控制在100字以内因为这是语音场景。 tts: voice: zh-CN-XiaoxiaoNeural output_path: output.mp3创建config/__init__.py并将配置加载封装成一个函数# 文件路径config/config.py import os import yaml def load_config(path: str config/settings.yaml) - dict: with open(path, r, encodingutf-8) as f: config yaml.safe_load(f) # 支持从环境变量读取大模型 API Key避免硬编码在配置文件里 api_key os.getenv(LLM_API_KEY) if api_key: config[llm][api_key] api_key return config配置文件的收益在项目后期非常明显。VAD 的静音阈值、ASR 的模型大小、TTS 的声音角色这些参数都不需要修改代码只需要改配置后重启程序即可。4.2 实现 VAD 与录音模块创建core/audio/vad.py实现基于 WebRTC VAD 的录音模块# 文件路径core/audio/vad.py import threading import numpy as np import sounddevice as sd import webrtcvad class VoiceActivityDetector: def __init__( self, sample_rate: int 16000, frame_ms: int 30, aggressiveness: int 2, ): self.sample_rate sample_rate self.frame_ms frame_ms self.frame_size int(sample_rate * frame_ms / 1000) self.vad webrtcvad.Vad(aggressiveness) def record_utterance( self, silence_seconds: float 1.2, max_seconds: float 15.0, ) - np.ndarray | None: 录制一句话 1. 等待用户开始说话 2. 检测到连续静音后停止 3. 返回 int16 格式的 PCM 音频数组 frames [] started False speech_count 0 silence_count 0 silence_threshold int(silence_seconds * 1000 / self.frame_ms) max_frames int(max_seconds * 1000 / self.frame_ms) done_event threading.Event() def callback(indata, frames_count, time_info, status): nonlocal started, speech_count, silence_count pcm indata.flatten() is_speech self.vad.is_speech(pcm.tobytes(), self.sample_rate) if not started: if is_speech: speech_count 1 # 连续 3 帧检测到语音才认为用户真正开始说话 if speech_count 3: started True silence_count 0 else: speech_count 0 else: frames.append(pcm.copy()) if is_speech: silence_count 0 else: silence_count 1 if silence_count silence_threshold: done_event.set() if len(frames) max_frames: done_event.set() with sd.InputStream( samplerateself.sample_rate, channels1, dtypeint16, blocksizeself.frame_size, callbackcallback, ): done_event.wait(timeoutmax_seconds 2) if not frames: return None return np.concatenate(frames)代码逻辑分为两个阶段。第一个阶段是等待语音开始。为了避免环境噪声误触发程序要求连续 3 帧都检测到语音才认定用户开始说话。第二个阶段是录音状态。程序持续把音频帧加入列表同时统计连续静音帧数。一旦静音帧数超过阈值说明用户说完了通过done_event通知主线程结束录音。webrtcvad.Vad(aggressiveness)参数表示 VAD 的激进程度取值范围 0 到 3。数值越小对语音的判断越宽松可能把噪声也当成语音数值越大过滤噪声越严格但可能把轻声细语当成静音。本文使用 2适合大多数室内环境。4.3 实现 ASR 模块创建core/asr/whisper_asr.py# 文件路径core/asr/whisper_asr.py import numpy as np from faster_whisper import WhisperModel class WhisperASR: def __init__( self, model_size: str small, device: str cpu, compute_type: str int8, language: str zh, ): self.model WhisperModel(model_size, devicedevice, compute_typecompute_type) self.language language def transcribe(self, audio_data: np.ndarray, sample_rate: int 16000) - str: 将 int16 PCM 音频转成文本。 audio_data 是 VAD 模块返回的 int16 数组。 audio_float audio_data.astype(np.float32) / 32768.0 segments, info self.model.transcribe( audio_float, languageself.language, beam_size1, vad_filterFalse, ) text .join(segment.text for segment in segments).strip() return text说明一个细节faster-whisper 的输入需要 float32 类型的音频数据而 VAD 模块返回的是 int16所以这里先将 int16 归一化到 [-1, 1] 区间的 float32。beam_size1表示使用贪心解码不启用束搜索速度更快适合语音交互场景。vad_filterFalse表示不再做二次 VAD 过滤因为上行层已经完成了端点检测这里是刻意避免重复处理。transcribe()方法返回的是纯文本字符串。上层流水线不需要关心 ASR 内部用的是哪个模型、什么量化方式只要拿到字符串就够了。这就是三明治架构中“层间只传文本”的体现。4.4 实现大模型智能体模块创建core/llm/agent.py实现支持工具调用的智能体# 文件路径core/llm/agent.py import json from openai import OpenAI class LLMAgent: def __init__( self, api_key: str, base_url: str, model: str, system_prompt: str , ): self.client OpenAI(api_keyapi_key, base_urlbase_url) self.model model self.system_prompt system_prompt self.history [] self.tools [] self.tool_functions {} def register_tool(self, name, description, parameters, func): 注册一个工具给大模型调用 self.tools.append( { type: function, function: { name: name, description: description, parameters: parameters, }, } ) self.tool_functions[name] func def chat(self, user_text: str) - str: messages [] if self.system_prompt: messages.append({role: system, content: self.system_prompt}) messages.extend(self.history) messages.append({role: user, content: user_text}) response self.client.chat.completions.create( modelself.model, messagesmessages, toolsself.tools if self.tools else None, tool_choiceauto if self.tools else None, ) message response.choices[0].message if message.tool_calls: # 执行工具调用 messages.append(message) for tool_call in message.tool_calls: function_name tool_call.function.name arguments json.loads(tool_call.function.arguments) if function_name in self.tool_functions: result self.tool_functions[function_name](**arguments) else: result f未知工具: {function_name} messages.append( { role: tool, tool_call_id: tool_call.id, content: str(result), } ) # 将工具结果返回给模型生成最终回复 second_response self.client.chat.completions.create( modelself.model, messagesmessages, ) answer second_response.choices[0].message.content else: answer message.content self.history.append({role: user, content: user_text}) self.history.append({role: assistant, content: answer}) return answer这个智能体的核心逻辑可以拆成四个步骤。第一步组装消息序列把系统提示词、历史对话、当前用户输入拼接起来。第二步调用大模型接口把已经注册的工具列表传给模型。第三步判断返回结果中是否包含工具调用请求。如果包含就执行相应的 Python 函数把执行结果封装成 tool 消息再次调用模型。第四步把最终的回答追加到历史记录中实现多轮对话。需要特别注意的是大模型 API 对工具调用的消息结构有严格要求。当出现 tool 消息时前面必须有一条 tool_calls 的 assistant 消息并且tool_call_id要对应上。本文代码已经正确处理了这一点。4.5 实现 TTS 模块创建core/tts/edge_tts_engine.py# 文件路径core/tts/edge_tts_engine.py import asyncio import edge_tts class EdgeTTSEngine: def __init__(self, voice: str zh-CN-XiaoxiaoNeural): self.voice voice async def _save_audio(self, text: str, output_path: str): communicate edge_tts.Communicate(text, self.voice) await communicate.save(output_path) def synthesize(self, text: str, output_path: str output.mp3) - str: 将文本转为 MP3 文件并返回文件路径 asyncio.run(self._save_audio(text, output_path)) return output_path再创建一个简单的播放工具放在core/tts/player.py# 文件路径core/tts/player.py import pygame def play_mp3(path: str): pygame.mixer.music.load(path) pygame.mixer.music.play() while pygame.mixer.music.get_busy(): pygame.time.wait(100)pygame 的mixer模块支持 MP3 解码。播放时通过循环检查播放状态来阻塞主线程确保音频播放完毕后再进入下一轮录音。如果不想引入 pygame也可以替换成调用系统播放器的方案例如 Windows 下执行start output.mp3macOS 下执行afplay output.mp3。4.6 组合级联式流水线核心的流水线文件放在core/pipeline/sandwich_pipeline.py# 文件路径core/pipeline/sandwich_pipeline.py import numpy as np from core.audio.vad import VoiceActivityDetector from core.asr.whisper_asr import WhisperASR from core.llm.agent import LLMAgent from core.tts.edge_tts_engine import EdgeTTSEngine from core.tts.player import play_mp3 class SandwichPipeline: def __init__(self, config: dict): audio_cfg config[audio] asr_cfg config[asr] llm_cfg config[llm] tts_cfg config[tts] self.vad VoiceActivityDetector( sample_rateaudio_cfg[sample_rate], frame_msaudio_cfg[frame_ms], ) self.asr WhisperASR( model_sizeasr_cfg[model_size], deviceasr_cfg[device], compute_typeasr_cfg[compute_type], languageasr_cfg[language], ) self.agent LLMAgent( api_keyllm_cfg[api_key], base_urlllm_cfg[base_url], modelllm_cfg[model], system_promptllm_cfg[system_prompt], ) self.tts EdgeTTSEngine(voicetts_cfg[voice]) # 注册内置工具 self._register_default_tools() def _register_default_tools(self): from datetime import datetime def get_current_time(): return datetime.now().strftime(%Y-%m-%d %H:%M:%S) def get_date(): return datetime.now().strftime(%Y-%m-%d) self.agent.register_tool( nameget_current_time, description获取当前精确时间格式为 年-月-日 时:分:秒, parameters{ type: object, properties: {}, }, funcget_current_time, ) self.agent.register_tool( nameget_date, description获取当前日期格式为 年-月-日, parameters{ type: object, properties: {}, }, funcget_date, ) def run_once(self) - tuple[str, str] | None: 执行一轮完整的 VoiceAgent 交互 录音 - 识别 - 大模型 - 合成 - 播放 # 上行感知层 audio self.vad.record_utterance( silence_seconds1.2, max_seconds15, ) if audio is None: return None # 上行感知层继续级联ASR user_text self.asr.transcribe(audio) # 认知决策层 reply_text self.agent.chat(user_text) # 下行表达层 audio_path self.tts.synthesize(reply_text) play_mp3(audio_path) return user_text, reply_textrun_once()方法把三层调用串联起来代码阅读顺序就是数据流动方向。VAD 录音返回 numpy 数组ASR 把它变成文本Agent 结合上下文和工具生成回复TTS 把回复变成声音。每两个模块之间传递的数据类型都是单一且明确的因此即使某个模块内部实现出现 bug也只需要替换对应模块。这里顺便注册了两个最简单也最常用的工具获取当前时间和获取当前日期。大模型在回答“现在几点了”“今天几号”这类问题时会主动调用这两个工具而不是基于训练数据猜测。4.7 编写入口文件并运行创建main.py# 文件路径main.py import pygame from config.config import load_config from core.pipeline.sandwich_pipeline import SandwichPipeline def main(): config load_config() pygame.mixer.init() pipeline SandwichPipeline(config) print(VoiceAgent 已启动请开始说话...) print(提示如果只想输入文本可以直接回复文本内容。) try: while True: result pipeline.run_once() if result is None: continue user_text, reply_text result print(f[用户]: {user_text}) print(f[Agent]: {reply_text}) if user_text and 退出 in user_text: print(再见) break except KeyboardInterrupt: print(\n程序已被手动终止) finally: pygame.mixer.quit() if __name__ __main__: main()运行程序前需要先设置大模型 API Key。建议使用环境变量而不是把 Key 写死在配置文件里。export LLM_API_KEYyour-api-key python main.pyWindows PowerShell 下使用$env:LLM_API_KEY your-api-key python main.py启动后程序会打开麦克风并进入监听状态。对着麦克风说“现在几点了”可以看到类似下面的输出VoiceAgent 已启动请开始说话... [用户]: 现在几点了 [Agent]: 现在是 2026-01-15 14:32:08。如果说完“退出”程序播放完语音后会自动结束。5. 常见问题与排查思路VoiceAgent 涉及音频设备、模型推理、网络请求等多个环节任何一个环节出错都会导致整个链路失败。以下是实际开发中最高频的几个问题。问题现象常见原因解决思路录音一直没有结束VAD 灵敏度太高或静音阈值太长降低 aggressiveness 参数缩短 silence_seconds录音刚开口就被切断静音阈值太短调大 silence_seconds例如从 1.2 改为 1.5ASR 识别结果为空输入音频采样率不是 16kHz确认 sounddevice 的 samplerate 参数为 16000ASR 识别不准模型太小或环境噪声大换 medium 模型重新测试 aggressiveness大模型返回超时base_url 或 api_key 配置错误确认接口地址和 Key 是否有效检查网络连接工具调用后回复异常工具参数类型与函数不匹配检查 parameters 中的 properties 类型使用 json.dumps 调试edge-tts 合成失败网络不可用或音色名称错误确认能访问服务换一个合法的 voice 参数pygame 播放没有声音音频输出设备未初始化调用 pygame.mixer.init() 并检查系统音量再补充一个高频问题程序启动后sounddevice.PortAudioError: Error opening InputStream。这通常表示麦克风被其他程序占用或者系统默认录音设备不可用。可以在系统设置中检查麦克风权限并关闭其他占用麦克风的软件。如果使用虚拟机运行还需要确保宿主机把麦克风设备透传给了虚机。ASR 加载模型时如果内存不足程序会直接退出。faster-whisper 在 CPU int8 下加载 small 模型大约需要 500MB 内存medium 模型约 1.5GB。如果内存紧张可以把模型降级为 base。6. 最佳实践与工程建议6.1 分层解耦与接口设计三明治架构的核心价值在于分层。工程落地时建议为每一层的核心模块先定义抽象接口再写具体实现。例如给 ASR 定义一个ASRBase抽象类其中包含transcribe()方法给 TTS 定义一个TTSBase抽象类包含synthesize()方法。这样后续从 faster-whisper 切换到其他 ASR 引擎时只需要新增一个实现类原有代码全部保留。层与层之间传递的数据结构也要保持稳定。本文的简化版本直接传字符串在更复杂的项目中可以定义统一的数据结构例如dataclass class ASRResult: text: str start_time: float end_time: float confidence: float这样下游的大模型层可以获取更多信息例如根据置信度决定是否需要再次确认。6.2 流式处理与延迟优化语音交互对延迟非常敏感。本文的简化版本是 VAD 录音完整结束之后才调用 ASR这在大模型回复较长时体验尚可但如果目标是实时对话建议做两个方向的优化。第一个方向是 ASR 流式化。faster-whisper 原生不支持流式识别但可以通过分段识别加结果融合的方式实现半流式效果VAD 每积累 1 秒语音就做一次部分识别把识别到的文字先展示给用户同时等待最终结果。这样用户还在说话时大模型就可以开始推理。第二个方向是 TTS 流式播放。当大模型生成回复文本时可以先把完整的回复切分成短句每生成一个短句就开始合成并播放而不是等所有文本生成完毕再统一合成。这种“边说边播”的体验会明显降低用户的等待感。6.3 安全与权限调用大模型 API 时API Key 是敏感信息绝对不要提交到 Git 仓库。使用环境变量或者密钥管理服务保存。工具调用功能虽然强大但也会带来安全风险。如果让大模型能够调用执行系统命令、修改文件、删除数据等危险操作建议增加人工确认环节并且按照最小权限原则控制每个工具的能力。例如查询天气的工具只需要只读权限不需要访问系统关键路径。麦克风权限同样需要关注。隐私合规要求录音前必须明确告知用户并且不要保存不必要的历史音频。本文项目在录音结束后没有持久化音频文件只在内存中完成处理这是更安全的设计。6.4 上下文管理与记忆智能体的多轮对话依赖历史记录。本文的LLMAgent直接将所有历史追加到请求中对话轮数一多token 消耗会快速上涨最终超过大模型上下文窗口上限。生产环境必须实现历史管理策略。常见的方案有两种滑动窗口截断只保留最近 N 轮对话摘要压缩把较早的对话先交给大模型生成一份摘要再把它作为系统提示词的一部分。如果对话中涉及到具体用户的长期偏好信息可以引入向量数据库做记忆检索而不是无脑塞进上下文。6.5 日志与可观测性级联式架构排查问题虽然方便但仍然需要日志来定位具体环节。建议在每一层处理完数据时输出一条结构化日志至少包含当前层的名称、耗时、数据摘要。例如[VAD] duration2.31s frames77 [ASR] text现在几点了 latency1.02s [AGENT] reply现在是... latency0.85s tool_calls[get_current_time] [TTS] savedoutput.mp3 latency0.94s这组日志可以直接用来统计每层的延迟占比找出整个链路中的性能瓶颈。如果某个环节耗时异常也可以立即定位到具体组件。6.6 模块替换建议级联式三明治架构的优势是每个模块可替换。在项目演进过程中你可能会遇到以下替换场景模块替换方案说明VADsilero-vad基于深度学习的 VAD对噪声鲁棒性更强ASRFunASR阿里的语音识别工具包中文场景准确率高支持流式LLM本地部署 Ollama数据不出内网隐私更好但硬件要求高TTSCosyVoice / GPT-SoVITS支持声音克隆生成效果更自然但部署成本高播放ffplay更轻量无额外 Python 依赖替换模块时只需要保证新实现遵循既有接口系统其他部分完全不用改动。有一次把 ASR 从 whisper 换成 FunASR只花了一下午时间就是因为接口预先做了抽象这就是三明治架构在工程上最直接的收益。7. 总结与学习路线通过这篇文章我从零搭建了一个完整的 VoiceAgent 项目核心思路就是级联式三明治架构上行感知层用 VAD ASR 把语音变成文本认知决策层用大模型智能体生成回复下行表达层用 TTS 播放器把文本变成语音。三层之间松耦合每一层都可以独立替换和升级。配套代码涵盖了音频采集、语音活动检测、语音识别、大模型工具调用、语音合成与播放的完整链路可以直接复制运行也可以在此基础上继续扩展更复杂的工具集。接下来可以往几个方向深入。如果想提升识别效果建议研究流式 ASR 和基于深度学习的 VAD比如 silero-vad如果想增强智能体能力可以接入搜索、日历、数据库等真实工具并完善工具权限控制如果关注产品体验可以研究 TTS 流式播放、声音克隆和情感合成。语音交互是一条链路很长的赛道把每一个环节的边界摸清楚你就能在大模型浪潮中做出真正可落地的产品。如果本文对你搭建自己的语音智能体有帮助可以收藏备用后续遇到具体报错也可以回来对照排查。接下来就打开终端把第一句“你好”跑起来吧。 VoiceAgent级联式三明治架构：从ASR到TTS的语音交互全链路实现
去年在做一个语音交互 Demo 时卡住我的并不是大模型本身而是如何把 ASR 语音识别、LLM 推理、TTS 语音合成这三段能力稳定地串成一条链路。网上资料大多是单点讲解讲 ASR 的只讲 ASR讲 Agent 的只讲工具调用很少有人把“语音进 → 文本理解 → 语音出”的完整闭环讲清楚。这篇文章是我把整个链路跑通后的工程记录基于一套可运行的 VoiceAgent 项目拆解级联式三明治架构的原理、代码实现和排错思路。无论你是刚入门语音智能体的新手还是想把现有 Agent 项目升级成语音交互的老手都能直接复用这套方案。1. 背景与核心概念1.1 什么是 VoiceAgentVoiceAgent 是一个以语音为主要交互入口的智能体系统。用户说话系统识别语音大模型理解意图并生成回复系统再把回复转成语音播放出来。整个过程模拟了人与人之间的自然对话也是当前智能语音助手、智能客服、AI 陪伴、智能硬件交互等场景的核心技术底座。与纯文本 Agent 相比VoiceAgent 多出了音频采集、端点检测、语音识别和语音合成这几个环节。这不仅仅是模块变多了还带来一系列实际问题录音什么时候开始、什么时候结束识别结果有错别字怎么处理用户说话很快系统能不能流式响应TTS 合成的声音是否自然整个链路的延迟能不能控制在可接受范围内。这些问题的答案都体现在系统架构设计上。1.2 什么是级联式三明治架构级联式三明治架构是本文 VoiceAgent 项目的核心设计思路。它的结构可以概括为“三层面包 内部级联”。把整个 VoiceAgent 的完整调用链想象成一个三明治上层面包上行感知层。负责把麦克风采集到的原始音频转成文本内部按照 VAD语音活动检测→ ASR语音识别的顺序级联执行。中间夹心认知决策层。负责接收文本结合对话历史、系统提示词和外部工具生成最合适的回复文本。这里是大模型和智能体的核心区域。下层面包下行表达层。负责把回复文本转成语音并播放内部按照 TTS语音合成→ 音频播放的顺序级联执行。三个层次之间只通过标准接口传递数据上行层输出文本字符串中间层输出文本字符串下行层接收文本字符串。每一层内部又可以继续拆成多个细粒度组件这就是“级联”的含义。层与层之间是松耦合的任何一层都可以独立替换不影响其他层。1.3 为什么需要这种架构最初尝试把所有功能写到一个大循环里结果就是改一个识别参数要动全局代码排查问题也不知道从哪下手。级联式三明治架构能解决几个核心痛点。第一关注点分离。每一层只关心自己的职责。VAD 不需要知道大模型用了什么提示词TTS 也不需要知道工具调用是怎么实现的。第二组件可替换。今天用 faster-whisper 做 ASR明天想换更轻量的模型只需要保证新模型实现了相同的transcribe()接口即可完全不影响上层逻辑。第三延迟可控。语音交互对时延很敏感。三层架构可以把耗时分散在各个阶段的处理中例如 VAD 在用户还没说完话时就已经开始检测端点ASR 可以做到边说边识别TTS 可以在大模型生成完整回复后再合成也可以设计成流式合成。第四便于排错。如果用户反馈“识别不准”问题大概率在上行感知层如果回答内容不对问题大概率在认知决策层如果声音卡顿、破音问题大概率在下行表达层。分层后排查范围迅速缩小。2. 环境准备与项目结构2.1 开发环境说明本文示例项目使用 Python 3.10 及以上版本在 Windows、macOS、Linux 上均可运行。语音识别部分使用 faster-whisper它基于 CTranslate2 推理框架CPU 上也能跑不需要独立安装 PyTorch 全家桶TTS 部分使用 edge-tts微软 Edge 的在线语音合成服务语音活动检测使用 webrtcvad这是 WebRTC 的 VAD 模块封装的 Python 库。大模型部分通过 OpenAI 兼容的 API 协议调用支持接入云端大模型服务也支持接入本地部署的 vLLM 或 Ollama 服务。版本需要根据你的项目实际情况调整本文示例以常见环境为例重点演示配置思路。2.2 项目目录结构先来看完整的项目目录结构voiceagent/ ├── config/ │ ├── settings.yaml │ └── __init__.py ├── core/ │ ├── audio/ │ │ ├── __init__.py │ │ └── vad.py │ ├── asr/ │ │ ├── __init__.py │ │ └── whisper_asr.py │ ├── llm/ │ │ ├── __init__.py │ │ └── agent.py │ ├── tts/ │ │ ├── __init__.py │ │ └── edge_tts_engine.py │ └── pipeline/ │ ├── __init__.py │ └── sandwich_pipeline.py ├── main.py └── requirements.txt每个模块的职责很清晰config/settings.yaml项目配置文件集中管理音频参数、ASR 参数、大模型参数和 TTS 参数。core/audio/vad.py音频采集与语音活动检测负责判断用户开始说话和结束说话。core/asr/whisper_asr.py语音识别把 PCM 音频转成文本。core/llm/agent.py大模型智能体支持多轮对话和工具调用。core/tts/edge_tts_engine.py语音合成把文本转为 MP3 音频。core/pipeline/sandwich_pipeline.py级联式流水线把上述模块按三明治架构组装起来。main.py程序入口。2.3 创建虚拟环境与安装依赖推荐先创建虚拟环境避免污染系统 Python 环境。mkdir voiceagent cd voiceagent python -m venv venv # Windows venv\Scripts\activate # macOS / Linux source venv/bin/activate创建requirements.txt内容如下faster-whisper1.0.0 edge-tts6.1.0 webrtcvad2.0.10 openai1.30.0 PyYAML6.0 sounddevice0.4.6 numpy1.24.0 pygame2.5.0安装依赖pip install -r requirements.txt这里简单解释几个依赖的用途。faster-whisper 负责 ASR它比原始 Whisper 在 CPU 上的推理速度快很多edge-tts 是纯 Python 的在线 TTS 库不需要本地模型权重sounddevice 负责读取麦克风音频pygame 负责播放 TTS 生成的 MP3 文件。3. 级联式三明治架构分层拆解3.1 上行感知层音频采集与 VAD上行感知层解决的核心问题是“从麦克风流中找出用户说话的有效片段”。麦克风源源不断地产生音频数据里面既有语音也有静音和环境噪声。如果把这堆数据全部送给大模型不仅浪费算力还会让识别结果变得混乱。VAD 的作用就是在音频流中标记哪些片段包含人声哪些是静音。VAD 的工程实现有两个关键参数帧长度和静音阈值。WebRTC VAD 要求每次检测的音频帧长度必须是 10ms、20ms 或 30ms 的 16kHz 16bit 单声道 PCM 数据。在本文项目中我们把帧长度设置为 30ms采样率 16kHz因此每帧包含 480 个采样点对应的字节数是 960 字节。静音阈值表示连续多少帧没有检测到人声就认为一句话已经结束了这个值直接决定了用户体验阈值太短容易把句子从中间切断阈值太长又会让交互变得拖沓。本文项目使用 1.2 秒的连续静音作为结束条件。也就是说用户说完一句话后停顿超过 1.2 秒系统就认为这句话说完了开始进入识别流程。这是经过多次测试比较合适的值既不会频繁切断长句也不会让用户在停顿后等待太久。3.2 上行感知层ASR 语音识别VAD 完成后我们得到一段干净的 16kHz 单声道 PCM 音频。接下来要交给 ASR 模块转成文本。本文使用 faster-whisper 作为识别引擎。faster-whisper 支持多种模型大小从 tiny、base、small 到 medium、large。模型越大识别准确率越高但推理速度越慢内存占用也越大。在实际项目中如果运行在 CPU 上建议从small模型起步如果机器性能较强可以尝试medium如果对延迟非常敏感可以使用base甚至tiny。faster-whisper 的另一个重要参数是compute_type。CPU 推理时使用int8量化可以显著提升速度同时识别准确率损失很小。如果使用 GPU 推理float16通常是更好的选择。这里需要注意faster-whisper 要求输入音频的采样率是 16kHzVAD 阶段输出的音频正好满足这个条件所以在三明治架构中上行感知层内部的两个级联组件之间数据类型是天然匹配的。3.3 认知决策层大模型驱动的智能体认知决策层是整个三明治的“夹心”也是 VoiceAgent 的“大脑”。这一层接收上行层传输过来的用户文本结合对话历史、系统提示词和外部工具最终生成回复文本。在本文项目中认知决策层被设计成一个支持工具调用的智能体。所谓工具调用是指大模型在回答时如果发现自己需要实时信息或执行特定操作可以生成一个结构化的工具调用请求。例如用户问“现在几点了”模型可以调用get_current_time工具用户问“北京天气怎么样”模型可以调用天气查询工具。工具执行的结果会再次返回给模型模型基于真实结果组织最终回复。这里要理解一个关键点大模型本身不直接执行工具它只是“决定”需要调用什么工具、传什么参数。真正执行工具的代码是外部函数执行结果再拼接到对话上下文里。这种设计让智能体的能力边界可以无限扩展接入订单查询、日程管理、设备控制、数据库查询等任何外部系统只要以工具的形式注册给智能体即可。3.4 下行表达层TTS 语音合成认知决策层输出的是文本字符串。下行表达层负责把这串文本变成声音。本文使用 edge-tts 实现语音合成。edge-tts 是 Python 语言实现的微软 Edge 在线语音合成客户端支持多种中文发音人例如zh-CN-XiaoxiaoNeural是女声zh-CN-YunxiNeural是男声。开发者可以根据产品调性选择不同的音色。edge-tts 的使用方式非常简单创建一个Communicate对象调用save()方法就可以保存音频文件。但需要注意它是异步接口所以外层需要用asyncio.run()来驱动。下行表达层内部也可以做级联。本文项目在 TTS 合成完 MP3 文件后使用 pygame 播放音频。在实际产品中下行层还可以进一步拆分成文本润色、韵律预测、流式音频块合成等多个子模块形成更深层的级联结构。4. 完整实战从 0 到 1 搭建 VoiceAgent4.1 编写配置文件在config/settings.yaml中集中管理所有参数audio: sample_rate: 16000 frame_ms: 30 silence_seconds: 1.2 max_seconds: 15 asr: model_size: small device: cpu compute_type: int8 language: zh llm: api_key: ${LLM_API_KEY} base_url: https://api.openai.com/v1 model: gpt-4o-mini system_prompt: | 你是一个友好的语音助手请用简洁自然的中文回答用户的问题。 回答尽量控制在100字以内因为这是语音场景。 tts: voice: zh-CN-XiaoxiaoNeural output_path: output.mp3创建config/__init__.py并将配置加载封装成一个函数# 文件路径config/config.py import os import yaml def load_config(path: str config/settings.yaml) - dict: with open(path, r, encodingutf-8) as f: config yaml.safe_load(f) # 支持从环境变量读取大模型 API Key避免硬编码在配置文件里 api_key os.getenv(LLM_API_KEY) if api_key: config[llm][api_key] api_key return config配置文件的收益在项目后期非常明显。VAD 的静音阈值、ASR 的模型大小、TTS 的声音角色这些参数都不需要修改代码只需要改配置后重启程序即可。4.2 实现 VAD 与录音模块创建core/audio/vad.py实现基于 WebRTC VAD 的录音模块# 文件路径core/audio/vad.py import threading import numpy as np import sounddevice as sd import webrtcvad class VoiceActivityDetector: def __init__( self, sample_rate: int 16000, frame_ms: int 30, aggressiveness: int 2, ): self.sample_rate sample_rate self.frame_ms frame_ms self.frame_size int(sample_rate * frame_ms / 1000) self.vad webrtcvad.Vad(aggressiveness) def record_utterance( self, silence_seconds: float 1.2, max_seconds: float 15.0, ) - np.ndarray | None: 录制一句话 1. 等待用户开始说话 2. 检测到连续静音后停止 3. 返回 int16 格式的 PCM 音频数组 frames [] started False speech_count 0 silence_count 0 silence_threshold int(silence_seconds * 1000 / self.frame_ms) max_frames int(max_seconds * 1000 / self.frame_ms) done_event threading.Event() def callback(indata, frames_count, time_info, status): nonlocal started, speech_count, silence_count pcm indata.flatten() is_speech self.vad.is_speech(pcm.tobytes(), self.sample_rate) if not started: if is_speech: speech_count 1 # 连续 3 帧检测到语音才认为用户真正开始说话 if speech_count 3: started True silence_count 0 else: speech_count 0 else: frames.append(pcm.copy()) if is_speech: silence_count 0 else: silence_count 1 if silence_count silence_threshold: done_event.set() if len(frames) max_frames: done_event.set() with sd.InputStream( samplerateself.sample_rate, channels1, dtypeint16, blocksizeself.frame_size, callbackcallback, ): done_event.wait(timeoutmax_seconds 2) if not frames: return None return np.concatenate(frames)代码逻辑分为两个阶段。第一个阶段是等待语音开始。为了避免环境噪声误触发程序要求连续 3 帧都检测到语音才认定用户开始说话。第二个阶段是录音状态。程序持续把音频帧加入列表同时统计连续静音帧数。一旦静音帧数超过阈值说明用户说完了通过done_event通知主线程结束录音。webrtcvad.Vad(aggressiveness)参数表示 VAD 的激进程度取值范围 0 到 3。数值越小对语音的判断越宽松可能把噪声也当成语音数值越大过滤噪声越严格但可能把轻声细语当成静音。本文使用 2适合大多数室内环境。4.3 实现 ASR 模块创建core/asr/whisper_asr.py# 文件路径core/asr/whisper_asr.py import numpy as np from faster_whisper import WhisperModel class WhisperASR: def __init__( self, model_size: str small, device: str cpu, compute_type: str int8, language: str zh, ): self.model WhisperModel(model_size, devicedevice, compute_typecompute_type) self.language language def transcribe(self, audio_data: np.ndarray, sample_rate: int 16000) - str: 将 int16 PCM 音频转成文本。 audio_data 是 VAD 模块返回的 int16 数组。 audio_float audio_data.astype(np.float32) / 32768.0 segments, info self.model.transcribe( audio_float, languageself.language, beam_size1, vad_filterFalse, ) text .join(segment.text for segment in segments).strip() return text说明一个细节faster-whisper 的输入需要 float32 类型的音频数据而 VAD 模块返回的是 int16所以这里先将 int16 归一化到 [-1, 1] 区间的 float32。beam_size1表示使用贪心解码不启用束搜索速度更快适合语音交互场景。vad_filterFalse表示不再做二次 VAD 过滤因为上行层已经完成了端点检测这里是刻意避免重复处理。transcribe()方法返回的是纯文本字符串。上层流水线不需要关心 ASR 内部用的是哪个模型、什么量化方式只要拿到字符串就够了。这就是三明治架构中“层间只传文本”的体现。4.4 实现大模型智能体模块创建core/llm/agent.py实现支持工具调用的智能体# 文件路径core/llm/agent.py import json from openai import OpenAI class LLMAgent: def __init__( self, api_key: str, base_url: str, model: str, system_prompt: str , ): self.client OpenAI(api_keyapi_key, base_urlbase_url) self.model model self.system_prompt system_prompt self.history [] self.tools [] self.tool_functions {} def register_tool(self, name, description, parameters, func): 注册一个工具给大模型调用 self.tools.append( { type: function, function: { name: name, description: description, parameters: parameters, }, } ) self.tool_functions[name] func def chat(self, user_text: str) - str: messages [] if self.system_prompt: messages.append({role: system, content: self.system_prompt}) messages.extend(self.history) messages.append({role: user, content: user_text}) response self.client.chat.completions.create( modelself.model, messagesmessages, toolsself.tools if self.tools else None, tool_choiceauto if self.tools else None, ) message response.choices[0].message if message.tool_calls: # 执行工具调用 messages.append(message) for tool_call in message.tool_calls: function_name tool_call.function.name arguments json.loads(tool_call.function.arguments) if function_name in self.tool_functions: result self.tool_functions[function_name](**arguments) else: result f未知工具: {function_name} messages.append( { role: tool, tool_call_id: tool_call.id, content: str(result), } ) # 将工具结果返回给模型生成最终回复 second_response self.client.chat.completions.create( modelself.model, messagesmessages, ) answer second_response.choices[0].message.content else: answer message.content self.history.append({role: user, content: user_text}) self.history.append({role: assistant, content: answer}) return answer这个智能体的核心逻辑可以拆成四个步骤。第一步组装消息序列把系统提示词、历史对话、当前用户输入拼接起来。第二步调用大模型接口把已经注册的工具列表传给模型。第三步判断返回结果中是否包含工具调用请求。如果包含就执行相应的 Python 函数把执行结果封装成 tool 消息再次调用模型。第四步把最终的回答追加到历史记录中实现多轮对话。需要特别注意的是大模型 API 对工具调用的消息结构有严格要求。当出现 tool 消息时前面必须有一条 tool_calls 的 assistant 消息并且tool_call_id要对应上。本文代码已经正确处理了这一点。4.5 实现 TTS 模块创建core/tts/edge_tts_engine.py# 文件路径core/tts/edge_tts_engine.py import asyncio import edge_tts class EdgeTTSEngine: def __init__(self, voice: str zh-CN-XiaoxiaoNeural): self.voice voice async def _save_audio(self, text: str, output_path: str): communicate edge_tts.Communicate(text, self.voice) await communicate.save(output_path) def synthesize(self, text: str, output_path: str output.mp3) - str: 将文本转为 MP3 文件并返回文件路径 asyncio.run(self._save_audio(text, output_path)) return output_path再创建一个简单的播放工具放在core/tts/player.py# 文件路径core/tts/player.py import pygame def play_mp3(path: str): pygame.mixer.music.load(path) pygame.mixer.music.play() while pygame.mixer.music.get_busy(): pygame.time.wait(100)pygame 的mixer模块支持 MP3 解码。播放时通过循环检查播放状态来阻塞主线程确保音频播放完毕后再进入下一轮录音。如果不想引入 pygame也可以替换成调用系统播放器的方案例如 Windows 下执行start output.mp3macOS 下执行afplay output.mp3。4.6 组合级联式流水线核心的流水线文件放在core/pipeline/sandwich_pipeline.py# 文件路径core/pipeline/sandwich_pipeline.py import numpy as np from core.audio.vad import VoiceActivityDetector from core.asr.whisper_asr import WhisperASR from core.llm.agent import LLMAgent from core.tts.edge_tts_engine import EdgeTTSEngine from core.tts.player import play_mp3 class SandwichPipeline: def __init__(self, config: dict): audio_cfg config[audio] asr_cfg config[asr] llm_cfg config[llm] tts_cfg config[tts] self.vad VoiceActivityDetector( sample_rateaudio_cfg[sample_rate], frame_msaudio_cfg[frame_ms], ) self.asr WhisperASR( model_sizeasr_cfg[model_size], deviceasr_cfg[device], compute_typeasr_cfg[compute_type], languageasr_cfg[language], ) self.agent LLMAgent( api_keyllm_cfg[api_key], base_urlllm_cfg[base_url], modelllm_cfg[model], system_promptllm_cfg[system_prompt], ) self.tts EdgeTTSEngine(voicetts_cfg[voice]) # 注册内置工具 self._register_default_tools() def _register_default_tools(self): from datetime import datetime def get_current_time(): return datetime.now().strftime(%Y-%m-%d %H:%M:%S) def get_date(): return datetime.now().strftime(%Y-%m-%d) self.agent.register_tool( nameget_current_time, description获取当前精确时间格式为 年-月-日 时:分:秒, parameters{ type: object, properties: {}, }, funcget_current_time, ) self.agent.register_tool( nameget_date, description获取当前日期格式为 年-月-日, parameters{ type: object, properties: {}, }, funcget_date, ) def run_once(self) - tuple[str, str] | None: 执行一轮完整的 VoiceAgent 交互 录音 - 识别 - 大模型 - 合成 - 播放 # 上行感知层 audio self.vad.record_utterance( silence_seconds1.2, max_seconds15, ) if audio is None: return None # 上行感知层继续级联ASR user_text self.asr.transcribe(audio) # 认知决策层 reply_text self.agent.chat(user_text) # 下行表达层 audio_path self.tts.synthesize(reply_text) play_mp3(audio_path) return user_text, reply_textrun_once()方法把三层调用串联起来代码阅读顺序就是数据流动方向。VAD 录音返回 numpy 数组ASR 把它变成文本Agent 结合上下文和工具生成回复TTS 把回复变成声音。每两个模块之间传递的数据类型都是单一且明确的因此即使某个模块内部实现出现 bug也只需要替换对应模块。这里顺便注册了两个最简单也最常用的工具获取当前时间和获取当前日期。大模型在回答“现在几点了”“今天几号”这类问题时会主动调用这两个工具而不是基于训练数据猜测。4.7 编写入口文件并运行创建main.py# 文件路径main.py import pygame from config.config import load_config from core.pipeline.sandwich_pipeline import SandwichPipeline def main(): config load_config() pygame.mixer.init() pipeline SandwichPipeline(config) print(VoiceAgent 已启动请开始说话...) print(提示如果只想输入文本可以直接回复文本内容。) try: while True: result pipeline.run_once() if result is None: continue user_text, reply_text result print(f[用户]: {user_text}) print(f[Agent]: {reply_text}) if user_text and 退出 in user_text: print(再见) break except KeyboardInterrupt: print(\n程序已被手动终止) finally: pygame.mixer.quit() if __name__ __main__: main()运行程序前需要先设置大模型 API Key。建议使用环境变量而不是把 Key 写死在配置文件里。export LLM_API_KEYyour-api-key python main.pyWindows PowerShell 下使用$env:LLM_API_KEY your-api-key python main.py启动后程序会打开麦克风并进入监听状态。对着麦克风说“现在几点了”可以看到类似下面的输出VoiceAgent 已启动请开始说话... [用户]: 现在几点了 [Agent]: 现在是 2026-01-15 14:32:08。如果说完“退出”程序播放完语音后会自动结束。5. 常见问题与排查思路VoiceAgent 涉及音频设备、模型推理、网络请求等多个环节任何一个环节出错都会导致整个链路失败。以下是实际开发中最高频的几个问题。问题现象常见原因解决思路录音一直没有结束VAD 灵敏度太高或静音阈值太长降低 aggressiveness 参数缩短 silence_seconds录音刚开口就被切断静音阈值太短调大 silence_seconds例如从 1.2 改为 1.5ASR 识别结果为空输入音频采样率不是 16kHz确认 sounddevice 的 samplerate 参数为 16000ASR 识别不准模型太小或环境噪声大换 medium 模型重新测试 aggressiveness大模型返回超时base_url 或 api_key 配置错误确认接口地址和 Key 是否有效检查网络连接工具调用后回复异常工具参数类型与函数不匹配检查 parameters 中的 properties 类型使用 json.dumps 调试edge-tts 合成失败网络不可用或音色名称错误确认能访问服务换一个合法的 voice 参数pygame 播放没有声音音频输出设备未初始化调用 pygame.mixer.init() 并检查系统音量再补充一个高频问题程序启动后sounddevice.PortAudioError: Error opening InputStream。这通常表示麦克风被其他程序占用或者系统默认录音设备不可用。可以在系统设置中检查麦克风权限并关闭其他占用麦克风的软件。如果使用虚拟机运行还需要确保宿主机把麦克风设备透传给了虚机。ASR 加载模型时如果内存不足程序会直接退出。faster-whisper 在 CPU int8 下加载 small 模型大约需要 500MB 内存medium 模型约 1.5GB。如果内存紧张可以把模型降级为 base。6. 最佳实践与工程建议6.1 分层解耦与接口设计三明治架构的核心价值在于分层。工程落地时建议为每一层的核心模块先定义抽象接口再写具体实现。例如给 ASR 定义一个ASRBase抽象类其中包含transcribe()方法给 TTS 定义一个TTSBase抽象类包含synthesize()方法。这样后续从 faster-whisper 切换到其他 ASR 引擎时只需要新增一个实现类原有代码全部保留。层与层之间传递的数据结构也要保持稳定。本文的简化版本直接传字符串在更复杂的项目中可以定义统一的数据结构例如dataclass class ASRResult: text: str start_time: float end_time: float confidence: float这样下游的大模型层可以获取更多信息例如根据置信度决定是否需要再次确认。6.2 流式处理与延迟优化语音交互对延迟非常敏感。本文的简化版本是 VAD 录音完整结束之后才调用 ASR这在大模型回复较长时体验尚可但如果目标是实时对话建议做两个方向的优化。第一个方向是 ASR 流式化。faster-whisper 原生不支持流式识别但可以通过分段识别加结果融合的方式实现半流式效果VAD 每积累 1 秒语音就做一次部分识别把识别到的文字先展示给用户同时等待最终结果。这样用户还在说话时大模型就可以开始推理。第二个方向是 TTS 流式播放。当大模型生成回复文本时可以先把完整的回复切分成短句每生成一个短句就开始合成并播放而不是等所有文本生成完毕再统一合成。这种“边说边播”的体验会明显降低用户的等待感。6.3 安全与权限调用大模型 API 时API Key 是敏感信息绝对不要提交到 Git 仓库。使用环境变量或者密钥管理服务保存。工具调用功能虽然强大但也会带来安全风险。如果让大模型能够调用执行系统命令、修改文件、删除数据等危险操作建议增加人工确认环节并且按照最小权限原则控制每个工具的能力。例如查询天气的工具只需要只读权限不需要访问系统关键路径。麦克风权限同样需要关注。隐私合规要求录音前必须明确告知用户并且不要保存不必要的历史音频。本文项目在录音结束后没有持久化音频文件只在内存中完成处理这是更安全的设计。6.4 上下文管理与记忆智能体的多轮对话依赖历史记录。本文的LLMAgent直接将所有历史追加到请求中对话轮数一多token 消耗会快速上涨最终超过大模型上下文窗口上限。生产环境必须实现历史管理策略。常见的方案有两种滑动窗口截断只保留最近 N 轮对话摘要压缩把较早的对话先交给大模型生成一份摘要再把它作为系统提示词的一部分。如果对话中涉及到具体用户的长期偏好信息可以引入向量数据库做记忆检索而不是无脑塞进上下文。6.5 日志与可观测性级联式架构排查问题虽然方便但仍然需要日志来定位具体环节。建议在每一层处理完数据时输出一条结构化日志至少包含当前层的名称、耗时、数据摘要。例如[VAD] duration2.31s frames77 [ASR] text现在几点了 latency1.02s [AGENT] reply现在是... latency0.85s tool_calls[get_current_time] [TTS] savedoutput.mp3 latency0.94s这组日志可以直接用来统计每层的延迟占比找出整个链路中的性能瓶颈。如果某个环节耗时异常也可以立即定位到具体组件。6.6 模块替换建议级联式三明治架构的优势是每个模块可替换。在项目演进过程中你可能会遇到以下替换场景模块替换方案说明VADsilero-vad基于深度学习的 VAD对噪声鲁棒性更强ASRFunASR阿里的语音识别工具包中文场景准确率高支持流式LLM本地部署 Ollama数据不出内网隐私更好但硬件要求高TTSCosyVoice / GPT-SoVITS支持声音克隆生成效果更自然但部署成本高播放ffplay更轻量无额外 Python 依赖替换模块时只需要保证新实现遵循既有接口系统其他部分完全不用改动。有一次把 ASR 从 whisper 换成 FunASR只花了一下午时间就是因为接口预先做了抽象这就是三明治架构在工程上最直接的收益。7. 总结与学习路线通过这篇文章我从零搭建了一个完整的 VoiceAgent 项目核心思路就是级联式三明治架构上行感知层用 VAD ASR 把语音变成文本认知决策层用大模型智能体生成回复下行表达层用 TTS 播放器把文本变成语音。三层之间松耦合每一层都可以独立替换和升级。配套代码涵盖了音频采集、语音活动检测、语音识别、大模型工具调用、语音合成与播放的完整链路可以直接复制运行也可以在此基础上继续扩展更复杂的工具集。接下来可以往几个方向深入。如果想提升识别效果建议研究流式 ASR 和基于深度学习的 VAD比如 silero-vad如果想增强智能体能力可以接入搜索、日历、数据库等真实工具并完善工具权限控制如果关注产品体验可以研究 TTS 流式播放、声音克隆和情感合成。语音交互是一条链路很长的赛道把每一个环节的边界摸清楚你就能在大模型浪潮中做出真正可落地的产品。如果本文对你搭建自己的语音智能体有帮助可以收藏备用后续遇到具体报错也可以回来对照排查。接下来就打开终端把第一句“你好”跑起来吧。