話者分離の評価指標を完全解説｜DER・JER・Purity・Boundary Errorの計算方法と使い分け

from pyannote.metrics.diarization import DiarizationErrorRate
from pyannote.core import Annotation, Segment

# 正解アノテーション
reference = Annotation()
reference[Segment(0, 5)] = 'SPEAKER_A'
reference[Segment(5, 10)] = 'SPEAKER_B'
reference[Segment(10, 15)] = 'SPEAKER_A'

# システム出力
hypothesis = Annotation()
hypothesis[Segment(0, 4)] = 'SPEAKER_1'   # SPEAKER_Aに対応
hypothesis[Segment(4, 11)] = 'SPEAKER_2'  # 一部混在
hypothesis[Segment(11, 15)] = 'SPEAKER_1'

metric = DiarizationErrorRate()
der = metric(reference, hypothesis)
print(f"DER: {der:.3f}")  # 例: 0.133 (13.3%)

# 成分ごとの詳細
detail = metric(reference, hypothesis, detailed=True)
print(f"  Miss: {detail['missed detection']:.3f}")
print(f"  FA:   {detail['false alarm']:.3f}")
print(f"  Conf: {detail['confusion']:.3f}")

# DERは長い話者の誤りが総合スコアを支配する
# 例: SPEAKER_Aが900秒、SPEAKER_Bが100秒
# SPEAKER_Bを完全に誤っても DER への影響は小さい

# JERはすべての話者を等重みで扱う
# → 少ない発話量の話者の精度も公平に評価される

from pyannote.metrics.diarization import JaccardErrorRate

jer = JaccardErrorRate()
jer_score = jer(reference, hypothesis)
print(f"JER: {jer_score:.3f}")

def compute_speaker_confusion_matrix(
    reference: Annotation,
    hypothesis: Annotation,
) -> dict:
    ref_labels = sorted(map(str, reference.labels()))
    hyp_labels = sorted(map(str, hypothesis.labels()))

    # 全ペアの重複時間（秒）を計算
    overlap_by_pair = {
        ref_label: {hyp_label: 0.0 for hyp_label in hyp_labels}
        for ref_label in ref_labels
    }
    for ref_seg, _, ref_label in reference.itertracks(yield_label=True):
        for hyp_seg, _, hyp_label in hypothesis.itertracks(yield_label=True):
            intersection = ref_seg & hyp_seg
            if intersection:
                overlap_by_pair[str(ref_label)][str(hyp_label)] += float(
                    intersection.duration
                )

    # 正解話者ごとにシェアを計算してサマリーを作る
    per_reference_speaker = {}
    for ref_label, row in overlap_by_pair.items():
        ref_duration = sum(
            float(seg.duration)
            for seg, _, lbl in reference.itertracks(yield_label=True)
            if str(lbl) == ref_label
        )
        overlaps = sorted(
            [
                {"hyp_speaker": h, "overlap_sec": s,
                 "share": s / ref_duration if ref_duration > 0 else 0.0}
                for h, s in row.items() if s > 0.0
            ],
            key=lambda x: -x["overlap_sec"],
        )
        dominant = overlaps[0] if overlaps else None
        per_reference_speaker[ref_label] = {
            "ref_duration_sec": ref_duration,
            "hyp_count": len(overlaps),
            "dominant": dominant["hyp_speaker"] if dominant else None,
            "dominant_share": dominant["share"] if dominant else 0.0,
            "overlaps": overlaps,
        }

    return {"seconds": overlap_by_pair, "per_reference_speaker": per_reference_speaker}

GT 6: hyp_count=4 dominant=speaker_9 share=49.6%
      overlaps=[speaker_9:49.6%, speaker_7:43.9%, speaker_10:3.4%, speaker_1:1.3%]

from pyannote.metrics.diarization import (
    DiarizationPurity,
    DiarizationCoverage,
    DiarizationErrorRate,
)

# DERと同じcollar/skip_overlap設定を揃えることが重要
collar = 0.0
skip_overlap = False

purity_metric   = DiarizationPurity(collar=collar, skip_overlap=skip_overlap)
coverage_metric = DiarizationCoverage(collar=collar, skip_overlap=skip_overlap)

purity   = float(purity_metric(reference, hypothesis))
coverage = float(coverage_metric(reference, hypothesis))

print(f"Purity:   {purity:.3f}")    # 1.0 が理想
print(f"Coverage: {coverage:.3f}")  # 1.0 が理想

# Purityが高くCoverageが低い → split / fragmentation が支配的
# Purityが低くCoverageが高い → merge が支配的（複数話者が1クラスタに混在）

# 実測値（15話者会議音声）
# GT 1: split_count=57, hyp_count=8  ← 最長話者(144s)が8クラスタに分裂
# GT 2: split_count=42, hyp_count=4
# GT 6: split_count=17, hyp_count=4

# duration_weighted_mean_fragment_count で全体傾向を把握する
# → 平均的に1人が何クラスタに分裂しているか

def compute_count_error(
    reference: Annotation,
    hypothesis: Annotation,
    timeline,  # DER の uemify() が返す評価タイムライン
) -> dict:
    """
    話者数誤りをタイムライン上で時間加重計算する。
    単純な ref/hyp 話者総数の差ではなく、
    各時刻フレームでの活性話者数のずれを集計する。
    """
    total_duration   = 0.0
    weighted_abs_err = 0.0
    weighted_sgn_err = 0.0
    exact_match_dur  = 0.0

    for segment in timeline:
        duration  = float(segment.duration)
        ref_count = len(reference.get_labels(segment, unique=False))
        hyp_count = len(hypothesis.get_labels(segment, unique=False))
        diff = hyp_count - ref_count

        total_duration   += duration
        weighted_abs_err += duration * abs(diff)
        weighted_sgn_err += duration * diff
        if diff == 0:
            exact_match_dur += duration

    return {
        "mean_abs_speaker_count_error": weighted_abs_err / total_duration,
        "mean_signed_speaker_count_error": weighted_sgn_err / total_duration,
        "exact_count_match_ratio": exact_match_dur / total_duration,
    }

# DERの評価タイムラインを取得してから呼ぶ
der_metric = DiarizationErrorRate(collar=0.0, skip_overlap=False)
ref_proj, hyp_proj, timeline = der_metric.uemify(
    reference, hypothesis,
    collar=0.0, skip_overlap=False, returns_timeline=True,
)
result = compute_count_error(ref_proj, hyp_proj, timeline)
print(f"mean_abs:          {result['mean_abs_speaker_count_error']:.4f}")
print(f"exact_match_ratio: {result['exact_count_match_ratio']:.4f}")

正解: A----A----A|B-----B|A---A
推定: A----A------|B----B-----|A---
                ↑           ↑
              境界の時間ずれ（Boundary Error）

def extract_boundaries(annotation: Annotation) -> list[float]:
    """アノテーション内の全セグメントの start/end を境界として返す"""
    boundaries: set[float] = set()
    for segment in annotation.itersegments():
        boundaries.add(round(float(segment.start), 9))
        boundaries.add(round(float(segment.end), 9))
    return sorted(boundaries)


def compute_boundary_error(
    reference: Annotation,
    hypothesis: Annotation,
    tolerance: float = 0.5,  # デフォルト 500ms
) -> dict:
    ref_boundaries = extract_boundaries(reference)
    hyp_boundaries = list(extract_boundaries(hypothesis))  # mutable: マッチしたら除去

    matched_pairs = []
    hyp_idx = 0

    for ref_b in ref_boundaries:
        # スキャン位置を tolerance 範囲の左端まで進める
        while (hyp_idx < len(hyp_boundaries)
               and hyp_boundaries[hyp_idx] < ref_b - tolerance):
            hyp_idx += 1

        # 近傍候補の中で最も近い仮説境界を1つ選ぶ
        candidates = []
        for ci in (hyp_idx - 1, hyp_idx, hyp_idx + 1):
            if 0 <= ci < len(hyp_boundaries):
                dist = abs(hyp_boundaries[ci] - ref_b)
                if dist <= tolerance:
                    candidates.append((dist, ci))

        if not candidates:
            continue

        _, best_ci = min(candidates)
        matched_pairs.append((ref_b, hyp_boundaries[best_ci]))
        hyp_boundaries.pop(best_ci)      # 使用済み境界を除去
        if best_ci < hyp_idx:
            hyp_idx -= 1

    matched   = len(matched_pairs)
    ref_total = len(ref_boundaries)
    hyp_total = matched + len(hyp_boundaries)  # 残りはFP（過剰検出）

    precision = matched / hyp_total if hyp_total else 1.0
    recall    = matched / ref_total if ref_total else 1.0
    f1        = 2 * precision * recall / (precision + recall) if precision + recall else 0.0
    errors    = [abs(r - h) for r, h in matched_pairs]

    return {
        "precision": precision,
        "recall": recall,
        "f1": f1,
        "mean_abs_error_sec": sum(errors) / matched if matched else None,
        "max_abs_error_sec": max(errors) if errors else None,
    }

result = compute_boundary_error(reference, hypothesis, tolerance=0.5)
print(f"Precision: {result['precision']:.3f}")
print(f"Recall:    {result['recall']:.3f}")
print(f"F1:        {result['f1']:.3f}")
if result["mean_abs_error_sec"]:
    print(f"Mean loc error: {result['mean_abs_error_sec']*1000:.1f}ms")

UTTERANCE_LENGTH_BINS = [
    ("0_to_1s",   0.0,  1.0),
    ("1_to_2s",   1.0,  2.0),
    ("2_to_5s",   2.0,  5.0),
    ("5_to_10s",  5.0, 10.0),
    ("10s_plus", 10.0, None),
]

def compute_utterance_length_recall(
    reference: Annotation,
    mapped_hypothesis: Annotation,  # optimal_mapping 適用済みの仮説
) -> dict:
    """
    発話長ビンごとの Recall を計算する。
    Recall の定義:
      その発話区間のうち「同一話者として」カバーされた時間 / 発話長
    """
    # 仮説側を話者ラベルでインデックス化（区間リストに整形）
    hyp_by_label: dict[str, list[tuple[float, float]]] = {}
    for seg, _, lbl in mapped_hypothesis.itertracks(yield_label=True):
        hyp_by_label.setdefault(str(lbl), []).append(
            (float(seg.start), float(seg.end))
        )

    grouped = {name: {"count": 0, "ref_dur": 0.0, "matched_dur": 0.0}
               for name, *_ in UTTERANCE_LENGTH_BINS}

    for ref_seg, _, ref_lbl in reference.itertracks(yield_label=True):
        dur = float(ref_seg.duration)
        lbl = str(ref_lbl)

        # 仮説側の同一話者ラベル区間との重複を計算
        matched = sum(
            max(0.0, min(float(ref_seg.end), h_end)
                   - max(float(ref_seg.start), h_start))
            for h_start, h_end in hyp_by_label.get(lbl, [])
        )
        bin_name = next(
            name for name, lo, hi in UTTERANCE_LENGTH_BINS
            if dur >= lo and (hi is None or dur < hi)
        )
        grouped[bin_name]["count"]       += 1
        grouped[bin_name]["ref_dur"]     += dur
        grouped[bin_name]["matched_dur"] += matched

    results = {}
    total_ref = total_matched = 0.0
    for name, stats in grouped.items():
        rd = stats["ref_dur"]
        results[name] = {
            "utterance_count": stats["count"],
            "duration_weighted_recall": stats["matched_dur"] / rd if rd > 0 else 0.0,
        }
        total_ref     += rd
        total_matched += stats["matched_dur"]

    results["overall_duration_weighted"] = total_matched / total_ref if total_ref else 0.0
    return results

utterance_length_recall:
  duration_weighted = 0.8473  ← 全体の時間ベース再現率
  macro             = 0.8071  ← 発話本数で等重みの再現率

DERが高い
    │
    ├─ MISS成分が大 → VADの感度を上げる
    │
    ├─ FA成分が大  → VADの閾値を上げる・ノイズフィルタ強化
    │
    └─ CONF成分が大
           │
           ├─ Count Errorも大 → 話者数推定の改善（クラスタリングのk選択）
           │
           ├─ Purityが低い  → クラスタが複数話者を含む → クラスタリング粒度を細かく
           │
           ├─ Coverageが低い → 1話者が複数クラスタに分散 → マージ戦略を見直す
           │
           ├─ Boundary F1が低い → 発話境界検出の精度問題
           │
           └─ Short recall が低い → VAD・セグメンテーションの短発話対応

| Metric            | System A | System B | Baseline |
|-------------------|----------|----------|----------|
| DER (%)           | 12.3     | 10.8     | 18.5     |
| JER (%)           | 15.6     | 13.2     | 22.1     |
| Purity            | 0.921    | 0.943    | 0.872    |
| Coverage          | 0.884    | 0.901    | 0.821    |
| Count Error (avg) | 0.4      | 0.3      | 1.2      |
| Boundary F1       | 0.823    | 0.871    | 0.741    |
| Short ULR (<1s)   | 0.51     | 0.68     | 0.38     |
| Collar (sec)      | 0.25     | 0.25     | 0.25     |
| Dataset           | AMI      | AMI      | AMI      |

DER:       17.69%  → 内訳: confusion 12.61% / missed 2.67% / FA 2.42%
JER:       37.36%
Purity:    97.31%
Coverage:  88.63%
ref/hyp:   15人 / 17人
Count Error: mean_abs=0.050  exact_match=95.0%
Boundary:  precision=0.785  recall=0.886  f1=0.833
ULR:       duration_weighted=0.847  macro=0.807

Purity:   97.31%  ← 非常に高い
Coverage: 88.63%  ← やや低い

GT 1:  split_count=57  hyp_count=8   ← 最長話者(144s)が8クラスタに散らばる
GT 2:  split_count=42  hyp_count=4
GT 6:  split_count=17  hyp_count=4

ref_speakers: 15
hyp_speakers: 17
count_error:  mean_abs=0.050  exact_match_ratio=0.950

GT 6:  dominant=speaker_9 share=49.6%
        overlaps=[speaker_9:49.6%, speaker_7:43.9%, ...]
        → 1人の話者が speaker_9 と speaker_7 の2クラスタに引き裂かれている

GT 9:  dominant=speaker_15 share=57.7%
        overlaps=[speaker_15:57.7%, speaker_10:21.3%, ...]
        → GT 6 と GT 9 がともに speaker_10 を共有
          （speaker_10 が複数GT話者の「受け皿」になっている）

GT 11: dominant=speaker_14 share=36.1%  ← dominant share が最低
        detected=53.8%  ← 音声の半分近くが未検出
        → 発話総量 2.38s という極端な短さが要因

boundary_error: precision=0.785  recall=0.886  f1=0.833

start+0.5s: coverage=76.1%  conf=18.1%
end-0.5s:   coverage=67.1%  conf=25.0%  ← 終端の方が一貫して悪い

utterance_length_conditioned_errors:
  0.5〜1s:  coverage=43.5%  conf=56.5%  miss=0.0%   (n=4)
  1〜2s:    coverage=70.7%  conf=23.8%  miss=5.5%   (n=16)
  2s+:      coverage=85.9%  conf=11.6%  miss=2.5%   (n=78)