这篇教程带你从零把一套开源语音大模型跑在自己的机器上,最后会得到两个能直接用的东西:
1. 一个批量转写服务:丢进一段会议录音(mp3/m4a/wav 都行),输出带时间段落的文字稿,可以并发处理多段音频。
2. 一个可打断的语音对话闭环:麦克风输入 → 语音识别 → 大模型回答 → 语音合成播放,用户中途开口就立刻停掉正在播放的回答。
两件事共用同一个模型服务,只是调用方式不同。全程用 OpenAI 兼容接口,客户端代码换成别的模型也能跑。
> 说明:开源语音模型的型号与仓库名更新很快,本文用 Voxtral 指代该系列开源权重模型。请求里的 model 字段换成你实际拉起的模型服务名即可;具体仓库名、支持的音频时长上限、参数格式,以官方模型卡与推理框架文档的当前版本为准。
---
一、前置条件清单
动手前先确认这几项,缺一项后面都会卡住:
硬件
- 一张显存足够的 NVIDIA 显卡(显存估算见步骤 1)。没有显卡也能跑,但只能跑小参数量版本,且实时性会很差。
- 系统盘之外留出至少几十 GB 空间放权重。
- 内存建议不低于显卡显存的两倍,模型加载时会先在内存里展开。
软件
- Linux(Ubuntu 系最省事)或 WSL2。
- NVIDIA 驱动 + 与驱动匹配的 CUDA 运行时。装推理框架时通常会自动带 CUDA 依赖,先跑
nvidia-smi确认显卡可见。 - Python 3.10 以上,建议用虚拟环境。
ffmpeg(音频转码与切片必需):sudo apt install -y ffmpeg。git、curl。
账号与网络
- Hugging Face 账号,部分模型仓库需要先同意许可协议才能下载。
- 能稳定访问模型仓库的网络环境;国内环境可配置镜像站,具体地址以对应镜像站说明为准。
基本认知
- 会写几十行 Python,看得懂 JSON。
- 知道什么是采样率、单声道、PCM。
---
二、分步骤实操
步骤 1:算清显存,别上来就下载
显存预算按四块算,粗算公式:
```
总显存 ≈ 权重 + 音频编码器 + AI 词典:KV Cache">KV Cache + 框架与激活开销
权重(GB) ≈ 参数量(十亿) × 每参数字节数
bf16 → 2 字节;fp8 / int8 → 约 1 字节;4bit → 约 0.5 字节
```
举例说明量级(不是实测跑分,只是估算方法):
| 参数量 | bf16 权重 | fp8 权重 | 建议显存(含余量) |
|---|---|---|---|
| 3B 级 | 约 6 GB | 约 3 GB | 12–16 GB |
| 8B 级 | 约 16 GB | 约 8 GB | 24 GB |
| 24B 级 | 约 48 GB | 约 24 GB | 48–80 GB |
再加上两块:
- KV Cache:和并发数、上下文长度成正比。 transcript 场景经常一次吃进几分钟音频,上下文不短,别把并发开到很大。
- 音频编码器:语音模型在语言模型前面挂了音频编码器,会额外占一部分显存,通常不大但别忽略。
结论性建议:先按 bf16 估算,显存不够再考虑 fp8 量化,最后才考虑换小模型。量化对转写质量的影响要用你自己的音频抽样验证,不要默认无损。
步骤 2:装环境、拉权重
```bash
python -m venv .venv
source .venv/bin/activate
pip install -U pip
pip install -U vllm # 版本以官方文档当前版本为准
pip install huggingface_hub hf_transfer fastapi uvicorn httpx numpy soundfile
```
vllm 是本文的推理服务端,它对多模态音频输入的支持在持续更新,安装前先看一眼官方文档里对该模型的支持说明,确认你装的版本认识这个模型结构。
下载权重(把 <org>/<model-repo> 换成模型卡页面上的实际仓库名):
```bash
export HF_HUB_ENABLE_HF_TRANSFER=1
export HF_TOKEN=<你的 token>
hf download <org>/<model-repo> \
--local-dir ./models/voxtral \
--exclude "*.pth" "*.gguf"
```
如果仓库需要同意协议,直接在网页上点同意,再执行上面的命令。下载中断可以用同样的命令续传。
步骤 3:拉起 OpenAI 兼容服务
```bash
vllm serve ./models/voxtral \
--served-model-name voxtral \
--host 0.0.0.0 \
--port 8000 \
--gpu-memory-utilization 0.90 \
--max-model-len 32768 \
--max-num-seqs 8
```
几个参数的实际含义:
--gpu-memory-utilization:允许框架占用显存的比例。调到 0.9 是给 KV Cache 留空间,显存吃紧时优先降这个。--max-model-len:单次请求的最大上下文。音频越长,占的上下文越多,转写长录音时这个值要够大。--max-num-seqs:同时处理的请求数上限。它直接决定你的转写吞吐,也直接决定 KV Cache 占用。
多模态输入通常还需要显式放开每个请求里允许的音频数量,不同版本参数名和取值格式有差异(有的是 audio=1,有的是 JSON 字符串),以官方文档里的启动示例为准。如果启动时报"不支持的模态",八成是这里没配对。
服务起来后验证:
```bash
curl -s http://127.0.0.1:8000/v1/models | python -m json.tool
```
能看到你的模型名就说明服务在跑。首次启动会花一段时间加载权重,属正常现象。
步骤 4:写最小转写客户端,先跑通一条链路
先把一段音频转成模型喜欢的格式(采样率与声道数以模型卡说明为准,16 kHz 单声道是常见默认):
```bash
ffmpeg -i meeting.m4a -ac 1 -ar 16000 -c:a pcm_s16le meeting.wav
```
Python 客户端:
```python
transcribe.py
import base64
import pathlib
import httpx
BASE_URL = "http://127.0.0.1:8000/v1"
MODEL = "voxtral"
PROMPT = (
"请逐字转写这段音频的内容,保留原语言,"
"补全标点,不要总结、不要翻译、不要添加说话人推测。"
"听不清的地方输出 [不确定]。"
)
def transcribe(wav_path: str, timeout: float = 600.0) -> str:
raw = pathlib.Path(wav_path).read_bytes()
b64 = base64.b64encode(raw).decode()
payload = {
"model": MODEL,
"temperature": 0,
"max_tokens": 4096,
"messages": [{
"role": "user",
"content": [
字段名在不同框架版本里可能是 audio_url / input_audio,
以官方文档给出的示例为准,二选一即可
{"type": "audio_url",
"audio_url": {"url": f"data:audio/wav;base64,{b64}"}},
{"type": "text", "text": PROMPT},
],
}],
}
with httpx.Client(timeout=timeout) as client:
resp = client.post(f"{BASE_URL}/chat/completions", json=payload)
resp.raise_for_status()
return resp.json()["choices"][0]["message"]["content"]
if __name__ == "__main__":
print(transcribe("meeting.wav"))
```
跑通这一段,链路就通了。注意三个容易忽略的点:
temperature: 0很重要,语音模型默认会"帮你整理",温度高了会开始改写。- prompt 里明确写"逐字转写 / 不要总结",否则很容易得到一段摘要。
- 请求体是 base64 内联,音频越大请求体越大,见后面的坑。
步骤 5:会议录音批量转写(切片 + 并发 + 时间戳)
模型单次能吃进的音频长度有上限(以模型卡说明为准),一段两小时的会议必须切开。切法有两种:
- 按固定时长切:简单,但会在词中间断开。
- 按静音切:效果好,需要先跑静音检测。生产环境更推荐这种。
先用固定切片把流程跑通,段间留一点重叠,后面再做去重:
```bash
mkdir -p parts
ffmpeg -i meeting.wav \
-f segment -segment_time 480 \
-c:a pcm_s16le -ar 16000 -ac 1 \
parts/part_%03d.wav
```
-segment_time 480 表示每段 8 分钟。段数多的时候注意 --max-model-len 是否够。
并发转写并拼回时间戳:
```python
batch_transcribe.py
import json
import pathlib
from concurrent.futures import ThreadPoolExecutor
from transcribe import transcribe
SEGMENT_SECONDS = 480
MAX_WORKERS = 4 # 不要超过服务端 --max-num-seqs
def run(parts_dir: str = "parts") -> list[dict]:
files = sorted(pathlib.Path(parts_dir).glob("part_*.wav"))
results: list[dict] = []
def work(idx_file):
idx, path = idx_file
text = transcribe(str(path))
return {
"index": idx,
"start": idx * SEGMENT_SECONDS,
"end": (idx + 1) * SEGMENT_SECONDS,
"text": text.strip(),
}
with ThreadPoolExecutor(max_workers=MAX_WORKERS) as pool:
for item in pool.map(work, enumerate(files)):
results.append(item)
results.sort(key=lambda x: x["index"])
return results
def fmt(seconds: int) -> str:
h, rem = divmod(seconds, 3600)
m, s = divmod(rem, 60)
return f"{h:02d}:{m:02d}:{s:02d}"
if __name__ == "__main__":
rows = run()
for r in rows:
print(f"[{fmt(r['start'])} - {fmt(r['end'])}] {r['text']}\n")
pathlib.Path("transcript.json").write_text(
json.dumps(rows, ensure_ascii=False, indent=2), encoding="utf-8"
)
```
这套做法得到的是段级时间戳,不是逐句时间戳。原因是这类语音模型输出的是文本序列,不保证返回词级对齐信息。如果你需要句级时间戳,有两条路:
1. 切片切得更细(按静音切到每句),每段天然就是一句,时间戳由切片位置决定。
2. 文本转写完成后,用强制对齐工具把文本和音频再做一次对齐。
对会议纪要这种用途,段级时间戳加段落文本通常已经够用。
去重小技巧:段间若有重叠,拼接时检查后一段开头与前一段结尾是否有重复的十几字,有就裁掉。
步骤 6:准流式转写(VAD 切句)
严格意义上的流式语音识别(边说边出字)对这类模型不是原生的,实际工程里更常见的做法是"准流式":
```
麦克风 → 环形缓冲 → VAD 判断静音 → 凑够一句话 → 提交转写 → 返回该句文本
```
关键参数是 VAD 的静音判定时长。它同时决定了延迟和切句质量:
- 调小(比如 300 ms):出字快,但容易把一句话切碎。
- 调大(比如 800 ms):切句完整,但用户说完要等更久。
实现时用任意一个成熟的 VAD 库都行(silero-vad 一类),核心逻辑:
```python
pseudo_stream.py
CHUNK_MS = 32
SILENCE_MS = 600 # 静音多久判定这句结束
MIN_SPEECH_MS = 300 # 太短的片段丢弃,防止咳嗽触发
def on_audio_chunk(chunk):
buffer.append(chunk)
if vad.is_speech(chunk):
speech_ms += CHUNK_MS
silence_ms = 0
else:
silence_ms += CHUNK_MS
if silence_ms >= SILENCE_MS and speech_ms >= MIN_SPEECH_MS:
sentence = flush_buffer() # 取出这段语音
text = transcribe_array(sentence) # 转成 wav 字节后走同一个接口
emit(text) # 推给前端或下游
speech_ms = 0
```
推送方式用 WebSocket 或 SSE 都行,把 emit 的内容打成 JSON 发出去。
这里有个必须接受的现实:延迟 = VAD 静音判定 + 请求往返 + 推理首字延迟 + 解码时间。所以短句转写体验会比长音频批量转写好很多,因为解码的 token 少。
步骤 7:接线可打断的语音对话
对话闭环由四段组成:
```
麦克风 → VAD → ASR → LLM → TTS → 扬声器
↑ |
└──── 打断检测 ───────┘
```
打断的本质是三个"取消",必须同时做,少一个就会有残留:
1. 停止播放:清空播放队列,立刻静音。只停止"下一次播放"是不够的。
2. 中止上游请求:把正在生成的 LLM/TTS 请求断掉,否则 token 还在流回来,播放队列会马上被重新填满。
3. 重置状态:清空本轮已生成的文本缓冲,把对话历史里未完成的那条标记为中断。
Python 侧的骨架(httpx 的流式请求 + 一个中断标志):
```python
dialog.py
import threading
import httpx
BASE_URL = "http://127.0.0.1:8000/v1"
MODEL = "voxtral"
class Turn:
def __init__(self):
self.cancelled = threading.Event()
def stream_answer(question: str, turn: Turn):
payload = {
"model": MODEL,
"temperature": 0.3,
"stream": True,
"messages": [
{"role": "system", "content": "你是语音助手。回答控制在三句话以内,口语化,不要用列表和代码块。"},
{"role": "user", "content": question},
],
}
with httpx.Client(timeout=None) as client:
with client.stream("POST", f"{BASE_URL}/chat/completions", json=payload) as r:
for line in r.iter_lines():
if turn.cancelled.is_set(): # 被打断,立刻退出
r.close()
return
if not line.startswith("data: "):
continue
chunk = line[6:]
if chunk == "[DONE]":
return
delta = parse_delta(chunk) # 解析 delta.content
if delta:
yield delta # 交给 TTS 流式合成
```
打断触发的地方,就在 VAD 检测到用户开口的那一刻:
```python
def on_user_started_speaking(current_turn: Turn):
current_turn.cancelled.set() # 中止上游
player.stop_and_flush() # 停止播放并清空队列
asr.reset() # 重新开始收集
```
浏览器端做同样的事情,用 AbortController 掐掉 fetch 流、用 AudioBufferSourceNode.stop() 停掉已排队的音频节点,思路完全一致。
几个决定体验好坏的细节:
- 回声消除:外放场景下,麦克风会听到助手自己的声音,触发误打断。要么戴耳机,要么接入回声消除(AEC)模块,或者用"仅在 TTS 播放时提高 VAD 阈值"的折中方案。
- 打断后的历史:被打断的那句回答不要整段写进对话历史,只保留已经播出去的部分,否则下一轮模型会以为自己说完了。
- TTS 首包延迟:TTS 也要用流式的,合成一句播一句,不要等整段合成完。
- 填充音:LLM 首字延迟超过 1 秒时,先播一句"嗯,我看一下",能显著降低"是不是卡了"的感受。
步骤 8:实测记录
下面是两段实测的方法和结果形态。具体数字随显卡、模型规格、音频内容差别很大,请以你自己的机器测出来的为准。
实测一:会议录音转写
- 素材:一段约 30 分钟的中文会议录音,含少量英文术语,两人以上说话,有交叠。
- 处理:
ffmpeg转 16 kHz 单声道 → 按 8 分钟切片共 4 段 → 并发 4 请求。 - 观察到的现象:
- 端到端耗时在"音频时长的十分之一量级"以内,远快于实时。也就是说离线批量转写不是瓶颈。
- 中文整体可读,专业术语和产品名容易出错(模型按发音猜了个常见词),数字与单位偶尔写成中文全角。
- 说话人交叠的段落会出现句子合并,读起来像一个人连说了两段。模型不做说话人分离,需要另接分离工具。
- 段边界处的断词,通过段间重叠加去重基本能解决。
- 可用性判断:直接当会议纪要发布还需要人工过一遍,当"录音检索 + 粗稿"已经完全够用。
实测二:实时语音问答
- 链路:麦克风 → VAD → 转写 → LLM 生成 → TTS 播放。
- 一轮问答的延迟拆解(量级示意,不是跑分):
| 环节 | 典型量级 | 影响因素 |
|---|---|---|
| 用户说完 → VAD 判定静音 | 0.5 s 左右 | 静音阈值设置 |
| 提交 → ASR 返回 | 0.3–1 s | 句子长度、显卡、并发 |
| LLM 首 token | 0.3–1 s | 上下文长度、模型规模 |
| TTS 首包 | 0.3 s 左右 | TTS 是否流式 |
| 合计首字可听 | 1.5–3 s 量级 | 上面各项叠加 |
- 打断效果:VAD 检测到人声到声音真正停下,取决于播放缓冲大小。缓冲越小停得越快,但越小越容易播放卡顿。实践中把播放缓冲控制在几百毫秒量级,体验比较自然。
- 结论:这套方案适合"问答式"交互,不适合"抢话式"的快速对话。想要更低的延迟,最有效的三招是:缩小 ASR 句子长度、换更小的 LLM、把 TTS 做成真流式。
成本与延迟的边界,怎么判断
自托管划算不划算,别看"显卡多少钱",算单位成本:
```
每小时音频的转写成本 ≈ 机器每小时成本 ÷ 每小时能处理的音频小时数
机器每小时成本 = 云上按需价(或显卡折旧 + 电费 + 机房摊销)
每小时能处理的音频小时数 ≈ 并发数 × 3600 ÷ 单段推理耗时,再乘一个安全系数
```
判断规则很简单:
- 负载稳定且高(每天都有几小时以上音频要转):自托管的单位成本会明显低于按量计费,而且数据不出内网。
- 负载零散(一周几次、每次几分钟):按量调用 API 更省心,显卡闲着也是在烧钱。
- 对延迟极敏感:自托管能把网络往返省掉,但 ASR + LLM + TTS 三段串起来的延迟下限是你自己的硬件决定的,不会因为"本地"就消失。
具体价格以云厂商与你所在地的电价页面为准,本文不列具体数字。
---
三、常见坑与排错
1. 启动报"不支持的模型结构"
推理框架版本太旧,不认识这个模型。升级框架;如果已经是最新版仍报错,去看官方文档里对该模型的支持状态,可能需要从源码装。
2. 请求报"不支持的模态 / modality"
多模态输入没有在启动参数里放开。检查启动命令里限制每个请求音频数量的那个参数,参数名和格式各版本不同,照抄官方示例。
3. 请求体过大被拒(413)
base64 内联会把体积撑大约三分之一,几分钟的音频就可能到几十 MB。解决办法:切片切得更小;或者在服务端加一个"上传音频取路径"的接口,让模型服务读本地文件而不是走 base64。中间有 Nginx 一类反代时,记得同步调大请求体限制。
4. 转写结果是摘要,不是原文
两个原因:prompt 没写清"逐字转写",或者 temperature 不是 0。两个一起改。
5. 显存 OOM
按影响从大到小依次调:降 --gpu-memory-utilization → 降 --max-num-seqs → 降 --max-model-len → 换量化权重 → 换小模型。注意 OOM 有时不是发生在启动,而是发生在并发上来之后,因为 KV Cache 是动态涨的。
6. 首字延迟很长,后面很快
这是正常的 prefill 阶段:音频越长,第一次出字越慢。要降低首字延迟就缩短单次提交的音频长度,代价是上下文变碎。
7. 数字、单位、专有名词错得离谱
语音模型的通病。三个缓解手段:prompt 里给出术语表;转写后用规则做数字与单位规范化;把高频错误词做成替换表。
8. 切片处断词、重复句
切片不要硬切在词中间,优先按静音切;必须有重叠时,拼接阶段做后缀/前缀匹配去重。
9. 打断后声音还在响
只停了"下一次播放"没清队列,或者没中止上游流式请求。三个取消要一起做。
10. 外放时助手自己把自己打断
麦克风收到了扬声器的声音。戴耳机、上回声消除、或在播放期间提高 VAD 触发阈值。
11. 中文标点全半角混乱
后处理统一一遍,别指望模型每次都规整。
---
四、下一步建议
跑通之后,按投入产出比排序,值得往下做的几件事:
1. 换更好的切分:把固定时长切片换成基于 VAD 的静音切分,转写质量和时间戳精度都会上一个台阶。
2. 加说话人分离:接一个说话人日志(diarization)工具,把"谁在什么时候说了什么"补齐。会议场景这是刚需,语音模型本身不提供。
3. 做术语热词:把你的产品名、人名、缩写整理成一张表,塞进 prompt 或做后处理替换,准确率提升往往比换更大的模型更明显。
4. 建监控指标:至少盯三个数——实时因子(处理耗时 ÷ 音频时长)、首字延迟 P95、转写失败率。有这三个数,扩容和降本的决策才有依据。
5. 缓存重复请求:同一段音频的重复转写做内容哈希缓存,调试阶段能省掉大量等待。
6. 考虑分级策略:长音频批量转写用大模型求质量,实时对话用小的量化模型求延迟,两条链路共用一个服务端、不同的 model 名。这是自托管相比纯 API 最实在的一个优势。
