基于 Whisper 与 pyannote.audio 构建说话人分离转写系统

当你把这两条时间轴对齐之后，就能输出更有'人味'的结构：

SPEAKER_00 [0.2–2.1] 大家好，今天我们来聊一下...
SPEAKER_01 [2.3–5.0] 我先简单介绍一下项目背景。
SPEAKER_00 [5.2–8.4] 好的，那我先从整体架构开始讲...

这就是我们真正想要的'谁在说什么'。

• 上游：音频文件（甚至是视频提取的音频轨）
• 中间：Whisper + pyannote.audio
• 下游：检索、质检、摘要、问答、BI 报表……

组合拳打完，一个普通 .wav 文件，瞬间就变成了可结构化分析的数据源。

二、整体架构：从'原始音频'到'可用数据'的流水线

先把整个流程画成一条简单的'数据管道'，心里有个大致地图：

1. 音频输入
   • 一段多说话人的音频，例如 meeting.wav、call.mp3 等；
2. Whisper 语音识别
   • 输出一串带时间戳的文本片段：
   • [{start, end, text}, ...]
3. pyannote.audio 说话人分离
   • 输出一串带说话人 ID 的时间片段：
   • [{start, end, speaker}, ...]
4. 时间轴对齐 & 融合
   • 按时间重叠度，把每条文本片段'分配'给最可能的说话人 ID；
5. 结构化输出
   • 可以是 JSON、Markdown、纯文本：
   • [{start, end, speaker, text}, ...]

这条流水线有几个关键点：

• Whisper 和 pyannote.audio 各自独立运行，只在'时间轴'上交汇；
• 整合步骤是纯 Python 逻辑，不依赖大模型；
• 易于封装成一个函数 / 服务，对外暴露统一接口。

下面我们按模块拆开讲。

三、Whisper 部分：要的是'带时间戳的转写结果'

Whisper 用法有两大类：

1. 用 OpenAI 官方 API 调云端模型
2. 在本地部署开源版（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 上：

1. 搜索并接受使用条款：pyannote/speaker-diarization-community-1
2. 在个人设置里创建一个 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}, ...]

接下来，就是时间轴融合的重头戏。

五、关键步骤：用时间重叠度给文本片段'认爹'（分配说话人）

融合的核心思想可以用一句话概括：

'这句话，大部分时间是谁在说，就归谁。'

更形式化一点：

1. 对于每个 Whisper 文本片段 seg：
2. 找出所有与之有时间重叠的说话人片段 turn；
3. 计算重叠时长 overlap(seg, turn)；
4. 把重叠时长最大的那个 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 把你的时间轴'填上文字'。

当系统开始真正回答'谁在什么时候说了什么'， 你会发现，后面很多曾经看起来很难的需求，其实离落地也就差一个好点子和几段代码了。