每日基础课

说话人日志（Speaker Diarization）解决的是“谁在什么时间说了话”（Who spoke when）。

生活类比： 想象你在整理一场没看画面的圆桌会议录音： 1. 第一步：把没人说话的喝水、翻纸等空白杂音用剪刀剪掉（VAD 语音活动检测）。 2. 第二步：把剩下的说话声按每 1.5 秒剪成一条条小纸条（分段 Segmentation）。 3. 第三步：给每张纸条贴上音色特征标签，比如“低沉男声”、“清脆女声”（声纹嵌入 Embedding）。 4. 第四步：把音色相似的纸条堆在同一堆，标上“发言人 A”、“发言人 B”（聚类 Clustering）。

核心流水线公式： $$S = \text{Cluster}(\text{Embed}(\text{Segment}(\text{VAD}(x[n]))))$$

符号解释： - $x[n]$：输入的原始音频波形采样点。 - $\text{VAD}(\cdot)$：语音活动检测器，输出包含人声的时间段，丢弃静音。 - $\text{Segment}(\cdot)$：把连续人声切分成有重叠的均匀短语音块（如 1.5 秒窗，0.75 秒步长）。 - $\text{Embed}(\cdot)$：声纹提取网络（如 x-vector、ECAPA-TDNN），把每个短语音块映射为一个定长声纹向量 $\mathbf{e}_i \in \mathbb{R}^D$。 - $\text{Cluster}(\cdot)$：聚类算法（如谱聚类、层次凝聚聚类 AHC），将几何距离相近的向量归为同一说话人编号 $spk_k$。 - $S$：最终输出的元组集合 $\{(t_{\text{start}}^{(i)}, t_{\text{end}}^{(i)}, spk_i)\}$。

它解决的问题： 在一整段多人混杂长音频中，盲分说话人角色和发言起止点。 没有它会怎样： 语音识别（ASR）只能输出一大段分不清谁说的流水账文字；智能会议、法庭审讯、音视频多角色对话分析将彻底无法区分责任人。

核心思路： 1. 构造信号：用正弦波+高斯白噪合成一段 8 秒音频（0-3秒 Speaker 1, 3-4秒静音, 4-7秒 Speaker 2, 7-8秒静音）。 2. VAD 阶段：计算帧能量，滤除静音帧。 3. 分段与声纹：滑窗截取 1 秒小段，送入一个极简的 PyTorch 卷积网络提取 16 维声纹嵌入。 4. 聚类阶段：用 KMeans 将嵌入向量聚为 2 类，合并连续帧输出最终时间区间。

完整可运行代码如下：

import numpy as np
import torch
import torch.nn as nn
from sklearn.cluster import KMeans

# 1. 合成测试音频 (采样率 16000Hz, 总长 8 秒)
sample_rate = 16000
duration = 8
t = np.linspace(0, duration, duration * sample_rate, endpoint=False)

# Speaker 1: 200Hz 基频; Speaker 2: 600Hz 基频; 其余为静音
audio = np.zeros_like(t)
audio[0*sample_rate : 3*sample_rate] = 0.5 * np.sin(2 * np.pi * 200 * t[0*sample_rate : 3*sample_rate])  # 0~3s Spk1
audio[4*sample_rate : 7*sample_rate] = 0.5 * np.sin(2 * np.pi * 600 * t[4*sample_rate : 7*sample_rate])  # 4~7s Spk2
audio += 0.01 * np.random.randn(len(audio))  # 背景白噪

# 2. 模拟声纹特征提取网络 (Conv1d 提取频域特征 -> 16维 Embedding)
class ToySpeakerEncoder(nn.Module):
    def __init__(self):
        super().__init__()
        self.conv = nn.Sequential(
            nn.Conv1d(1, 8, kernel_size=15, stride=5),
            nn.ReLU(),
            nn.AdaptiveAvgPool1d(16)
        )
    def forward(self, x):
        # x: (B, 1, T) -> out: (B, 16)
        feat = self.conv(x).squeeze(1)
        return nn.functional.normalize(feat, p=2, dim=-1)

encoder = ToySpeakerEncoder().eval()

# 3. 滑动窗口切片 + 能量 VAD + 声纹提取
window_sec = 1.0
hop_sec = 0.5
win_samples = int(window_sec * sample_rate)
hop_samples = int(hop_sec * sample_rate)

segments = []
embeddings = []
energy_threshold = 0.05

for start_idx in range(0, len(audio) - win_samples + 1, hop_samples):
    end_idx = start_idx + win_samples
    chunk = audio[start_idx:end_idx]
    
    # 能量 VAD 判断
    energy = np.sqrt(np.mean(chunk**2))
    if energy < energy_threshold:
        continue  # 静音段直接跳过
    
    # 提取 Embedding
    tensor_chunk = torch.tensor(chunk, dtype=torch.float32).unsqueeze(0).unsqueeze(0)
    with torch.no_grad():
        emb = encoder(tensor_chunk).squeeze(0).numpy()
    
    start_time = start_idx / sample_rate
    end_time = end_idx / sample_rate
    segments.append((start_time, end_time))
    embeddings.append(emb)

# 4. 聚类 (设说话人数 K=2)
embeddings = np.array(embeddings)
kmeans = KMeans(n_clusters=2, random_state=42, n_init=10).fit(embeddings)
labels = kmeans.labels_

# 5. 格式化输出日志
print(f"{'Start (s)':<10}{'End (s)':<10}{'Speaker ID':<10}")
print("-" * 30)
for (st, et), spk in zip(segments, labels):
    print(f"{st:<10.2f}{et:<10.2f}Speaker_{spk}")