AudioStreamer 与 AsyncAudioStreamer 在 VibeVoice 里怎么分工

2026-08-18

翻 VibeVoice 仓库的时候，vibevoice/modular/streamer.py 是那种很容易被跳过的文件——篇幅不长，没有模型结构，没有配置类。但只要你想搞明白 demo/web/app.py 里那个 WebSocket 是怎么做到「一边生成一边往浏览器推音频」的，绕不开它。

问题很具体：VibeVoiceStreamingForConditionalGenerationInference.generate() 是一个大循环，它跑完之前不会返回。而 WebSocket 那头等着一段一段拿数据。中间必须有一个东西，让生成循环每产出一块音频就能立刻交出去，同时不阻塞自己。这个东西就是 streamer。

下面沿着代码走一遍。所有内容来自仓库文件，我们没有下载权重也没有跑过推理。

生产端：`put` 和 `end` 是给生成循环调的

AudioStreamer 继承自 transformers.generation 的 BaseStreamer。构造函数三个参数：

def __init__(
    self, 
    batch_size: int,
    stop_signal: Optional[any] = None,
    timeout: Optional[float] = None,
):

docstring 里写明 stop_signal 是「生成结束时放进队列的信号」，timeout 是「音频队列的超时；如果为 None，队列会无限阻塞」。构造函数里按 batch_size 建了一组 Queue()，一个样本一条队列，另外还有一个 finished_flags 的布尔列表。

生产端只有两个方法。put(audio_chunks, sample_indices) 按 sample_indices 把每块音频分发到对应队列，落队之前做了 audio_chunks[i].detach().cpu()；而且带一个前置判断 if idx < self.batch_size and not self.finished_flags[idx]——已经标记结束的样本，再 put 也会被丢掉。end(sample_indices=None) 往队列里塞 stop_signal 并把 finished_flags 置 True，传 None 就是全部结束。

这两个方法在生成循环里被调用的位置，在 vibevoice/modular/modeling_vibevoice_streaming_inference.py 里能一处一处找到：拿到一块音频后是 audio_streamer.put(audio_chunk, diffusion_indices)；EOS 分类器判定某个样本说完了，走 audio_streamer.end(diffusion_indices)；外部传进来的 stop_check_fn 返回 True 时先 audio_streamer.end() 再 break；整个循环退出后还有一次兜底的 audio_streamer.end()。也就是说，无论正常结束、被 EOS 截停还是被外部叫停，队列里最终都会出现停止信号——消费端不至于永远挂着等。

消费端：两条路，语义不一样

AudioStreamer 给了两个入口，很容易混。

get_stream(sample_idx) 返回 AudioSampleIterator，只盯一条队列。它的 __next__ 是 self.streamer.audio_queues[self.sample_idx].get(timeout=self.streamer.timeout)，也就是阻塞取，取到的值等于 stop_signal 就抛 StopIteration。这条路适合单路消费，代价是它会占住调用它的那个线程。

__iter__ 返回的是 AudioBatchIterator，行为完全不同：它对所有活跃样本用 get(block=False)，捕获 Empty 就跳过这一轮；一轮下来如果一块都没拿到但还有活跃样本，它会 time.sleep(0.01) 然后 return self.__next__()——递归调用自己。拿到东西时返回的是一个 {样本下标: 音频块} 的字典，而不是单块音频。所以这两个入口的返回类型是不一样的，写消费端代码之前先看清你用的是哪个。

还有一个细节值得留意：判断结束用的是 value == self.streamer.stop_signal，是 == 不是 is。仓库里的 demo 传的 stop_signal 就是 None；如果你要换成别的对象，这里的相等语义得自己确认一遍。

`AsyncAudioStreamer` 改了哪几处

AsyncAudioStreamer 直接继承 AudioStreamer，class docstring 只有一句「Async version of AudioStreamer for use in async contexts.」。它先调 super().__init__(...)，然后把队列整组换掉：

super().__init__(batch_size, stop_signal, timeout)
# Replace regular queues with async queues
self.audio_queues = [asyncio.Queue() for _ in range(batch_size)]
self.loop = asyncio.get_running_loop()

第二行注释是仓库原文。注意 asyncio.get_running_loop() 写在构造函数里，按标准库语义，它要求调用时当前线程已经有运行中的事件循环——这意味着这个对象不能随便在哪个线程里 new 出来。

put 的实现也随之改写：不再直接 queue.put(...)，而是

self.loop.call_soon_threadsafe(
    self.audio_queues[idx].put_nowait, audio_chunk
)

这是整个类的关键一步。生成循环通常跑在别的线程里，asyncio.Queue 不是线程安全的，call_soon_threadsafe 把入队动作调度回事件循环所在线程。顺带的后果是：同步版 put 里那个 timeout=self.timeout 参数在异步版里没有了，因为 put_nowait 不接受超时。end 同理，也是走 call_soon_threadsafe + put_nowait。构造时传的 timeout 在异步版里只剩一个用处——AsyncAudioBatchIterator 里 asyncio.wait(...) 的超时。

get_stream 这一处是两个类之间差别最大、也最容易踩的地方。父类里它是普通方法，返回一个迭代器对象；子类里它写成了 async def 并且函数体里有 yield，这在 Python 里是一个异步生成器函数——调用它拿到的是异步生成器，用法是 async for，而不是 await。同名、同参数，但调用方式不能互换。所以「把 AudioStreamer 换成 AsyncAudioStreamer 就行」这句话在这里不成立，消费端代码要跟着改。

批量入口也换成了 __aiter__ → AsyncAudioBatchIterator。它的 __anext__ 给每条活跃队列建一个 task，用 asyncio.wait(..., return_when=asyncio.FIRST_COMPLETED, timeout=self.streamer.timeout) 等第一个就绪，然后 task.cancel() 掉其余的，下一轮重建。同步版那边靠 sleep(0.01) 轮询，异步版靠事件循环等待，这是两者在批量迭代上的结构性差异。

仓库里实际用的是哪一个

grep 一遍就会发现：仓库里唯一导入 streamer 的地方是 demo/web/app.py，而且只导了同步版 from vibevoice.modular.streamer import AudioStreamer。AsyncAudioStreamer 只出现在两个地方：一是 vibevoice/modular/__init__.py 把它和 AudioStreamer 一起写进了 __all__，二是 modeling_vibevoice_streaming_inference.py 里 generate() 的类型标注 Optional[Union[AudioStreamer, AsyncAudioStreamer]]。也就是说它是对外导出的公开符号，但我们在仓库里没有找到实际构造它的代码。

更有意思的是这个 demo 明明是 FastAPI + WebSocket 的异步服务，却选了同步版。它的做法是：

audio_streamer = AudioStreamer(batch_size=1, stop_signal=None, timeout=None)

然后把 model.generate(...) 连同 audio_streamer=audio_streamer、stop_check_fn=stop_event.is_set 一起丢进一个 threading.Thread(..., daemon=True)，主协程这边 stream = audio_streamer.get_stream(0) 逐块取；WebSocket 处理函数再用 chunk = await asyncio.to_thread(next, iterator, sentinel) 把这个阻塞迭代器桥到事件循环上。也就是说，异步性是靠 asyncio.to_thread 补的，不是靠 AsyncAudioStreamer。

收尾那段同样值得抄下来：finally 里依次做 stop_signal.set()、audio_streamer.end()、thread.join()。按代码里的语义，这三步各管一头：set() 让生成循环里的 stop_check_fn 在下一轮返回 True，end() 保证队列里一定落下停止信号，thread.join() 等生成线程自己收尾。特别注意 demo 传的是 timeout=None，而 AudioStreamer 的 docstring 写明「如果为 None，队列会无限阻塞」——停止信号这一步在这种配置下就没有退路可走。

以上代码片段均原样取自仓库文件，为按仓库代码中的接口语义组合的示例，未经实测，以仓库最新代码为准。

三处「代码里写了但没启用」

读这个文件顺手会撞到几处半成品，写代码前最好知道：

AudioStreamer.__init__ 里有 self.sample_indices_map = {}，注释写着 # Maps from sample index to queue index，但我们在仓库里没有找到任何地方读写它。
generate() 里有一段被注释掉的代码，原本是检查 audio_streamer.finished_flags 来判断是否被外部停止；现在生效的是 stop_check_fn 这条路。
AudioStreamer 按 batch_size 开多条队列、put/end 都按 sample_indices 分发，看上去是为批量设计的；但 generate() 的 docstring 里写明「The function only supports batch size = 1 currently.」。demo 里传的也确实是 batch_size=1。这两处放在一起看，多路能力目前只在 streamer 这一侧成立。

跑这个 demo 的前置条件与 Windows 侧

docs/vibevoice-realtime-0.5b.md 里给的安装方式是 pip install -e .[streamingtts]，启动命令是 python demo/vibevoice_realtime_demo.py --model_path microsoft/VibeVoice-Realtime-0.5B。这个启动脚本本身只有十几行，用 uvicorn.run("web.app:app", ...) 拉起上面那个 app，--device 的取值在 argparse 里限定为 cpu、cuda、mpx、mps。以上为仓库文档与代码里的原文，随版本可能变动。

Windows 上有一处仓库自己写明的坑。demo/web/app.py 的 StreamingTTSService.__init__ 里有一行注释：# Keep model_path as string for HuggingFace repo IDs (Path() converts / to \ on Windows)——模型路径被刻意保留成字符串而不是 Path，因为在 Windows 上 Path() 会把 Hugging Face repo id 里的 / 转成 \。如果你要改这个 demo，别顺手把它包成 Path。Linux/macOS 侧没有这个问题，但同一段代码是共用的，改坏了两边都受影响。

小结一句

分工其实很清楚：AudioStreamer 面向「生成在别的线程、消费在同步代码里」，AsyncAudioStreamer 面向「消费在协程里」，两者的生产端签名一致（都是 put/end），消费端签名不一致（get_stream 一个返迭代器一个是异步生成器）。而仓库当前的 demo 选的是前者加线程桥。你要接自己的服务，先决定消费端在哪一侧，再回头选类，比反过来省事。

本文依据 github.com/microsoft/VibeVoice 仓库与 Hugging Face 模型卡于 2026-08-18 的公开内容整理，事实来自仓库内的文档与源码。我们没有下载权重、没有跑过推理、也没有做过训练，因此不涉及显存占用、推理速度、识别准确率与音质的任何描述，也不与其它模型做比较或排名。该项目持续更新，文中涉及的模块路径、配置字段与接口写法随版本变动，请以仓库最新内容为准。

需要说明的是：仓库 README 记载，2025-09-05 微软因发现有与既定意图不符的使用方式，基于负责任 AI 原则从该仓库移除了 VibeVoice-TTS 代码；当前 vibevoice/modular/modeling_vibevoice.py 首行注释标明其来自社区 fork，且该模块未被 vibevoice/modular/__init__.py 的 __all__ 导出。本文只讲代码与架构，不构成 TTS 推理的可用性保证。

仓库 README 的风险与限制一节写明：该模型仅供研究与开发用途，未经进一步测试与开发不建议用于商业或真实场景，并特别提示了合成语音被用于伪造与虚假信息的风险。使用合成语音时应遵守所在司法辖区的法律法规，并在分享 AI 生成内容时主动披露。