多人会议、客服通话和访谈录音的难点，不只是把声音转成文字，还要回答两个问题：这句话是谁说的，以及他说话的时间范围是什么。NVIDIA Nemotron 3 Diarization 聚焦的正是这层能力——说话人分离（speaker diarization）。
不过，从模型输出到真正可用的多说话人 AI，中间还隔着流式音频、增量修订、说话人状态和下游 Agent 等一整条工程链路。由于现有摘要没有给出具体 SDK、端点或字段定义，下面不假设某个固定 API，而是给出一套可直接改造的事件协议和处理方式；接入实际服务时，只需替换适配器。
说话人分离不是语音识别
自动语音识别（ASR）解决的是“说了什么”，说话人分离解决的是“谁在什么时候说话”。一个典型的分离结果可以抽象为：
{
"segment_id": "seg-42",
"speaker": "SPEAKER_01",
"start_ms": 1240,
"end_ms": 3820,
"text": "我们下周发布这个版本",
"final": false
}
这里有三个容易混淆的边界：
SPEAKER_01通常只是当前会话内的聚类标签，不等于真实姓名。- 说话人分离不一定自带文本；文本可能来自独立 ASR，再通过时间戳与说话人区间对齐。
- 实时结果可能被修订。模型在收到更多上下文后，可能调整片段边界或说话人标签。
因此，业务层不能把第一条增量结果直接当成不可变记录。更稳妥的办法是使用稳定的 segment_id，让后续事件覆盖同一片段，而不是不断追加文本。
一条可落地的实时处理链路
多人语音应用可以拆成六层：
- 音频入口：接收麦克风、WebRTC、电话或媒体流，统一采样率和声道格式。
- 切块与语音活动检测：按较短窗口推送音频，同时避免把长时间静音送入推理链路。
- 说话人分离与 ASR：获得说话人区间和文本。两者可以由同一服务产生，也可以并行运行后再对齐。
- 修订协调器：用片段 ID 合并 partial/final 事件，处理边界变化、迟到事件和重复消息。
- 会话状态：保存“当前谁在说话”、已确认转写、临时说话人标签和业务身份映射。
- 下游 AI：生成会议纪要、提取行动项、分析客服轮次，或把带说话人信息的上下文交给 LLM。
不要只统计模型推理时间。用户感受到的延迟还包括音频缓冲、网络传输、ASR、事件合并和前端渲染。建议给每一段打时间戳，分别观察：
- 首个增量结果延迟；
- 最终片段确认延迟；
- 说话人标签被修订的次数；
- 音频断流后的恢复时间；
- 重叠说话片段的比例与准确性。
固定一个漂亮的总延迟数字意义不大。会议字幕和通话打断检测对延迟的容忍度不同，应根据产品交互设定预算。
可运行的增量事件协调器
下面的 Python 程序不依赖第三方包，可以直接运行。它模拟接收 Nemotron 3 Diarization 适配器输出的标准化事件，并展示如何覆盖 partial 结果、保留 final 状态以及生成按时间排序的会话快照。
实际接入时，需要把供应商返回字段转换为示例中的 segment_id、speaker、时间戳和 final。这些字段是本文定义的适配层协议，并非对某个官方接口的声明。
#!/usr/bin/env python3
import argparse
import json
import sys
from dataclasses import dataclass
from typing import Dict
@dataclass
class Segment:
segment_id: str
speaker: str
start_ms: int
end_ms: int
text: str
final: bool
class TranscriptState:
def __init__(self) -> None:
self.segments: Dict[str, Segment] = {}
self.speaker_names = {
"SPEAKER_00": "主持人",
"SPEAKER_01": "嘉宾",
}
def apply(self, event: dict) -> None:
required = {
"segment_id", "speaker", "start_ms",
"end_ms", "text", "final"
}
missing = required - event.keys()
if missing:
raise ValueError(f"missing fields: {sorted(missing)}")
old = self.segments.get(event["segment_id"])
if old is not None and old.final:
# 已确认片段默认不可被普通增量事件回退。
return
self.segments[event["segment_id"]] = Segment(**event)
def snapshot(self) -> list[dict]:
ordered = sorted(
self.segments.values(),
key=lambda item: (item.start_ms, item.end_ms)
)
return [
{
"speaker": self.speaker_names.get(item.speaker, item.speaker),
"start_ms": item.start_ms,
"end_ms": item.end_ms,
"text": item.text,
"final": item.final,
}
for item in ordered
]
DEMO_EVENTS = [
{
"segment_id": "seg-1",
"speaker": "SPEAKER_00",
"start_ms": 0,
"end_ms": 1500,
"text": "今天讨论发布计划",
"final": False
},
{
"segment_id": "seg-1",
"speaker": "SPEAKER_00",
"start_ms": 0,
"end_ms": 1820,
"text": "今天讨论下周的发布计划",
"final": True
},
{
"segment_id": "seg-2",
"speaker": "SPEAKER_01",
"start_ms": 1900,
"end_ms": 3600,
"text": "我会负责上线检查",
"final": True
}
]
def read_events(demo: bool):
if demo:
yield from DEMO_EVENTS
return
for line in sys.stdin:
line = line.strip()
if line:
yield json.loads(line)
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument(
"--demo",
action="store_true",
help="consume built-in sample events"
)
args = parser.parse_args()
state = TranscriptState()
for revision, event in enumerate(read_events(args.demo), start=1):
state.apply(event)
print(json.dumps(
{"revision": revision, "segments": state.snapshot()},
ensure_ascii=False
))
if __name__ == "__main__":
main()
保存为 diarization_consumer.py 后运行：
python3 diarization_consumer.py --demo
也可以从标准输入发送 JSON Lines：
cat <<'JSONL' | python3 diarization_consumer.py
{"segment_id":"s1","speaker":"SPEAKER_00","start_ms":0,"end_ms":900,"text":"可以开始了","final":false}
{"segment_id":"s1","speaker":"SPEAKER_00","start_ms":0,"end_ms":1200,"text":"我们可以开始了","final":true}
{"segment_id":"s2","speaker":"SPEAKER_01","start_ms":1250,"end_ms":2300,"text":"我先介绍方案","final":true}
JSONL
生产环境还应在事件里加入 session_id、单调递增的序列号和模型版本。若服务可能重发消息，可以用 (session_id, segment_id, revision) 做幂等键。
从 SPEAKER_00 到真实身份
说话人分离得到的是“声音 A”和“声音 B”，不能仅凭标签断言某人的真实身份。把标签映射到姓名，通常需要额外信号：
- 会议席位或登录账号；
- 用户主动确认“我是张三”；
- 独立且获得授权的说话人识别系统；
- 主持人在会话中的人工标注。
这种映射涉及隐私和生物特征数据。需要明确获得授权，设置保存期限，并把原始音频、声纹、转写文本和业务身份分开控制权限。对于高风险场景，不能把自动标签当作身份认证依据。
重叠说话也是必须单独设计的数据结构。不要强制要求任意时刻只能有一个活跃说话人；同一时间区间可能对应两个片段。界面可以叠放两条字幕，下游统计则要避免把重叠时长重复计入总通话时长。
上线前的检查清单
接入 Nemotron 3 Diarization 时，建议先以适配层隔离具体 API，再逐步替换模拟事件：
- 确认音频编码、采样率、声道和切块大小符合实际服务要求；
- 明确服务是否同时返回文本，还是只返回说话人时间区间；
- 区分 partial 与 final，并允许片段边界和标签被修订；
- 使用会话 ID、序列号和幂等键处理断线重连；
- 用真实场景测试插话、噪声、口音、远场麦克风和相似音色；
- 分别评估转写准确率、说话人归属错误和端到端延迟；
- 对身份映射、音频保存和日志访问设置隐私边界。
真正可靠的多说话人 AI，不是简单地在每句话前加一个 SPEAKER_01。关键在于接受流式结果会变化这一事实，并用可修订事件、稳定片段 ID 和清晰的身份边界，把模型能力变成可维护的实时系统。