每日基础课

【生活类比】 想象你坐在隔壁房间听隔壁开会，你看不到人，也没有事先录入他们的声音： 1. 第一步：你先用笔划掉完全没人说话的安静死寂时间（静音过滤 / VAD）。 2. 第二步：你把剩下的谈话切成一段段几秒钟的录音卡片（分段 / Segmentation）。 3. 第三步：你仔细听每张卡片的声音特质（音调、共振峰），在卡片背面写下一个“声音特征码”（提取声纹 Embedding）。 4. 第四步：你把特征码接近的卡片堆在一起，贴上“蓝色标签（人A）”、“黄色标签（人B）”（聚类 / Clustering）。 5. 最后：按时间轴把卡片拼回去，就得到了一份完整的发言时间表。

【核心公式与符号】 说话人日志的核心可以用两步数学映射来概括： 1. 声纹抽取映射： $$\mathbf{e}_i = f_{\theta}(\mathbf{x}_i)$$ - $\mathbf{x}_i$：第 $i$ 个切片窗口的音频特征（如梅尔频谱）。 - $f_{\theta}$：预训练声纹编码器（如 ECAPA-TDNN 或 x-vector 网络），$\theta$ 是网络参数。 - $\mathbf{e}_i$：提取出的固定维度声纹向量（通常为 192 或 512 维）。

2. 聚类与打标： $$\hat{y}_i = \arg\min_{k \in \{1,\dots,K\}} \mathcal{D}(\mathbf{e}_i, \mathbf{c}_k)$$ - $\mathbf{c}_k$：第 $k$ 个说话人的聚类中心向量。 - $\mathcal{D}(\cdot, \cdot)$：距离度量函数（通常是余弦距离 $1 - \cos(\mathbf{e}_i, \mathbf{c}_k)$）。 - $\hat{y}_i$：第 $i$ 个切片最终被判定的说话人编号。

【解决什么问题？没有它会怎样？】 - 它解决会议纪要、庭审记录、音视频访谈中“发言人归属”问题。 - 如果没有说话人日志：自动语音识别（ASR）吐出来的就是一长串没有标明发言人的文字流水账。你无法区分哪句话是面试官问的，哪句话是求职者答的，后续的大模型总结和视频人物高亮将完全乱套。

【在主流系统中的位置】 在语音处理全流程中，它通常与 ASR 并行或作为后处理模块运行： 音频输入 $\to$ [ASR 转写文字] + [说话人日志打时间标签] $\to$ [时间对齐模块] $\to$ 输出“张三 00:01-00:05：大家早上好”。

【实现思路】 1. 构造合成特征：生成一段包含 说话人A（高频分布） $\to$ 静音 $\to$ 说话人B（低频分布） 的时序特征。 2. 能量 VAD：计算每帧能量，过滤掉低于阈值的静音帧。 3. 窗口声纹编码：用一个简单的 PyTorch 编码器将窗口内多帧压缩为一个固定维度的 Embedding。 4. 余弦相似度聚类：计算切片之间的相似度矩阵，使用层次聚类（或简单的余弦阈值聚类）打上标签。 5. 结果聚合：将连续相同标签的时间步合并为时间段区间。

【完整可运行代码】

import torch
import torch.nn as nn
import torch.nn.functional as F
import numpy as np

# 固定随机种子，保证每次运行结果一致
torch.manual_seed(42)
np.random.seed(42)

# 1. 模拟生成音频梅尔特征 (时间帧数=100, 频带数=40)
# 0-35帧: 说话人A (均值偏正)
# 35-55帧: 静音段 (极小能量噪声)
# 55-100帧: 说话人B (均值偏负)
time_frames = 100
feat_dim = 40

spk_a = torch.randn(35, feat_dim) + 2.0
silence = torch.randn(20, feat_dim) * 0.01
spk_b = torch.randn(45, feat_dim) - 2.0
features = torch.cat([spk_a, silence, spk_b], dim=0)  # Shape: [100, 40]

# 2. 简易 VAD (计算每帧能量/L2模长)
frame_energy = torch.norm(features, dim=-1)
vad_mask = frame_energy > 0.5  # 过滤静音
print(f"总帧数: {time_frames}, 活跃语音帧数: {vad_mask.sum().item()}")

# 3. 简易声纹提取网络 (将 [Window_Size, Feat_Dim] 压缩为 [Embed_Dim])
class ToySpeakerEncoder(nn.Module):
    def __init__(self, in_dim=40, embed_dim=16):
        super().__init__()
        self.net = nn.Sequential(
            nn.Linear(in_dim, 32),
            nn.ReLU(),
            nn.Linear(32, embed_dim)
        )
    def forward(self, x):
        # x: [Window_Len, Feat_Dim] -> 帧平均池化 -> 映射
        pooled = torch.mean(x, dim=0)
        embed = self.net(pooled)
        return F.normalize(embed, p=2, dim=0)  # L2归一化，方便余弦相似度计算

encoder = ToySpeakerEncoder()
encoder.eval()

# 4. 滑动窗口提取 Embedding
window_size = 10
step = 5
segments = []
embeddings = []

for start in range(0, time_frames - window_size + 1, step):
    end = start + window_size
    # 如果窗口内大部分是有效语音，则保留
    if vad_mask[start:end].float().mean() > 0.6:
        win_feat = features[start:end]
        with torch.no_grad():
            emb = encoder(win_feat)
        embeddings.append(emb.numpy())
        segments.append((start, end))

embeddings = np.array(embeddings)  # Shape: [Num_Segments, 16]
print(f"有效切片数量: {len(segments)}")

# 5. 简单的阈值聚类 (基于余弦相似度)
speaker_labels = []
cluster_centers = []
sim_threshold = 0.7  # 余弦相似度阈值

for emb in embeddings:
    if len(cluster_centers) == 0:
        cluster_centers.append(emb)
        speaker_labels.append(0)
    else:
        # 计算与已有聚类中心的余弦相似度 (已归一化，直接点积)
        sims = [np.dot(emb, c) for c in cluster_centers]
        max_sim = max(sims)
        best_spk = int(np.argmax(sims))
        
        if max_sim >= sim_threshold:
            speaker_labels.append(best_spk)
            # 简易更新聚类中心
            cluster_centers[best_spk] = (cluster_centers[best_spk] + emb) / 2.0
            cluster_centers[best_spk] /= np.linalg.norm(cluster_centers[best_spk])
        else:
            # 发现新说话人
            new_id = len(cluster_centers)
            cluster_centers.append(emb)
            speaker_labels.append(new_id)

# 6. 输出时序日志结果
print("\n--- 说话人日志时间线 ---")
current_spk = speaker_labels[0]
seg_start = segments[0][0]
seg_end = segments[0][1]

for i in range(1, len(speaker_labels)):
    spk = speaker_labels[i]
    start, end = segments[i]
    if spk == current_spk and start <= seg_end:
        seg_end = end  # 连续同人，合并区间
    else:
        print(f"帧 {seg_start:03d} -> {seg_end:03d} | 说话人: Speaker_{current_spk}")
        current_spk = spk
        seg_start = start
        seg_end = end
print(f"帧 {seg_start:03d} -> {seg_end:03d} | 说话人: Speaker_{current_spk}")