Nemotron-3-Diarization 流式 ASR 集成实战：搭建「谁说了什么」的多说话人实时转录管线

• 人工智能
• 语音
• 音频

nvidia/Nemotron-3-Diarization回答的是「谁在什么时候说了话（who spoke when）」。要想得到带说话人归属的完整转录文本——「谁说了什么（who said what）」——必须把它与一个兼容的流式 ASR 模型配对使用。本文基于本仓库的 ASR_INTEGRATION_GUIDE.md，系统讲解两条官方支持的流式 ASR 集成路径（Multitalker Parakeet 与 Nemotron 3.5 ASR），从环境安装、音频预处理、参数调优，到离线批量推理与在线实时服务的完整实现，帮助你在自己的对话录音、会议或通话场景中构建可运行的说话人归属转录系统。

一、耦合式流式管线：Diarization 与 ASR 如何协同工作

这是一条耦合的流式（coupled streaming）管线：Nemotron-3-Diarization 在流式推理过程中输出帧级说话人活动（frame-level speaker activity），而 ASR 阶段为每一个被检测到的说话人维护一条独立的转录流（separate transcription stream）。

音频流(16 kHz 单声道 PCM) │ ▼ ┌─────────────────────────────┐ │ Nemotron-3-Diarization │ 输出: 帧级说话人活动 │ (Sortformer 编码器) │ (T×8 活动概率, 按首次到达排序) └─────────────────────────────┘ │ 说话人活动 / 掩码 ▼ ┌─────────────────────────────┐ │ 流式 ASR (每说话人独立流) │ 输出: 带说话人标签的转录 │ Option 1: Multitalker Parakeet│ │ Option 2: Nemotron 3.5 ASR │ └─────────────────────────────┘ │ ▼ speaker_attributed_output.json (SegLST 格式)

从模型本身看，Nemotron-3-Diarization 是一个约 100M 参数的 Transformer 编码器（详见 explainability.md）：输入 10 ms 的 Mel 频谱特征，按 8 倍下采样堆叠成 80 ms 的编码器帧；流式推理时借助Arrival-Order Speaker Cache（AOSC）与FIFO 队列在 chunk 之间保留说话人上下文；最终输出[T, 8]的逐说话人活动概率张量，8 个通道按说话人在输入音频中的首次到达顺序排列，可后处理成带起止时间的通用说话人标签（本仓库 README.md 对该架构有完整说明）。

本仓库的 streaming_diarization_demo.png（及同内容的动图）正是模型卡中标注的「Nemotron-Diarization with Streaming ASR」在线演示画面，展示了该耦合管线的实际运行效果：说话人分离后的转录文本按说话人着色实时呈现。

官方推荐的流式 ASR 配对方案如下：

二、环境准备：安装 NeMo Speech

因为多说话人流式辅助代码与模型是同步开发的，请使用 NeMo Speech 仓库的较新 checkout。

apt-get update && apt-get install -y libsndfile1 ffmpeg git git clone --branch main --single-branch https://github.com/NVIDIA-NeMo/Speech.git cd Speech curl -LsSf https://astral.sh/uv/install.sh | sh uv sync --python 3.13 --extra asr --extra cu13 --no-dev source .venv/bin/activate

• 若你本机安装的 NVIDIA 驱动与 CUDA 环境需要 CUDA 12，请将--extra cu13换成--extra cu12。
• 如果 Hugging Face 仓库需要鉴权，请提供一个对两个模型仓库都有读取权限的 token：

export HF_TOKEN="your-hugging-face-token"

不要把 token 写进源码或提交到仓库。

NeMo 示例脚本直接接受 Hugging Face 仓库 ID：

diar_model=nvidia/Nemotron-3-Diarization

若希望把 checkpoint 保留在本地，可用hf download下载：

hf download nvidia/Nemotron-3-Diarization \ --local-dir /path/to/Nemotron-3-Diarization

然后用本地路径替换下文命令中的仓库 ID，即把diar_model=nvidia/Nemotron-3-Diarization替换为/path/to/Nemotron-3-Diarization/Nemotron-3-Diarization.nemo。注意：本仓库根目录下的 Nemotron-3-Diarization.nemo 是一个 Git LFS 指针文件（约 198 MB 的 checkpoint 本体由 LFS 托管），实际使用时按上述方式下载或通过SortformerEncLabelModel.restore_from加载。

三、音频准备：统一为 16 kHz 单声道 PCM

配对模型期望单声道（mono）、16 kHz的音频输入，推理前需要转换其他格式的输入：

ffmpeg -i input.mp3 -ac 1 -ar 16000 -c:a pcm_s16le conversation.wav

-ac 1强制单声道、-ar 16000重采样到 16 kHz、-c:a pcm_s16le输出 16-bit PCM。音频格式不达标通常会导致推理被拒绝或结果质量明显下降（见文末故障排查）。

四、ASR 模型选型与运行命令

官方提供两条集成路线，请在下面两者中择一使用。

方案 1：Multitalker Parakeet（面向重叠语音）

nvidia/multitalker-parakeet-streaming-0.6b-v1是一个为重叠语音微调过的流式多说话人 ASR 模型，仅支持英语，且必须搭配masked_asr=false使用——它把说话人活动当作条件信息（conditioning）消费，而不是用它做掩码。

在 NeMo Speech 仓库根目录执行：

python examples/asr/asr_cache_aware_streaming/speech_to_text_multitalker_streaming_infer.py \ asr_model=nvidia/multitalker-parakeet-streaming-0.6b-v1 \ diar_model=nvidia/Nemotron-3-Diarization \ audio_file=/absolute/path/to/conversation.wav \ max_num_of_spks=8 \ single_speaker_mode=false \ masked_asr=false \ parallel_speaker_strategy=true \ cache_gating=true \ binary_diar_preds=true \ att_context_size='[70,13]' \ fifo_len=264 \ spkcache_update_period=222 \ generate_realtime_scripts=false \ output_path=./speaker_attributed_output.json

方案 2：Nemotron 3.5 ASR（掩码增强的单说话人流式模型）

nvidia/nemotron-3.5-asr-streaming-0.6b是一个常规的单说话人流式 ASR 模型；在这条说话人归属管线中，它利用 diarization 提供的掩码说话人活动（masked speaker activity）来在一定程度上识别重叠语音。它开箱支持32 个语言-locale，tokenizer 还额外支持8 个需要微调的 adaptation-ready locale。此方案必须搭配masked_asr=true。

target_lang应设置为输入音频的语言-locale；当希望模型自动推断语言提示时，使用target_lang=auto。

python examples/asr/asr_cache_aware_streaming/speech_to_text_multitalker_streaming_infer.py \ asr_model=nvidia/nemotron-3.5-asr-streaming-0.6b \ diar_model=nvidia/Nemotron-3-Diarization \ audio_file=/absolute/path/to/conversation.wav \ max_num_of_spks=8 \ single_speaker_mode=false \ masked_asr=true \ parallel_speaker_strategy=true \ cache_gating=true \ binary_diar_preds=true \ att_context_size='[56,13]' \ fifo_len=264 \ spkcache_update_period=222 \ target_lang=auto \ generate_realtime_scripts=false \ output_path=./speaker_attributed_output.json

注意：两套命令中att_context_size不同（Parakeet 推荐[70,13]，Nemotron 3.5 ASR 推荐[56,13]），这是两模型编码器帧几何（encoder frame geometry）不同导致的，务必按方案取用，不要混配。

输出与结果解读

脚本把最终结果写入speaker_attributed_output.json，格式为 NeMo 的SegLST 格式：每个 finalized 条目包含其文本（text）、时间范围（time range）和匿名说话人标签（anonymous speaker label）。设置generate_realtime_scripts=true还会打印流式中间假设（streaming hypotheses）。

典型的渲染结果如下：

Speaker 0: Welcome, everyone. Let us begin. Speaker 1: I have the latest numbers. Speaker 2: I agree, but there is one remaining issue.

⚠️ 需要特别强调：说话人编号只是该次流式会话中「发现的身份」，并不对应现实世界的真实姓名。如果应用需要显示人名，必须通过额外的注册（enrollment）或应用层映射步骤把编号关联到具体身份，切忌把匿名标签当作生物特征身份直接展示（标签在会话之间还会变化，见故障排查）。

核心参数含义速查表

需要理解：max_num_of_spks是一个上限，不是「每个配置的说话人都会出现」的承诺——活跃说话人的身份是由 diarization 模型从音频中动态发现的（本仓库 README.md 也确认模型最多支持 8 个说话人，超过 8 人时部分人将不可避免地被漏检或错配）。参数fifo_len=264与spkcache_update_period=222直接对应 README 中「低延迟 1.04 s」档位（SPKCACHE_LEN=264, FIFO_LEN=264, UPDATE_PERIOD=222）的流式几何，输入缓冲延迟按(CHUNK_LEN + RIGHT_CONTEXT) × 80 ms计算。

五、实时麦克风 / 服务集成：从离线回放走向生产服务

speech_to_text_multitalker_streaming_infer.py示例最适合被理解为一个离线流式回放与评估驱动（offline streaming replay and evaluation driver）：它逐 chunk 走通流式推理路径，但读取的是已经就绪的文件或 manifest。生产环境中的 Web、麦克风或 websocket 服务需要为每个客户端维护一个持久会话（persistent session），并把新到达的 PCM chunk 源源不断地喂入该会话。

一条实时接口的典型链路是：客户端把音频 chunk 发送到服务器 → 服务器按模型要求的 hop/cache 几何缓冲这些 chunk → 把每个完整的帧交给 NeMo Speech 的SpeakerTaggedASR。Gradio 只是众多可选界面之一，其余替代方案包括：自定义 Web UI、桌面或移动应用、HTTP API、原生 websocket 客户端。

服务端结构（5 步）

1. 每个 worker 进程只加载一次ASR 与 diarization checkpoint。
2. 每条实时客户端连接到来时，新建一个流式会话。
3. 对每个到达的单声道 PCM chunk：如需则重采样到 16 kHz，追加到该客户端的待处理音频缓冲，并处理每一个完整的流式帧。
4. 向客户端返回最新的说话人归属转录，以及（如果需要）最新的 diarization 活动状态。
5. 录音结束或 websocket 关闭时，重置该客户端的会话。

参考实现：LiveMultitalkerSession

下面的流式循环代码专门按方案 1（Multitalker Parakeet）配置；方案 2 需换成其对应的模型专属设置（如att_context_size=[56,13]、masked_asr=true）。

import os import numpy as np import torch from omegaconf import OmegaConf import nemo.collections.asr as nemo_asr from nemo.collections.asr.models.sortformer_diar_models import SortformerEncLabelModel from nemo.collections.asr.parts.utils.multispk_transcribe_utils import ( SpeakerTaggedASR, configure_diar_streaming, validate_feature_frame_strides, ) from nemo.collections.asr.parts.utils.streaming_utils import CacheAwareStreamingAudioBuffer def load_models(asr_model_path, diar_model_path, device="cuda"): if os.path.isfile(asr_model_path): asr_model = nemo_asr.models.ASRModel.restore_from(restore_path=asr_model_path) else: asr_model = nemo_asr.models.ASRModel.from_pretrained(asr_model_path) if os.path.isfile(diar_model_path): diar_model = SortformerEncLabelModel.restore_from(restore_path=diar_model_path, map_location=device) else: diar_model = SortformerEncLabelModel.from_pretrained(diar_model_path) asr_model.eval().to(device) diar_model.eval().to(device) asr_model.encoder.set_default_att_context_size([70, 13]) validate_feature_frame_strides(asr_model=asr_model, diar_model=diar_model) return asr_model, diar_model class LiveMultitalkerSession: def __init__(self, asr_model, diar_model, sample_rate=16000): self.cfg = OmegaConf.create( { "device": str(asr_model.device), "sample_rate": sample_rate, "deploy_mode": True, "streaming_mode": True, "max_num_of_spks": 8, "batch_size": 32, "parallel_speaker_strategy": True, "masked_asr": False, "mask_preencode": False, "single_speaker_mode": False, "cache_gating": True, "cache_gating_buffer_size": 2, "binary_diar_preds": True, "spkcache_len": None, "spkcache_update_period": 222, "fifo_len": 264, "diar_right_context": 0, "att_context_size": [70, 13], "use_amp": True, "precision": "bf16", "online_normalization": False, "pad_and_drop_preencoded": False, "feat_len_sec": 0.01, "discarded_frames": 8, "word_window": 50, "sent_break_sec": 1.0, "fix_prev_words_count": 5, "update_prev_words_sentence": 5, "left_frame_shift": -1, "right_frame_shift": 0, "min_sigmoid_val": 1e-2, "ignored_initial_frame_steps": 5, "generate_realtime_scripts": True, "print_sample_indices": [0], "colored_text": True, "verbose": False, "print_time": False, "log": False, } ) self.asr_model = asr_model self.diar_model = diar_model streaming_cfg = asr_model.encoder.streaming_cfg diar_chunk_len = streaming_cfg.valid_out_len + streaming_cfg.cache_drop_size configure_diar_streaming( diar_model=diar_model, cfg=self.cfg, output_subsampling_factor=asr_model.encoder.subsampling_factor, diar_chunk_len=diar_chunk_len, ) self.cfg.spkcache_len = int(diar_model.sortformer_modules.spkcache_len) self.streamer = SpeakerTaggedASR(self.cfg, asr_model, diar_model) self.audio_buffer = CacheAwareStreamingAudioBuffer( model=asr_model, online_normalization=self.cfg.online_normalization, ) feature_stride = float(asr_model.cfg.preprocessor.window_stride) hop_feature_frames = streaming_cfg.valid_out_len * asr_model.encoder.subsampling_factor self.hop_samples = round(hop_feature_frames * feature_stride * sample_rate) cache_frames = streaming_cfg.pre_encode_cache_size if isinstance(cache_frames, (list, tuple)): cache_frames = cache_frames[-1] cache_samples = round(cache_frames * feature_stride * sample_rate) self.frame_samples = self.hop_samples + cache_samples self.pending_audio = np.zeros(cache_samples, dtype=np.float32) self.step_num = 0 @torch.inference_mode() def accept_audio(self, pcm_chunk, sample_rate): """Accept one live mono PCM chunk and return the latest transcript text.""" if pcm_chunk.dtype == np.int16: pcm_chunk = pcm_chunk.astype(np.float32) / 32768.0 elif pcm_chunk.dtype == np.int32: pcm_chunk = pcm_chunk.astype(np.float32) / 2147483648.0 else: pcm_chunk = pcm_chunk.astype(np.float32) # Resample here when sample_rate != self.cfg.sample_rate. self.pending_audio = np.concatenate([self.pending_audio, pcm_chunk]) latest = None while len(self.pending_audio) >= self.frame_samples: frame = self.pending_audio[: self.frame_samples] self.pending_audio = self.pending_audio[self.hop_samples :] chunk_audio, chunk_lengths = self.audio_buffer.preprocess_audio(frame) chunk_audio = chunk_audio[:, :, : chunk_lengths[0]] drop_extra_pre_encoded = ( 0 if self.step_num == 0 else self.asr_model.encoder.streaming_cfg.drop_extra_pre_encoded ) latest = self.streamer.perform_parallel_streaming_stt_spk( step_num=self.step_num, chunk_audio=chunk_audio, chunk_lengths=chunk_lengths, is_buffer_empty=False, drop_extra_pre_encoded=drop_extra_pre_encoded, ) self.step_num += 1 return "" if latest is None else latest[0]

接入应用界面与并发注意事项

把accept_audio()连接到你的应用所用接口或传输层：对网络服务，在客户端连接时创建会话，对每条解码后的 PCM 消息调用一次accept_audio()，客户端断开时删除会话。如果服务器可能对同一客户端并发回调，请用一把小锁保护每个会话。

严禁在客户端之间共享会话对象：说话人缓存、ASR 解码器状态、时间戳、待处理音频缓冲与转录历史都是会话专属状态。

模型权重则可以在连接之间共享以节省 GPU 显存——前提是每条连接都持有独立的SpeakerTaggedASR、流式缓冲与说话人缓存状态。

关键几何参数从哪里来

从源码结构看，上述实现的核心几何推导如下（对应 NeMo Speechparts/utils/multispk_transcribe_utils.py与parts/utils/streaming_utils.py的协作方式）：

• hop_samples：由 ASR 编码器的valid_out_len × subsampling_factor换算成音频采样点数，决定每步前进的帧长；
• frame_samples = hop_samples + cache_samples：完整帧 = 前进步长 + 预编码缓存（pre_encode_cache_size），这正是模型所需的hop/cache 几何；
• configure_diar_streaming()：把 ASR 侧的帧几何（output_subsampling_factor、diar_chunk_len）同步给 diarization 模型，保证两条模型流的 chunk 几何对齐——这也是故障排查中「不要各自独立调参」的根本原因；
• validate_feature_frame_strides()：启动时校验 ASR 与 diarization 的特征帧步长是否兼容（explainability.md 指出模型以 80 ms 帧率工作，10 ms 特征下采样 8 倍，因此这类校验是强耦合管线的必要防线）。

六、更换其他 ASR 模型的考量

一种朴素做法是：用普通单说话人 ASR，在 diarization 之后按说话人片段切音频再转录（cut-and-transcribe）。当说话人很少重叠时这种方法可行；但它不能等价替代 Multitalker Parakeet——因为切出的时间片里仍然包含所有与之重叠的声纹，常规 ASR 可能把不同说话人的词合并或选错。

要做重叠感知（overlap-aware）的说话人归属转录，请遵循两条原则：

1. 选用专为消费说话人活动（speaker activity）而设计的 ASR 模型（如 Multitalker Parakeet 的条件式输入、Nemotron 3.5 ASR 的掩码式输入）；
2. 核实其编码器帧几何与 diarization 模型兼容（这正是att_context_size、fifo_len、spkcache_update_period等参数不能随意混配的原因）。

七、故障排查速查表

八、进一步延伸：在仓库内继续深挖

• README.md：模型总体说明、SortformerEncLabelModel的加载/推理示例、流式配置参数（SPKCACHE_LEN/FIFO_LEN/CHUNK_LEN/RIGHT_CONTEXT/UPDATE_PERIOD）与延迟档位表、输入输出格式、性能评估（DER/SCA/MAE/RTFx）数据，以及「Integration with Streaming ASR」对本指南的引用。
• diarization_evaluation.md：若需对集成后的管线做量化评估，可使用 NeMo Speech 的e2e_diarize_speech.py与score_diarization.py计算 DER（false alarm + missed speech + speaker confusion）、Speaker Counting Accuracy（SCA）与说话人计数 MAE，并注意报告 collar、overlap、流式参数等协议细节。
• explainability.md：模型工作原理（10 ms Mel 特征 8 倍堆叠 → 80 ms 帧 → Transformer 编码 → Conv1D 上采样回 10 ms）、AOSC/FIFO 机制、输出张量[T, 8]及技术限制（最多 8 人、长录音与高噪/强混响下性能可能下降、低延迟与精度/速度的权衡）。

综合本指南与仓库资料，一条可落地的实践路径是：先按第二节安装 NeMo Speech 并准备 16 kHz 单声道音频，用第四节两条命令中的任一条跑通离线流式回放验证效果与参数组合，再按第五节的服务结构把LiveMultitalkerSession接入你自己的 Web/websocket/桌面接口，最后用第八节提到的评估脚本对 DER 等指标做量化验收。这样，你就拥有了一套从「谁在什么时候说话」到「谁说了什么」的完整多说话人实时转录能力。

• 人工智能
• 语音
• 音频