Muse Voice Transcribe 是 Meta 推出的实时语音转写模型名字里的 Voice Transcribe 直接点明核心任务把语音转成文本而“实时”这个定语说明它面向的不是录完整段再识别的离线流程而是音频流边进边出的在线场景。对工程开发者来说这类消息真正值得关注的不是模型演示效果而是实时语音转写链路里的输入格式、分帧策略、静音检测、推理服务、延迟评估和异常恢复如何配合起来。这篇文章以 Muse Voice Transcribe 作为技术场景入口从模型定位讲到工程接入再用一个 WebSocket 实时转写服务的最小版本把链路串起来最后补充评测指标、常见问题和上线检查清单。适合已经掌握基础 Python正在做语音交互、实时字幕、会议纪要或客服质检的开发者参考。需要先说明一个原则不同团队拿到的 Muse Voice Transcribe 形态可能不一样可能是模型权重可能是推理服务也可能是 API。这篇文章不把某个具体接口当作不变的官方事实而是把工程接入过程中必须处理的公共问题讲清楚。实际项目使用前要以你拿到的模型仓库、模型卡或服务文档为准。1. Muse Voice Transcribe 想解决什么实时语音转写不再是“录完再识别”1.1 模型定位与典型应用场景Muse Voice Transcribe 所属的赛道是实时语音转写。它要解决的是“一段音频正在产生系统能不能同步把已经说出来的内容先转成文字”的问题。传统的离线识别需要等待整段音频提交完成再一次性返回全文。这样做准确率容易控制但在语音输入法、直播字幕、远程会议字幕这类场景里不可接受用户不可能等你说完整句话再看到结果。实时语音转写的典型使用场景可以归纳为几类会议软件实时字幕边说边显示并允许随时修正。直播或视频内容的实时字幕要求延迟尽量低。语音输入法和可穿戴设备命令识别需要边说边出词。呼叫中心客服质检对通话实时转写并触发关键词提醒。语音助手的中间态理解先得到局部文本再等待完整指令。在这些场景里用户对“延迟”的容忍度通常按秒级计算。会议字幕如果每句话结束两秒后才冒出来体验已经大打折扣而语音输入法如果出现候选词延迟会直接影响打字效率。正因为如此实时语音转写系统不能简单套用离线识别的代码工程侧需要单独设计。1.2 实时转写与离线转写的关键差异很多人第一次接触实时语音转写时会默认“只要模型足够快离线代码改一改也能做实时”。这种思路在小型验证里也许碰巧能跑通生产环境很快会出问题。原因是离线与实时在输入方式、输出粒度、错误恢复上有本质差异。下表把两类流程放在一起比较维度离线语音转写实时语音转写输入方式一次性提交完整音频文件音频流分片持续进入输出方式整段音频处理完成后返回全文边说边返回片段或候选结果端点检测可选可手动裁剪音频段必须可靠切分点决定输出质量延迟要求不敏感分钟级也可接受敏感通常要求秒级或更低错误处理可整体重新识别已输出片段可能需要后端修正或覆盖并发特征任务型按文件并发长连接型按会话并发离线转写可以把一个 10 分钟的录音交给后台任务处理 30 秒后集中返回结果。实时语音转写则会同时面临大量 WebSocket 长连接每路连接上是一个持续到达的音频流服务端还要处理连接中断、音频无法对齐、静音段过长等情况。1.3 一条完整的实时转写链路包含哪些环节从麦克风采集到最终字幕显示中间至少要经过这些环节音频采集设备或录音文件读取。重采样、声道合并、位深转换生成模型能接受的 PCM 数据。按固定时长分包送入流式缓冲区。静音检测或端点检测判断当前音频段何时结束。将有效音频段送入实时语音转写模型。对模型输出的裸文本做标点恢复、数字格式化、热词替换等后处理。后端通过 WebSocket、HTTP 长轮询或消息队列把文本推送到前端。任何一个环节出现问题最终表现都可能是“没有文字”或“文字断断续续”。排查时必须顺着这条链路逐层检查而不是只盯着模型准确率。这里要注意一个容易被误解的地方实时语音转写不等于“每 40 毫秒调用一次模型”。模型调用有固定开销而且没有上下文边界的零散音频片段会导致识别效果变差。工程上通常采用“短分段 滑窗 静音判断”的组合手段让模型的每次输入都有相对完整的语义单元。2. 接入前准备环境、音频格式与最小项目结构2.1 先分清学习环境与生产环境在本地跑通 Muse Voice Transcribe 的接入链路主要目的是验证接口、观察延迟、准备测试集。这个阶段可以接受手动装依赖、CPU 推理、单客户端测试。生产环境则需要把模型服务独立出去并处理好权限、日志、监控、并发和回滚。学习环境建议按以下清单准备组件建议配置用途Python3.10 或更高运行接口服务和脚本ffmpeg4.4 或更高音频转码、重采样、PCM 导出模型服务模型仓库说明为准提供推理能力GPU可选显存足够时可加速推理测试音频16kHz / 16bit / 单声道 PCM保证格式对齐如果本地没有 GPU实时性可能不达标但不影响先验证接口协议和代码链路。真正的性能验收要放在带 GPU 或 NPU 的测试机上进行。项目依赖可以先放在requirements.txt中。下面的依赖清单只列出公共部分是否安装torch、tensorflow或特定推理库取决于 Muse Voice Transcribe 的部署形态。numpy1.24 soundfile0.12 ffmpeg-python0.2.0 fastapi0.104 uvicorn0.24 websockets12.0 requests2.31实际项目如果模型服务不在本地torch这类推理依赖可以不加在 Web 服务里。推荐做法是把“音频接入服务”和“模型推理服务”分离两者的演进速度和故障边界都不相同。2.2 音频数据约束采样率、位深和声道必须先对齐大多数语音识别模型在训练时会要求声音按固定采样率输入。16kHz 是语音识别领域最常见的采样率16kHz 意味着每秒采样 16000 个点一个 16bit 单声道采样点占 2 字节。如果输入音频是 48kHz 的录音直接喂给模型会导致音高偏移和特征不匹配识别效果严重下降。接入 Muse Voice Transcribe 之前要先把音频统一成模型需要的格式。把一段 m4a 录音转成 16kHz / 16bit / 单声道 PCM可以执行这样的命令ffmpeg -i test_record.m4a -ar 16000 -ac 1 -f s16le -acodec pcm_s16le test_record.pcm参数含义如下参数含义说明-i test_record.m4a输入文件名需要转码的原始文件-ar 16000输出采样率设为 16000 Hz-ac 1输出声道数1 代表单声道-f s16le输出容器格式16bit 小端 PCM-acodec pcm_s16le输出编码强制使用 PCM 编码如果音频本身是多声道直接用-ac 1会做降混音。注意不能只降混音而不重采样否则后续 PCM 字节长度和采样时间对不上日志里的音频时长可能偏差很大。2.3 最小项目结构如何设计实时语音转写服务常见的分层是WebSocket 接入层、音频处理层、模型适配层。为了让代码不依赖某一个具体模型建议把所有模型交互收拢到一个适配器里。一个最小但可演化的目录结构如下muse-voice-transcribe-demo/ ├── app/ │ ├── __init__.py │ ├── asr_engine.py # 模型适配层 │ ├── audio_processor.py # PCM 分包、静音检测 │ └── server.py # FastAPI WebSocket 服务 ├── scripts/ │ ├── stream_client.py # 发送 PCM 流的客户端 │ └── convert_audio.sh # 音频转 16k 单声道 PCM ├── tests/ │ └── test_audio_processor.py ├── requirements.txt └── README.mdasr_engine.py隔离模型服务audio_processor.py处理音频流的字节切分server.py只关心连接、接收数据和返回文本。这样即使后续从 HTTP 模型服务换成进程内推理引擎改动范围也能被控制在一个模块内。3. 用 WebSocket 服务把音频流转成文字3.1 模型适配层把 Muse Voice Transcribe 的接口差异挡在外面实际接入时你拿到的 Muse Voice Transcribe 可能是一个本地模型也可能是一个远程推理服务。为了避免业务代码到处写模型 SDK先定义抽象接口。下面是一份适配层示例。它假设模型服务暴露了一个接收 PCM 原始音频并返回 JSON 文本的 HTTP 接口。# app/asr_engine.py import requests class MuseVoiceTranscribeHTTP: Muse Voice Transcribe 的 HTTP 适配示例。 具体路径、鉴权方式、返回字段以模型服务文档为准。 如果官方提供 Python SDK直接在此类内部调用 SDK 即可。 def __init__(self, endpoint: str, token: str ): self.endpoint endpoint self.session requests.Session() if token: self.session.headers.update({Authorization: fBearer {token}}) def transcribe_segment( self, pcm: bytes, is_final: bool False, context: dict | None None ) - str: 输入一段 PCM返回识别文本。 pcm 一段 16kHz/16bit/单声道 的 PCM is_final 当前是否为最终分段模型服务可据此做端点处理 context 预留字段可以放说话人ID、领域、热词等 params {is_final: str(is_final).lower()} response self.session.post( self.endpoint, paramsparams, datapcm, headers{Content-Type: application/octet-stream}, timeout3 ) response.raise_for_status() payload response.json() return payload.get(text, ) def close(self): self.session.close()这段代码的关键点有两个。第一它禁止在业务代码里散落 HTTP 调用所有模型交互的细节集中在一个类中。第二is_final参数被显式暴露出来这是流式识别里容易忽视的概念同一个音频片段在静音尚未出现时是中间状态在静音到来后变成可提交的最终状态。实际服务里不一定有/v1/transcribe这个路径也可能不用 HTTP 而用 gRPC。保留这个适配层的价值在于后续替换协议时transcribe_ws函数不需要改。3.2 音频处理分包、滑动累积与静音切分实时音频流到达服务端时是连续字节流不能直接把整串字节一次性交给模型。需要先按固定帧长切分再做静音检测。下面代码把收到的字节流切分成 40ms 一帧并累计语音段等待静音达到一定时长后返回一个完整音频段。# app/audio_processor.py import numpy as np class PCMStreamProcessor: 按字节处理 16kHz/16bit/单声道 PCM 流。 def __init__( self, sample_rate: int 16000, frame_ms: int 40, silence_ms: int 700, min_segment_ms: int 400, energy_threshold: float 300.0 ): self.sample_rate sample_rate self.frame_size int(sample_rate * frame_ms / 1000) * 2 self.min_segment_bytes int(sample_rate * min_segment_ms / 1000) * 2 self.silence_bytes_limit int(sample_rate * silence_ms / 1000) * 2 self.energy_threshold energy_threshold self.pending b self.segment b self.silent_bytes 0 def push(self, data: bytes): 送入新到达的 PCM 字节返回已完成的语音段列表。 self.pending data finished_segments [] while len(self.pending) self.frame_size: frame, self.pending self.pending[:self.frame_size], self.pending[self.frame_size:] self.segment frame if self._is_silence(frame): self.silent_bytes len(frame) if ( len(self.segment) self.min_segment_bytes and self.silent_bytes self.silence_bytes_limit ): finished_segments.append(self.segment) self.segment b self.silent_bytes 0 else: self.silent_bytes 0 return finished_segments def flush(self): 客户端断开前调用返回尚未提交的剩余音频段。 if not self.segment: return None segment, self.segment self.segment, b self.silent_bytes 0 return segment def _is_silence(self, frame: bytes) - bool: samples np.frombuffer(frame, dtypenp.int16).astype(np.float32) rms float(np.sqrt(np.mean(samples * samples))) return rms self.energy_threshold这个处理器采用的切分逻辑是先保证至少积累了 400ms 音频再判断是否出现 700ms 静音。只要满足条件就把当前累积的整段音频作为一个“句子”返回。能量阈值300.0是示例值不同麦克风和环境噪声不同需要在测试中调整。这里的常见误区是把静音检测简单理解成“能量低就是静音”。真实环境中呼吸声、键盘声、空调声都可能超过阈值。静音检测做得粗糙模型就会被切成碎片。后续如果要提高质量可以用 WebRTC VAD 或自训练 VAD 替换这里的能量阈值方案但整体字节处理逻辑可以保留。3.3 服务端WebSocket 接收流式 PCM 并返回文本为了让前端或音频采集端能持续推送 PCM服务端使用 FastAPI 的 WebSocket 接口。下面代码只要处理器返回一个语音段就调用模型适配层并返回 JSON 结果。# app/server.py import asyncio import json from contextlib import asynccontextmanager from fastapi import FastAPI, WebSocket, WebSocketDisconnect from app.asr_engine import MuseVoiceTranscribeHTTP from app.audio_processor import PCMStreamProcessor transcriber None asynccontextmanager async def lifespan(app: FastAPI): global transcriber # 用实际模型服务地址替换token 按部署要求填写 transcriber MuseVoiceTranscribeHTTP( endpointhttp://127.0.0.1:9001/v1/transcribe, token ) yield if transcriber is not None: transcriber.close() app FastAPI(lifespanlifespan) app.websocket(/transcribe) async def transcribe_ws(ws: WebSocket): await ws.accept() processor PCMStreamProcessor() try: while True: data await ws.receive_bytes() for segment in processor.push(data): text await asyncio.to_thread( transcriber.transcribe_segment, segment, True ) await ws.send_text(json.dumps({ type: Muse Voice Transcribe实时语音转写服务构建指南
Muse Voice Transcribe 是 Meta 推出的实时语音转写模型名字里的 Voice Transcribe 直接点明核心任务把语音转成文本而“实时”这个定语说明它面向的不是录完整段再识别的离线流程而是音频流边进边出的在线场景。对工程开发者来说这类消息真正值得关注的不是模型演示效果而是实时语音转写链路里的输入格式、分帧策略、静音检测、推理服务、延迟评估和异常恢复如何配合起来。这篇文章以 Muse Voice Transcribe 作为技术场景入口从模型定位讲到工程接入再用一个 WebSocket 实时转写服务的最小版本把链路串起来最后补充评测指标、常见问题和上线检查清单。适合已经掌握基础 Python正在做语音交互、实时字幕、会议纪要或客服质检的开发者参考。需要先说明一个原则不同团队拿到的 Muse Voice Transcribe 形态可能不一样可能是模型权重可能是推理服务也可能是 API。这篇文章不把某个具体接口当作不变的官方事实而是把工程接入过程中必须处理的公共问题讲清楚。实际项目使用前要以你拿到的模型仓库、模型卡或服务文档为准。1. Muse Voice Transcribe 想解决什么实时语音转写不再是“录完再识别”1.1 模型定位与典型应用场景Muse Voice Transcribe 所属的赛道是实时语音转写。它要解决的是“一段音频正在产生系统能不能同步把已经说出来的内容先转成文字”的问题。传统的离线识别需要等待整段音频提交完成再一次性返回全文。这样做准确率容易控制但在语音输入法、直播字幕、远程会议字幕这类场景里不可接受用户不可能等你说完整句话再看到结果。实时语音转写的典型使用场景可以归纳为几类会议软件实时字幕边说边显示并允许随时修正。直播或视频内容的实时字幕要求延迟尽量低。语音输入法和可穿戴设备命令识别需要边说边出词。呼叫中心客服质检对通话实时转写并触发关键词提醒。语音助手的中间态理解先得到局部文本再等待完整指令。在这些场景里用户对“延迟”的容忍度通常按秒级计算。会议字幕如果每句话结束两秒后才冒出来体验已经大打折扣而语音输入法如果出现候选词延迟会直接影响打字效率。正因为如此实时语音转写系统不能简单套用离线识别的代码工程侧需要单独设计。1.2 实时转写与离线转写的关键差异很多人第一次接触实时语音转写时会默认“只要模型足够快离线代码改一改也能做实时”。这种思路在小型验证里也许碰巧能跑通生产环境很快会出问题。原因是离线与实时在输入方式、输出粒度、错误恢复上有本质差异。下表把两类流程放在一起比较维度离线语音转写实时语音转写输入方式一次性提交完整音频文件音频流分片持续进入输出方式整段音频处理完成后返回全文边说边返回片段或候选结果端点检测可选可手动裁剪音频段必须可靠切分点决定输出质量延迟要求不敏感分钟级也可接受敏感通常要求秒级或更低错误处理可整体重新识别已输出片段可能需要后端修正或覆盖并发特征任务型按文件并发长连接型按会话并发离线转写可以把一个 10 分钟的录音交给后台任务处理 30 秒后集中返回结果。实时语音转写则会同时面临大量 WebSocket 长连接每路连接上是一个持续到达的音频流服务端还要处理连接中断、音频无法对齐、静音段过长等情况。1.3 一条完整的实时转写链路包含哪些环节从麦克风采集到最终字幕显示中间至少要经过这些环节音频采集设备或录音文件读取。重采样、声道合并、位深转换生成模型能接受的 PCM 数据。按固定时长分包送入流式缓冲区。静音检测或端点检测判断当前音频段何时结束。将有效音频段送入实时语音转写模型。对模型输出的裸文本做标点恢复、数字格式化、热词替换等后处理。后端通过 WebSocket、HTTP 长轮询或消息队列把文本推送到前端。任何一个环节出现问题最终表现都可能是“没有文字”或“文字断断续续”。排查时必须顺着这条链路逐层检查而不是只盯着模型准确率。这里要注意一个容易被误解的地方实时语音转写不等于“每 40 毫秒调用一次模型”。模型调用有固定开销而且没有上下文边界的零散音频片段会导致识别效果变差。工程上通常采用“短分段 滑窗 静音判断”的组合手段让模型的每次输入都有相对完整的语义单元。2. 接入前准备环境、音频格式与最小项目结构2.1 先分清学习环境与生产环境在本地跑通 Muse Voice Transcribe 的接入链路主要目的是验证接口、观察延迟、准备测试集。这个阶段可以接受手动装依赖、CPU 推理、单客户端测试。生产环境则需要把模型服务独立出去并处理好权限、日志、监控、并发和回滚。学习环境建议按以下清单准备组件建议配置用途Python3.10 或更高运行接口服务和脚本ffmpeg4.4 或更高音频转码、重采样、PCM 导出模型服务模型仓库说明为准提供推理能力GPU可选显存足够时可加速推理测试音频16kHz / 16bit / 单声道 PCM保证格式对齐如果本地没有 GPU实时性可能不达标但不影响先验证接口协议和代码链路。真正的性能验收要放在带 GPU 或 NPU 的测试机上进行。项目依赖可以先放在requirements.txt中。下面的依赖清单只列出公共部分是否安装torch、tensorflow或特定推理库取决于 Muse Voice Transcribe 的部署形态。numpy1.24 soundfile0.12 ffmpeg-python0.2.0 fastapi0.104 uvicorn0.24 websockets12.0 requests2.31实际项目如果模型服务不在本地torch这类推理依赖可以不加在 Web 服务里。推荐做法是把“音频接入服务”和“模型推理服务”分离两者的演进速度和故障边界都不相同。2.2 音频数据约束采样率、位深和声道必须先对齐大多数语音识别模型在训练时会要求声音按固定采样率输入。16kHz 是语音识别领域最常见的采样率16kHz 意味着每秒采样 16000 个点一个 16bit 单声道采样点占 2 字节。如果输入音频是 48kHz 的录音直接喂给模型会导致音高偏移和特征不匹配识别效果严重下降。接入 Muse Voice Transcribe 之前要先把音频统一成模型需要的格式。把一段 m4a 录音转成 16kHz / 16bit / 单声道 PCM可以执行这样的命令ffmpeg -i test_record.m4a -ar 16000 -ac 1 -f s16le -acodec pcm_s16le test_record.pcm参数含义如下参数含义说明-i test_record.m4a输入文件名需要转码的原始文件-ar 16000输出采样率设为 16000 Hz-ac 1输出声道数1 代表单声道-f s16le输出容器格式16bit 小端 PCM-acodec pcm_s16le输出编码强制使用 PCM 编码如果音频本身是多声道直接用-ac 1会做降混音。注意不能只降混音而不重采样否则后续 PCM 字节长度和采样时间对不上日志里的音频时长可能偏差很大。2.3 最小项目结构如何设计实时语音转写服务常见的分层是WebSocket 接入层、音频处理层、模型适配层。为了让代码不依赖某一个具体模型建议把所有模型交互收拢到一个适配器里。一个最小但可演化的目录结构如下muse-voice-transcribe-demo/ ├── app/ │ ├── __init__.py │ ├── asr_engine.py # 模型适配层 │ ├── audio_processor.py # PCM 分包、静音检测 │ └── server.py # FastAPI WebSocket 服务 ├── scripts/ │ ├── stream_client.py # 发送 PCM 流的客户端 │ └── convert_audio.sh # 音频转 16k 单声道 PCM ├── tests/ │ └── test_audio_processor.py ├── requirements.txt └── README.mdasr_engine.py隔离模型服务audio_processor.py处理音频流的字节切分server.py只关心连接、接收数据和返回文本。这样即使后续从 HTTP 模型服务换成进程内推理引擎改动范围也能被控制在一个模块内。3. 用 WebSocket 服务把音频流转成文字3.1 模型适配层把 Muse Voice Transcribe 的接口差异挡在外面实际接入时你拿到的 Muse Voice Transcribe 可能是一个本地模型也可能是一个远程推理服务。为了避免业务代码到处写模型 SDK先定义抽象接口。下面是一份适配层示例。它假设模型服务暴露了一个接收 PCM 原始音频并返回 JSON 文本的 HTTP 接口。# app/asr_engine.py import requests class MuseVoiceTranscribeHTTP: Muse Voice Transcribe 的 HTTP 适配示例。 具体路径、鉴权方式、返回字段以模型服务文档为准。 如果官方提供 Python SDK直接在此类内部调用 SDK 即可。 def __init__(self, endpoint: str, token: str ): self.endpoint endpoint self.session requests.Session() if token: self.session.headers.update({Authorization: fBearer {token}}) def transcribe_segment( self, pcm: bytes, is_final: bool False, context: dict | None None ) - str: 输入一段 PCM返回识别文本。 pcm 一段 16kHz/16bit/单声道 的 PCM is_final 当前是否为最终分段模型服务可据此做端点处理 context 预留字段可以放说话人ID、领域、热词等 params {is_final: str(is_final).lower()} response self.session.post( self.endpoint, paramsparams, datapcm, headers{Content-Type: application/octet-stream}, timeout3 ) response.raise_for_status() payload response.json() return payload.get(text, ) def close(self): self.session.close()这段代码的关键点有两个。第一它禁止在业务代码里散落 HTTP 调用所有模型交互的细节集中在一个类中。第二is_final参数被显式暴露出来这是流式识别里容易忽视的概念同一个音频片段在静音尚未出现时是中间状态在静音到来后变成可提交的最终状态。实际服务里不一定有/v1/transcribe这个路径也可能不用 HTTP 而用 gRPC。保留这个适配层的价值在于后续替换协议时transcribe_ws函数不需要改。3.2 音频处理分包、滑动累积与静音切分实时音频流到达服务端时是连续字节流不能直接把整串字节一次性交给模型。需要先按固定帧长切分再做静音检测。下面代码把收到的字节流切分成 40ms 一帧并累计语音段等待静音达到一定时长后返回一个完整音频段。# app/audio_processor.py import numpy as np class PCMStreamProcessor: 按字节处理 16kHz/16bit/单声道 PCM 流。 def __init__( self, sample_rate: int 16000, frame_ms: int 40, silence_ms: int 700, min_segment_ms: int 400, energy_threshold: float 300.0 ): self.sample_rate sample_rate self.frame_size int(sample_rate * frame_ms / 1000) * 2 self.min_segment_bytes int(sample_rate * min_segment_ms / 1000) * 2 self.silence_bytes_limit int(sample_rate * silence_ms / 1000) * 2 self.energy_threshold energy_threshold self.pending b self.segment b self.silent_bytes 0 def push(self, data: bytes): 送入新到达的 PCM 字节返回已完成的语音段列表。 self.pending data finished_segments [] while len(self.pending) self.frame_size: frame, self.pending self.pending[:self.frame_size], self.pending[self.frame_size:] self.segment frame if self._is_silence(frame): self.silent_bytes len(frame) if ( len(self.segment) self.min_segment_bytes and self.silent_bytes self.silence_bytes_limit ): finished_segments.append(self.segment) self.segment b self.silent_bytes 0 else: self.silent_bytes 0 return finished_segments def flush(self): 客户端断开前调用返回尚未提交的剩余音频段。 if not self.segment: return None segment, self.segment self.segment, b self.silent_bytes 0 return segment def _is_silence(self, frame: bytes) - bool: samples np.frombuffer(frame, dtypenp.int16).astype(np.float32) rms float(np.sqrt(np.mean(samples * samples))) return rms self.energy_threshold这个处理器采用的切分逻辑是先保证至少积累了 400ms 音频再判断是否出现 700ms 静音。只要满足条件就把当前累积的整段音频作为一个“句子”返回。能量阈值300.0是示例值不同麦克风和环境噪声不同需要在测试中调整。这里的常见误区是把静音检测简单理解成“能量低就是静音”。真实环境中呼吸声、键盘声、空调声都可能超过阈值。静音检测做得粗糙模型就会被切成碎片。后续如果要提高质量可以用 WebRTC VAD 或自训练 VAD 替换这里的能量阈值方案但整体字节处理逻辑可以保留。3.3 服务端WebSocket 接收流式 PCM 并返回文本为了让前端或音频采集端能持续推送 PCM服务端使用 FastAPI 的 WebSocket 接口。下面代码只要处理器返回一个语音段就调用模型适配层并返回 JSON 结果。# app/server.py import asyncio import json from contextlib import asynccontextmanager from fastapi import FastAPI, WebSocket, WebSocketDisconnect from app.asr_engine import MuseVoiceTranscribeHTTP from app.audio_processor import PCMStreamProcessor transcriber None asynccontextmanager async def lifespan(app: FastAPI): global transcriber # 用实际模型服务地址替换token 按部署要求填写 transcriber MuseVoiceTranscribeHTTP( endpointhttp://127.0.0.1:9001/v1/transcribe, token ) yield if transcriber is not None: transcriber.close() app FastAPI(lifespanlifespan) app.websocket(/transcribe) async def transcribe_ws(ws: WebSocket): await ws.accept() processor PCMStreamProcessor() try: while True: data await ws.receive_bytes() for segment in processor.push(data): text await asyncio.to_thread( transcriber.transcribe_segment, segment, True ) await ws.send_text(json.dumps({ type: