最近不少做语音助手、实时通信和 AI Gateway 的朋友都在讨论“赫尔墨斯代理”的新版本。这次更新最吸引人的地方就是它把“语音激活”能力内置到了代理层而不是像以前那样把逻辑散落在各个业务服务里。简单说代理不再只是转发 HTTP 请求它开始能“听懂”语音流知道什么时候该唤醒、什么时候该把音频转给后端处理。这篇文章我不打算只停留在功能介绍上而是会从代理服务的定位讲起拆解语音激活链路中的关键模块再用 Python 手写一个精简版“赫尔墨斯代理”的语音激活转发示例最后把常见的坑和工程化建议一起整理出来。无论你是正在做语音助手后端还是想给现有代理网关增加音频处理能力这篇都值得收藏慢慢看。1. 赫尔墨斯代理是什么从“信使”到“语音入口”1.1 代理服务的核心定位在说“赫尔墨斯代理”之前先聊聊代理本身。赫尔墨斯是希腊神话里的信使神负责传递消息。软件领域里的代理服务Proxy做的事情很相似接收客户端请求按照规则做鉴权、路由、转发、限流再把后端的响应返回给客户端。传统的 API 代理处理的是结构化数据比如 JSON、XML、表单参数。它不关心请求体里的音频二进制流是什么含义也不关心这段音频里有没有人说话。代理层的判断依据主要是 URL、Header、Token、参数这些“元信息”。1.2 语音激活能力是什么语音激活Voice Activation指的是系统通过麦克风持续采集音频在检测到有人说话时自动触发后续流程的能力。最常见的应用就是智能音箱的“唤醒词”比如“小爱同学”“Hey Siri”平时设备处于待机状态只有检测到唤醒词才开始上传音频、执行指令。但“语音激活”并不等于“唤醒词”。它可以更宽泛检测到人声就开始录音人声停止就结束录音VAD语音活动检测。检测到特定声纹特征才放行声纹激活。检测到语义意图后启动某个技能意图激活。赫尔墨斯代理新版本强调的“语音激活更新”重点在于把这一类检测能力下沉到代理层让所有接入代理的业务都能复用。1.3 代理层做语音激活和客户端做有什么区别很多客户端本身可以做 VAD比如手机端的 WebRTC VAD、iOS 的 SFSpeechRecognizer。那为什么还要在代理层做主要有三个原因统一策略不同客户端网页、小程序、App、硬件设备的算法能力和算力差异很大把语音激活放在代理层可以保证所有端的行为一致。节省资源客户端只负责传原始音频流代理层判断没有语音时就不往后端转发减少 ASR 服务和 LLM 服务的调用成本。动态升级算法模型更新时只需要更新代理服务不需要重新发版每一个客户端。2. 语音激活代理的整体架构设计2.1 分层设计一个带语音激活能力的代理服务我建议至少分成四层层级职责关键组件接入层接收客户端 WebSocket 音频流处理连接生命周期WebSocket Server、连接管理器语音处理层对音频流做 VAD、降噪、格式转换、语音激活判定VAD 模块、音频格式器路由转发层把激活后的音频交给 ASR 服务把文本交给 LLM 服务并返回响应HTTP/WebSocket Client、路由策略会话管理层维护用户会话、上下文、激活状态、超时控制会话存储、状态机2.2 音频数据流一次完整的语音交互流程可以简单概括为客户端麦克风 - 音频帧 - WebSocket - 代理服务 - VAD 检测 - 检测到语音 - 转发 ASR - ASR 文本 - LLM 推理 - 应答文本 - TTS - 返回客户端在设计代理时需要注意音频流是“持续不断”的。如果每帧都转发给 ASR会造成极大的资源浪费。正确的做法是VAD 模块负责判断当前说话状态静音状态丢弃或只做能量统计。说话开始缓存音频并通知会话进入“语音激活”状态。说话中持续缓存并转发。说话结束把整段音频提交给 ASR。2.3 为什么把语音激活放在代理层把语音激活放在代理层核心收益是“可插拔”。后端服务不需要关心音频是怎么来的只需要接收代理已经处理好的文本客户端也不需要关心 VAD 阈值和敏感度参数代理可以统一配置。这种架构在面对多端接入、多模型切换时非常灵活。3. 环境准备与项目结构3.1 运行环境本文示例以 Python 3.10 为例操作系统不限Windows、macOS、Linux 都可以运行。重点演示的是链路和思路所以不会依赖具体的商业语音服务。你可以把代码中的 mock 部分替换成真实的 ASR 服务。依赖库如下库名用途安装命令websockets提供 WebSocket Server 和 Clientpip install websocketsnumpy处理音频帧的数值计算用于简单能量检测pip install numpyasyncioPython 异步编程库内置无需安装struct解析二进制音频数据内置无需安装版本需要根据你的项目实际情况调整本文示例以常见环境为例重点演示配置思路。3.2 项目结构为了让代码清晰我按模块拆分hermes-proxy-demo/ ├── requirements.txt ├── config.py ├── vad.py ├── audio_utils.py ├── proxy_server.py ├── mock_backend.py └── test_client.py下面逐一实现。4. 核心模块实现4.1 配置文件# 文件路径config.py # WebSocket 服务端口 WS_HOST 0.0.0.0 WS_PORT 8765 # 音频参数 SAMPLE_RATE 16000 # 采样率常见语音识别场景使用 16k CHANNELS 1 # 单声道 SAMPLE_WIDTH 2 # 16bit 采样2 字节 FRAME_DURATION_MS 20 # 每帧时长WebRTC VAD 常用 20ms/30ms FRAME_SIZE int(SAMPLE_RATE * FRAME_DURATION_MS / 1000) # 每帧采样点数 # VAD 参数 VAD_THRESHOLD 500 # 能量阈值根据实际环境调整 SILENCE_TIMEOUT_MS 600 # 尾音静默多久判定为一句结束 MIN_SPEECH_MS 200 # 最短语音时长过滤环境噪音 # 后端 ASR/LLM 地址 ASR_URL http://127.0.0.1:9000/asr LLM_URL http://127.0.0.1:9001/chat说明一下SAMPLE_RATE和FRAME_DURATION_MS是音频处理中两个非常关键的参数。16kHz 采样率是目前大多数语音识别服务支持的采样率既保留了足够的语音信息又不会像 44.1kHz 那样产生过多数据量。每帧 20ms 是 VAD 判断延迟和精度之间的折中帧太短误判多帧太长延迟高。4.2 音频工具模块# 文件路径audio_utils.py import struct import numpy as np def bytes_to_pcm(audio_bytes: bytes) - np.ndarray: 将二进制音频数据转换为 float 类型的 PCM 数组。 这里假设输入是 16bit 单声道小端字节序。 # 每两个字节组成一个采样点 sample_count len(audio_bytes) // 2 # struct.unpack 一次解包所有采样点 fmt h * sample_count samples struct.unpack(fmt, audio_bytes[: sample_count * 2]) # 转换为 float 数组方便计算能量 return np.array(samples, dtypenp.float32) def pcm_to_bytes(samples: np.ndarray) - bytes: 将 float 类型的 PCM 数组转回 bytes方便通过 WebSocket 发送。 samples_int np.clip(samples, -32768, 32767).astype(np.int16) return samples_int.tobytes() def calculate_energy(samples: np.ndarray) - float: 计算音频帧的能量值。这里使用 RMS均方根 比简单的绝对值求和更稳定。 if len(samples) 0: return 0.0 # 避免除零 rms np.sqrt(np.mean(np.square(samples))) return float(rms)这里用numpy的好处是计算速度快代码也简洁。如果你的环境不允许装numpy也可以纯用math和struct计算能量但代码会啰嗦不少。4.3 VAD 模块VAD 是整个语音激活链路的“守门员”。它的任务不是识别内容而是判断“有没有人在说话”。这里我用能量检测实现一个可运行的 VAD实际项目中可以替换成webrtcvad或者 AI 模型。# 文件路径vad.py import time from enum import Enum class VadState(Enum): SILENCE 0 # 静音 SPEECH_START 1 # 检测到语音开始 SPEAKING 2 # 说话中 SPEECH_END 3 # 一句话结束 class EnergyVAD: 基于能量的语音活动检测器。 工作方式 1. 每帧计算能量。 2. 能量超过阈值且持续一段时间判定为语音开始。 3. 语音开始后如果连续静音超过 timeout判定为语音结束。 def __init__(self, threshold: float, silence_timeout_ms: int, min_speech_ms: int): self.threshold threshold self.silence_timeout_s silence_timeout_ms / 1000.0 self.min_speech_s min_speech_ms / 1000.0 self.state VadState.SILENCE self.speech_start_ts None self.last_voice_ts None def process_frame(self, energy: float, frame_ts: float) - VadState: 逐帧处理。 frame_ts 是这一帧的时间戳单位秒。 返回值是当前帧处理后的状态变化。 is_speech energy self.threshold event None if self.state VadState.SILENCE: if is_speech: self.speech_start_ts frame_ts self.last_voice_ts frame_ts self.state VadState.SPEAKING event VadState.SPEECH_START elif self.state VadState.SPEAKING: if is_speech: self.last_voice_ts frame_ts else: # 检查静音是否超时 silent_duration frame_ts - self.last_voice_ts if silent_duration self.silence_timeout_s: speech_duration self.last_voice_ts - self.speech_start_ts if speech_duration self.min_speech_s: self.state VadState.SILENCE event VadState.SPEECH_END else: # 语音太短当作误触发 self.state VadState.SILENCE return event这段代码的逻辑很直观。process_frame每收到一帧音频就调用一次返回三种可能的事件SPEECH_START、SPEECH_END、None。代理层只需要根据事件类型决定是否开始缓存音频、是否把整段音频送去 ASR。有一点需要注意speech_start_ts和last_voice_ts使用的是frame_ts也就是音频帧自身的时间戳而不是系统当前时间。这样即使处理速度有波动也不会影响断句判断。4.4 模拟后端服务在真正开发时ASR 和 LLM 通常是外部服务。为了演示完整链路我先写一个 mock 后端# 文件路径mock_backend.py import asyncio import json from websockets.asyncio.server import serve async def handle_asr(websocket): 模拟 ASR 服务。收到音频字节后返回一段固定文本。 真实场景中这里应该调用语音识别 API 或本地模型。 async for message in websocket: if isinstance(message, bytes): # 模拟识别耗时 await asyncio.sleep(0.2) result { code: 0, text: 帮我查一下明天的天气, } await websocket.send(json.dumps(result, ensure_asciiFalse)) async def handle_llm(websocket): 模拟 LLM 服务。收到用户文本返回一个应答文本。 async for message in websocket: if isinstance(message, str): await asyncio.sleep(0.3) result { code: 0, reply: 好的明天北京市区天气晴气温 12 到 24 摄氏度。, } await websocket.send(json.dumps(result, ensure_asciiFalse)) async def start_mock_backend(): async with serve(handle_asr, 127.0.0.1, 9000) as asr_server: async with serve(handle_llm, 127.0.0.1, 9001) as llm_server: print(Mock ASR 服务运行在 ws://127.0.0.1:9000) print(Mock LLM 服务运行在 ws://127.0.0.1:9001) await asyncio.gather(asr_server.serve_forever(), llm_server.serve_forever()) if __name__ __main__: asyncio.run(start_mock_backend())这里我把 ASR 和 LLM 都封装成了 WebSocket 服务方便和代理服务统一通信。真实项目中后端可能是 HTTP 接口也可能是 gRPC逻辑是一样的接受音频/文本返回结构化的结果。4.5 代理服务主程序代理服务是整个示例的核心。它要做的事情可以拆成几块监听客户端的 WebSocket 连接。接收二进制音频帧。对每帧做能量计算和 VAD 状态判断。语音开始后把音频帧缓存到一个队列。语音结束后把缓存音频通过 WebSocket 发给 ASR。拿到 ASR 文本后转发给 LLM。把 LLM 的应答结果返回给客户端。# 文件路径proxy_server.py import asyncio import json import time import uuid from websockets.asyncio.server import serve from websockets.asyncio.client import connect import config from audio_utils import bytes_to_pcm, calculate_energy from vad import EnergyVAD, VadState class HermesProxy: 赫尔墨斯代理 - 语音激活版本。 负责接收客户端音频流做 VAD 判断激活后转发给后端 ASR 和 LLM。 def __init__(self): self.sessions {} async def handle_client(self, websocket): 处理一个客户端的 WebSocket 连接。 session_id str(uuid.uuid4()) self.sessions[session_id] { vad: EnergyVAD( thresholdconfig.VAD_THRESHOLD, silence_timeout_msconfig.SILENCE_TIMEOUT_MS, min_speech_msconfig.MIN_SPEECH_MS, ), speech_buffer: bytearray(), speech_active: False, } print(f[{session_id}] 客户端已连接) try: async for message in websocket: if isinstance(message, bytes): await self._process_audio(session_id, message, websocket) elif isinstance(message, str): await self._process_text(session_id, message, websocket) except Exception as exc: print(f[{session_id}] 连接异常: {exc}) finally: self.sessions.pop(session_id, None) print(f[{session_id}] 客户端已断开) async def _process_audio(self, session_id, audio_bytes, client_ws): session self.sessions[session_id] vad session[vad] # 转成 PCM 数组并计算能量 samples bytes_to_pcm(audio_bytes) energy calculate_energy(samples) # 当前时间戳这里简化使用系统时间实际项目中应当使用 # 音频自己的时间戳以免网络延迟影响断句。 frame_ts time.time() event vad.process_frame(energy, frame_ts) if event VadState.SPEECH_START: session[speech_active] True session[speech_buffer] bytearray() print(f[{session_id}] 语音激活开始接收语音) if session[speech_active]: session[speech_buffer].extend(audio_bytes) if event VadState.SPEECH_END: print(f[{session_id}] 语音结束发送到 ASR) session[speech_active] False audio_payload bytes(session[speech_buffer]) session[speech_buffer] bytearray() # 有足够长度的语音才交给 ASR if len(audio_payload) config.FRAME_SIZE: await self._send_to_asr(session_id, audio_payload, client_ws) else: print(f[{session_id}] 语音段过短丢弃) async def _send_to_asr(self, session_id, audio_payload, client_ws): 把缓存音频交给 ASR 服务识别完成后把文本交给 LLM。 try: async with connect(config.ASR_URL) as asr_ws: await asr_ws.send(audio_payload) asr_response await asr_ws.recv() asr_result json.loads(asr_response) user_text asr_result.get(text, ) print(f[{session_id}] ASR 识别结果: {user_text}) if not user_text: await client_ws.send(json.dumps({ type: error, message: 没有识别到有效的语音内容 }, ensure_asciiFalse)) return async with connect(config.LLM_URL) as llm_ws: await llm_ws.send(user_text) llm_response await llm_ws.recv() llm_result json.loads(llm_response) reply llm_result.get(reply, ) print(f[{session_id}] LLM 回复: {reply}) # 应答返回给客户端 await client_ws.send(json.dumps({ type: reply, text: reply, }, ensure_asciiFalse)) except Exception as exc: print(f[{session_id}] 调用后端服务失败: {exc}) await client_ws.send(json.dumps({ type: error, message: 后端服务暂时不可用 }, ensure_asciiFalse)) async def _process_text(self, session_id, text, client_ws): 支持客户端直接发送文本方便调试不走 VAD。 data json.loads(text) if data.get(type) ping: await client_ws.send(json.dumps({type: pong}, ensure_asciiFalse)) return async def main(): proxy HermesProxy() print(f赫尔墨斯代理运行在 ws://{config.WS_HOST}:{config.WS_PORT}) async with serve(proxy.handle_client, config.WS_HOST, config.WS_PORT) as server: await server.serve_forever() if __name__ __main__: asyncio.run(main())这个主程序有两个值得关注的点。第一个是会话管理。每个客户端连接对应一个HermesProxy里的 sessionsession 保存了独立的 VAD 实例和语音缓存。这样多个用户同时说话时各自的检测状态不会互相干扰。第二个是_send_to_asr方法。它把“VAD 收集音频”和“后端调用”彻底解耦。VAD 一旦检测到一句话结束就把音频交给 ASRASR 返回文本后再调用 LLM。整个过程对客户端来说是异步的客户端不需要自己分割音频。4.6 测试客户端没有客户端模拟器的话服务端代码没法验证。我写一个简单的测试客户端# 文件路径test_client.py import asyncio import random import struct from websockets.asyncio.client import connect def generate_audio_frame(duration_ms: int, sample_rate: int, with_voice: bool): 生成一帧音频数据。with_voiceTrue 时模拟人声带波形 with_voiceFalse 时模拟静音低能量。 frame_samples int(sample_rate * duration_ms / 1000) samples [] for i in range(frame_samples): if with_voice: # 模拟一个 200Hz 左右的声音 value int(3000 * (i % 80) / 40 - 1500) # 加点随机噪声更接近真实音频 value random.randint(-50, 50) else: value random.randint(-5, 5) samples.append(value) return struct.pack( h * len(samples), *samples) async def run_client(): uri ws://127.0.0.1:8765 async with connect(uri) as ws: print(已连接到赫尔墨斯代理) # 先发 500ms 静音 for _ in range(10): frame generate_audio_frame(20, 16000, with_voiceFalse) await ws.send(frame) await asyncio.sleep(0.02) # 发送 1000ms 模拟语音 for _ in range(20): frame generate_audio_frame(20, 16000, with_voiceTrue) await ws.send(frame) await asyncio.sleep(0.02) # 发送 1000ms 静音触发 VAD 结束 for _ in range(20): frame generate_audio_frame(20, 16000, with_voiceFalse) await ws.send(frame) await asyncio.sleep(0.02) # 等待后端返回结果 try: response await asyncio.wait_for(ws.recv(), timeout5) print(f收到代理响应: {response}) except asyncio.TimeoutError: print(等待响应超时) if __name__ __main__: asyncio.run(run_client())测试客户端的核心是模拟音频帧。我需要用struct.pack生成标准 16bit PCM 数据而不是发送字符串因为服务端是按二进制解析的。4.7 运行与验证按照下面的顺序启动# 安装依赖 pip install -r requirements.txt # 终端 1启动模拟后端 python mock_backend.py # 终端 2启动代理服务 python proxy_server.py # 终端 3运行测试客户端 python test_client.pyrequirements.txt内容websockets12.0 numpy1.24.0预期你会看到类似输出赫尔墨斯代理运行在 ws://0.0.0.0:8765 [会话] 客户端已连接 [会话] 语音激活开始接收语音 [会话] 语音结束发送到 ASR [会话] ASR 识别结果: 帮我查一下明天的天气 [会话] LLM 回复: 好的明天北京市区天气晴气温 12 到 24 摄氏度。到这里一条完整的“语音激活 - 代理转发 - ASR - LLM - 返回客户端”链路就通了。5. 常见问题与排查思路在实际部署中你可能会遇到下面这些情况。我按“现象 - 原因 - 解决思路”整理成表格便于快速定位。问题现象常见原因解决思路客户端一说话就断连发送了非 16bit PCM 格式的数据服务端解析异常统一定义音频格式在接入层做格式校验语音一直不激活能量阈值过高或者音量太小调低VAD_THRESHOLD同时输出日志观察每帧能量值没有声音时频繁误触发环境噪音大阈值过低开启降噪或提高阈值增加“最短语音时长”过滤一句话被切成两段静音超时时间设置太短说话人停顿稍长就断开适当增大SILENCE_TIMEOUT_MS比如 800ms 到 1200msASR 收到的音频不完整语音开始的前几帧被丢弃或结束判定过早在SPEECH_START事件前缓存一小段音频作为前导缓冲后端返回慢导致连接超时LLM 推理耗时较长客户端等待超时代理改为“先返回已收到再异步推送结果”或增加超时时间多用户同时说话时串音session 管理不当多个连接共用同一个 VAD 实例确保每个客户端连接创建独立的 session 和 VAD 实例这里单独说一下“语音开始前导缓冲”。人的发音不是突然一下就满能量的。很多真实语音从“准备发声”到“完全清晰”有几十毫秒的过渡期。如果 VAD 判定条件太严格可能丢掉声母部分。常见做法是在SPEECH_START事件触发时把最近 100ms 到 200ms 的缓存音频一起拼接到语音段开头。6. 工程实践与优化建议6.1 音频数据格式与采样率统一代理层最怕的就是“音频格式混乱”。不同客户端可能使用不同的采样率、位深、声道数。建议在接入层强制统一采样率语音识别场景推荐 16kHz。位深16bit。声道单声道。编码PCM 裸流或 Opus 压缩后再在代理层解码。如果客户端传上来的是压缩格式代理需要先解码再交给 VAD否则能量检测会失真。6.2 VAD 阈值需要动态调节VAD_THRESHOLD 500只是一个示例值。在真实场景中不同麦克风、不同环境的背景噪音差异非常大。上线前建议做一次音量校准让用户静默 5 秒统计环境噪音的平均能量。让用户正常说话 5 秒统计语音平均能量。阈值取两者之间的中间偏上值。代码层面可以在 session 启动时自动收集前几十帧作为噪音底噪动态初始化阈值避免所有用户共用同一个固定值。6.3 安全与鉴权语音代理一旦暴露到公网可能面临恶意刷量、语音注入等风险。需要注意连接必须鉴权。可以在 WebSocket 握手时校验 JWT 或 API Key。对单连接做音频时长和调用频率限制。涉及语音数据的存储和转发时确保链路使用 TLS/WSS。不要随意转发来自不可信客户端的“文本指令”给 LLM防止提示词注入。我一般会在代理层增加一个简单的规则客户端上传音频前必须先发送一个携带 Token 的认证文本帧代理验证通过后才开始接收音频流。6.4 可观测性语音链路比普通 API 链路更难排查因为你不能“看参数”就知道问题出在哪里。建议在代理层增加结构化日志[2025-01-10 10:00:01] sessionabc123 eventvad_start energy2340 [2025-01-10 10:00:02] sessionabc123 eventvad_end duration_ms860 [2025-01-10 10:00:02] sessionabc123 eventasr_start audio_size27520 [2025-01-10 10:00:02] sessionabc123 eventasr_end text_length12 [2025-01-10 10:00:03] sessionabc123 eventllm_end latency_ms320日志里至少包含会话 ID、事件类型、关键数值。这样即使出现“用户说了一句话但代理没有反应”的问题也能快速判断是 VAD 没触发还是 ASR 没调用还是 LLM 返回超时。6.5 性能优化语音代理是 IO 密集和计算密集的混合体。主要优化点有三个使用异步非阻塞框架不要在请求处理函数里做同步阻塞操作。VAD 计算不要使用 Python 循环逐采样点遍历。上面例子里用的是numpy向量化计算性能远好于 Python 循环。如果并发量高可以在代理层做一个“帧聚合”多个音频帧凑成一个大包再交给 VAD 一次处理减少函数调用开销。6.6 不要忽略前端体验这个不完全是后端问题但和语音激活的成败强相关。如果代理层检测到语音结束在向后端请求 ASR 和 LLM 的这段时间里客户端可能没有任何反馈。用户会以为设备坏了。建议在客户端加一个“语音识别中”的过渡状态如果整体响应时间超过 1 秒还可以考虑先把“正在听”的动画播起来提升感知速度。7. 总结赫尔墨斯代理这次“语音激活更新”的核心价值不是单纯加了一个 VAD 功能而是让代理从“转发请求”进化成“理解语音入口”。架构上语音处理、VAD 判定、ASR/LLM 路由可以在代理层统一编排业务代码不需要关心音频细节只关注文本交互。这篇文章里我带着大家从零写了一个精简版语音激活代理重点代码包括audio_utils.py二进制音频转 PCM 数组计算能量。vad.py基于能量检测的语音活动检测器。proxy_server.py代理主服务连接客户端、VAD、ASR、LLM。mock_backend.py模拟 ASR 和 LLM方便本地跑通流程。test_client.py模拟客户端发送音频帧。你可以直接运行这套代码感受整个语音激活链路然后逐步替换成真实服务。如果后面有余力我建议优先补三个方向把 VAD 换成webrtcvad或者神经网络 VAD提高复杂环境下的准确率给代理增加基于 Redis 的分布式会话管理支持多实例水平扩展把SPEECH_START前导缓存策略加上能明显提升 ASR 对首字识别率。语音交互的链路很复杂但只要把代理层这一环做好后面接多少种 ASR、多少种 LLM都是在配置层面解决的问题。希望这篇文章能帮你把“语音激活代理”从概念落地成代码。如果你在运行中遇到问题或者发现更好的工程思路欢迎在评论区一起交流。 赫尔墨斯代理语音激活实战：基于VAD的AI网关升级指南
最近不少做语音助手、实时通信和 AI Gateway 的朋友都在讨论“赫尔墨斯代理”的新版本。这次更新最吸引人的地方就是它把“语音激活”能力内置到了代理层而不是像以前那样把逻辑散落在各个业务服务里。简单说代理不再只是转发 HTTP 请求它开始能“听懂”语音流知道什么时候该唤醒、什么时候该把音频转给后端处理。这篇文章我不打算只停留在功能介绍上而是会从代理服务的定位讲起拆解语音激活链路中的关键模块再用 Python 手写一个精简版“赫尔墨斯代理”的语音激活转发示例最后把常见的坑和工程化建议一起整理出来。无论你是正在做语音助手后端还是想给现有代理网关增加音频处理能力这篇都值得收藏慢慢看。1. 赫尔墨斯代理是什么从“信使”到“语音入口”1.1 代理服务的核心定位在说“赫尔墨斯代理”之前先聊聊代理本身。赫尔墨斯是希腊神话里的信使神负责传递消息。软件领域里的代理服务Proxy做的事情很相似接收客户端请求按照规则做鉴权、路由、转发、限流再把后端的响应返回给客户端。传统的 API 代理处理的是结构化数据比如 JSON、XML、表单参数。它不关心请求体里的音频二进制流是什么含义也不关心这段音频里有没有人说话。代理层的判断依据主要是 URL、Header、Token、参数这些“元信息”。1.2 语音激活能力是什么语音激活Voice Activation指的是系统通过麦克风持续采集音频在检测到有人说话时自动触发后续流程的能力。最常见的应用就是智能音箱的“唤醒词”比如“小爱同学”“Hey Siri”平时设备处于待机状态只有检测到唤醒词才开始上传音频、执行指令。但“语音激活”并不等于“唤醒词”。它可以更宽泛检测到人声就开始录音人声停止就结束录音VAD语音活动检测。检测到特定声纹特征才放行声纹激活。检测到语义意图后启动某个技能意图激活。赫尔墨斯代理新版本强调的“语音激活更新”重点在于把这一类检测能力下沉到代理层让所有接入代理的业务都能复用。1.3 代理层做语音激活和客户端做有什么区别很多客户端本身可以做 VAD比如手机端的 WebRTC VAD、iOS 的 SFSpeechRecognizer。那为什么还要在代理层做主要有三个原因统一策略不同客户端网页、小程序、App、硬件设备的算法能力和算力差异很大把语音激活放在代理层可以保证所有端的行为一致。节省资源客户端只负责传原始音频流代理层判断没有语音时就不往后端转发减少 ASR 服务和 LLM 服务的调用成本。动态升级算法模型更新时只需要更新代理服务不需要重新发版每一个客户端。2. 语音激活代理的整体架构设计2.1 分层设计一个带语音激活能力的代理服务我建议至少分成四层层级职责关键组件接入层接收客户端 WebSocket 音频流处理连接生命周期WebSocket Server、连接管理器语音处理层对音频流做 VAD、降噪、格式转换、语音激活判定VAD 模块、音频格式器路由转发层把激活后的音频交给 ASR 服务把文本交给 LLM 服务并返回响应HTTP/WebSocket Client、路由策略会话管理层维护用户会话、上下文、激活状态、超时控制会话存储、状态机2.2 音频数据流一次完整的语音交互流程可以简单概括为客户端麦克风 - 音频帧 - WebSocket - 代理服务 - VAD 检测 - 检测到语音 - 转发 ASR - ASR 文本 - LLM 推理 - 应答文本 - TTS - 返回客户端在设计代理时需要注意音频流是“持续不断”的。如果每帧都转发给 ASR会造成极大的资源浪费。正确的做法是VAD 模块负责判断当前说话状态静音状态丢弃或只做能量统计。说话开始缓存音频并通知会话进入“语音激活”状态。说话中持续缓存并转发。说话结束把整段音频提交给 ASR。2.3 为什么把语音激活放在代理层把语音激活放在代理层核心收益是“可插拔”。后端服务不需要关心音频是怎么来的只需要接收代理已经处理好的文本客户端也不需要关心 VAD 阈值和敏感度参数代理可以统一配置。这种架构在面对多端接入、多模型切换时非常灵活。3. 环境准备与项目结构3.1 运行环境本文示例以 Python 3.10 为例操作系统不限Windows、macOS、Linux 都可以运行。重点演示的是链路和思路所以不会依赖具体的商业语音服务。你可以把代码中的 mock 部分替换成真实的 ASR 服务。依赖库如下库名用途安装命令websockets提供 WebSocket Server 和 Clientpip install websocketsnumpy处理音频帧的数值计算用于简单能量检测pip install numpyasyncioPython 异步编程库内置无需安装struct解析二进制音频数据内置无需安装版本需要根据你的项目实际情况调整本文示例以常见环境为例重点演示配置思路。3.2 项目结构为了让代码清晰我按模块拆分hermes-proxy-demo/ ├── requirements.txt ├── config.py ├── vad.py ├── audio_utils.py ├── proxy_server.py ├── mock_backend.py └── test_client.py下面逐一实现。4. 核心模块实现4.1 配置文件# 文件路径config.py # WebSocket 服务端口 WS_HOST 0.0.0.0 WS_PORT 8765 # 音频参数 SAMPLE_RATE 16000 # 采样率常见语音识别场景使用 16k CHANNELS 1 # 单声道 SAMPLE_WIDTH 2 # 16bit 采样2 字节 FRAME_DURATION_MS 20 # 每帧时长WebRTC VAD 常用 20ms/30ms FRAME_SIZE int(SAMPLE_RATE * FRAME_DURATION_MS / 1000) # 每帧采样点数 # VAD 参数 VAD_THRESHOLD 500 # 能量阈值根据实际环境调整 SILENCE_TIMEOUT_MS 600 # 尾音静默多久判定为一句结束 MIN_SPEECH_MS 200 # 最短语音时长过滤环境噪音 # 后端 ASR/LLM 地址 ASR_URL http://127.0.0.1:9000/asr LLM_URL http://127.0.0.1:9001/chat说明一下SAMPLE_RATE和FRAME_DURATION_MS是音频处理中两个非常关键的参数。16kHz 采样率是目前大多数语音识别服务支持的采样率既保留了足够的语音信息又不会像 44.1kHz 那样产生过多数据量。每帧 20ms 是 VAD 判断延迟和精度之间的折中帧太短误判多帧太长延迟高。4.2 音频工具模块# 文件路径audio_utils.py import struct import numpy as np def bytes_to_pcm(audio_bytes: bytes) - np.ndarray: 将二进制音频数据转换为 float 类型的 PCM 数组。 这里假设输入是 16bit 单声道小端字节序。 # 每两个字节组成一个采样点 sample_count len(audio_bytes) // 2 # struct.unpack 一次解包所有采样点 fmt h * sample_count samples struct.unpack(fmt, audio_bytes[: sample_count * 2]) # 转换为 float 数组方便计算能量 return np.array(samples, dtypenp.float32) def pcm_to_bytes(samples: np.ndarray) - bytes: 将 float 类型的 PCM 数组转回 bytes方便通过 WebSocket 发送。 samples_int np.clip(samples, -32768, 32767).astype(np.int16) return samples_int.tobytes() def calculate_energy(samples: np.ndarray) - float: 计算音频帧的能量值。这里使用 RMS均方根 比简单的绝对值求和更稳定。 if len(samples) 0: return 0.0 # 避免除零 rms np.sqrt(np.mean(np.square(samples))) return float(rms)这里用numpy的好处是计算速度快代码也简洁。如果你的环境不允许装numpy也可以纯用math和struct计算能量但代码会啰嗦不少。4.3 VAD 模块VAD 是整个语音激活链路的“守门员”。它的任务不是识别内容而是判断“有没有人在说话”。这里我用能量检测实现一个可运行的 VAD实际项目中可以替换成webrtcvad或者 AI 模型。# 文件路径vad.py import time from enum import Enum class VadState(Enum): SILENCE 0 # 静音 SPEECH_START 1 # 检测到语音开始 SPEAKING 2 # 说话中 SPEECH_END 3 # 一句话结束 class EnergyVAD: 基于能量的语音活动检测器。 工作方式 1. 每帧计算能量。 2. 能量超过阈值且持续一段时间判定为语音开始。 3. 语音开始后如果连续静音超过 timeout判定为语音结束。 def __init__(self, threshold: float, silence_timeout_ms: int, min_speech_ms: int): self.threshold threshold self.silence_timeout_s silence_timeout_ms / 1000.0 self.min_speech_s min_speech_ms / 1000.0 self.state VadState.SILENCE self.speech_start_ts None self.last_voice_ts None def process_frame(self, energy: float, frame_ts: float) - VadState: 逐帧处理。 frame_ts 是这一帧的时间戳单位秒。 返回值是当前帧处理后的状态变化。 is_speech energy self.threshold event None if self.state VadState.SILENCE: if is_speech: self.speech_start_ts frame_ts self.last_voice_ts frame_ts self.state VadState.SPEAKING event VadState.SPEECH_START elif self.state VadState.SPEAKING: if is_speech: self.last_voice_ts frame_ts else: # 检查静音是否超时 silent_duration frame_ts - self.last_voice_ts if silent_duration self.silence_timeout_s: speech_duration self.last_voice_ts - self.speech_start_ts if speech_duration self.min_speech_s: self.state VadState.SILENCE event VadState.SPEECH_END else: # 语音太短当作误触发 self.state VadState.SILENCE return event这段代码的逻辑很直观。process_frame每收到一帧音频就调用一次返回三种可能的事件SPEECH_START、SPEECH_END、None。代理层只需要根据事件类型决定是否开始缓存音频、是否把整段音频送去 ASR。有一点需要注意speech_start_ts和last_voice_ts使用的是frame_ts也就是音频帧自身的时间戳而不是系统当前时间。这样即使处理速度有波动也不会影响断句判断。4.4 模拟后端服务在真正开发时ASR 和 LLM 通常是外部服务。为了演示完整链路我先写一个 mock 后端# 文件路径mock_backend.py import asyncio import json from websockets.asyncio.server import serve async def handle_asr(websocket): 模拟 ASR 服务。收到音频字节后返回一段固定文本。 真实场景中这里应该调用语音识别 API 或本地模型。 async for message in websocket: if isinstance(message, bytes): # 模拟识别耗时 await asyncio.sleep(0.2) result { code: 0, text: 帮我查一下明天的天气, } await websocket.send(json.dumps(result, ensure_asciiFalse)) async def handle_llm(websocket): 模拟 LLM 服务。收到用户文本返回一个应答文本。 async for message in websocket: if isinstance(message, str): await asyncio.sleep(0.3) result { code: 0, reply: 好的明天北京市区天气晴气温 12 到 24 摄氏度。, } await websocket.send(json.dumps(result, ensure_asciiFalse)) async def start_mock_backend(): async with serve(handle_asr, 127.0.0.1, 9000) as asr_server: async with serve(handle_llm, 127.0.0.1, 9001) as llm_server: print(Mock ASR 服务运行在 ws://127.0.0.1:9000) print(Mock LLM 服务运行在 ws://127.0.0.1:9001) await asyncio.gather(asr_server.serve_forever(), llm_server.serve_forever()) if __name__ __main__: asyncio.run(start_mock_backend())这里我把 ASR 和 LLM 都封装成了 WebSocket 服务方便和代理服务统一通信。真实项目中后端可能是 HTTP 接口也可能是 gRPC逻辑是一样的接受音频/文本返回结构化的结果。4.5 代理服务主程序代理服务是整个示例的核心。它要做的事情可以拆成几块监听客户端的 WebSocket 连接。接收二进制音频帧。对每帧做能量计算和 VAD 状态判断。语音开始后把音频帧缓存到一个队列。语音结束后把缓存音频通过 WebSocket 发给 ASR。拿到 ASR 文本后转发给 LLM。把 LLM 的应答结果返回给客户端。# 文件路径proxy_server.py import asyncio import json import time import uuid from websockets.asyncio.server import serve from websockets.asyncio.client import connect import config from audio_utils import bytes_to_pcm, calculate_energy from vad import EnergyVAD, VadState class HermesProxy: 赫尔墨斯代理 - 语音激活版本。 负责接收客户端音频流做 VAD 判断激活后转发给后端 ASR 和 LLM。 def __init__(self): self.sessions {} async def handle_client(self, websocket): 处理一个客户端的 WebSocket 连接。 session_id str(uuid.uuid4()) self.sessions[session_id] { vad: EnergyVAD( thresholdconfig.VAD_THRESHOLD, silence_timeout_msconfig.SILENCE_TIMEOUT_MS, min_speech_msconfig.MIN_SPEECH_MS, ), speech_buffer: bytearray(), speech_active: False, } print(f[{session_id}] 客户端已连接) try: async for message in websocket: if isinstance(message, bytes): await self._process_audio(session_id, message, websocket) elif isinstance(message, str): await self._process_text(session_id, message, websocket) except Exception as exc: print(f[{session_id}] 连接异常: {exc}) finally: self.sessions.pop(session_id, None) print(f[{session_id}] 客户端已断开) async def _process_audio(self, session_id, audio_bytes, client_ws): session self.sessions[session_id] vad session[vad] # 转成 PCM 数组并计算能量 samples bytes_to_pcm(audio_bytes) energy calculate_energy(samples) # 当前时间戳这里简化使用系统时间实际项目中应当使用 # 音频自己的时间戳以免网络延迟影响断句。 frame_ts time.time() event vad.process_frame(energy, frame_ts) if event VadState.SPEECH_START: session[speech_active] True session[speech_buffer] bytearray() print(f[{session_id}] 语音激活开始接收语音) if session[speech_active]: session[speech_buffer].extend(audio_bytes) if event VadState.SPEECH_END: print(f[{session_id}] 语音结束发送到 ASR) session[speech_active] False audio_payload bytes(session[speech_buffer]) session[speech_buffer] bytearray() # 有足够长度的语音才交给 ASR if len(audio_payload) config.FRAME_SIZE: await self._send_to_asr(session_id, audio_payload, client_ws) else: print(f[{session_id}] 语音段过短丢弃) async def _send_to_asr(self, session_id, audio_payload, client_ws): 把缓存音频交给 ASR 服务识别完成后把文本交给 LLM。 try: async with connect(config.ASR_URL) as asr_ws: await asr_ws.send(audio_payload) asr_response await asr_ws.recv() asr_result json.loads(asr_response) user_text asr_result.get(text, ) print(f[{session_id}] ASR 识别结果: {user_text}) if not user_text: await client_ws.send(json.dumps({ type: error, message: 没有识别到有效的语音内容 }, ensure_asciiFalse)) return async with connect(config.LLM_URL) as llm_ws: await llm_ws.send(user_text) llm_response await llm_ws.recv() llm_result json.loads(llm_response) reply llm_result.get(reply, ) print(f[{session_id}] LLM 回复: {reply}) # 应答返回给客户端 await client_ws.send(json.dumps({ type: reply, text: reply, }, ensure_asciiFalse)) except Exception as exc: print(f[{session_id}] 调用后端服务失败: {exc}) await client_ws.send(json.dumps({ type: error, message: 后端服务暂时不可用 }, ensure_asciiFalse)) async def _process_text(self, session_id, text, client_ws): 支持客户端直接发送文本方便调试不走 VAD。 data json.loads(text) if data.get(type) ping: await client_ws.send(json.dumps({type: pong}, ensure_asciiFalse)) return async def main(): proxy HermesProxy() print(f赫尔墨斯代理运行在 ws://{config.WS_HOST}:{config.WS_PORT}) async with serve(proxy.handle_client, config.WS_HOST, config.WS_PORT) as server: await server.serve_forever() if __name__ __main__: asyncio.run(main())这个主程序有两个值得关注的点。第一个是会话管理。每个客户端连接对应一个HermesProxy里的 sessionsession 保存了独立的 VAD 实例和语音缓存。这样多个用户同时说话时各自的检测状态不会互相干扰。第二个是_send_to_asr方法。它把“VAD 收集音频”和“后端调用”彻底解耦。VAD 一旦检测到一句话结束就把音频交给 ASRASR 返回文本后再调用 LLM。整个过程对客户端来说是异步的客户端不需要自己分割音频。4.6 测试客户端没有客户端模拟器的话服务端代码没法验证。我写一个简单的测试客户端# 文件路径test_client.py import asyncio import random import struct from websockets.asyncio.client import connect def generate_audio_frame(duration_ms: int, sample_rate: int, with_voice: bool): 生成一帧音频数据。with_voiceTrue 时模拟人声带波形 with_voiceFalse 时模拟静音低能量。 frame_samples int(sample_rate * duration_ms / 1000) samples [] for i in range(frame_samples): if with_voice: # 模拟一个 200Hz 左右的声音 value int(3000 * (i % 80) / 40 - 1500) # 加点随机噪声更接近真实音频 value random.randint(-50, 50) else: value random.randint(-5, 5) samples.append(value) return struct.pack( h * len(samples), *samples) async def run_client(): uri ws://127.0.0.1:8765 async with connect(uri) as ws: print(已连接到赫尔墨斯代理) # 先发 500ms 静音 for _ in range(10): frame generate_audio_frame(20, 16000, with_voiceFalse) await ws.send(frame) await asyncio.sleep(0.02) # 发送 1000ms 模拟语音 for _ in range(20): frame generate_audio_frame(20, 16000, with_voiceTrue) await ws.send(frame) await asyncio.sleep(0.02) # 发送 1000ms 静音触发 VAD 结束 for _ in range(20): frame generate_audio_frame(20, 16000, with_voiceFalse) await ws.send(frame) await asyncio.sleep(0.02) # 等待后端返回结果 try: response await asyncio.wait_for(ws.recv(), timeout5) print(f收到代理响应: {response}) except asyncio.TimeoutError: print(等待响应超时) if __name__ __main__: asyncio.run(run_client())测试客户端的核心是模拟音频帧。我需要用struct.pack生成标准 16bit PCM 数据而不是发送字符串因为服务端是按二进制解析的。4.7 运行与验证按照下面的顺序启动# 安装依赖 pip install -r requirements.txt # 终端 1启动模拟后端 python mock_backend.py # 终端 2启动代理服务 python proxy_server.py # 终端 3运行测试客户端 python test_client.pyrequirements.txt内容websockets12.0 numpy1.24.0预期你会看到类似输出赫尔墨斯代理运行在 ws://0.0.0.0:8765 [会话] 客户端已连接 [会话] 语音激活开始接收语音 [会话] 语音结束发送到 ASR [会话] ASR 识别结果: 帮我查一下明天的天气 [会话] LLM 回复: 好的明天北京市区天气晴气温 12 到 24 摄氏度。到这里一条完整的“语音激活 - 代理转发 - ASR - LLM - 返回客户端”链路就通了。5. 常见问题与排查思路在实际部署中你可能会遇到下面这些情况。我按“现象 - 原因 - 解决思路”整理成表格便于快速定位。问题现象常见原因解决思路客户端一说话就断连发送了非 16bit PCM 格式的数据服务端解析异常统一定义音频格式在接入层做格式校验语音一直不激活能量阈值过高或者音量太小调低VAD_THRESHOLD同时输出日志观察每帧能量值没有声音时频繁误触发环境噪音大阈值过低开启降噪或提高阈值增加“最短语音时长”过滤一句话被切成两段静音超时时间设置太短说话人停顿稍长就断开适当增大SILENCE_TIMEOUT_MS比如 800ms 到 1200msASR 收到的音频不完整语音开始的前几帧被丢弃或结束判定过早在SPEECH_START事件前缓存一小段音频作为前导缓冲后端返回慢导致连接超时LLM 推理耗时较长客户端等待超时代理改为“先返回已收到再异步推送结果”或增加超时时间多用户同时说话时串音session 管理不当多个连接共用同一个 VAD 实例确保每个客户端连接创建独立的 session 和 VAD 实例这里单独说一下“语音开始前导缓冲”。人的发音不是突然一下就满能量的。很多真实语音从“准备发声”到“完全清晰”有几十毫秒的过渡期。如果 VAD 判定条件太严格可能丢掉声母部分。常见做法是在SPEECH_START事件触发时把最近 100ms 到 200ms 的缓存音频一起拼接到语音段开头。6. 工程实践与优化建议6.1 音频数据格式与采样率统一代理层最怕的就是“音频格式混乱”。不同客户端可能使用不同的采样率、位深、声道数。建议在接入层强制统一采样率语音识别场景推荐 16kHz。位深16bit。声道单声道。编码PCM 裸流或 Opus 压缩后再在代理层解码。如果客户端传上来的是压缩格式代理需要先解码再交给 VAD否则能量检测会失真。6.2 VAD 阈值需要动态调节VAD_THRESHOLD 500只是一个示例值。在真实场景中不同麦克风、不同环境的背景噪音差异非常大。上线前建议做一次音量校准让用户静默 5 秒统计环境噪音的平均能量。让用户正常说话 5 秒统计语音平均能量。阈值取两者之间的中间偏上值。代码层面可以在 session 启动时自动收集前几十帧作为噪音底噪动态初始化阈值避免所有用户共用同一个固定值。6.3 安全与鉴权语音代理一旦暴露到公网可能面临恶意刷量、语音注入等风险。需要注意连接必须鉴权。可以在 WebSocket 握手时校验 JWT 或 API Key。对单连接做音频时长和调用频率限制。涉及语音数据的存储和转发时确保链路使用 TLS/WSS。不要随意转发来自不可信客户端的“文本指令”给 LLM防止提示词注入。我一般会在代理层增加一个简单的规则客户端上传音频前必须先发送一个携带 Token 的认证文本帧代理验证通过后才开始接收音频流。6.4 可观测性语音链路比普通 API 链路更难排查因为你不能“看参数”就知道问题出在哪里。建议在代理层增加结构化日志[2025-01-10 10:00:01] sessionabc123 eventvad_start energy2340 [2025-01-10 10:00:02] sessionabc123 eventvad_end duration_ms860 [2025-01-10 10:00:02] sessionabc123 eventasr_start audio_size27520 [2025-01-10 10:00:02] sessionabc123 eventasr_end text_length12 [2025-01-10 10:00:03] sessionabc123 eventllm_end latency_ms320日志里至少包含会话 ID、事件类型、关键数值。这样即使出现“用户说了一句话但代理没有反应”的问题也能快速判断是 VAD 没触发还是 ASR 没调用还是 LLM 返回超时。6.5 性能优化语音代理是 IO 密集和计算密集的混合体。主要优化点有三个使用异步非阻塞框架不要在请求处理函数里做同步阻塞操作。VAD 计算不要使用 Python 循环逐采样点遍历。上面例子里用的是numpy向量化计算性能远好于 Python 循环。如果并发量高可以在代理层做一个“帧聚合”多个音频帧凑成一个大包再交给 VAD 一次处理减少函数调用开销。6.6 不要忽略前端体验这个不完全是后端问题但和语音激活的成败强相关。如果代理层检测到语音结束在向后端请求 ASR 和 LLM 的这段时间里客户端可能没有任何反馈。用户会以为设备坏了。建议在客户端加一个“语音识别中”的过渡状态如果整体响应时间超过 1 秒还可以考虑先把“正在听”的动画播起来提升感知速度。7. 总结赫尔墨斯代理这次“语音激活更新”的核心价值不是单纯加了一个 VAD 功能而是让代理从“转发请求”进化成“理解语音入口”。架构上语音处理、VAD 判定、ASR/LLM 路由可以在代理层统一编排业务代码不需要关心音频细节只关注文本交互。这篇文章里我带着大家从零写了一个精简版语音激活代理重点代码包括audio_utils.py二进制音频转 PCM 数组计算能量。vad.py基于能量检测的语音活动检测器。proxy_server.py代理主服务连接客户端、VAD、ASR、LLM。mock_backend.py模拟 ASR 和 LLM方便本地跑通流程。test_client.py模拟客户端发送音频帧。你可以直接运行这套代码感受整个语音激活链路然后逐步替换成真实服务。如果后面有余力我建议优先补三个方向把 VAD 换成webrtcvad或者神经网络 VAD提高复杂环境下的准确率给代理增加基于 Redis 的分布式会话管理支持多实例水平扩展把SPEECH_START前导缓存策略加上能明显提升 ASR 对首字识别率。语音交互的链路很复杂但只要把代理层这一环做好后面接多少种 ASR、多少种 LLM都是在配置层面解决的问题。希望这篇文章能帮你把“语音激活代理”从概念落地成代码。如果你在运行中遇到问题或者发现更好的工程思路欢迎在评论区一起交流。