Voxlocal：一个用 Rust 编写的极简语音代理

# Voxlocal：用 Rust 编写的极简语音代理 来源：https://samkhawase.com/blog/voxlocal-minimal-voice-agent/ > "*man muss* immer umkehren" – Carl Gustav Jacob Jacobi > "当有疑问时，做减法" 我研究语音代理已有一段时间，它们是一项迷人的技术。与语音代理对话并完成工作，这种体验十分奇妙。你是否好奇过语音代理在幕后是如何工作的？这篇博客文章是我尝试理解其内部运作机制的记录。 ### 什么是语音代理？ AI 代理是一个自主软件系统，它利用 AI 模型进行推理、规划步骤并执行任务以达成特定目标。语音代理是它的一种变体——一个能够进行自然的类人语音对话来完成任务的系统。简而言之，语音代理的工作流程是：听、想、做、说。 ### 鸟瞰全局 一张展示电话通话流程的图表，涵盖电话运营商、语音识别、带有工具的 LLM 和语音合成 这个图表信息量很大。出于对工作原理的好奇，我用 Rust 创建了一个精简项目，以深入了解幕后发生的细节。在深入探讨语音代理部分之前，让我们先了解一些传统组件。 #### 电话运营商 电话网络是一个复杂的话题，远超本文讨论范围。在我们的场景中，电话运营商模块负责连接传统公共电话网络（PSTN/SIP）与云端 AI 应用。它承担着处理呼叫生命周期、双向流式传输、转码以及保持低延迟等复杂任务。该模块的主要任务之一是针对丢包或延迟数据包进行抖动缓冲，以及音频上采样以保证语音转文字的准确性。这一模块致力于将网络延迟控制在阈值以下。 现代电话通信是一项工程奇迹，与现代互联网密切相关。了解完这些之后，让我们看看我们的"杰作"。 ### 介绍 voxlocal：先锋式语音代理 voxlocal 是一个完全本地化、低延迟的语音代理，适用于 macOS。它是一个用 Rust 编写的 CLI 工具，用户可以与语音代理对话并获得响应。出于实验精神，我使用了最基础的组件和微型模型，尽可能接近其"魔力"本质。**请勿在生产环境中使用！** 它的应用范围非常狭窄：通过英语进行汽车服务场景的对话。我还使用了一些现成组件，如 Whisper 和 Piper，而不是重复造轮子。为了保持流程的可检查性和透明度，`voxlocal` 故意避免使用复杂的异步流式传输或 VAD（语音活动检测）。我可能会在后续版本中深入探讨这些内容。 探索源代码：开源项目 · Rust · macOS samkhawase / voxlocal 一个极简的、完全本地化的语音代理，旨在让对话的每个步骤都易于检查。 [查看仓库](https://github.com/samkhawase/voxlocal) 让我们一步步来探索 ### 1. 当你对语音代理说话时，会发生什么？ 当你与语音代理对话或给它一段音频样本时，它会将其分解为多个更小的任务来完成工作。它从音频中识别文字，通过去除"嗯"、"啊"等语气词来清理文本。它利用这些文字查找相关信息并选择相应的操作。最后，它将响应转换回语音。 在 `voxlocal` 中，首先捕获麦克风音频并传递给语音识别模块： ``` let capture = record_until(RECORD_SECS)?; let transcript = stt.transcribe(&capture.samples)?; ``` 然后，转录文本进入共享的回合处理流程。在流程中，代理会规范化文本、搜索匹配的上下文、将请求路由到相关工具，最后准备回复： ``` run_turn( &norm, &rag, &embedder, &mut llm, tts.as_mut(), transcript, &mut budget, )?; ``` 每个阶段使用不同的技术来执行其操作： - 语音识别确定音频中的词语 - 检索通过数学方法（[余弦相似度](https://samkhawase.com/blog/voxlocal-minimal-voice-agent/#user-content-fn-1)）以数值形式比较文本块 - LLM 预测一个动作 - 普通的 Rust 代码执行工具调用和动作执行 - 语音合成为最终回复生成音频 不过这些描述太过枯燥和学术化。让我们通过一个实际例子来层层解构——用户向代理提问："换机油多少钱？" ### 2. 从声波到文字 令我惊讶的是，麦克风通过持续测量气压来捕获声音，并将读数转换为音频采样。`voxlocal` 然后将捕获的音频转换为单声道（[注2](https://samkhawase.com/blog/voxlocal-minimal-voice-agent/#user-content-fn-2)），采样率为 16,000 样本每秒，然后传递给 Whisper 进行进一步处理。 Whisper 是我们训练好的神经网络，它被设计来寻找音频中类似语音的模式。然后它估计这些声音是否构成一段词序列。Whisper 的创建者为这一任务对其进行了训练和微调，我们的程序只是搭便车，调用模型（[注3](https://samkhawase.com/blog/voxlocal-minimal-voice-agent/#user-content-fn-3)）。 ``` let mut params = whisper_rs::FullParams::new( whisper_rs::SamplingStrategy::Greedy { best_of: 1 } ); params.set_language(Some("en")); params.set_temperature(0.0); state.full(params, audio) .map_err(|e| anyhow!("whisper decode failed: {e}"))?; ``` 在这个例子中，我们使用贪心解码而非采样来选择最可能的下一个 token。温度设为 `0` 意味着它只会返回最高似然度的首个匹配，比如 `oil` 或 `change`。我们可以这样做是因为我们的上下文非常有限，且我们了解我们的对话内容。 当我们从 Whisper 获得分段响应后，`voxlocal` 的任务是将它们拼接成一个完整的转录文本。 ``` let mut out = String::new(); for i in 0..state.full_n_segments() { if let Some(segment) = state.get_segment(i) { if let Ok(text) = segment.to_str_lossy() { out.push_str(&text); } } } let transcript = out.trim().to_string(); ``` 在我们的例子中，输出结果为： ``` "What does an oil change cost?" ``` 在这个阶段，输出是基于信号做出的估计，可能与实际所说的话略有不同。由于噪声、口音、节奏、音色等各种因素，模拟到数字的转写总是棘手的。因此我们需要额外的阶段来清理它，使其在后续处理中更加一致。 ### 3. 清理识别出的语音 人类说话时并不总是保持一致，常常添加"嗯"这样的填充词、尴尬的停顿，或对数字的特殊发音（比如把"Nine Thirty"说成"noyynTherTee"）。voxlocal 在将文本输入下一阶段之前，应用简单且可预测的规则来清理文本。 例如，`voxlocal` 中的规范化器会移除常见的填充短语、去除多余空格并合并分隔符： ``` s = self.thousands.replace_all(&s, "$1$2").into_owned(); s = self.fillers.replace_all(&s, " ").into_owned(); s = self.space.replace_all(&s, " ").into_owned(); ``` 当用户说"Um yeah, what does an oil change cost?"时，它可能会变成"What does an oil change cost"。程序只是使用已知的现有转换规则，让下一阶段来解释其含义。这就是简单的正则表达式，在每一轮对话中复用，符合我们极简的做法。 ### 4. 将词语转化为向量 既然 `voxlocal` 已经获得了清理过的文本，它需要从中找出用户的意图。用户说"What does an oil change cost"和"How much for an oil service"具有相同的意图。那么我们如何确定意图呢？ 答案是使用嵌入模型（embedding model），它将文本转换为一组数字（你可以说 `向量` 听起来更酷）。用这个技巧，含义相关的文本往往具有指向相同方向的向量。 我们在 `voxlocal` 中使用 MiniLM 为规范化后的查询生成一个向量。 ``` pub fn embed_one(&self, text: &str) -> Result<Vec<f32>> { Ok(self.embed_batch(&[text])?.remove(0)) } ``` 在 `embed_batch` 函数内部，MiniLM 为每个 token 生成一个表示。然后代码通过掩码均值池化（masked mean pooling）将这些 token 表示组合成一个句子向量。简单来说：它对真实 token 求平均值，并忽略为使批次长度一致而添加的填充。 ``` let mask = attention.to_dtype(DType::F32)?.unsqueeze(2)?; let summed = hidden.broadcast_mul(&mask)?.sum(1)?; let counts = mask.sum(1)?; let pooled = summed.broadcast_div(&counts)?.contiguous()?; ``` 模型如何确定哪些模式对含义和意图重要？答案在于模型已学习的权重中，这些权重是预训练好的，可直接供 `voxlocal` 使用。`voxlocal` 的池化代码将 token 级别的输出归约为整个句子的单个向量。这是纯粹的模型输出表示；Rust 代码并不定义每个坐标的含义。 最后，向量经过 L2 归一化。 ``` let norm = row.iter().map(|v| v * v).sum::<f32>().sqrt(); let unit_vector = row.iter().map(|v| v / norm).collect::<Vec<f32>>(); ``` 简单来说，归一化确保每个非零向量的长度为 1。这是下一阶段所需的，下一阶段通过简单的点积来比较方向。 根据数学原理，即使措辞不同，相似的含义也会指向相似的方向。是的，如果你好奇的话，它确实在字面意义和几何意义上计算余弦值。我能看到我的高中数学老师 Mr. Bhalérao 正带着得意的笑容看着我。 ### 5. 查找相关知识 现在有了单位长度的查询向量，`voxlocal` 可以将其与现有的服务文档进行比较。`voxlocal` 将查询向量与每个文档向量进行比较，找到有用的上下文。向量的点积（即余弦相似度）被计算并比较以衡量相似度。更好的匹配意味着向量指向更相似的方向。 我们的搜索比较了得分并丢弃低于参考阈值的结果。剩余的匹配项被排序后，返回请求数量。 ``` let mut scored: Vec<(&Doc, f32)> = self.docs .iter() .map(|doc| { let sim: f32 = doc.embedding.iter() .zip(query.iter()) .map(|(a, b)| a * b) .sum(); (doc, sim) }) .filter(|(_, sim)| *sim >= RAG_MIN_SIM) .collect(); scored.sort_by(|a, b| { b.1.partial_cmp(&a.1).unwrap_or(std::cmp::Ordering::Equal) }); scored.truncate(top_k); ``` 我们需要阈值，因为我们希望排除排名较高但不相关的文档。如果这变得太数学化了，让我用简单的例子解释： 想象我们的查询向量是 `[1.0, 0.0]`，一个指向相似方向的文档向量为 `[0.8, 0.6]`。它们的点积是 `1.0×0.8 + 0.0×0.6 = 0.8`。一个主要指向另一个方向的向量（不相关的术语）`[0.1, 0.995]`，得分约为 `0.1`。哪个点积更接近 1.0，哪个就是赢家。 查询匹配得到 $\text{Similarity} = (1.0 \times 0.8) + (0.0 \times 0.6) = 0.8$。 文档还必须得分至少 `0.35` 才能被纳入。示例中的换机油查询得分为 `0.816`，而像"hi"这样的不相关问候语在这个语料库上的得分约为 `0.10–0.15`。这帮助我们排除不相关的术语。 这是一个最小化的 RAG 示例，搜索最多返回 2 个文档。流程为语言模型准备这些文档，并将上下文限制在 320 个字符以内。语言模型基于这些证据决定执行什么响应和动作。 ### 6. 语言模型如何选择动作 现在我们有了用户的问题和检索到的上下文，可以进入下一阶段。在 `voxlocal` 中，我们的语言模型（SmolLM2）被提示决定并返回结构化的工具调用，而非自由格式的回答。 例如，对于"What does an oil change cost?"，期望的输出是一个工具名称及其参数： ``` { "tool": "check_price", "args": { "service": "oil change" } } ``` 模型逐个 token 产生这个输出，一个 token 可以是一个完整的单词、单词的一部分、标点符号或一个 JSON 片段。在每一步中，模型根据前面的提示和 token 预测下一个最可能的 token。Rust 代码通过 Candle 运行这个生成过程，它使用包含已学习信息的模型文件。 路由器限制可以生成的 token 数量，然后将结果解析为 `ToolCall`： ``` let (raw, truncated) = self.generate(&prompt, max_new)?; let call = parse_tool_call(&raw) .or_else(|| truncated.then(|| repair_truncated(&raw)).flatten()) .ok_or_else(|| anyhow!("model did not emit valid tool JSON: {raw:?}"))?; ``` 上面的代码片段展示了生成与解释之间的边界。模型返回一个文本值，程序尝试将该文本解释为规范化的动作。模型的输出不会被 `voxlocal` 自动视为可执行命令。 小型语言模型可能会遇到问题，在 JSON 中途停止或产生格式错误的输出。当 `voxlocal` 检测到生成被截断时，会尝试修复以解决此问题。如果解析仍然失败，`voxlocal` 的路由器组件会在该轮返回错误。下一节解释了如何验证和执行有效的工具调用。 ### 7. 从模型输出到执行动作 正如我们所见，模型生成的工具调用在这个阶段仍然是文本。为了执行指定的动作，`voxlocal` 必须解析它并决定该工具被允许做什么。 `voxlocal` 解析模型的 JSON，然后将其提议的工具名称映射到受支持的工具集上。如果检索结果中有可用的服务信息，它也会将其纳入： ``` let call = parse_tool_call(&raw) .or_else(|| truncated.then(|| repair_truncated(&raw)).flatten()) .ok_or_else(|| anyhow!("model did not emit valid tool JSON: {raw:?}"))?; let mut call = call; call.tool = snap_tool(&call.tool, query); if let Some(hit) = top_hit { call.args.insert( "service".to_string(), serde_json::Value::String(hit.title.clone()), ); } ``` 我们在这里看到的是统计预测与程序行为之间的有趣边界。Rust 程序解析模型的建议，并在执行前应用规则。例如，路由器从当前查询中提取时间，而不是盲目信任模型可能从示例中复制的时间。 执行器随后处理我们为这个示例设置的支持的操作： ``` match call.tool.as_str() { "book_appointment" => { /* 准备预约回复 */ } "check_price" => { /* 查询服务价格 */ } other => format!("Unknown tool {other}."), } ``` 在这个项目中，由于是模拟实现，实际上什么也没有执行。在真实应用中，程序可能会调用日历或 API。 ### 8. 将回复转换回声音 在这个阶段，我们已经为用户准备好了答案，是时候将文本回复转换回语音了。Piper 将词语转换为音频表示，然后使用其训练好的语音模型生成最终音频。就像 Whisper 的情况一样，我们搭便车使用了已学习的模型权重，这些权重帮助我们处理时序、音高、音色等。 `voxlocal` 通过 `piper-rs` 调用 Piper 并合成语音： ``` pub fn speak(&mut self, text: &str) -> Result<(Vec<f32>, u32)> { self.piper .create(text, false, None, None, None, None) .map_err(|e| anyhow!("piper synth failed: {e}")) } ``` `voxlocal` 将生成的音频采样传递给 `rodio` 进行播放，并附带采样率信息。 ``` let source = rodio::buffer::SamplesBuffer::new( 1, sample_rate, samples.to_vec(), ); player.append(source); player.sleep_until_end(); ``` 我们需要记住，神经合成是在 Piper 的 ONNX 语音模型（[注4](https://samkhawase.com/blog/voxlocal-minimal-voice-agent/#user-content-fn-4)）内部完成的……