用声音控制Agent：从语音识别到工具调用的完整工程实现

在语音助手类产品里，我们常见的交互逻辑是“用户说一句——助手执行固定技能——语音回复结果”。这种模式虽然体验连贯，但底层能力其实是写死的：每个意图都要提前开发，遇到没覆盖到的说法就答非所问。而 Agent（智能体）出现之后，事情发生了变化，它可以自主规划、调用工具、观察结果、继续修正，本质上是一套“思考 + 行动”的循环。

那如果让 Agent 直接听我们说话，而不是在对话框里敲文字呢？这就把“语音交互的自然性”和“Agent 的自主性”结合在了一起。本文就把这套链路的完整工程实现拆开来讲：语音识别、大模型语义理解、工具调用、语音合成，以及中间的容错与安全问题。

无论你是刚开始接触 agent 开发的学习者，还是想在本地搭建一个可用的语音控制 agent 项目，这篇文章都会给你一条从零到一的可执行路径。废话不多说，我们直接开始。

1. 背景与核心概念

1.1 什么是“用声音控制 Agent”

传统意义上，我们操作软件靠的是鼠标键盘，操作手机靠的是触摸屏。随着大语言模型（LLM）能力增强，Agent 成为新的“操作入口”：用户给出一个目标，Agent 把目标拆解成多个步骤，调用外部工具（比如搜索、查数据库、发请求、操作文件），最终完成任务。

“用声音控制 Agent”则是把 Agent 的输入方式从“键盘输入”换成“语音输入”，同时把 Agent 的反馈从“文字回显”换成“语音播报”。它并不是一个独立的新技术，而是几条已有技术的组合：

1. 语音识别（ASR）：把用户的语音转成文字。
2. 大语言模型（LLM）：理解文字目标，做出规划和决策。
3. 工具调用（Function Calling / Tool Use）：让 Agent 能执行具体动作。
4. 语音合成（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 模块职责划分

把链路拆成模块，每个模块只做一件事：

2.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 很友好。
• openai：OpenAI 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( samplerate=samplerate, channels=1, dtype="int16", blocksize=block_size, callback=audio_callback, ) stream.start()

这里有几个关键点：

• samplerate必须设置为 16000，因为 Vosk 模型默认使用 16kHz 采样率。如果麦克风原始采样率不是 16kHz，sounddevice会自动重采样。
• channels=1表示单声道，语音识别并不需要立体声。
• dtype="int16"是 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(duration=5.0): """从音频队列中读取声音并识别，duration 为最大等待秒数""" import time start = time.time() full_text = "" while time.time() - start < duration: try: data = audio_queue.get(timeout=0.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（工具调用）是目前最通用的方案。下面是核心流程：

1. 定义工具函数，并给模型提供工具描述。
2. 模型根据用户指令决定调用哪个工具，并给出参数。
3. 程序执行工具函数，拿到结果。
4. 把结果返回给模型，生成最终回复。

在代码中，我们先定义工具：

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_url="http://localhost:8000/v1", # 本地推理服务地址 api_key="EMPTY", # 本地服务通常不需要真实 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( model="your-model-name", messages=messages, tools=tools, tool_choice="auto", ) 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数组用于维护多轮上下文，其中新增了role="tool"的消息，它是工具结果的载体。
• tool_choice="auto"表示让模型自主决定是否调用工具。
• 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, voice="zh-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.py

import 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( samplerate=samplerate, channels=1, dtype="int16", blocksize=block_size, callback=audio_callback, ) stream.start() return stream

5.3 创建 asr.py

import 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(timeout=0.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.py

import 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_url=base_url, api_key="EMPTY") 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( model=self.model, messages=messages, tools=TOOLS, tool_choice="auto", ) 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.py

import asyncio import edge_tts async def _save_audio(text: str, output_file: str): tts = edge_tts.Communicate(text, voice="zh-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_file

5.6 创建 main.py

import 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(f"start {file_path}") elif os.uname().sysname == "Darwin": os.system(f"afplay {file_path}") else: os.system(f"mpg123 {file_path}") def main(): print("正在初始化语音 Agent...") stream = start_stream() recognizer = SpeechRecognizer(MODEL_PATH) agent = VoiceAgent(base_url=AGENT_BASE_URL, model=AGENT_MODEL) print("系统就绪，请说话。按 Ctrl+C 退出。") print("示例指令：帮我查一下北京的天气 / 明天上午九点提醒我开会") try: while True: user_text = recognizer.recognize(timeout=8.0) if not user_text: print("没有识别到语音，继续等待...") continue print(f"\n识别结果: {user_text}") print("Agent 正在思考并执行...") response_text = agent.run(user_text) print(f"Agent 回复: {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 模型，然后进入监听状态。这时候你对着麦克风说“帮我查一下北京的天气”，系统会经过如下流程：

1. Vosk 把语音识别成文字。
2. Agent 收到文本，调用get_weather工具。
3. 工具返回北京天气结果。
4. LLM 将结果组织成自然语言回复。
5. edge-tts 合成语音并播放。

预期输出类似：

系统就绪，请说话。按 Ctrl+C 退出。 示例指令：帮我查一下北京的天气 / 明天上午九点提醒我开会 识别结果: 帮我查一下北京的天气 Agent 正在思考并执行... Agent 回复: 根据查询结果，北京目前晴天，气温 25 摄氏度，空气质量良。 正在合成语音...

到这里，一个“用声音控制 agent”的最小闭环就完成了。

6. 常见问题与排查思路

在实际运行中，新手最容易遇到下面几类问题。我整理成了一张排查表：

这里重点说一个常见错误：工具参数是字符串，不是字典。

OpenAI 的 Function Calling 返回的tool_call.function.arguments是 JSON 字符串，必须先json.loads()再传入函数。很多初学者直接拿字符串去调用函数，导致参数解包失败。

另一个常见问题是Vosk 模型加载速度慢。大模型文件在 CPU 上加载可能要十几秒到几十秒，这并不是死机，耐心等待初始化完成即可。如果觉得慢，可以在程序启动时加入模型加载进度提示。

7. 安全边界与工程最佳实践

声音控制 Agent 在带来便利的同时，也带来了新的安全挑战。语音本身就是一种敏感的个人数据，因此这一部分必须认真对待。

7.1 语音数据安全

使用麦克风意味着持续采集环境声音。在开发和生产环境中都应注意以下几点：

• 本地优先原则：尽可能在本地完成 ASR 识别，避免把原始音频送到云端。
• 数据最小化：录音完成后立即释放音频资源，不长期缓存。
• 日志脱敏：不要将完整的用户语音文本直接打进日志，尤其是涉及姓名、地址、账号等信息时，先做脱敏处理。
• 权限最小化：Agent 能调用的工具必须最小化。例如一个只查天气的 Agent，不应该具备操作数据库的权限。

7.2 Agent 工具调用的风险控制

Agent 本质上是一个能自主调用工具的智能体，这就意味着它可能执行开发者没有预料到的操作。对于工具调用，我建议加以下几层防护：

1. 白名单机制：只允许调用在TOOLS列表中注册的工具，不在列表内的函数一律不允许执行。
2. 参数校验：工具函数内部必须对参数做类型和取值范围校验，不要直接信任 LLM 生成的参数。
3. 人工确认机制：对于删除、更新、付款等敏感操作，加上二次确认环节。
4. 执行审计：记录每次工具调用的时间、参数、结果，便于事后排查。

下面是一个带参数校验的工具函数示例：

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, dtype=np.int16) volume = np.abs(audio_array).mean() return volume > threshold

7.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。遇到具体问题，欢迎在评论区交流。