STT-Agent-TTS：构建实时语音智能体的完整链路

在实际开发语音智能体时，很多人第一次听到 Voice Agent 都会以为它只是“语音版聊天机器人”：用户说一句，程序回一句。真正动手做才发现，问题远没有这么简单。语音信号怎么转成文本，转出来的文本带不带标点，大模型返回的内容是长句还是短句，合成语音时用哪个音色、要不要打断、能不能流式播放，每个环节都会影响最终体验。

本文围绕一个核心架构展开：STT-Agent-TTS。也就是把“语音转文字”“大模型语义处理”“文字转语音”三段能力串成一条完整链路，实现一个多模态语音智能体。文中会给出技术选型对比、最小可运行代码、关键参数说明、运行验证方法，以及一套从现象倒推原因的排查清单。无论你是刚接触语音应用开发，还是已经在做 RAG、Agent 项目想增加语音入口，这条链路都能直接借鉴。

1. 先理解 Voice Agent 为什么是“三条链路”而不是“一个模型”

1.1 一句话理解 Voice Agent

Voice Agent，可以理解为“能用自然语言对话的方式完成任务的语音智能体”。用户说话，系统识别意图，调用大模型推理，再把结果用语音反馈出来。和普通 Chatbot 的区别在于，输入输出都是语音，中间可以嵌入工具调用、知识库检索、状态记忆等能力。

从工程角度看，它并不是一个单一模型，而是由多个独立模块组合出来的系统。最常见的组合就是 STT（Speech-to-Text）、Agent（通常是大模型加上外部工具和记忆）、TTS（Text-to-Speech）三段式架构。三段各司其职，彼此通过标准文本接口通信。这种解耦设计的价值在于：某个环节升级时，不需要重写整个系统。今天用本地 Whisper，明天换成更快的流式识别服务，只改 STT 模块即可。

1.2 为什么很多人做出来的 Voice Agent 不好用

这里有一个很常见的认知误区：以为只要把 ASR、LLM、TTS 三个 API 串起来就能得到好产品。实际跑通一次会发现，结果往往是“识别出来了但答非所问”或者“回答内容很好但语音生硬还带读错字”。

问题通常出在以下三层：

• 文本质量问题。ASR 输出没有标点、没有分段、把同音字识别错，大模型拿到这种输入自然无法准确理解语义。
• 中间格式问题。大模型返回的是适合阅读的长段落，直接交给 TTS 会合成出没有停顿、节奏混乱的语音。需要做文本归一化和口语化改写。
• 交互体验问题。没有处理用户打断、没有做流式响应、缺少超时兜底，真实场景里根本聊不下去。

这三个问题恰好对应了 STT-Agent-TTS 链路中的三个关键接口。理解这条链路时，不能只看“怎么调 API”，更要把“文本在模块之间怎么流动”作为主线。

1.3 STT-Agent-TTS 一次完整调度里发生了什么

一次完整的语音对话，内部大致经历以下步骤：

1. 麦克风采集音频，做端点检测，判断用户是否说完。
2. 音频片段送入 STT 引擎，得到文本，必要时带上时间戳和置信度。
3. 文本进入 Agent 上下文管理器，拼接历史对话，形成完整 Prompt。
4. Agent 调用大模型，可能触发工具调用、知识库检索，最终得到回复文本。
5. 回复文本经过文本归一化，例如数字转中文、英文拼读改写、去除 Markdown 符号。
6. 归一化后的文本交给 TTS 引擎，合成语音并播放。

整个流程中，任何一步出错都会向下游传导。这也是为什么本文后面会反复强调：先保证每个模块单独可用，再串联成系统。

2. 技术选型：STT、Agent、TTS 分别用什么

选型的时候只有一个原则：学习环境先求跑通，生产环境再按延迟、成本、并发去优化。不要一开始就上分布式语音服务器，也不要因为某个模型“最新最热”就强行集成。

下面是一组适合从零到实战起步的技术组合，覆盖离线与在线两种思路。

2.1 STT 选型细节

学习阶段最推荐 faster-whisper。它内部使用 CTranslate2 推理，比原始 Whisper 快很多，且支持 CPU 运行。模型尺寸从 tiny 到 large-v3 都有，可以根据机器性能选择。

pip install faster-whisper

一个常见问题是：为什么不用openai-whisper原版？因为它依赖 PyTorch，显存和内存占用更大，推理速度也慢不少。faster-whisper 直接跑 CTranslate2 量化模型，CPU 上也能获得可用延迟。

2.2 Agent 选型细节

Agent 部分不一定非要复杂框架。如果只做语音对话，直接用大模型 API 也足够；如果后续要接数据库、查天气、订日程，再引入函数调用或 Agent 框架。

建议先使用 OpenAI 兼容接口，因为国内外的多种模型网关都提供该协议。代码里只需配置base_url和api_key，就能切换模型。

pip install openai

2.3 TTS 选型细节

学习阶段用 edge-tts 是性价比极高的选择。它不需要 API Key，也不需要 GPU，直接用微软 Edge 的语音合成接口，输出 mp3 格式，支持多种中文音色。

pip install edge-tts

生产环境如果对音色稳定性、并发、商业化协议有要求，再换云端 TTS 或自建 TTS 服务。

3. 环境准备与项目结构

建议在 Python 3.10 以上的虚拟环境中操作。创建一个干净的虚拟环境，避免和系统 Python 冲突。

python -m venv voice_agent_env source voice_agent_env/bin/activate

然后安装依赖：

pip install faster-whisper openai edge-tts pyaudio sounddevice numpy

pyaudio和sounddevice用于麦克风采集，numpy用于音频数据处理。如果只是先跑通链路，可以先用音频文件代替麦克风，减少音频设备兼容性问题。

项目目录建议如下：

voice_agent/ ├── main.py ├── stt_engine.py ├── agent_engine.py ├── tts_engine.py ├── config.py ├── audio/ │ └── test.wav └── requirements.txt

配置文件config.py集中管理所有可调参数，方便后续切换模型和调节行为。

# config.py STT_MODEL_SIZE = "small" STT_DEVICE = "cpu" STT_COMPUTE_TYPE = "int8" LLM_BASE_URL = "https://your-llm-gateway.example.com/v1" LLM_API_KEY = "your-api-key" LLM_MODEL = "your-model-name" LLM_TEMPERATURE = 0.7 LLM_MAX_TOKENS = 512 TTS_VOICE = "zh-CN-XiaoxiaoNeural" TTS_RATE = "+0%" TTS_VOLUME = "+0%"

这里把 STT 模型尺寸设为 small，是为了在 CPU 上获得速度和准确率的平衡。如果机器性能较好，可以换成 medium。

4. 核心实现：把 STT-Agent-TTS 三段串成最小闭环

4.1 STT 模块：从音频到文本

STT 模块负责将音频文件或麦克风输入转换为文本。下面是一个基于 faster-whisper 的实现，输入为本地音频文件路径，输出为识别文本。

# stt_engine.py from faster_whisper import WhisperModel class STTEngine: def __init__(self, model_size: str, device: str = "cpu", compute_type: str = "int8"): self.model = WhisperModel(model_size, device=device, compute_type=compute_type) def transcribe(self, audio_path: str) -> str: segments, info = self.model.transcribe(audio_path, language="zh", beam_size=5) text = "".join(segment.text for segment in segments) return text.strip()

关键参数说明：

在 CPU 环境下，如果解码很慢，优先把beam_size降到 1 或 3，而不是直接升级模型。

4.2 Agent 模块：从文本到回复

Agent 模块把用户文本变成回复文本。这里使用 OpenAI 兼容接口，内置一个简单的多轮记忆机制，后续可以做函数调用扩展。

# agent_engine.py from openai import OpenAI from config import LLM_BASE_URL, LLM_API_KEY, LLM_MODEL class AgentEngine: def __init__(self): self.client = OpenAI(base_url=LLM_BASE_URL, api_key=LLM_API_KEY) self.history = [] def chat(self, user_text: str) -> str: self.history.append({"role": "user", "content": user_text}) response = self.client.chat.completions.create( model=LLM_MODEL, messages=self.history, temperature=0.7, max_tokens=512, ) reply = response.choices[0].message.content self.history.append({"role": "assistant", "content": reply}) return reply

这里最关键的是history列表。如果没有它，每轮对话都是独立的，用户说“刚才那个问题再解释一下”时，模型根本不知道“刚才”指什么。

4.3 TTS 模块：从文本到语音

TTS 模块用 edge-tts 把回复文本变成语音文件并播放。为了减少首句延迟，可以将长文本先按标点切分，逐句合成，形成“边说边播”的效果。

# tts_engine.py import edge_tts import asyncio class TTSEngine: def __init__(self, voice: str = "zh-CN-XiaoxiaoNeural"): self.voice = voice async def synth_to_file(self, text: str, output_path: str) -> str: communicate = edge_tts.Communicate(text, self.voice) await communicate.save(output_path) return output_path async def synth_stream(self, text: str): communicate = edge_tts.Communicate(text, self.voice) async for chunk in communicate.stream(): if chunk["type"] == "audio": yield chunk["data"]

流式版与文件版的关键区别：文件版适合离线合成，流式版适合实时播放。下面的主流程会演示如何把两者结合。

4.4 主流程：一次完整对话

主流程把三个模块串起来。先处理音频文件，再调用 Agent，最后合成语音，并播放。

# main.py import asyncio import tempfile import edge_tts import pygame from stt_engine import STTEngine from agent_engine import AgentEngine from tts_engine import TTSEngine async def play_audio_bytes(audio_bytes: bytes): # 将音频字节写入临时文件，然后播放 with tempfile.NamedTemporaryFile(suffix=".mp3", delete=False) as f: f.write(audio_bytes) temp_path = f.name pygame.mixer.init() pygame.mixer.music.load(temp_path) pygame.mixer.music.play() while pygame.mixer.music.get_busy(): await asyncio.sleep(0.1) async def main(): stt = STTEngine(model_size="small") agent = AgentEngine() tts = TTSEngine() audio_path = "audio/test.wav" user_text = stt.transcribe(audio_path) print("识别结果:", user_text) reply_text = agent.chat(user_text) print("模型回复:", reply_text) async for audio_chunk in tts.synth_stream(reply_text): # 实际项目中会把 chunk 写入播放缓冲 pass output_file = "output.mp3" await tts.synth_to_file(reply_text, output_file) print("语音已保存:", output_file) if __name__ == "__main__": asyncio.run(main())

这个最小闭环里，synth_stream还没真正播放，只是演示数据流。生产项目会把音频块交给播放器队列，实现边说边播。

5. 多模态扩展：音频对齐、文本归一化与打断处理

5.1 让 ASR 输出更适合大模型

实际使用中，ASR 返回的文本往往没有标点。比如用户说“我想知道今天的天气怎么样”，如果识别成“我想知道今天的天气怎么样”丢失了语气停顿，大模型虽然能理解，但遇到多个意图混在一起时容易出现理解偏差。

推荐在 STT 输出后做一次文本清洗：

import re def clean_asr_text(text: str) -> str: # 去除多余的空白和识别噪声 text = re.sub(r"\s+", " ", text).strip() # 常见口语语气词可选择性保留或删除 text = text.replace("嗯", "").replace("那个", "") return text

注意不要过度清洗，有些语气词对大模型理解口语有帮助，尤其是用户表达犹豫和转折时。

5.2 让 Agent 输出更适合 TTS

大模型默认输出的文本包含 Markdown 符号、英文缩写、数字和换行，直接交给 TTS 会读成“星号 加粗 星号”之类的内容。

需要做一段面向口语合成的归一化：

def normalize_for_tts(text: str) -> str: # 去除 Markdown 语法 text = re.sub(r"[#*_>`\[\]]", "", text) # 连续换行合并为句号 text = re.sub(r"\n+", "。", text) # 简单处理数字，生产环境建议用专业库 text = text.replace("100%", "百分之百") return text

这里还能加一条约束：在 Agent 的系统提示词里要求模型“用适合朗读的短句回复，不要使用 Markdown”。

5.3 麦克风输入与端点检测

如果要实现真正的语音对话，需要从麦克风采集音频并检测用户是否说完。Python 里可以用sounddevice做实时采集，用能量阈值判断静音，实现简单的 VAD。

import sounddevice as sd import numpy as np SAMPLE_RATE = 16000 BLOCK_SIZE = 1600 SILENCE_THRESHOLD = 500 def record_until_silence(max_seconds: int = 10) -> np.ndarray: frames = [] silent_blocks = 0 max_silent_blocks = 5 def callback(indata, frames_count, time_info, status): volume = np.linalg.norm(indata) * 10 frames.append(indata.copy()) nonlocal silent_blocks if volume < SILENCE_THRESHOLD: silent_blocks += 1 else: silent_blocks = 0 with sd.InputStream(samplerate=SAMPLE_RATE, channels=1, blocksize=BLOCK_SIZE, callback=callback): while silent_blocks < max_silent_blocks: sd.sleep(100) return np.concatenate(frames)

这段代码只适合学习验证。生产环境的 VAD 应该用 WebRTC VAD、Silero VAD 或云端服务，因为它们对噪声、音乐、呼吸声的判断更稳定。

5.4 打断处理：语音智能体最核心的体验差异

真正好用的 Voice Agent 必须支持“用户随时插话”。实现思路是：

1. TTS 播放过程中，麦克风继续录音。
2. 一旦检测到用户语音能量超过阈值，立即停止 TTS 播放。
3. 把用户新说的话作为下一轮输入。

def stop_tts_if_interrupted(): # 伪代码，示意逻辑 while mixer.music.get_busy(): if is_user_speaking(): mixer.music.stop() break

打断处理是生产级 Voice Agent 的难点，涉及回声消除、双讲检测、状态机切换。学习阶段可以先从“手动回车打断”开始，再逐步迁移到自动 VAD。

6. 运行验证：如何判断链路是否正常

6.1 先测 STT 单点

准备一段包含明确数字和中文短句的音频，例如：“帮我查一下明天下午三点到上海的航班”。运行识别后，确认以下几个指标：

6.2 再测 Agent 单点

跳过 STT，直接给 Agent 引擎传一段文本，检查上下文连贯性。

agent = AgentEngine() print(agent.chat("我想订一张明天去北京的票")) print(agent.chat("几点出发比较好？"))

第二句话能理解“几点出发”指的是去北京的出发时间，说明多轮记忆生效。

6.3 最后测整链路

用测试音频跑完整流程，同时测三个指标：端到端耗时、回复语义正确性、合成语音可懂度。

time python main.py

如果端到端耗时超过 10 秒，优先看 STT 解码和 TTS 合成时间，而不是盲目换大模型。

6.4 预期输出示例

识别结果: 你好，请介绍一下你自己 模型回复: 你好，我是一个语音智能体，可以帮你完成信息查询、日程管理和知识问答。 语音已保存: output.mp3

出现这类输出说明 STT-Agent-TTS 三段的链路已经打通。

7. 常见问题排查：从现象倒推原因

下面这张表整理了这条链路里最容易遇到的问题，以及对应的排查路径。

7.1 排查顺序建议

遇到问题不要从中间开始查，严格按以下顺序推进：

1. 确认输入音频本身是否能正常播放。
2. 确认 STT 单点输出文本是否符合预期。
3. 确认 Agent 单点输入输出是否符合预期。
4. 确认 TTS 单点合成音频是否能播放。
5. 最后检查模块之间的数据格式是否被意外修改。

很多时候“链路不通”并不是某一个模块坏了，而是两个模块之间对文本格式的约定不一致。比如 Agent 返回了带\n的文本，TTS 没有归一化直接读出来，听起来就是一顿一顿的。

7.2 一个典型的定位过程

假设用户说“帮我查一下天气”，最后合成的语音是“帮我查一下 天气”。现象是停顿怪异。

第一步，打印 STT 输出，可能发现是“帮我查一下天气”，没有明显问题。

第二步，打印 Agent 输出，结果可能是“json\n{\"action\": \"weather\"}\n”，明显带了 Markdown JSON 格式。

第三步，打印 TTS 输入，发现没有做归一化，所以把反引号、换行都当成了文本。

解决方案：在 Agent 系统提示词里写死输出格式，同时在 TTS 前调用normalize_for_tts清理文本。

8. 生产化建议与学习路径

8.1 从“跑通”到“能上线”还要做什么

目前的最小闭环只能证明链路可行。想把它做成真实可用的 Voice Agent，还需要补齐以下能力：

• 流式 ASR：用户边说话边出文本，降低等待感。
• 流式 TTS：第一句话先播出来，后面边合成边播。
• 打断与轮次状态机：管理“用户说话”“模型回答”“被打断”三种状态。
• 工具调用：让 Agent 能查数据库、调接口，而不是只聊天。
• 会话持久化：把 history 存入 Redis 或数据库，支持多设备续聊。
• 安全与权限：语音指令涉及支付、删除、下单等操作时，必须二次确认。
• 可观测性：记录每轮 ASR 文本、Agent 回复、TTS 合成耗时，方便回溯。

其中，状态机设计是最容易被忽略的。没有它，系统容易出现“用户还在说话，TTS 已经开始播放”的抢话问题。

8.2 生产环境架构参考

麦克风 -> 回声消除 -> 流式ASR -> 意图路由 -> Agent | 播放器 <- 流式TTS <- 文本归一化 <- 回复文本 <---+

生产架构里，回声消除和打断检测通常放在音频层，Agent 层只处理文本，TTS 层只处理文本转语音。各层之间通过消息队列或 WebSocket 通信，而不是直接函数调用。

8.3 学习路径建议

如果刚接触 Voice Agent，建议按照下面顺序练习：

1. 用音频文件跑通 STT-Agent-TTS 离线链路。
2. 把音频文件换成麦克风实时采集。
3. 加入多轮记忆，测试上下文连贯性。
4. 加入一句话打断功能。
5. 接入工具调用，让 Agent 能查天气、查时间、查数据库。
6. 再做性能优化和监控告警。

每一步都单独验证，不要等到全部写完再调试。

8.4 最有价值的练习方向

对大多数学习者来说，与其追最新的语音大模型，不如先把“文本在三个模块之间如何清洗和传递”这件事吃透。因为 Voice Agent 的体验瓶颈通常不在单一模型，而在模块之间的文本流设计：ASR 输出脏文本怎么办，Agent 返回格式怎么约束，TTS 前怎么归一化，打断时正在处理的文本怎么丢弃。把这条链路打磨顺了，再换更好的模型，效果会立竿见影。