在语音助手类产品里我们常见的交互逻辑是“用户说一句——助手执行固定技能——语音回复结果”。这种模式虽然体验连贯但底层能力其实是写死的每个意图都要提前开发遇到没覆盖到的说法就答非所问。而 Agent智能体出现之后事情发生了变化它可以自主规划、调用工具、观察结果、继续修正本质上是一套“思考 行动”的循环。那如果让 Agent 直接听我们说话而不是在对话框里敲文字呢这就把“语音交互的自然性”和“Agent 的自主性”结合在了一起。本文就把这套链路的完整工程实现拆开来讲语音识别、大模型语义理解、工具调用、语音合成以及中间的容错与安全问题。无论你是刚开始接触 agent 开发的学习者还是想在本地搭建一个可用的语音控制 agent 项目这篇文章都会给你一条从零到一的可执行路径。废话不多说我们直接开始。1. 背景与核心概念1.1 什么是“用声音控制 Agent”传统意义上我们操作软件靠的是鼠标键盘操作手机靠的是触摸屏。随着大语言模型LLM能力增强Agent 成为新的“操作入口”用户给出一个目标Agent 把目标拆解成多个步骤调用外部工具比如搜索、查数据库、发请求、操作文件最终完成任务。“用声音控制 Agent”则是把 Agent 的输入方式从“键盘输入”换成“语音输入”同时把 Agent 的反馈从“文字回显”换成“语音播报”。它并不是一个独立的新技术而是几条已有技术的组合语音识别ASR把用户的语音转成文字。大语言模型LLM理解文字目标做出规划和决策。工具调用Function Calling / Tool Use让 Agent 能执行具体动作。语音合成TTS把 Agent 的回答或操作结果转成语音播放出来。一句话概括用声音控制 Agent 语音输入 Agent 决策循环 语音输出。1.2 Agent 解决的是什么问题要理解“用声音控制 Agent”的意义先要理解 Agent 解决了什么问题。以前的语音助手比如设置闹钟、播放音乐通常走的是“意图识别 槽位填充”的老路线。开发者要提前定义好“意图”列表和“参数”才能支撑一次对话。遇到模糊表达比如“明天早上帮我看看天气如果下雨就提醒我带伞”传统方案写起来很吃力因为这里既有条件判断又有隐式逻辑。Agent 不一样。LLM 能理解这种自然语言背后的逻辑链并自己规划出执行步骤明天早上是几点。天气从哪个接口获取。判断是否下雨。下雨则创建提醒不下雨则无需操作。Agent 在这种复杂指令、多步骤任务、跨工具协作场景中明显比传统对话系统灵活。1.3 常见应用场景声音控制 Agent 的场景可以覆盖很广举几个实际能落地的方向语音运维助手直接说“帮我查一下服务器的负载情况”Agent 调用监控 API返回结果并朗读。语音数据分析说“对比这两个月的订单量变化”Agent 查询数据库、生成结论。语音日程管理说“明天下午三点和王总开会提前半小时提醒我”Agent 创建日程和提醒。语音自动化测试说“打开登录页面输入错误密码检查报错提示”Agent 驱动自动化工具完成操作。这些场景的共同点是任务本身是需要多步决策的而语音让操作者可以解放双手在做饭、开车、运维看板前等环境下更方便地执行。2. 整体技术架构与链路设计在做任何 agent 开发之前我强烈建议先把整体链路画出来。声音控制 Agent 不是单一模型能搞定的它是一套管道式架构。2.1 完整链路一个最简单的语音 Agent 系统核心链路如下麦克风采集 - 语音识别(ASR) - 文本输入 - LLM Agent 规划与工具调用 - 工具执行结果返回 - 文本输出 - 语音合成(TTS) - 扬声器播放在这个链路中Agent 是大脑负责规划ASR 是耳朵负责把人话变成文本TTS 是嘴巴负责把结果说出来工具调用是手脚负责实际执行。2.2 模块职责划分把链路拆成模块每个模块只做一件事模块职责常见选型音频采集从麦克风读取音频流做分帧、降噪sounddevice、pyaudio语音识别ASR将音频转成文字Vosk、faster-whisper、FunASRAgent 核心理解意图、规划步骤、调用工具LangChain、自研循环、各类 Agent 框架大模型底座为 Agent 提供语义理解能力OpenAI 兼容接口、本地部署模型工具执行执行具体的函数或 API自研注册函数、requests、数据库客户端语音合成TTS将文字转成语音edge-tts、pyttsx3、CosyVoice2.3 同步执行还是异步循环在实现时我们需要确定 Agent 的调用方式。声音控制场景通常有两种模式单轮模式用户说一句话Agent 完整跑完“规划-执行-返回”再语音播报结果。多轮对话模式Agent 每执行一步都可能反问我补充信息形成持续对话。在实际项目中我建议第一版先做单轮模式把链路跑通后再扩展多轮。因为多轮语音对话会引入“打断检测”“语音活动检测VAD”“上下文维护”等额外复杂度容易让新手陷入细节无法专注核心逻辑。3. 环境准备与依赖安装3.1 环境要求本文示例使用 Python 实现版本建议 3.9 及以上。下面是我的推荐环境组合操作系统Windows 10/11、macOS、Ubuntu 均可Python 版本3.9推荐 3.10 或 3.11包管理工具pip 或 poetry麦克风任意可用麦克风笔记本自带或 USB 麦克风注意音频采集在不同操作系统下依赖不同如果在 Linux 服务器上运行需要额外处理 ALSA 和 PulseAudio 的权限问题。3.2 安装核心依赖创建一个新的虚拟环境python -m venv venv source venv/bin/activate # Windows 下是 venv\Scripts\activate然后安装以下依赖pip install sounddevice vosk openai edge-tts逐一说下这些库的用途sounddevice负责麦克风录音API 简单适合实时音频流处理。vosk离线语音识别工具包支持中文模型不需要联网即可完成 ASR对本地部署 agent 很友好。openaiOpenAI Python SDK不仅支持 OpenAI 官方接口也支持各种“OpenAI 兼容协议”的本地推理服务比如 vLLM、Ollama 等。edge-tts微软 Edge 的在线 TTS 库支持多种语言和音色生成语音自然度高无需申请额外 API Key。3.3 模型文件准备Vosk 需要单独下载模型文件。打开 Vosk 官网模型列表根据自己的需要下载中文或英文模型解压后放到项目models目录下。以中文模型为例mkdir models # 下载 vosk-model-small-cn-0.22 并解压到 models/ 目录如果希望提高识别准确率也可以选择更大的模型代价是识别延迟增加。在自己的项目里优先用小模型跑通流程再按需换大模型。4. 核心模块拆解与实现在写完整项目之前我们先单独拆解每个核心模块理解它们的输入输出和关键参数。4.1 语音采集模块语音采集是整个链路的起点。我们要从麦克风读取到音频数据并交给 ASR 识别。sounddevice提供的InputStream可以让我们以回调方式获取音频帧import sounddevice as sd import queue import numpy as np audio_queue queue.Queue() samplerate 16000 # Vosk 推荐使用 16kHz 采样率 block_size 8000 # 每次读取的采样点数约 0.5 秒 def audio_callback(indata, frames, time_info, status): if status: print(f录音状态异常: {status}) audio_queue.put(bytes(indata)) stream sd.InputStream( sampleratesamplerate, channels1, dtypeint16, blocksizeblock_size, callbackaudio_callback, ) stream.start()这里有几个关键点samplerate必须设置为 16000因为 Vosk 模型默认使用 16kHz 采样率。如果麦克风原始采样率不是 16kHzsounddevice会自动重采样。channels1表示单声道语音识别并不需要立体声。dtypeint16是 Vosk 能识别的音频格式16 位 PCM。4.2 语音识别模块Vosk 支持流式识别边录音边出结果。它的用法是创建KaldiRecognizer把音频数据喂进去通过PartialResult获取临时识别结果通过Result拿到最终结果。from vosk import Model, KaldiRecognizer model Model(models/vosk-model-small-cn-0.22) recognizer KaldiRecognizer(model, samplerate) recognizer.SetWords(False) def recognize_from_queue(duration5.0): 从音频队列中读取声音并识别duration 为最大等待秒数 import time start time.time() full_text while time.time() - start duration: try: data audio_queue.get(timeout0.5) except queue.Empty: continue if recognizer.AcceptWaveform(data): result_json recognizer.Result() # result_json 形如 {text: 你好世界} import json text json.loads(result_json).get(text, ) if text: full_text text else: # 部分识别结果可以在这里做实时交互提示 pass return full_text使用AcceptWaveform方法时它会返回布尔值True表示一段语音识别完成可以通过Result()取得稳定文本False表示还在识别过程中可以通过PartialResult()获取临时结果。在实现中我不会直接使用PartialResult来做最终判断而是等AcceptWaveform返回True后取最终结果这样更稳定。4.3 Agent 核心与工具调用Agent 的决策核心一般交给 LLM 完成。为了让 Agent 具备调用工具的能力我们需要给 LLM 提供“工具定义”并把工具的调用结果反馈给模型让它生成下一步动作。OpenAI 的 Function Calling工具调用是目前最通用的方案。下面是核心流程定义工具函数并给模型提供工具描述。模型根据用户指令决定调用哪个工具并给出参数。程序执行工具函数拿到结果。把结果返回给模型生成最终回复。在代码中我们先定义工具def get_weather(city: str) - str: 查询指定城市的天气情况 # 实际项目中这里替换为真实天气 API if 北京 in city: return 晴气温 25 摄氏度 elif 上海 in city: return 小雨气温 22 摄氏度 else: return f{city}的天气数据暂时无法获取请稍后再试 def create_reminder(time: str, content: str) - str: 创建一条提醒 return f已创建提醒在 {time} 提醒你 {content}然后通过 OpenAI 兼容接口调用模型from openai import OpenAI client OpenAI( base_urlhttp://localhost:8000/v1, # 本地推理服务地址 api_keyEMPTY, # 本地服务通常不需要真实 key ) tools [ { type: function, function: { name: get_weather, description: 查询天气, parameters: { type: object, properties: { city: {type: string, description: 城市名称} }, required: [city], }, }, }, { type: function, function: { name: create_reminder, description: 创建提醒, parameters: { type: object, properties: { time: {type: string, description: 提醒时间}, content: {type: string, description: 提醒内容}, }, required: [time, content], }, }, }, ] def run_agent(user_query: str, max_rounds: int 3): messages [{role: user, content: user_query}] for _ in range(max_rounds): response client.chat.completions.create( modelyour-model-name, messagesmessages, toolstools, tool_choiceauto, ) 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 tool_call.function.arguments # 这里需要将 arguments 从 JSON 字符串转为字典 import json args json.loads(arguments) if function_name get_weather: result get_weather(**args) elif function_name create_reminder: result create_reminder(**args) else: result f未知工具: {function_name} messages.append({ role: tool, tool_call_id: tool_call.id, content: result, }) else: # 没有工具调用说明模型已经生成最终答案 return message.content return Agent 执行轮次达到上限请简化指令后重试这段代码是整个语音 Agent 的核心。简单梳理一下messages数组用于维护多轮上下文其中新增了roletool的消息它是工具结果的载体。tool_choiceauto表示让模型自主决定是否调用工具。max_rounds限制了最大循环轮次防止 Agent 陷入死循环。4.4 语音合成模块Agent 返回文本结果后下一步是把文本转成语音。edge-tts使用起来非常简单import edge_tts import asyncio async def text_to_speech(text: str, output_file: str output.mp3): tts edge_tts.Communicate(text, voicezh-CN-XiaoxiaoNeural) await tts.save(output_file) return output_file if __name__ __main__: asyncio.run(text_to_speech(你好欢迎使用语音控制 Agent))edge-tts默认输出 MP3 格式。如果需要播放可以使用playsound库或者用系统命令播放# Linux mpg123 output.mp3 # macOS afplay output.mp3 # Windows start output.mp3在实际项目中我更推荐把 TTS 结果保存到一个临时文件然后异步播放这样不会阻塞 Agent 的主流程。5. 完整实战做一个本地语音控制 Agent在理解了各个模块之后我们把它们组装成一个可运行的完整项目。这个项目会支持两个工具查询天气、创建提醒。5.1 项目结构voice-agent/ ├── main.py # 主程序入口 ├── asr.py # 语音识别模块 ├── agent.py # Agent 核心逻辑 ├── tts.py # 语音合成模块 ├── audio.py # 麦克风采集模块 ├── requirements.txt # 依赖清单 └── models/ └── vosk-model-small-cn-0.22/5.2 创建 audio.pyimport queue import sounddevice as sd samplerate 16000 block_size 8000 audio_queue queue.Queue() def audio_callback(indata, frames, time_info, status): if status: print(f录音状态异常: {status}) audio_queue.put(bytes(indata)) def start_stream(): stream sd.InputStream( sampleratesamplerate, channels1, dtypeint16, blocksizeblock_size, callbackaudio_callback, ) stream.start() return stream5.3 创建 asr.pyimport json import queue from vosk import Model, KaldiRecognizer from audio import samplerate, audio_queue class SpeechRecognizer: def __init__(self, model_path: str): self.model Model(model_path) self.recognizer KaldiRecognizer(self.model, samplerate) def recognize(self, timeout: float 5.0) - str: import time start time.time() full_text while time.time() - start timeout: try: data audio_queue.get(timeout0.5) except queue.Empty: continue if self.recognizer.AcceptWaveform(data): result json.loads(self.recognizer.Result()) text result.get(text, ) if text: full_text text else: partial json.loads(self.recognizer.PartialResult()) partial_text partial.get(partial, ) if partial_text: print(f\r识别中: {partial_text}, end) print() return full_text.strip()5.4 创建 agent.pyimport json from openai import OpenAI def get_weather(city: str) - str: 查询天气 if 北京 in city: return 北京目前晴天气温 25 摄氏度空气质量良。 elif 上海 in city: return 上海目前小雨气温 22 摄氏度出门记得带伞。 else: return f暂时无法获取 {city} 的天气数据。 def create_reminder(time: str, content: str) - str: 创建提醒 return f已创建提醒{time} 提醒你 {content}。 TOOLS [ { type: function, function: { name: get_weather, description: 查询指定城市的天气情况, parameters: { type: object, properties: { city: {type: string, description: 城市名称例如北京、上海} }, required: [city], }, }, }, { type: function, function: { name: create_reminder, description: 创建一条提醒, parameters: { type: object, properties: { time: {type: string, description: 提醒时间例如明天上午9点}, content: {type: string, description: 提醒内容} }, required: [time, content], }, }, }, ] class VoiceAgent: def __init__(self, base_url: str, model: str): self.client OpenAI(base_urlbase_url, api_keyEMPTY) self.model model def execute_tool(self, function_name: str, arguments: str) - str: args json.loads(arguments) if function_name get_weather: return get_weather(**args) elif function_name create_reminder: return create_reminder(**args) return f未知工具: {function_name} def run(self, user_query: str, max_rounds: int 3) - str: messages [{role: user, content: user_query}] for _ in range(max_rounds): response self.client.chat.completions.create( modelself.model, messagesmessages, toolsTOOLS, tool_choiceauto, ) message response.choices[0].message if message.tool_calls: messages.append(message) for tool_call in message.tool_calls: result self.execute_tool( tool_call.function.name, tool_call.function.arguments, ) messages.append({ role: tool, tool_call_id: tool_call.id, content: result, }) else: return message.content or return Agent 执行轮次达到上限请简化指令后重试。5.5 创建 tts.pyimport asyncio import edge_tts async def _save_audio(text: str, output_file: str): tts edge_tts.Communicate(text, voicezh-CN-XiaoxiaoNeural) await tts.save(output_file) def speak(text: str, output_file: str response.mp3) - str: if not text.strip(): return asyncio.run(_save_audio(text, output_file)) return output_file5.6 创建 main.pyimport os from audio import start_stream from asr import SpeechRecognizer from agent import VoiceAgent from tts import speak MODEL_PATH models/vosk-model-small-cn-0.22 AGENT_BASE_URL http://localhost:8000/v1 AGENT_MODEL your-model-name # 替换为你实际使用的模型名称 def play_audio(file_path: str): # 按系统差异选择播放命令 if os.name nt: os.system(fstart {file_path}) elif os.uname().sysname Darwin: os.system(fafplay {file_path}) else: os.system(fmpg123 {file_path}) def main(): print(正在初始化语音 Agent...) stream start_stream() recognizer SpeechRecognizer(MODEL_PATH) agent VoiceAgent(base_urlAGENT_BASE_URL, modelAGENT_MODEL) print(系统就绪请说话。按 CtrlC 退出。) print(示例指令帮我查一下北京的天气 / 明天上午九点提醒我开会) try: while True: user_text recognizer.recognize(timeout8.0) if not user_text: print(没有识别到语音继续等待...) continue print(f\n识别结果: {user_text}) print(Agent 正在思考并执行...) response_text agent.run(user_text) print(fAgent 回复: {response_text}) print(正在合成语音...) audio_file speak(response_text) play_audio(audio_file) except KeyboardInterrupt: print(\n程序退出) finally: stream.stop() stream.close() if __name__ __main__: main()5.7 运行与验证启动本地推理服务之后运行主程序python main.py程序会先加载 Vosk 模型然后进入监听状态。这时候你对着麦克风说“帮我查一下北京的天气”系统会经过如下流程Vosk 把语音识别成文字。Agent 收到文本调用get_weather工具。工具返回北京天气结果。LLM 将结果组织成自然语言回复。edge-tts 合成语音并播放。预期输出类似系统就绪请说话。按 CtrlC 退出。 示例指令帮我查一下北京的天气 / 明天上午九点提醒我开会 识别结果: 帮我查一下北京的天气 Agent 正在思考并执行... Agent 回复: 根据查询结果北京目前晴天气温 25 摄氏度空气质量良。 正在合成语音...到这里一个“用声音控制 agent”的最小闭环就完成了。6. 常见问题与排查思路在实际运行中新手最容易遇到下面几类问题。我整理成了一张排查表问题现象常见原因解决思路程序启动后听不到声音麦克风权限未开启或默认输入设备不对检查系统麦克风权限使用sounddevice.query_devices()查看设备列表并指定输入设备识别结果一直是空采样率不匹配或模型路径错误确认samplerate16000确认 Vosk 模型解压路径正确识别准确率低使用了小模型或环境噪音较大更换大模型添加降噪处理靠近麦克风说话Agent 不调用工具直接给答案模型不支持 function calling或工具定义格式不兼容确认使用支持工具调用的模型检查工具定义是否符合 OpenAI 规范工具参数解析报错LLM 返回的arguments不是合法 JSON在解析时加入异常处理必要时提示模型重新生成TTS 播放没有声音系统缺少 mp3 播放器或播放命令不对手动执行mpg123 response.mp3或改用playsound播放Agent 执行超时LLM 接口地址不通或网络延迟过高检查base_url是否正确测试模型接口连通性循环执行多次没有结果模型理解指令有误工具定义不够清晰优化工具描述信息增加max_rounds控制这里重点说一个常见错误工具参数是字符串不是字典。OpenAI 的 Function Calling 返回的tool_call.function.arguments是 JSON 字符串必须先json.loads()再传入函数。很多初学者直接拿字符串去调用函数导致参数解包失败。另一个常见问题是Vosk 模型加载速度慢。大模型文件在 CPU 上加载可能要十几秒到几十秒这并不是死机耐心等待初始化完成即可。如果觉得慢可以在程序启动时加入模型加载进度提示。7. 安全边界与工程最佳实践声音控制 Agent 在带来便利的同时也带来了新的安全挑战。语音本身就是一种敏感的个人数据因此这一部分必须认真对待。7.1 语音数据安全使用麦克风意味着持续采集环境声音。在开发和生产环境中都应注意以下几点本地优先原则尽可能在本地完成 ASR 识别避免把原始音频送到云端。数据最小化录音完成后立即释放音频资源不长期缓存。日志脱敏不要将完整的用户语音文本直接打进日志尤其是涉及姓名、地址、账号等信息时先做脱敏处理。权限最小化Agent 能调用的工具必须最小化。例如一个只查天气的 Agent不应该具备操作数据库的权限。7.2 Agent 工具调用的风险控制Agent 本质上是一个能自主调用工具的智能体这就意味着它可能执行开发者没有预料到的操作。对于工具调用我建议加以下几层防护白名单机制只允许调用在TOOLS列表中注册的工具不在列表内的函数一律不允许执行。参数校验工具函数内部必须对参数做类型和取值范围校验不要直接信任 LLM 生成的参数。人工确认机制对于删除、更新、付款等敏感操作加上二次确认环节。执行审计记录每次工具调用的时间、参数、结果便于事后排查。下面是一个带参数校验的工具函数示例def create_reminder(time: str, content: str) - str: if not time or not content: return 提醒时间和内容不能为空请重新输入。 if len(content) 50: return 提醒内容过长请控制在 50 字以内。 return f已创建提醒{time} 提醒你 {content}。7.3 唤醒词与误触发处理在真实环境中持续监听麦克风会带来大量误触发。比如周围人聊天时Agent 可能突然“插嘴”。解决这个问题有两个思路加入唤醒词比如“小智小智”检测到唤醒词后才开始正式识别。使用语音活动检测VAD只有检测到超过一定音量阈值的语音才开始识别。第一版项目建议先加一个简单的音量阈值判断复杂度低效果立竿见影def is_speech(data: bytes, threshold: int 800) - bool: import numpy as np audio_array np.frombuffer(data, dtypenp.int16) volume np.abs(audio_array).mean() return volume threshold7.4 模型选型与响应速度在 agent 开发中模型选型直接影响体验。语音 Agent 对响应延迟要求比普通聊天更高因为用户等着听结果。建议ASR 使用本地小模型保证首包延迟低。LLM 使用支持 function calling 且推理速度较快的模型。如果本地部署优先考虑量化版本。TTS 在首次使用时可以预生成多句常用提示语缓存减少接口调用。对于复杂任务不要试图在一次交互中完成而是拆成多轮问答逐段反馈提升用户体验。7.5 可维护性把工具注册做成配置当工具越来越多时把工具定义和函数写死在代码里会非常难维护。建议把工具注册做成一张配置表统一管理。# tools_registry.py TOOL_FUNCTIONS { get_weather: get_weather, create_reminder: create_reminder, } TOOL_DEFINITIONS [ { type: function, function: { name: get_weather, description: 查询指定城市的天气情况, parameters: { type: object, properties: { city: {type: string, description: 城市名称} }, required: [city], }, }, }, # ... 更多工具定义 ]这样新增工具时只需要添加函数和定义不用改动 Agent 核心逻辑。8. 扩展与进阶方向如果你已经跑通上面这个最小系统可以继续往这几个方向扩展8.1 支持多轮对话当前版本每次识别后都从头开始一轮 Agent 循环没有保留历史记忆。你可以引入一个memory列表把每轮的识别文本和 Agent 回复追加进去在下一次调用时作为上下文传入。8.2 加入打断检测在 Agent 执行较长时间时用户可能会说“停一下”“换个问题”。通过持续监听麦克风在 Agent 执行阶段检测到新的语音输入时可以中断当前任务这能明显提升交互体验。8.3 接入更多工具天气、提醒只是两个演示工具。你可以继续接入搜索、数据库查询、企业内部 API 等把声音控制 Agent 变成真正的“语音操作入口”。8.4 可视化调试界面为 Agent 增加一个简单的 Web 界面展示“识别文本、工具调用过程、执行结果”对调试非常有帮助。用 Flask 或 FastAPI 包一层接口即可SSE 推送实时状态会更顺手。9. 结语声音控制 Agent 的核心并不是某一项黑科技而是把语音识别、大模型决策、工具调用、语音合成这四件事有机地串联起来。它的工程难度在于模块之间的协调和容错而不在于某个单独环节的实现。如果你是第一次接触 agent 开发建议不要一上来就追求复杂的多轮对话和联网搜索。先把本地链路完整跑通理解 ASR 的流式输出、Function Calling 的循环调用机制、TTS 的异步播放逻辑再逐步叠加功能。把最小闭环做扎实远比堆砌花哨的 Agent 框架更重要。希望这篇文章能帮你用声音真正“驾驭”自己的 Agent。遇到具体问题欢迎在评论区交流。