先给结论如果你正打算从零做一个能“听、想、说”的语音智能体应用但又不想被 ASR、LLM、TTS、流式协议、状态管理这些概念绕晕那么“级联式三明治架构”是目前最值得先吃透的一套落地范式。它不会让项目变成最酷的但大概率能让项目最快跑起来而且工程上可控、出错可排查、后续可扩展。这篇文章不聊虚的直接拆架构、给代码、走流程、讲排错目标只有一个让你今晚就能在自己电脑上跑通一个最小可用的 VoiceAgent。1. 这篇文章真正要解决的问题做语音智能体开发很多人第一反应是“接入一个大模型 API 不就行了”但实际动手之后会发现根本不是这么回事。语音交互链路比纯文本对话长得多麦克风采集到的音频要先变成文字文字进入大模型后要结合上下文和业务逻辑生成回复回复文字又要变成语音播放给用户。每一段都有独立的工具链、独立的延迟、独立的出错点。更麻烦的是当你把大模型、语音识别、语音合成串在一起时很容易写出一个“能跑但很难维护”的脚本所有逻辑堆在一个 Python 文件里回调套回调异常处理靠运气想加一个“打断当前播报”的功能都要重构半天。这就是为什么需要有一个清晰的架构而不是“把 API 串起来就行”。这篇文章要解决的核心问题有三个语音智能体的标准链路是什么每一层到底负责什么“级联式三明治架构”说的不是花哨概念而是输入层、大脑层、输出层三层分离的设计方法你该怎么用代码把它们拼起来并且能验证、能排错、能上生产。文中所有演示代码以 Python 为宿主语言语音识别和语音合成使用通用开源能力或云端 API大模型部分使用兼容 OpenAI 格式的服务。版本号不做死绑重点是跑通思路。2. 基础概念大模型、智能体、级联式三明治架构在进入代码之前先把文章里反复出现的三个词说清楚。理解到位了后面看代码会顺畅很多也避免在概念层面走偏。2.1 大模型不是“题库”而是“推理内核”大模型在这个项目里的定位不是知识库而是整个 VoiceAgent 的“意图决策中心”。它负责理解用户这次说话的真实诉求结合当前对话上下文决定应该回答什么、查询什么、调用什么工具。真正的大模型应用重点不是“它知道什么”而是“你如何引导它正确使用自己知道的东西”。在代码层面你只需要关注一点选一个接口兼容的大模型服务然后统一定义系统提示词和消息格式。后续想换模型只改配置不改业务逻辑。2.2 智能体不是“聊天机器人”而是“任务执行器”智能体的关键在于它能根据用户指令主动规划步骤、调用外部工具、读取状态最后把结果整理成对用户有用的话。语音场景下的智能体通常还要多处理一层如何把长文本结果转成适合“听”的表达例如去掉 Markdown 符号、让句子更口语化、避免一次性输出太长导致用户记不住。这就决定了你的 VoiceAgent 核心代码里需要有一个明确的“工具调用”设计。即使第一个版本只有一个查询时间、查询天气之类的工具也要把机制搭好。2.3 级联式三明治架构三层分离中间厚、两边薄先拆一下这个听起来有点唬人的名字。把整个语音智能体想象成一块三明治上层是“感知层”麦克风采集、语音识别、端点检测。作用是把用户的自然讲话变成文本中间是“大脑层”大模型对话管理、上下文记忆、工具调用、业务逻辑。这是整块三明治里最厚的一层也是核心价值所在下层是“表达层”回复文本的处理、语音合成、音频播放。所谓“级联”就是数据必须按顺序一层一层流下去音频进、文本出、文本进、决策出、文本进、语音出。每一层都是独立模块层与层之间只通过标准数据格式通信。这个架构的直观好处有三个。第一你可以单独替换任何一层今天用 A 厂商的语音识别明天换 B 厂商不会动到大模型代码。第二每一层都能单独测试方便定位问题。第三也是最重要的延迟优化有了明确方向用户感知的总延迟 识别耗时 大模型耗时 合成耗时优化哪个环节一目了然。3. 整体架构设计与模块划分这里先给出一个可以落地的模块划分后面所有代码都按这个结构来组织。为了照顾多数场景示例项目命名为 voiceagent_demo文件结构如下voiceagent_demo/ ├── main.py # 主控制流串联三层模块 ├── config.py # 全局配置模型、API Key、音频参数 ├── requirements.txt # 依赖列表 ├── modules/ │ ├── __init__.py │ ├── asr_module.py # 感知层语音识别 │ ├── llm_module.py # 大脑层大模型对话与工具调用 │ ├── tts_module.py # 表达层语音合成 │ └── audio_utils.py # 音频采集与播放工具三层模块可以简单对照如下层级模块输入输出核心职责感知层asr_module音频文件/音频流文本语音转文字模型输出包括中间状态大脑层llm_module用户文本 对话历史 工具定义回复文本/结构化动作理解意图、调用工具、组织话术表达层tts_module回复文本音频文本转语音支持播放与缓存模块之间不互相直接调内部函数而是走主控制流。也就是说main.py 负责依次调用 asr、llm、tts 的结果并处理中间可能出现的异常。这样的结构虽然看起来多了一个调度层但调试的时候你会庆幸有个唯一入口。4. 环境准备与前置条件4.1 运行环境操作系统Windows 10/11、macOS、Linux 均可本文以 Windows 和 Linux 通用命令演示Python 版本建议 3.10 或更高。语音库和 WebSocket 客户端对 3.10 以下版本支持参差不齐直接用新版本省心麦克风与扬声器如果你要做实时对话必须保证系统能采集和播放音频。如果是在没有麦克风的服务器上调试可以准备一个 wav 测试音频文件。4.2 依赖安装创建虚拟环境并安装依赖python -m venv venv source venv/bin/activate # Windows 下执行 venv\Scripts\activate pip install --upgrade pip本项目依赖包括openai调用兼容 OpenAI 接口的大模型websockets部分语音识别服务和实时语音交互需要pyaudio本地音频采集和播放soundfile、numpy音频读取、格式转换edge-tts免费的文本转语音方案网络环境可用时比较方便。安装命令pip install openai websockets pyaudio soundfile numpy edge-tts如果你的系统安装 pyaudio 报错Windows 可以先安装 wheel 包或者使用 pipwin 解决Linux 需要先安装 portaudio 开发库。下面是 Ubuntu/Debian 的参考命令sudo apt-get update sudo apt-get install -y portaudio19-dev python3-pyaudiomacOS 安装命令brew install portaudio4.3 大模型 API 配置大模型部分建议选择与 OpenAI Chat Completion 接口兼容的服务因为生态工具最成熟切换成本低。你需要准备API Base URL服务地址API Key模型名称。在 config.py 中集中管理不要写散在代码里。生产环境更推荐用环境变量示例为了直观写在配置文件里但会特别标注敏感信息不宜提交到代码仓库。5. 核心流程拆解整体流程按如下顺序执行。先看流程再对照代码理解会更清楚。启动程序初始化 ASR、LLM、TTS 模块和对话上下文从麦克风采集音频或读取测试音频文件将音频送入 ASR 模块得到用户文本将用户文本送入 LLM 模块附带对话历史和工具描述LLM 可能返回普通文本也可能返回工具调用指令。如果有工具调用执行工具后把结果回传让模型生成最终文本将最终文本送入 TTS 模块合成语音播放语音并把当轮文本加入历史记录回到第 2 步等待下一次输入。这个流程里最容易被忽略的是第 5 步。很多人只做“一句对话一句回复”不带工具调用导致语音助手只会聊天不会干活。做 VoiceAgent 不是做一个“会说话的搜索框”而是做一个“能执行任务的语音入口”所以工具调用机制无论如何都要预留。6. 完整代码实现与逐步讲解下面逐模块给出代码。为了保证可运行性每个文件都给出完整内容。6.1 依赖清单# 文件路径voiceagent_demo/requirements.txt openai1.30.0 websockets12.0 pyaudio0.2.14 soundfile0.12.1 numpy1.26.0 edge-tts6.1.06.2 全局配置# 文件路径voiceagent_demo/config.py import os # 大模型配置 LLM_API_KEY os.getenv(LLM_API_KEY, your-api-key) LLM_BASE_URL os.getenv(LLM_BASE_URL, https://api.openai.com/v1) LLM_MODEL os.getenv(LLM_MODEL, gpt-4o-mini) # 系统提示词 SYSTEM_PROMPT 你是一个语音助手的智能内核名叫小智。 你需要根据用户语音转写的文本给出自然、简洁、适合朗读的回答。 要求 1. 优先使用工具获取准确信息 2. 回答控制在3句话以内适合语音播报 3. 不要使用Markdown、列表、表格等不适合朗读的格式 4. 如果用户的问题不在能力范围内直接说明无法处理。 # 语音识别配置 ASR_API_URL os.getenv(ASR_API_URL, http://127.0.0.1:9001/asr) ASR_SAMPLE_RATE 16000 # 语音合成配置 TTS_VOICE os.getenv(TTS_VOICE, zh-CN-XiaoxiaoNeural) TTS_OUTPUT_DIR output # 音频参数 AUDIO_FORMAT wav注意这里 API Key 默认是占位符。你应该使用环境变量来注入真实 Key尤其是打算把项目传到 GitHub 时千万别硬编码在源码里。6.3 音频工具模块# 文件路径voiceagent_demo/modules/audio_utils.py import soundfile as sf import numpy as np import io def read_audio_file(file_path, target_sr16000): 读取音频文件并重采样到目标采样率。 返回格式numpy.ndarrayfloat32单声道。 data, sr sf.read(file_path) if len(data.shape) 1: data data.mean(axis1) if sr ! target_sr: # 简单的线性重采样生产环境建议用 librosa 或 soxr ratio target_sr / sr new_len int(len(data) * ratio) indices (np.arange(new_len) / ratio).astype(int) indices np.clip(indices, 0, len(data) - 1) data data[indices] return data.astype(np.float32) def float32_to_bytes(data): 将 float32 音频数据转为 pcm 字节流便于传输。 pcm16 (data * 32767).astype(np.int16) return pcm16.tobytes()这个模块当前只处理文件读取。在真实麦克风场景里pyaudio 采集到的数据流需要做类似格式转换这里预留了 float32_to_bytes 方法方便对接流式服务。6.4 感知层语音识别模块# 文件路径voiceagent_demo/modules/asr_module.py import json import requests import numpy as np from modules.audio_utils import float32_to_bytes class ASRModule: 语音识别模块感知层。 默认假设有一个本地或线上的 ASR HTTP 服务 接收 pcm 数据返回 JSON: {text: 识别结果} def __init__(self, api_url): self.api_url api_url def transcribe(self, audio_data: np.ndarray, sample_rate: int 16000) - str: pcm_bytes float32_to_bytes(audio_data) resp requests.post( self.api_url, params{sample_rate: sample_rate}, datapcm_bytes, timeout20, headers{Content-Type: application/octet-stream} ) resp.raise_for_status() result resp.json() return result.get(text, ) def transcribe_file(self, file_path: str) - str: 直接识别本地音频文件。 from modules.audio_utils import read_audio_file audio read_audio_file(file_path, target_sr16000) return self.transcribe(audio)这个模块设计的假设是你已经有一个可用的语音识别服务。实际项目里你可以用 FunASR 等开源框架在本地起一个服务或者调用商业语音识别 API只需把请求和响应格式改成对应服务的要求即可。这个模块体现了“级联架构”的价值替换 ASR 时不碰其他代码。6.5 大脑层大模型与工具调用模块这是整个项目最核心的模块。它不仅要完成普通对话还要支持工具调用。# 文件路径voiceagent_demo/modules/llm_module.py import json from openai import OpenAI # 工具定义示例工具获取指定城市当前时间 TOOLS [ { type: function, function: { name: get_current_time, description: 获取指定城市当前时间, parameters: { type: object, properties: { city: { type: string, description: 城市名称中文例如北京、上海 } }, required: [city] } } } ] def get_current_time(city: str) - str: 示例工具函数。 真实项目中应接入时间服务或者业务 API。 import datetime now datetime.datetime.now().strftime(%Y-%m-%d %H:%M:%S) return f{city}当前时间为{now} class LLMModule: def __init__(self, api_key, base_url, model, system_prompt): self.client OpenAI(api_keyapi_key, base_urlbase_url) self.model model self.system_prompt system_prompt def chat(self, messages, toolsNone, tool_choiceNone): 调用大模型。 messages 格式遵循 OpenAI 消息规范。 返回完整 response 对象由上层判断是文本还是工具调用。 payload { model: self.model, messages: messages, temperature: 0.3 } if tools: payload[tools] tools payload[tool_choice] tool_choice or auto response self.client.chat.completions.create(**payload) return response def run(self, user_text, history): 执行完整对话逻辑 1. 拼装消息 2. 调用模型 3. 如果返回工具调用执行工具并回传结果 4. 返回最终回复文本 messages [{role: system, content: self.system_prompt}] messages.extend(history) messages.append({role: user, content: user_text}) response self.chat(messages, toolsTOOLS) message response.choices[0].message # 检查是否有工具调用 if message.tool_calls: tool_call message.tool_calls[0] fn_name tool_call.function.name args json.loads(tool_call.function.arguments) print(f[LLM] 调用工具: {fn_name}, 参数: {args}) if fn_name get_current_time: tool_result get_current_time(cityargs.get(city, 未知)) else: tool_result 未支持的工具 # 追加工具调用消息和工具结果消息 messages.append({ role: assistant, content: None, tool_calls: [tool_call] }) messages.append({ role: tool, tool_call_id: tool_call.id, content: tool_result }) # 让模型基于工具结果生成最终回复 second_response self.chat(messages, toolsTOOLS) final_message second_response.choices[0].message return final_message.content or # 没有工具调用直接返回文本 return message.content or 这段代码的核心逻辑是模型不直接回答“现在几点”而是返回一个工具调用指令你的代码执行 get_current_time 之后把真实结果回传给模型模型再组织成自然语言。这在语音智能体里非常重要因为工具结果的时效性和准确性直接影响用户体验。同时注意代码里 message.tool_calls[0] 只处理了第一个工具调用。生产环境如果工具很多建议循环处理所有工具调用。6.6 表达层语音合成模块# 文件路径voiceagent_demo/modules/tts_module.py import asyncio import edge_tts class TTSModule: 文本转语音模块表达层。 使用 edge-tts 实现文本转语音输出 mp3 文件后播放。 def __init__(self, voicezh-CN-XiaoxiaoNeural, output_diroutput): self.voice voice self.output_dir output_dir async def _synthesize_to_file(self, text, output_path): communicate edge_tts.Communicate(text, self.voice) await communicate.save(output_path) def synthesize(self, text, output_pathNone): 同步入口方便主流程调用。 import os if output_path is None: os.makedirs(self.output_dir, exist_okTrue) output_path os.path.join(self.output_dir, reply.mp3) asyncio.run(self._synthesize_to_file(text, output_path)) return output_pathedge-tts 是微软 Edge 的在线语音合成接口优点是免费、发音自然、支持多种中文音色。缺点是依赖外网可用性和微软服务的稳定性。生产环境如果要可控建议替换为商用的 TTS 服务接口封装不变。如果需要播放音频可以直接用系统命令。Windows 下播放 mp3 可以这样import os def play_audio(file_path): if os.name nt: os.system(fstart {file_path}) else: os.system(fffplay -nodisp -autoexit {file_path})你也可以把这一步放在 main.py 里效果等同。6.7 主控流程# 文件路径voiceagent_demo/main.py import os import sys sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from config import ( ASR_API_URL, TTS_VOICE, LLM_API_KEY, LLM_BASE_URL, LLM_MODEL, SYSTEM_PROMPT ) from modules.asr_module import ASRModule from modules.llm_module import LLMModule from modules.tts_module import TTSModule from modules.audio_utils import read_audio_file class VoiceAgent: def __init__(self): self.asr ASRModule(api_urlASR_API_URL) self.llm LLMModule( api_keyLLM_API_KEY, base_urlLLM_BASE_URL, modelLLM_MODEL, system_promptSYSTEM_PROMPT ) self.tts TTSModule(voiceTTS_VOICE) self.history [] def handle_audio_file(self, audio_path): 离线模式识别音频文件并回复。 text self.asr.transcribe_file(audio_path) print(f[ASR] 识别结果: {text}) if not text: print([ASR] 未识别到内容) return None reply self.llm.run(user_texttext, historyself.history) print(f[LLM] 回复: {reply}) self.history.append({role: user, content: text}) self.history.append({role: assistant, content: reply}) # 限制历史长度避免超出模型上下文窗口 if len(self.history) 10: self.history self.history[-10:] audio_file self.tts.synthesize(reply) print(f[TTS] 语音已生成: {audio_file}) return audio_file def handle_text(self, user_text): 纯文本测试模式用于调试大模型链路。 reply self.llm.run(user_textuser_text, historyself.history) print(f[LLM] 回复: {reply}) self.history.append({role: user, content: user_text}) self.history.append({role: assistant, content: reply}) return reply if __name__ __main__: agent VoiceAgent() # 如果要测试语音文件识别传入音频文件路径 if len(sys.argv) 1: agent.handle_audio_file(sys.argv[1]) else: # 文本模式循环对话 while True: user_input input(你说: ) if user_input.lower() in [exit, quit, 退出]: break agent.handle_text(user_input)这里做了一个很实用的设计支持两种运行模式。有语音文件时走全链路没有语音文件时走文本模式。建议第一次运行时先用文本模式调通大模型再引入语音识别最后加入语音合成。分步调试能帮你快速区分问题是出在哪个环节。7. 运行效果与验证方法7.1 先验证大模型链路不传音频文件直接运行python main.py程序会进入文本循环你说: 北京现在几点 [LLM] 调用工具: get_current_time, 参数: {city: 北京} [LLM] 回复: 北京当前时间为2026-01-15 14:23:05。 你说: exit看到这段输出说明大模型链路和工具调用机制已经打通。这一步是整个项目的地基务必先确认它正常。7.2 验证语音识别链路准备一个包含清晰人声的 wav 文件采样率最好已经是 16k。然后运行python main.py test_audio/test.wav预期输出[ASR] 识别结果: 你好请问今天天气怎么样 [LLM] 回复: 你好我这边还没有接入天气查询服务暂时无法回答这个问题。 [TTS] 语音已生成: output/reply.mp3如果 ASR 识别为空先确认 ASR 服务是否启动、音频格式是否符合要求。如果 LLM 回复异常用文本模式直接验证模型是否可用。如果 TTS 输出不了文件检查 edge-tts 网络是否可用。7.3 如何判断链路是否健康可以从三个维度判断延迟是否可接受每一层打印出耗时便于定位瓶颈。示例代码未加计时建议第一个生产版本就给每一层加上耗时统计语义是否合理大模型是否理解了用户意图是否在工具调用后给出正确回答语音是否自然TTS 合成的句子是否断句合理、清晰可懂。整体跑通后你就有了一套最小可用的 VoiceAgent 骨架。接下来无论接入具体的业务 API还是优化实时交互体验都有明确的方向。8. 常见问题与排查思路问题现象可能原因排查方式解决方案ASR 识别结果为空ASR 服务地址不正确音频采样率不匹配音频静音太长先用 curl 或 Postman 直接调用 ASR 服务测试修正地址音频重采样到 16k调整端点检测参数大模型返回 401 或 403API Key 错误或没有权限检查环境变量和 config.py 中的 Key替换为有效 Key确认模型名称与账号权限匹配大模型返回超时网络代理问题模型上下文过长查看请求耗时缩短历史记录配置网络代理调整 history 最大长度工具调用报错工具参数不匹配JSON 解析失败打印 tool_calls 原始返回检查参数 definitions 和 args 解析逻辑增加 try exceptTTS 合成失败edge-tts 网络不可用音频路径无写权限单独运行 TTS 测试脚本检查网络改用其他 TTS 服务确认输出目录存在播放没有声音系统音频设备问题mp3 缺少播放器用其他播放器打开生成的 mp3安装播放器调整播放逻辑使用 pyaudio 播放除了表格里的问题还有一个典型的架构级错误让大模型模块直接调用 ASR 模块或者让 TTS 模块直接访问 LLM 内部方法。一旦出现这种耦合排错就会变难。保持单向数据流是这套架构能“排错快”的根本原因。9. 最佳实践与工程建议9.1 对话历史的窗口管理大模型的上下文窗口是有限资源。语音交互场景通常不需要很长的历史建议只保留最近 5 到 10 轮。更早的内容如果需要可以用摘要方式压缩进 system prompt而不是无脑拼接历史消息。9.2 延迟优化优先级用户对语音助手的耐心比对文字助手更短。延迟优化的优先级是这样排的大模型首字延迟最重要优先使用支持流式输出的模型服务识别阶段使用流式 ASR可以在用户说完之前就开始出文字TTS 可以提前合成固定话术例如“好的正在查询”这类缓冲语不要过度优化音频播放延迟那是最后一步。9.3 工具调用的容错设计工具函数返回的数据不一定是模型期望的格式。建议工具函数统一返回字符串并且对异常情况返回明确的错误描述例如“查询失败网络超时”。这样模型才能生成合理的用户提示而不是凭空白编一个答案。9.4 环境变量与密钥管理示例为了方便直接写在 config.py 里生产环境必须改成环境变量注入。还可以引入配置中心或密钥管理服务确保 API Key、数据库连接串等敏感信息不进入代码仓库。.gitignore 必须加入 .env 文件。9.5 日志与监控VoiceAgent 调试时最烦的是“不知道卡在哪一层”。建议从第一版就给每个模块加上统一的日志埋点。日志格式可以这样设计模块名、时间戳、请求内容摘要、响应内容摘要、耗时、状态码。后续接入指标监控时这些日志直接就能用。9.6 从示例到生产的演进路径第一步跑通文本模式第二步接入真实 ASR第三步接入真实 TTS第四步把“一问一答”改成“多轮会话”第五步引入打断和流式语音第六步对接业务 API让助手真正干活。每一步都建议单独发布一个可运行的版本不要一口气从 demo 跳到全功能产品。10. 总结与后续学习方向这篇文章做了一件很具体的事把一个“语音智能体项目”从概念拆成了三层模块并给出了可执行的最小代码骨架。你现在应该能回答这几个问题了ASR 层、LLM 层、TTS 层各自的职责边界是什么工具调用在主流程中是怎么工作的音频数据如何逐层流转为什么“级联式三明治架构”能减少开发过程中的混乱。下一步可以按自己的业务场景继续深入如果你的重点是助手能力可以去研究 Function Calling 的进阶用法和多工具调度如果重点是交互体验可以研究流式 ASR 和语音打断如果重点是系统性能可以研究并发会话、缓存策略和模型部署优化。代码骨架放在那里不会让你变强基于它跑通一个自己的场景才会。找一个你日常最想自动化的事情给它写一个工具函数让 VoiceAgent 帮你把它干了你会对这个架构的理解完全不一样。 语音智能体开发实战：级联式三明治架构从零搭建
先给结论如果你正打算从零做一个能“听、想、说”的语音智能体应用但又不想被 ASR、LLM、TTS、流式协议、状态管理这些概念绕晕那么“级联式三明治架构”是目前最值得先吃透的一套落地范式。它不会让项目变成最酷的但大概率能让项目最快跑起来而且工程上可控、出错可排查、后续可扩展。这篇文章不聊虚的直接拆架构、给代码、走流程、讲排错目标只有一个让你今晚就能在自己电脑上跑通一个最小可用的 VoiceAgent。1. 这篇文章真正要解决的问题做语音智能体开发很多人第一反应是“接入一个大模型 API 不就行了”但实际动手之后会发现根本不是这么回事。语音交互链路比纯文本对话长得多麦克风采集到的音频要先变成文字文字进入大模型后要结合上下文和业务逻辑生成回复回复文字又要变成语音播放给用户。每一段都有独立的工具链、独立的延迟、独立的出错点。更麻烦的是当你把大模型、语音识别、语音合成串在一起时很容易写出一个“能跑但很难维护”的脚本所有逻辑堆在一个 Python 文件里回调套回调异常处理靠运气想加一个“打断当前播报”的功能都要重构半天。这就是为什么需要有一个清晰的架构而不是“把 API 串起来就行”。这篇文章要解决的核心问题有三个语音智能体的标准链路是什么每一层到底负责什么“级联式三明治架构”说的不是花哨概念而是输入层、大脑层、输出层三层分离的设计方法你该怎么用代码把它们拼起来并且能验证、能排错、能上生产。文中所有演示代码以 Python 为宿主语言语音识别和语音合成使用通用开源能力或云端 API大模型部分使用兼容 OpenAI 格式的服务。版本号不做死绑重点是跑通思路。2. 基础概念大模型、智能体、级联式三明治架构在进入代码之前先把文章里反复出现的三个词说清楚。理解到位了后面看代码会顺畅很多也避免在概念层面走偏。2.1 大模型不是“题库”而是“推理内核”大模型在这个项目里的定位不是知识库而是整个 VoiceAgent 的“意图决策中心”。它负责理解用户这次说话的真实诉求结合当前对话上下文决定应该回答什么、查询什么、调用什么工具。真正的大模型应用重点不是“它知道什么”而是“你如何引导它正确使用自己知道的东西”。在代码层面你只需要关注一点选一个接口兼容的大模型服务然后统一定义系统提示词和消息格式。后续想换模型只改配置不改业务逻辑。2.2 智能体不是“聊天机器人”而是“任务执行器”智能体的关键在于它能根据用户指令主动规划步骤、调用外部工具、读取状态最后把结果整理成对用户有用的话。语音场景下的智能体通常还要多处理一层如何把长文本结果转成适合“听”的表达例如去掉 Markdown 符号、让句子更口语化、避免一次性输出太长导致用户记不住。这就决定了你的 VoiceAgent 核心代码里需要有一个明确的“工具调用”设计。即使第一个版本只有一个查询时间、查询天气之类的工具也要把机制搭好。2.3 级联式三明治架构三层分离中间厚、两边薄先拆一下这个听起来有点唬人的名字。把整个语音智能体想象成一块三明治上层是“感知层”麦克风采集、语音识别、端点检测。作用是把用户的自然讲话变成文本中间是“大脑层”大模型对话管理、上下文记忆、工具调用、业务逻辑。这是整块三明治里最厚的一层也是核心价值所在下层是“表达层”回复文本的处理、语音合成、音频播放。所谓“级联”就是数据必须按顺序一层一层流下去音频进、文本出、文本进、决策出、文本进、语音出。每一层都是独立模块层与层之间只通过标准数据格式通信。这个架构的直观好处有三个。第一你可以单独替换任何一层今天用 A 厂商的语音识别明天换 B 厂商不会动到大模型代码。第二每一层都能单独测试方便定位问题。第三也是最重要的延迟优化有了明确方向用户感知的总延迟 识别耗时 大模型耗时 合成耗时优化哪个环节一目了然。3. 整体架构设计与模块划分这里先给出一个可以落地的模块划分后面所有代码都按这个结构来组织。为了照顾多数场景示例项目命名为 voiceagent_demo文件结构如下voiceagent_demo/ ├── main.py # 主控制流串联三层模块 ├── config.py # 全局配置模型、API Key、音频参数 ├── requirements.txt # 依赖列表 ├── modules/ │ ├── __init__.py │ ├── asr_module.py # 感知层语音识别 │ ├── llm_module.py # 大脑层大模型对话与工具调用 │ ├── tts_module.py # 表达层语音合成 │ └── audio_utils.py # 音频采集与播放工具三层模块可以简单对照如下层级模块输入输出核心职责感知层asr_module音频文件/音频流文本语音转文字模型输出包括中间状态大脑层llm_module用户文本 对话历史 工具定义回复文本/结构化动作理解意图、调用工具、组织话术表达层tts_module回复文本音频文本转语音支持播放与缓存模块之间不互相直接调内部函数而是走主控制流。也就是说main.py 负责依次调用 asr、llm、tts 的结果并处理中间可能出现的异常。这样的结构虽然看起来多了一个调度层但调试的时候你会庆幸有个唯一入口。4. 环境准备与前置条件4.1 运行环境操作系统Windows 10/11、macOS、Linux 均可本文以 Windows 和 Linux 通用命令演示Python 版本建议 3.10 或更高。语音库和 WebSocket 客户端对 3.10 以下版本支持参差不齐直接用新版本省心麦克风与扬声器如果你要做实时对话必须保证系统能采集和播放音频。如果是在没有麦克风的服务器上调试可以准备一个 wav 测试音频文件。4.2 依赖安装创建虚拟环境并安装依赖python -m venv venv source venv/bin/activate # Windows 下执行 venv\Scripts\activate pip install --upgrade pip本项目依赖包括openai调用兼容 OpenAI 接口的大模型websockets部分语音识别服务和实时语音交互需要pyaudio本地音频采集和播放soundfile、numpy音频读取、格式转换edge-tts免费的文本转语音方案网络环境可用时比较方便。安装命令pip install openai websockets pyaudio soundfile numpy edge-tts如果你的系统安装 pyaudio 报错Windows 可以先安装 wheel 包或者使用 pipwin 解决Linux 需要先安装 portaudio 开发库。下面是 Ubuntu/Debian 的参考命令sudo apt-get update sudo apt-get install -y portaudio19-dev python3-pyaudiomacOS 安装命令brew install portaudio4.3 大模型 API 配置大模型部分建议选择与 OpenAI Chat Completion 接口兼容的服务因为生态工具最成熟切换成本低。你需要准备API Base URL服务地址API Key模型名称。在 config.py 中集中管理不要写散在代码里。生产环境更推荐用环境变量示例为了直观写在配置文件里但会特别标注敏感信息不宜提交到代码仓库。5. 核心流程拆解整体流程按如下顺序执行。先看流程再对照代码理解会更清楚。启动程序初始化 ASR、LLM、TTS 模块和对话上下文从麦克风采集音频或读取测试音频文件将音频送入 ASR 模块得到用户文本将用户文本送入 LLM 模块附带对话历史和工具描述LLM 可能返回普通文本也可能返回工具调用指令。如果有工具调用执行工具后把结果回传让模型生成最终文本将最终文本送入 TTS 模块合成语音播放语音并把当轮文本加入历史记录回到第 2 步等待下一次输入。这个流程里最容易被忽略的是第 5 步。很多人只做“一句对话一句回复”不带工具调用导致语音助手只会聊天不会干活。做 VoiceAgent 不是做一个“会说话的搜索框”而是做一个“能执行任务的语音入口”所以工具调用机制无论如何都要预留。6. 完整代码实现与逐步讲解下面逐模块给出代码。为了保证可运行性每个文件都给出完整内容。6.1 依赖清单# 文件路径voiceagent_demo/requirements.txt openai1.30.0 websockets12.0 pyaudio0.2.14 soundfile0.12.1 numpy1.26.0 edge-tts6.1.06.2 全局配置# 文件路径voiceagent_demo/config.py import os # 大模型配置 LLM_API_KEY os.getenv(LLM_API_KEY, your-api-key) LLM_BASE_URL os.getenv(LLM_BASE_URL, https://api.openai.com/v1) LLM_MODEL os.getenv(LLM_MODEL, gpt-4o-mini) # 系统提示词 SYSTEM_PROMPT 你是一个语音助手的智能内核名叫小智。 你需要根据用户语音转写的文本给出自然、简洁、适合朗读的回答。 要求 1. 优先使用工具获取准确信息 2. 回答控制在3句话以内适合语音播报 3. 不要使用Markdown、列表、表格等不适合朗读的格式 4. 如果用户的问题不在能力范围内直接说明无法处理。 # 语音识别配置 ASR_API_URL os.getenv(ASR_API_URL, http://127.0.0.1:9001/asr) ASR_SAMPLE_RATE 16000 # 语音合成配置 TTS_VOICE os.getenv(TTS_VOICE, zh-CN-XiaoxiaoNeural) TTS_OUTPUT_DIR output # 音频参数 AUDIO_FORMAT wav注意这里 API Key 默认是占位符。你应该使用环境变量来注入真实 Key尤其是打算把项目传到 GitHub 时千万别硬编码在源码里。6.3 音频工具模块# 文件路径voiceagent_demo/modules/audio_utils.py import soundfile as sf import numpy as np import io def read_audio_file(file_path, target_sr16000): 读取音频文件并重采样到目标采样率。 返回格式numpy.ndarrayfloat32单声道。 data, sr sf.read(file_path) if len(data.shape) 1: data data.mean(axis1) if sr ! target_sr: # 简单的线性重采样生产环境建议用 librosa 或 soxr ratio target_sr / sr new_len int(len(data) * ratio) indices (np.arange(new_len) / ratio).astype(int) indices np.clip(indices, 0, len(data) - 1) data data[indices] return data.astype(np.float32) def float32_to_bytes(data): 将 float32 音频数据转为 pcm 字节流便于传输。 pcm16 (data * 32767).astype(np.int16) return pcm16.tobytes()这个模块当前只处理文件读取。在真实麦克风场景里pyaudio 采集到的数据流需要做类似格式转换这里预留了 float32_to_bytes 方法方便对接流式服务。6.4 感知层语音识别模块# 文件路径voiceagent_demo/modules/asr_module.py import json import requests import numpy as np from modules.audio_utils import float32_to_bytes class ASRModule: 语音识别模块感知层。 默认假设有一个本地或线上的 ASR HTTP 服务 接收 pcm 数据返回 JSON: {text: 识别结果} def __init__(self, api_url): self.api_url api_url def transcribe(self, audio_data: np.ndarray, sample_rate: int 16000) - str: pcm_bytes float32_to_bytes(audio_data) resp requests.post( self.api_url, params{sample_rate: sample_rate}, datapcm_bytes, timeout20, headers{Content-Type: application/octet-stream} ) resp.raise_for_status() result resp.json() return result.get(text, ) def transcribe_file(self, file_path: str) - str: 直接识别本地音频文件。 from modules.audio_utils import read_audio_file audio read_audio_file(file_path, target_sr16000) return self.transcribe(audio)这个模块设计的假设是你已经有一个可用的语音识别服务。实际项目里你可以用 FunASR 等开源框架在本地起一个服务或者调用商业语音识别 API只需把请求和响应格式改成对应服务的要求即可。这个模块体现了“级联架构”的价值替换 ASR 时不碰其他代码。6.5 大脑层大模型与工具调用模块这是整个项目最核心的模块。它不仅要完成普通对话还要支持工具调用。# 文件路径voiceagent_demo/modules/llm_module.py import json from openai import OpenAI # 工具定义示例工具获取指定城市当前时间 TOOLS [ { type: function, function: { name: get_current_time, description: 获取指定城市当前时间, parameters: { type: object, properties: { city: { type: string, description: 城市名称中文例如北京、上海 } }, required: [city] } } } ] def get_current_time(city: str) - str: 示例工具函数。 真实项目中应接入时间服务或者业务 API。 import datetime now datetime.datetime.now().strftime(%Y-%m-%d %H:%M:%S) return f{city}当前时间为{now} class LLMModule: def __init__(self, api_key, base_url, model, system_prompt): self.client OpenAI(api_keyapi_key, base_urlbase_url) self.model model self.system_prompt system_prompt def chat(self, messages, toolsNone, tool_choiceNone): 调用大模型。 messages 格式遵循 OpenAI 消息规范。 返回完整 response 对象由上层判断是文本还是工具调用。 payload { model: self.model, messages: messages, temperature: 0.3 } if tools: payload[tools] tools payload[tool_choice] tool_choice or auto response self.client.chat.completions.create(**payload) return response def run(self, user_text, history): 执行完整对话逻辑 1. 拼装消息 2. 调用模型 3. 如果返回工具调用执行工具并回传结果 4. 返回最终回复文本 messages [{role: system, content: self.system_prompt}] messages.extend(history) messages.append({role: user, content: user_text}) response self.chat(messages, toolsTOOLS) message response.choices[0].message # 检查是否有工具调用 if message.tool_calls: tool_call message.tool_calls[0] fn_name tool_call.function.name args json.loads(tool_call.function.arguments) print(f[LLM] 调用工具: {fn_name}, 参数: {args}) if fn_name get_current_time: tool_result get_current_time(cityargs.get(city, 未知)) else: tool_result 未支持的工具 # 追加工具调用消息和工具结果消息 messages.append({ role: assistant, content: None, tool_calls: [tool_call] }) messages.append({ role: tool, tool_call_id: tool_call.id, content: tool_result }) # 让模型基于工具结果生成最终回复 second_response self.chat(messages, toolsTOOLS) final_message second_response.choices[0].message return final_message.content or # 没有工具调用直接返回文本 return message.content or 这段代码的核心逻辑是模型不直接回答“现在几点”而是返回一个工具调用指令你的代码执行 get_current_time 之后把真实结果回传给模型模型再组织成自然语言。这在语音智能体里非常重要因为工具结果的时效性和准确性直接影响用户体验。同时注意代码里 message.tool_calls[0] 只处理了第一个工具调用。生产环境如果工具很多建议循环处理所有工具调用。6.6 表达层语音合成模块# 文件路径voiceagent_demo/modules/tts_module.py import asyncio import edge_tts class TTSModule: 文本转语音模块表达层。 使用 edge-tts 实现文本转语音输出 mp3 文件后播放。 def __init__(self, voicezh-CN-XiaoxiaoNeural, output_diroutput): self.voice voice self.output_dir output_dir async def _synthesize_to_file(self, text, output_path): communicate edge_tts.Communicate(text, self.voice) await communicate.save(output_path) def synthesize(self, text, output_pathNone): 同步入口方便主流程调用。 import os if output_path is None: os.makedirs(self.output_dir, exist_okTrue) output_path os.path.join(self.output_dir, reply.mp3) asyncio.run(self._synthesize_to_file(text, output_path)) return output_pathedge-tts 是微软 Edge 的在线语音合成接口优点是免费、发音自然、支持多种中文音色。缺点是依赖外网可用性和微软服务的稳定性。生产环境如果要可控建议替换为商用的 TTS 服务接口封装不变。如果需要播放音频可以直接用系统命令。Windows 下播放 mp3 可以这样import os def play_audio(file_path): if os.name nt: os.system(fstart {file_path}) else: os.system(fffplay -nodisp -autoexit {file_path})你也可以把这一步放在 main.py 里效果等同。6.7 主控流程# 文件路径voiceagent_demo/main.py import os import sys sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from config import ( ASR_API_URL, TTS_VOICE, LLM_API_KEY, LLM_BASE_URL, LLM_MODEL, SYSTEM_PROMPT ) from modules.asr_module import ASRModule from modules.llm_module import LLMModule from modules.tts_module import TTSModule from modules.audio_utils import read_audio_file class VoiceAgent: def __init__(self): self.asr ASRModule(api_urlASR_API_URL) self.llm LLMModule( api_keyLLM_API_KEY, base_urlLLM_BASE_URL, modelLLM_MODEL, system_promptSYSTEM_PROMPT ) self.tts TTSModule(voiceTTS_VOICE) self.history [] def handle_audio_file(self, audio_path): 离线模式识别音频文件并回复。 text self.asr.transcribe_file(audio_path) print(f[ASR] 识别结果: {text}) if not text: print([ASR] 未识别到内容) return None reply self.llm.run(user_texttext, historyself.history) print(f[LLM] 回复: {reply}) self.history.append({role: user, content: text}) self.history.append({role: assistant, content: reply}) # 限制历史长度避免超出模型上下文窗口 if len(self.history) 10: self.history self.history[-10:] audio_file self.tts.synthesize(reply) print(f[TTS] 语音已生成: {audio_file}) return audio_file def handle_text(self, user_text): 纯文本测试模式用于调试大模型链路。 reply self.llm.run(user_textuser_text, historyself.history) print(f[LLM] 回复: {reply}) self.history.append({role: user, content: user_text}) self.history.append({role: assistant, content: reply}) return reply if __name__ __main__: agent VoiceAgent() # 如果要测试语音文件识别传入音频文件路径 if len(sys.argv) 1: agent.handle_audio_file(sys.argv[1]) else: # 文本模式循环对话 while True: user_input input(你说: ) if user_input.lower() in [exit, quit, 退出]: break agent.handle_text(user_input)这里做了一个很实用的设计支持两种运行模式。有语音文件时走全链路没有语音文件时走文本模式。建议第一次运行时先用文本模式调通大模型再引入语音识别最后加入语音合成。分步调试能帮你快速区分问题是出在哪个环节。7. 运行效果与验证方法7.1 先验证大模型链路不传音频文件直接运行python main.py程序会进入文本循环你说: 北京现在几点 [LLM] 调用工具: get_current_time, 参数: {city: 北京} [LLM] 回复: 北京当前时间为2026-01-15 14:23:05。 你说: exit看到这段输出说明大模型链路和工具调用机制已经打通。这一步是整个项目的地基务必先确认它正常。7.2 验证语音识别链路准备一个包含清晰人声的 wav 文件采样率最好已经是 16k。然后运行python main.py test_audio/test.wav预期输出[ASR] 识别结果: 你好请问今天天气怎么样 [LLM] 回复: 你好我这边还没有接入天气查询服务暂时无法回答这个问题。 [TTS] 语音已生成: output/reply.mp3如果 ASR 识别为空先确认 ASR 服务是否启动、音频格式是否符合要求。如果 LLM 回复异常用文本模式直接验证模型是否可用。如果 TTS 输出不了文件检查 edge-tts 网络是否可用。7.3 如何判断链路是否健康可以从三个维度判断延迟是否可接受每一层打印出耗时便于定位瓶颈。示例代码未加计时建议第一个生产版本就给每一层加上耗时统计语义是否合理大模型是否理解了用户意图是否在工具调用后给出正确回答语音是否自然TTS 合成的句子是否断句合理、清晰可懂。整体跑通后你就有了一套最小可用的 VoiceAgent 骨架。接下来无论接入具体的业务 API还是优化实时交互体验都有明确的方向。8. 常见问题与排查思路问题现象可能原因排查方式解决方案ASR 识别结果为空ASR 服务地址不正确音频采样率不匹配音频静音太长先用 curl 或 Postman 直接调用 ASR 服务测试修正地址音频重采样到 16k调整端点检测参数大模型返回 401 或 403API Key 错误或没有权限检查环境变量和 config.py 中的 Key替换为有效 Key确认模型名称与账号权限匹配大模型返回超时网络代理问题模型上下文过长查看请求耗时缩短历史记录配置网络代理调整 history 最大长度工具调用报错工具参数不匹配JSON 解析失败打印 tool_calls 原始返回检查参数 definitions 和 args 解析逻辑增加 try exceptTTS 合成失败edge-tts 网络不可用音频路径无写权限单独运行 TTS 测试脚本检查网络改用其他 TTS 服务确认输出目录存在播放没有声音系统音频设备问题mp3 缺少播放器用其他播放器打开生成的 mp3安装播放器调整播放逻辑使用 pyaudio 播放除了表格里的问题还有一个典型的架构级错误让大模型模块直接调用 ASR 模块或者让 TTS 模块直接访问 LLM 内部方法。一旦出现这种耦合排错就会变难。保持单向数据流是这套架构能“排错快”的根本原因。9. 最佳实践与工程建议9.1 对话历史的窗口管理大模型的上下文窗口是有限资源。语音交互场景通常不需要很长的历史建议只保留最近 5 到 10 轮。更早的内容如果需要可以用摘要方式压缩进 system prompt而不是无脑拼接历史消息。9.2 延迟优化优先级用户对语音助手的耐心比对文字助手更短。延迟优化的优先级是这样排的大模型首字延迟最重要优先使用支持流式输出的模型服务识别阶段使用流式 ASR可以在用户说完之前就开始出文字TTS 可以提前合成固定话术例如“好的正在查询”这类缓冲语不要过度优化音频播放延迟那是最后一步。9.3 工具调用的容错设计工具函数返回的数据不一定是模型期望的格式。建议工具函数统一返回字符串并且对异常情况返回明确的错误描述例如“查询失败网络超时”。这样模型才能生成合理的用户提示而不是凭空白编一个答案。9.4 环境变量与密钥管理示例为了方便直接写在 config.py 里生产环境必须改成环境变量注入。还可以引入配置中心或密钥管理服务确保 API Key、数据库连接串等敏感信息不进入代码仓库。.gitignore 必须加入 .env 文件。9.5 日志与监控VoiceAgent 调试时最烦的是“不知道卡在哪一层”。建议从第一版就给每个模块加上统一的日志埋点。日志格式可以这样设计模块名、时间戳、请求内容摘要、响应内容摘要、耗时、状态码。后续接入指标监控时这些日志直接就能用。9.6 从示例到生产的演进路径第一步跑通文本模式第二步接入真实 ASR第三步接入真实 TTS第四步把“一问一答”改成“多轮会话”第五步引入打断和流式语音第六步对接业务 API让助手真正干活。每一步都建议单独发布一个可运行的版本不要一口气从 demo 跳到全功能产品。10. 总结与后续学习方向这篇文章做了一件很具体的事把一个“语音智能体项目”从概念拆成了三层模块并给出了可执行的最小代码骨架。你现在应该能回答这几个问题了ASR 层、LLM 层、TTS 层各自的职责边界是什么工具调用在主流程中是怎么工作的音频数据如何逐层流转为什么“级联式三明治架构”能减少开发过程中的混乱。下一步可以按自己的业务场景继续深入如果你的重点是助手能力可以去研究 Function Calling 的进阶用法和多工具调度如果重点是交互体验可以研究流式 ASR 和语音打断如果重点是系统性能可以研究并发会话、缓存策略和模型部署优化。代码骨架放在那里不会让你变强基于它跑通一个自己的场景才会。找一个你日常最想自动化的事情给它写一个工具函数让 VoiceAgent 帮你把它干了你会对这个架构的理解完全不一样。