PythonAI算法
使用 OpenAI Whisper 与 pyannote.audio 构建说话人分离语音识别系统
介绍如何结合 OpenAI Whisper 语音识别模型与 pyannote.audio 说话人分离管线，构建完整的语音理解系统。内容涵盖技术思路、工程实现流程（音频输入、ASR 转写、说话人分离、时间轴对齐融合）、代码示例及实战中的工程取舍（云端 vs 本地、身份映射、误差处理）。最终输出带说话人标签的结构化文本，适用于客服质检、会议纪要等场景。
介绍如何结合 OpenAI Whisper 语音识别模型与 pyannote.audio 说话人分离管线，构建完整的语音理解系统。内容涵盖技术思路、工程实现流程（音频输入、ASR 转写、说话人分离、时间轴对齐融合）、代码示例及实战中的工程取舍（云端 vs 本地、身份映射、误差处理）。最终输出带说话人标签的结构化文本，适用于客服质检、会议纪要等场景。
本文从工程落地的角度，介绍如何将 OpenAI 的 Whisper 语音识别模型和 pyannote.audio 的说话人分离管线拼成一个'谁在什么时候说了什么'的完整解决方案。
我们会回答三个核心问题：
全文尽量站在'要上线一个能工作的系统'的视角，而不是'能跑就行的 demo'。
通俗来说，场景如下：
这背后的统一问题是：
在一段多说话人的音频 / 视频里，准确回答： 谁 在 什么时候 说了 什么。
拆开来看：
如果只用 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] 好的，那我先从整体架构开始讲...
这就是我们真正想要的'谁在说什么'。
组合拳打完，一个普通 .wav 文件，瞬间就变成了可结构化分析的数据源。
先把整个流程画成一条简单的'数据管道'，心里有个大致地图：
meeting.wav、call.mp3 等；[{start, end, text}, ...][{start, end, speaker}, ...][{start, end, speaker, text}, ...]这条流水线有几个关键点：
下面我们按模块拆开讲。
Whisper 用法有两大类：
openai-whisper 或 faster-whisper）从我们这个任务的角度看，只关心一件事：
能否拿到一串形如
[{start, end, text}, ...]的分段结果。
先安装依赖（示意）：
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 三个字段，后面就可以无缝进入融合步骤。
如果你出于成本 / 隐私考虑想在本地跑 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 的架构，这里只站在'用户视角'看使用方法。
pip install pyannote.audio
然后在 Hugging Face 上：
pyannote/speaker-diarization-community-1YOUR_HF_TOKEN）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']}")
现在你手上有两套时间片段：
segments = [{start, end, text}, ...]speaker_turns = [{start, end, speaker}, ...]接下来，就是时间轴融合的重头戏。
融合的核心思想可以用一句话概括：
'这句话，大部分时间是谁在说，就归谁。'
更形式化一点：
seg：turn；overlap(seg, turn)；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)
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",
})
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 的'成品数据格式'了。
为了避免在项目里四处复制粘贴，我们可以把转写 + 说话人分离 + 融合封装成一个统一函数。
transcribe_and_diarizefrom 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)
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']}")
这样，一整条处理链路就被藏进了一个函数里，外层只需要关心：
其余的，都交给这层封装搞定。
理论路线图画完，落地的时候，通常会遇到一堆非常现实的问题。提前帮你打几个'预防针'。
云端（OpenAI API）优点：
本地 Whisper 优点：
一个常见的折中策略是：
pyannote.audio 给你的 SPEAKER_00 / SPEAKER_01 等，只是'时间上同一说话人的聚类 ID'，它并不知道这个人到底是谁。
如果你需要'识别出张三 / 李四'，还有一整条'说话人识别 / 声纹识别'的路线要走：
建议是：
Whisper 和 pyannote.audio 在时间戳上往往有小量误差：
在大多数业务场景，这种 0.1~0.3 秒级的误差是可以接受的； 但如果你要做的是：
那就需要更谨慎，可以用一些方式做'缓冲'：
start/end 前后各扩展 0.1s；实际部署时，还会遇到这些问题：
这里的经验是：
当你已经拥有 [{start, end, speaker, text}, ...] 这样的结构之后，后面能玩的东西就多了。
给 LLM 喂上下文时，不再只是干巴巴一长串文本，而是明确标出说话人：
SPEAKER_00: 大家好，今天我们来聊一下...
SPEAKER_01: 我先简单介绍一下项目背景。
SPEAKER_00: 好的，那我先从整体架构开始讲...
...
你可以让模型：
SPEAKER_CUSTOMER 的发言）；有了说话人时间轴，这些事情就顺理成章了：
很多'自动会议纪要 + 行动项追踪'的产品，核心其实就是： 说话人分离 + 语音识别 + 一层比较聪明的业务逻辑。
在客服场景里，'谁在说什么'是无数质检规则的底座：
这些本质上都是'基于时间轴的行为分析'，而 Whisper + pyannote.audio 正好给了你构建这条时间轴的工具。
Whisper 让机器听懂了'说了什么'； pyannote.audio 让机器知道'谁在什么时候说话'。
把这两者拼在一起，机器就开始慢慢具备一种更接近人类的'听觉理解能力'——它不再只是一堆文本，而是一场有角色、有结构、有互动的对话。
表面上看，我们只是给转写结果多加了一个 speaker 字段；
实际上，这一列信息往往是从'能用'到'好用'的那一步关键跨越。
如果你已经在用 Whisper 做语音识别，非常建议顺手把 pyannote.audio 串进来试一试；
如果你在玩说话人分离，也不妨用 Whisper 把你的时间轴'填上文字'。
当系统开始真正回答'谁在什么时候说了什么'， 你会发现，后面很多曾经看起来很难的需求，其实离落地也就差一个好点子和几段代码了。
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
使用加密算法（如AES、TripleDES、Rabbit或RC4）加密和解密文本明文。 在线工具，加密/解密文本在线工具，online
生成新的随机RSA私钥和公钥pem证书。 在线工具，RSA密钥对生成器在线工具，online
基于 Mermaid.js 实时预览流程图、时序图等图表，支持源码编辑与即时渲染。 在线工具，Mermaid 预览与可视化编辑在线工具，online
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具，curl 转代码在线工具，online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online