用 OpenAI Whisper + pyannote.audio 打造“谁在说什么”的全栈语音理解系统
只做语音识别的系统，只能回答“说了什么”；
只有说话人分离的系统，只能回答“谁在什么时候说话”；
把两者拼在一起，你就有了一个真正能看懂对话的机器。
这篇文章，我们从工程落地的角度，聊一聊：如何把 OpenAI 的 Whisper 语音识别模型，和 pyannote.audio 的说话人分离管线拼成一个“谁在什么时候说了什么”的完整解决方案。
我们会回答这三个核心问题：
- 技术思路：Whisper + pyannote.audio 的组合到底在解决什么问题？
- 工程实现：从一段音频到“带说话人标签的转写结果”，需要哪些关键步骤？
- 实战建议：在真实业务里，这种方案要怎么做取舍、怎么优化？
全文尽量站在“要上线一个能工作的系统”的视角，而不是“能跑就行的 demo”。
一、为什么一定要把 Whisper 和 pyannote.audio 拼在一起？
把场景先说人话一点：
- 客服中心想知道：客户在什么时候提了哪些问题，座席是怎么回应的？
- B 端会议系统想自动生成：带说话人标签的会议纪要，谁提出了什么决策，谁接了什么任务。
- 播客 / 访谈节目希望自动生成：按嘉宾分角色的文字稿，甚至还能按人检索“这个嘉宾都说了啥”。
这背后的统一问题是：
在一段多说话人的音频 / 视频里，准确回答：
谁 在 什么时候 说了 什么。
拆开来看：
- Whisper 负责把「声音 → 文本」，告诉你内容；
- pyannote.audio 负责把「声音 → 说话人时间轴」，告诉你结构（谁在什么时候说话）。
如果只用 Whisper，通常拿到的是这样的结构：
[ {"start": 0.5, "end": 3.2, "text": "大家好，今天我们来聊一下..."}, {"start": 3.3, "end": 7.8, "text": "我先简单介绍一下项目背景。"} ] 如果只用 pyannote.audio，说话人分离给你的是这样的：
0.20s–2.10s SPEAKER_00 2.30s–5.00s SPEAKER_01 5.20s–8.40s SPEAKER_00 ... 当你把这两条时间轴对齐之后，就能输出更有“人味”的结构：
SPEAKER_00 [0.2–2.1] 大家好，今天我们来聊一下... SPEAKER_01 [2.3–5.0] 我先简单介绍一下项目背景。 SPEAKER_00 [5.2–8.4] 好的，那我先从整体架构开始讲... 这就是我们真正想要的“谁在说什么”。
- 上游：音频文件（甚至是视频提取的音频轨）
- 中间：Whisper + pyannote.audio
- 下游：检索、质检、摘要、问答、BI 报表……
组合拳打完，一个普通 .wav 文件，瞬间就变成了可结构化分析的数据源。
二、整体架构：从“原始音频”到“可用数据”的流水线
先把整个流程画成一条简单的“数据管道”，心里有个大致地图：
- 音频输入
- 一段多说话人的音频，例如
meeting.wav、call.mp3等；
- 一段多说话人的音频，例如
- Whisper 语音识别
- 输出一串带时间戳的文本片段：
[{start, end, text}, ...]
- pyannote.audio 说话人分离
- 输出一串带说话人 ID 的时间片段：
[{start, end, speaker}, ...]
- 时间轴对齐 & 融合
- 按时间重叠度，把每条文本片段“分配”给最可能的说话人 ID；
- 结构化输出
- 可以是 JSON、Markdown、纯文本：
[{start, end, speaker, text}, ...]
这条流水线有几个关键点：
- Whisper 和 pyannote.audio 各自独立运行，只在“时间轴”上交汇；
- 整合步骤是纯 Python 逻辑，不依赖大模型；
- 易于封装成一个函数 / 服务，对外暴露统一接口。
下面我们按模块拆开讲。
三、Whisper 部分：要的是“带时间戳的转写结果”
Whisper 用法有两大类：
- 用 OpenAI 官方 API 调云端模型
- 在本地部署开源版（
openai-whisper或faster-whisper）
从我们这个任务的角度看，只关心一件事：
能否拿到一串形如 [{start, end, text}, ...] 的分段结果。3.1 用 OpenAI 官方 API
先安装依赖（示意）：
pip install openai pip install python-dotenv # 用来管理 API Key（可选） 下面是一个典型的调用方式（注意：具体参数名需根据你当前使用的 OpenAI SDK 版本调整，这里强调的是思路和结构）：
from openai import OpenAI client = OpenAI(api_key="YOUR_OPENAI_API_KEY") audio_file_path = "audio.wav" with open(audio_file_path, "rb") as f: transcription = client.audio.transcriptions.create( model="whisper-1", # 或其他支持语音识别的模型 file=f, response_format="verbose_json", # 拿到详细分段和时间戳 timestamp_granularities=["segment"], language="zh" # 或 "en" / "auto" ) segments = [ { "start": seg["start"], "end": seg["end"], "text": seg["text"].strip(), } for seg in transcription.segments ] for seg in segments: print(f"[{seg['start']:.2f}–{seg['end']:.2f}] {seg['text']}") 这里有两个关键点：
response_format="verbose_json"：拿到分段信息；timestamp_granularities=["segment"]：告诉服务“我要时间戳”。
只要 segments 里有 start / end / text 三个字段，后面就可以无缝进入融合步骤。
3.2 用本地 Whisper（可选）
如果你出于成本 / 隐私考虑想在本地跑 Whisper，大致调用方式是这样的：
import whisper model = whisper.load_model("medium") # 或 tiny/base/small/large result = model.transcribe("audio.wav", language="zh") segments = [ { "start": seg["start"], "end": seg["end"], "text": seg["text"].strip(), } for seg in result["segments"] ] 只要输出结构类似，后面的代码不用任何改动。
四、pyannote.audio 部分：要的是“谁在什么时候说话”
前一篇我们已经拆过 pyannote.audio 的架构，这里只站在“用户视角”看使用方法。
4.1 安装和授权
pip install pyannote.audio 然后在 Hugging Face 上：
- 搜索并接受使用条款：
pyannote/speaker-diarization-community-1 - 在个人设置里创建一个 Access Token（记为
YOUR_HF_TOKEN）
4.2 调用说话人分离管线
from pyannote.audio import Pipeline pipeline = Pipeline.from_pretrained( "pyannote/speaker-diarization-community-1", use_auth_token="YOUR_HF_TOKEN", # 新版可用 token=... ) diarization = pipeline("audio.wav") speaker_turns = [] for turn, speaker in diarization.itertracks(yield_label=True): speaker_turns.append({ "start": float(turn.start), "end": float(turn.end), "speaker": str(speaker), }) for t in speaker_turns: print(f"[{t['start']:.2f}–{t['end']:.2f}] {t['speaker']}") 现在你手上有两套时间片段：
- Whisper：
segments = [{start, end, text}, ...] - pyannote：
speaker_turns = [{start, end, speaker}, ...]
接下来，就是时间轴融合的重头戏。
五、关键步骤：用时间重叠度给文本片段“认爹”（分配说话人）
融合的核心思想可以用一句话概括：
“这句话，大部分时间是谁在说，就归谁。”
更形式化一点：
- 对于每个 Whisper 文本片段
seg： - 找出所有与之有时间重叠的说话人片段
turn； - 计算重叠时长
overlap(seg, turn)； - 把重叠时长最大的那个
speaker赋给该文本片段。
5.1 计算时间重叠的辅助函数
def overlap(a_start, a_end, b_start, b_end) -> float: left = max(a_start, b_start) right = min(a_end, b_end) return max(0.0, right - left) 5.2 完整的融合函数
from typing import List, Dict def assign_speaker_to_segments( segments: List[Dict], speaker_turns: List[Dict], ) -> List[Dict]: """为每个 Whisper 文本片段分配说话人 ID。 Parameters ---------- segments : list of dict 每个元素形如 {"start": float, "end": float, "text": str} speaker_turns : list of dict 每个元素形如 {"start": float, "end": float, "speaker": str} Returns ------- list of dict 每个元素形如 {"start", "end", "text", "speaker"} """ def overlap(a_start, a_end, b_start, b_end) -> float: left = max(a_start, b_start) right = min(a_end, b_end) return max(0.0, right - left) results = [] for seg in segments: seg_start, seg_end = seg["start"], seg["end"] best_speaker = None best_overlap = 0.0 for turn in speaker_turns: ov = overlap(seg_start, seg_end, turn["start"], turn["end"]) if ov > best_overlap: best_overlap = ov best_speaker = turn["speaker"] results.append({ "start": seg_start, "end": seg_end, "text": seg["text"], "speaker": best_speaker or "UNKNOWN", }) return results 调用示例：
final_segments = assign_speaker_to_segments(segments, speaker_turns) for seg in final_segments: print(f"{seg['speaker']} [{seg['start']:.2f}–{seg['end']:.2f}] {seg['text']}") 这样你就得到了一份结构大致如下的结果：
[ { "start": 0.5, "end": 3.2, "text": "大家好，今天我们来聊一下...", "speaker": "SPEAKER_00" }, { "start": 3.3, "end": 7.8, "text": "我先简单介绍一下项目背景。", "speaker": "SPEAKER_01" } ] ——这就已经是一个可以直接喂给前端、数据库、或者下游 LLM 的“成品数据格式”了。
六、封装成一个可复用的高层 API
为了避免在项目里四处复制粘贴，我们可以把转写 + 说话人分离 + 融合封装成一个统一函数。
6.1 高层封装：transcribe_and_diarize
from typing import List, Dict from openai import OpenAI from pyannote.audio import Pipeline def transcribe_and_diarize( audio_path: str, openai_client: OpenAI, whisper_model: str, diarization_pipeline: Pipeline, ) -> List[Dict]: """对单个音频做转写 + 说话人分离，并融合结果。 返回形如 [{start, end, speaker, text}, ...] 的列表。 """ # 1) Whisper 转写 with open(audio_path, "rb") as f: transcription = openai_client.audio.transcriptions.create( model=whisper_model, file=f, response_format="verbose_json", timestamp_granularities=["segment"], ) segments = [ { "start": seg["start"], "end": seg["end"], "text": seg["text"].strip(), } for seg in transcription.segments ] # 2) 说话人分离 diarization = diarization_pipeline(audio_path) speaker_turns = [ { "start": float(turn.start), "end": float(turn.end), "speaker": str(speaker), } for turn, speaker in diarization.itertracks(yield_label=True) ] # 3) 时间轴融合 return assign_speaker_to_segments(segments, speaker_turns) 6.2 实际调用长什么样？
from openai import OpenAI from pyannote.audio import Pipeline client = OpenAI(api_key="YOUR_OPENAI_API_KEY") diar_pipeline = Pipeline.from_pretrained( "pyannote/speaker-diarization-community-1", use_auth_token="YOUR_HF_TOKEN", ) results = transcribe_and_diarize( "audio.wav", openai_client=client, whisper_model="whisper-1", # 或其他支持的模型 diarization_pipeline=diar_pipeline, ) for r in results: print(f"{r['speaker']} [{r['start']:.2f}–{r['end']:.2f}] {r['text']}") 这样，一整条处理链路就被藏进了一个函数里，外层只需要关心：
- 音频在哪；
- 用哪个 Whisper 模型；
- 用哪个说话人分离管线。
其余的，都交给这层封装搞定。
七、实战中的几个现实问题与工程取舍
理论路线图画完，落地的时候，通常会遇到一堆非常现实的问题。提前帮你打几个“预防针”。
7.1 Whisper：云端 vs 本地
云端（OpenAI API）优点：
- 不用管模型部署、GPU 资源、负载均衡；
- 模型持续更新，新版本上线你直接可用；
- 对于中小规模调用来说，开发效率极高。
本地 Whisper 优点：
- 大规模离线处理时，长期成本可控；
- 对数据合规 / 隐私要求高时更安心（音频不出内网）；
- 可以更细致地控制 batch、并发、缓存等细节。
一个常见的折中策略是：
- POC / 内部试点 / 小流量阶段：先用 OpenAI API；
- 确认效果、场景、ROI 后，再评估是否迁移到本地部署。
7.2 说话人 ID 与“真实身份”的映射问题
pyannote.audio 给你的 SPEAKER_00 / SPEAKER_01 等，只是“时间上同一说话人的聚类 ID”，它并不知道这个人到底是谁。
如果你需要“识别出张三 / 李四”，还有一整条“说话人识别 / 声纹识别”的路线要走：
- 用说话人验证模型（Speaker Verification）对比声纹；
- 或结合视频做人脸识别，然后做跨模态匹配；
- 或者在业务侧某些角色是已知的（例如：坐席是已知 ID，客户是未知 ID）。
建议是：
- 先把“分人说话”的问题做好，即我们这篇文章解决的事情；
- 再按需一点点加上“谁是谁”的逻辑，而不是一上来就同时搞定。
7.3 时间戳误差与边界模糊
Whisper 和 pyannote.audio 在时间戳上往往有小量误差：
- 前处理方式不同（重采样、静音截断等）；
- 模型对边界的判断不同；
- Whisper 的 segment 粒度有时会比较粗。
在大多数业务场景，这种 0.1~0.3 秒级的误差是可以接受的；
但如果你要做的是：
- 合规审计（比如“打断时长超过 1 秒是否违规”）；
- 精确到帧的裁剪 / 对齐；
那就需要更谨慎，可以用一些方式做“缓冲”：
- 在计算重叠时，把 Whisper 文本片段的
start/end前后各扩展 0.1s； - 对特别敏感的规则，统一以 pyannote.audio 的 VAD / 分段时间轴 为基准。
7.4 性能与并发
实际部署时，还会遇到这些问题：
- 如何同时处理多路音频（线程池 / 进程池 / 队列 / K8s）；
- 如何避免重复加载模型（Whisper / pyannote 模型常驻内存）；
- 如何缓存处理结果（同一文件多次被查询时直接走缓存）。
这里的经验是：
- 把“处理单个音频”的逻辑写成纯函数风格；
- 把模型实例、客户端（OpenAI Client、Pipeline）放在更高一层管理；
- 预留日志、监控、指标埋点，方便后面排查“哪一步慢 / 哪一步出错”。
八、延伸玩法：有了“谁在说什么”，还能玩什么花样？
当你已经拥有 [{start, end, speaker, text}, ...] 这样的结构之后，后面能玩的东西就多了。
8.1 带说话人语境感知的摘要 & 问答
给 LLM 喂上下文时，不再只是干巴巴一长串文本，而是明确标出说话人：
SPEAKER_00: 大家好，今天我们来聊一下... SPEAKER_01: 我先简单介绍一下项目背景。 SPEAKER_00: 好的，那我先从整体架构开始讲... ... 你可以让模型：
- 总结“客户”说了什么（只看
SPEAKER_CUSTOMER的发言）； - 总结某个嘉宾的观点合集；
- 针对某个说话人的发言做评价或建议（例如“给销售的反馈”）。
8.2 会议信息结构化
有了说话人时间轴，这些事情就顺理成章了：
- 每个人的发言时长、轮次数量；
- 谁提出了议题，谁给出了决策；
- 发言打断、插话频次（尤其在销售、谈判、教练等场景）。
很多“自动会议纪要 + 行动项追踪”的产品，核心其实就是：
说话人分离 + 语音识别 + 一层比较聪明的业务逻辑。
8.3 客服质检与智能辅导
在客服场景里，“谁在说什么”是无数质检规则的底座：
- 是否出现“长时间客户独自讲话而坐席没反馈”？
- 是否频繁出现“坐席打断客户”？
- 是否按要求完成了“身份核验 / 风险提示 / 总结回顾”？
这些本质上都是“基于时间轴的行为分析”，而 Whisper + pyannote.audio 正好给了你构建这条时间轴的工具。
九、结语：让时间轴长出“人”的轮廓
Whisper 让机器听懂了“说了什么”；
pyannote.audio 让机器知道“谁在什么时候说话”。
把这两者拼在一起，机器就开始慢慢具备一种更接近人类的“听觉理解能力”——它不再只是一堆文本，而是一场有角色、有结构、有互动的对话。
表面上看，我们只是给转写结果多加了一个 speaker 字段；
实际上，这一列信息往往是从“能用”到“好用”的那一步关键跨越。
如果你已经在用 Whisper 做语音识别，非常建议顺手把 pyannote.audio 串进来试一试；
如果你在玩说话人分离，也不妨用 Whisper 把你的时间轴“填上文字”。
当系统开始真正回答“谁在什么时候说了什么”，
你会发现，后面很多曾经看起来很难的需求，其实离落地也就差一个好点子和几段代码了。
