# 蚂蚁 inclusionAI 开源全双工音视频交互模型 Realtime-Venus

- 来源：蚂蚁 inclusionAI：HuggingFace 新模型
- 发布时间：2026-09-16 16:02
- AIHOT 分数：60
- AIHOT 标记：同事件
- AIHOT 链接：https://aihot.news/items/cmu72qy2c0o5qrowk0rgihc9e
- 原文链接：https://huggingface.co/inclusionAI/Realtime-Venus

## AI 摘要

蚂蚁集团 inclusionAI 在 Hugging Face 发布 Realtime-Venus，包含 Realtime-Venus-Omni 和 Realtime-Venus-Audio 两个 9B 检查点，基于 MiniCPM-o 4.5 / Omni-Flow 架构，语言骨干为 Qwen3-8B。

## 正文

Realtime-Venus 支持主动式音视频交互、异步委派以及可感知打断的全双工对话。

1. 🧭 概述

本仓库托管了 Realtime-Venus 系统的两个 checkpoint：

Realtime-Venus-Omni（Realtime-Venus-Omni/）：9B 音视频交互模型。它持续观看和聆听，判断是否以及何时做出回应，并在共享的因果时间线上生成文本和语音。它基于 MiniCPM-o 4.5 适配而来，支持主动交互、语义打断处理以及免训练的长视频记忆。

Realtime-Venus-Audio（Realtime-Venus-Audio/）：基于同一流式骨干网络的音频聚焦 checkpoint，用于音频理解以及以文本或语音输出的音频驱动对话。

两个目录均包含模型权重和自定义的 Hugging Face Transformers 代码。异步的 Realtime-Venus-Harness 及其外部工具集成位于 GitHub 仓库中。

2. ✨ 亮点

原生全双工对话：在说话的同时持续感知，并能区分附和、打断、纠正和改向。

Omni-Proactive 交互：持续处理时间对齐的视频和音频，并在事件需要时主动发起响应——无需等待用户提示词。

任务委派：在共享的因果时间线上以流式方式发出 <delegate> 请求，并以同样的方式消费异步后端结果，因此外部任务永远不会阻塞正在进行的对话。（执行请求需要 Realtime-Venus-Harness 运行时，可在 GitHub 仓库中获取。）

免训练长视频记忆：归档具有视觉信息量的时刻，检索与查询相关且非冗余的证据，并重新组装对应的音视频上下文——无需额外训练。

文本与语音输出：通过捆绑的 Token2wav 资源和参考语音，同时生成响应文本和原生语音。

3. 📋 模型详情

项目 Realtime-Venus-Omni Realtime-Venus-Audio

参数量 9B 9B

基础架构 MiniCPM-o 4.5 / Omni-Flow MiniCPM-o 4.5 / Omni-Flow

视觉编码器 SigLIP2 推理时不使用

音频编码器 Whisper-Medium Whisper-Medium

语言主干 Qwen3-8B Qwen3-8B

语音生成 离散 S3 语音 token，搭配流式流匹配解码器 同一解码器，在全双工模式下启用

输入 视频/图像、音频和文本 音频和文本

输出 文本和可选语音波形 文本和语音波形

上下文长度 40,960 tokens 40,960 tokens

权重数据类型 BF16 BF16

4. 📊 评估

所有数值均报告于Realtime-Venus 技术报告中。

图 1. 来自论文的视频和音频理解结果。

图 2. 来自论文的全双工交互结果。

5. 🗂️ 仓库结构

. ├── Realtime-Venus-Omni/ # Audio-visual full-duplex checkpoint │ ├── model-*.safetensors # Sharded model weights │ ├── config.json, *.py # Model config and custom Transformers code │ ├── realtime_venus_omni_memory.py # Public Memory entry point │ ├── memory_adapter/ # Chat and Duplex Memory runtime │ ├── assets/ # Reference voice, Token2wav, demo videos │ └── requirements.txt ├── Realtime-Venus-Audio/ # Audio-focused checkpoint │ ├── model-*.safetensors # Sharded model weights │ ├── config.json, *.py # Model config and custom Transformers code │ └── assets/ # Reference voice, Token2wav, demo audio ├── assets/ # Brand resources (logo) ├── README.md ├── README_zh.md └── LICENSE

以下示例将生成的媒体写入output/。重复实验时，请使用新的文件名或新的输出目录。

6. 🛠️ 安装

需要 Python 3.10、CUDA 和 FFmpeg。下载该仓库（两个 checkpoint 位于其子目录中）并安装 Python 依赖：

huggingface-cli download inclusionAI/Realtime-Venus --local-dir .

python -m pip install -r Realtime-Venus-Omni/requirements.txt

下面所有示例路径均相对于该仓库的根目录。from_pretrained 无法访问 Hub 仓库的子目录，因此这些示例在下载后指向本地的 Realtime-Venus-Omni/ 和 Realtime-Venus-Audio/ 路径。

7. 🎙️ Realtime-Venus-Omni 用法

这些示例的可独立运行版本位于 GitHub 上的 Omni cookbook。

7.1 🧱 模型初始化

下面的示例共用以下模型初始化方式；请在全新的 Python 进程中运行每个示例。Chat 和 Duplex 会自动加载默认参考音色。

点击查看 Omni 模型加载代码。

from pathlib import Path

import torch from transformers import AutoModel, set_seed

Path("output").mkdir(exist_ok=True) set_seed(42) print("Loading model ...") model = AutoModel.from_pretrained( "./Realtime-Venus-Omni", trust_remote_code=True, local_files_only=True, attn_implementation="sdpa", torch_dtype=torch.bfloat16, ) model.eval().cuda() print("Model loaded.")

7.2 🔊 双向全模态模式

model = model.as_duplex()将模型切换为全双工流式模式：prepare()初始化会话，随后每一秒的输入由一个streaming_prefill() + streaming_generate()对来处理，然后as_simplex()切换回离线模式。设置MAX_NUM_FRAMES在导入之前minicpmo.utils，否则超过 64 秒的视频会被截断至默认帧数上限。

字幕字体说明：Duplex 示例通过 FFmpeg/libass 将回复文本烧录进输出视频，而 FFmpeg/libass 通过 fontconfig 解析字体。渲染非拉丁语系回复（例如中文）需要系统上安装支持 CJK 的字体，否则这些字形会显示为空方框。在任何 Linux 发行版上，无需 root 即可安装一款字体并刷新字体缓存：

mkdir -p ~/.local/share/fonts curl --fail --location --retry 3 \ --output ~/.local/share/fonts/NotoSansCJKsc-Regular.otf \ https://raw.githubusercontent.com/notofonts/noto-cjk/main/Sans/OTF/SimplifiedChinese/NotoSansCJKsc-Regular.otf fc-cache -f

包管理器对应命令：apt install -y fonts-noto-cjk（Debian/Ubuntu）或 yum install -y cjkuni-ukai-fonts cjkuni-uming-fonts（RHEL/Alibaba Cloud Linux）。无需修改代码。

7.2.1 Duplex Chat

逐秒流式播放演示视频，并在 question_times 给出的秒数处注入文本问题（与 questions 配合使用）。模型持续聆听，并在回答时开口说话。

点击显示 Duplex Chat 代码。

import os

os.environ["MAX_NUM_FRAMES"] = "100000"

from minicpmo.utils import get_video_frame_audio_segments, generate_duplex_video

model = model.as_duplex() model.prepare()

video_path = "Realtime-Venus-Omni/assets/sample_1_real.mp4"

question_times = [60, 128] questions = [ "What do you see in the video so far?", "What is the color of the cooler labeled PRIME near the team bench?", ] question_plan = dict(zip(question_times, questions)) print(f"Extracting per-second audio and frames from {video_path} ...") frames, audios, _ = get_video_frame_audio_segments( video_path, stack_frames=1, use_ffmpeg=True, adjust_audio_length=True ) print(f"Streaming {len(audios)} seconds; questions are injected at {question_times}.") results, output_audio = [], [] for second, (frame, audio) in enumerate(zip(frames, audios), start=1): model.streaming_prefill( audio_waveform=audio, frame_list=[frame] if frame is not None else None, text_list=[question_plan[second]] if second in question_plan else None, ) result = model.streaming_generate() print( f"[{second}/{len(audios)}]", "listen..." if result["is_listen"] else f"speak> {result['text']}", flush=True, ) results.append({"chunk_idx": second - 1, **result}) if result["audio_waveform"] is not None: output_audio.append((second - 1, result["audio_waveform"]))

model = model.as_simplex() print("Muxing the spoken responses into the output video ...") generate_duplex_video( video_path=video_path, output_video_path="output/duplex_chat.mp4", results_log=results, timed_output_audio=output_audio, )

7.2.2 Speech-In Duplex Chat

与上述相同，区别在于问题是通过语音说出的，并且已经混入视频的音轨中（约在第 3 秒，要求在水烧开时发出提醒），因此不注入任何文本——模型必须自己听到。

点击显示 Speech-In Duplex Chat 代码。

import os

os.environ["MAX_NUM_FRAMES"] = "100000"

from minicpmo.utils import get_video_frame_audio_segments, generate_duplex_video

model = model.as_duplex() model.prepare()

video_path = "Realtime-Venus-Omni/assets/speech_in.mp4" print(f"Extracting per-second audio and frames from {video_path} ...") frames, audios, _ = get_video_frame_audio_segments( video_path, stack_frames=1, use_ffmpeg=True, adjust_audio_length=True ) print(f"Streaming {len(audios)} seconds; the spoken question is already in the audio track.") results, output_audio = [], [] for second, (frame, audio) in enumerate(zip(frames, audios), start=1): model.streaming_prefill( audio_waveform=audio, frame_list=[frame] if frame is not None else None, ) result = model.streaming_generate() print( f"[{second}/{len(audios)}]", "listen..." if result["is_listen"] else result["text"], flush=True, ) results.append({"chunk_idx": second - 1, **result}) if result["audio_waveform"] is not None: output_audio.append((second - 1, result["audio_waveform"]))

model = model.as_simplex() print("Muxing the spoken responses into the output video ...") generate_duplex_video( video_path=video_path, output_video_path="output/duplex_speech_in_chat.mp4", results_log=results, timed_output_audio=output_audio, )

7.2.3 Memory Duplex Chat

model.use_memory(memory_minutes=40) 在进入 duplex 模式前启用长视频 Memory。

点击查看 Memory Duplex Chat 代码。

import os

os.environ["MAX_NUM_FRAMES"] = "100000"

from minicpmo.utils import get_video_frame_audio_segments, generate_duplex_video

model.use_memory(memory_minutes=40) model = model.as_duplex() model.prepare()

video_path = "Realtime-Venus-Omni/assets/sample_1_real.mp4" question = "What is the color of the cooler labeled PRIME near the team bench?" print(f"Extracting per-second audio and frames from {video_path} ...") frames, audios, _ = get_video_frame_audio_segments( video_path, stack_frames=1, use_ffmpeg=True, adjust_audio_length=True ) print(f"Streaming {len(audios)} seconds; the text question is injected at second 128.") results, output_audio = [], [] for second, (frame, audio) in enumerate(zip(frames, audios), start=1): model.streaming_prefill( audio_waveform=audio, frame_list=[frame] if frame is not None else None, text_list=[question] if second == 128 else None, ) result = model.streaming_generate() print( f"[{second}/{len(audios)}]", "listen..." if result["is_listen"] else f"speak> {result['text']}", flush=True, ) results.append({"chunk_idx": second - 1, **result}) if result["audio_waveform"] is not None: output_audio.append((second - 1, result["audio_waveform"]))

model = model.as_simplex() print("Muxing the spoken responses into the output video ...") generate_duplex_video( video_path=video_path, output_video_path="output/duplex_memory_chat.mp4", results_log=results, timed_output_audio=output_audio, )

7.3 💬 半双工全模态模式

model.chat(...) 针对整个视频一次回答一轮。model.init_tts() 启用语音输出。

7.3.1 离线对话

采样帧、逐秒音频和问题一起进入单次 chat() 调用。128 帧上限（MAX_NUM_FRAMES）限制了视觉负载，而 max_inp_length=32768 则设定输入 token 预算。完整音频仍会保留，因此即使采用帧采样，超长视频也可能超出该预算。

点击查看离线对话代码。

import os

os.environ.setdefault("MAX_NUM_FRAMES", "128")

from minicpmo.utils import get_video_frame_audio_segments

model.init_tts()

video_path = "Realtime-Venus-Omni/assets/sample_1_real.mp4" question = "What is the color of the cooler labeled PRIME near the team bench?" print(f"Extracting audio and frames from {video_path} ...") frames, audios, _ = get_video_frame_audio_segments( video_path, stack_frames=1 ) content = [] for frame, audio in zip(frames, audios): if frame is not None: content.append(frame) content.append(audio) content.append(question)

print("Running chat inference ...") response = model.chat( msgs=[{"role": "user", "content": content}], max_new_tokens=4096, max_inp_length=32768, do_sample=True, temperature=0.7, use_image_id=False, max_slice_nums=1, use_tts_template=True, enable_thinking=False, omni_mode=True, generate_audio=True, output_audio_path="output/offline_chat.wav", ) print(response)

7.3.2 记忆离线对话

model.use_memory() 在对话调用前启用记忆；检索最多选取 96 个历史帧加 4 个近期帧，每帧附带 ±1 秒音频。

点击查看记忆离线对话代码。

import os

os.environ["MAX_NUM_FRAMES"] = "100000"

from minicpmo.utils import get_video_frame_audio_segments

model.use_memory() model.init_tts()

video_path = "Realtime-Venus-Omni/assets/sample_1_real.mp4" question = "What is the color of the cooler labeled PRIME near the team bench?" print(f"Extracting audio and frames from {video_path} ...") frames, audios, _ = get_video_frame_audio_segments( video_path, stack_frames=1, use_ffmpeg=True, adjust_audio_length=True ) content = [] for frame, audio in zip(frames, audios): if frame is not None: content.append(frame) content.append(audio) content.append(question)

print("Running chat inference ...") response = model.chat( msgs=[{"role": "user", "content": content}], max_new_tokens=4096, max_inp_length=32768, do_sample=True, temperature=0.7, use_image_id=False, max_slice_nums=1, use_tts_template=True, enable_thinking=False, omni_mode=True, generate_audio=True, output_audio_path="output/offline_memory_chat.wav", ) print(response)

8. 🎧 Realtime-Venus-Audio 用法

这些示例的可独立运行版本位于 GitHub 上的 Audio cookbook。

Audio checkpoint 通过两种方式运行纯音频推理：基于轮次的 model.chat（文本回复）以及全双工流式 API（语音回复）。输入会从任意音频或视频文件中解码为 16 kHz 单声道音频。

8.1 🧱 模型初始化

通过 init_tts=True 启用语音输出，因此同一个 model 可同时服务两个示例；若仅需纯文本对话，可使用 init_tts=False 以加快加载速度。

from pathlib import Path

import torch from transformers import AutoModel, AutoTokenizer, set_seed

Path("output").mkdir(exist_ok=True) set_seed(42) print("Loading model ...") tokenizer = AutoTokenizer.from_pretrained( "./Realtime-Venus-Audio", trust_remote_code=True, local_files_only=True, fix_mistral_regex=True, ) model = AutoModel.from_pretrained( "./Realtime-Venus-Audio", trust_remote_code=True, local_files_only=True, attn_implementation="sdpa", torch_dtype=torch.bfloat16, init_vision=False, init_audio=True, init_tts=True, ).eval().cuda() print("Model loaded.")

8.2 💭 离线对话

对完整音频输入执行一次确定性轮次：音频（外加可选的文本指令）通过单次 model.chat() 调用处理。

点击查看离线对话代码。

import librosa

print("Loading audio ...") audio, _ = librosa.load( "Realtime-Venus-Audio/assets/case_offline.wav", sr=16000, mono=True ) msgs = [ {"role": "system", "content": "You are a helpful assistant."}, {"role": "user", "content": [audio, "What is the speaker asking about?"]}, ]

print("Running chat inference ...") answer = model.chat( msgs=msgs, tokenizer=tokenizer, do_sample=False, max_new_tokens=2048, enable_thinking=False, use_tts_template=True, generate_audio=False, ) print(answer)

8.3 🎙️ 双工对话

model.as_duplex(generate_audio=True) 会切换至全双工流式模式：音频逐秒输入，模型持续聆听，并在回答时开口说话。该示例会在末尾追加 10 秒的静音，以便模型在输入结束后完成回复，并将生成的语音写入 output/audio_full_duplex.wav。

点击查看双工对话代码。

import librosa import numpy as np import soundfile as sf

duplex = model.as_duplex(generate_audio=True) duplex.prepare(prompt_wav_path="Realtime-Venus-Audio/assets/HT_ref_audio.wav")

audio, _ = librosa.load( "Realtime-Venus-Audio/assets/case_duplex.wav", sr=16000, mono=True ) audio = np.concatenate([audio, np.zeros(10 * 16000, dtype=np.float32)])

chunk_samples = int(duplex.CHUNK_MS * duplex.SAMPLE_RATE / 1000) total_chunks = max(1, (len(audio) + chunk_samples - 1) // chunk_samples) timed_audio = [] for chunk_index in range(total_chunks): chunk = audio[chunk_index * chunk_samples:(chunk_index + 1) * chunk_samples] if len(chunk) < chunk_samples: chunk = np.pad(chunk, (0, chunk_samples - len(chunk))) duplex.streaming_prefill(audio_waveform=chunk) result = duplex.streaming_generate( max_new_speak_tokens_per_chunk=20, decode_mode="sampling", temperature=0.7, top_k=20, top_p=0.8, listen_prob_scale=1.0, ) state = "listen" if result["is_listen"] else f"speak> {result['text']}" print(f"[{chunk_index + 1}/{total_chunks}] {state}", flush=True) if result["audio_waveform"] is not None and not result["is_listen"]: timed_audio.append((chunk_index, result["audio_waveform"]))

sample_rate = 24000 total_samples = max( t * sample_rate + len(np.asarray(w, dtype=np.float32).squeeze()) for t, w in timed_audio ) output = np.zeros(total_samples, dtype=np.float32) for t, waveform in timed_audio: w = np.asarray(waveform, dtype=np.float32).squeeze() output[t * sample_rate: t * sample_rate + len(w)] += w sf.write("output/audio_full_duplex.wav", np.clip(output, -1.0, 1.0), sample_rate) print("Saved generated speech to output/audio_full_duplex.wav")

9. 📝 引用

如果你觉得 Realtime-Venus 有用，请引用该技术报告：

@article{zhao2026realtime, title={{Realtime-Venus}: A full-duplex interaction system with asynchronous delegation}, author={{Venus Team,Ant Group;Tsinghua University}}, journal={arXiv preprint arXiv:2609.13814}, year={2026} }

10. 📄 许可证

本仓库包含一份 Apache License 2.0。请同时查阅上游模型、第三方库以及与本检查点搭配使用的任何数据的许可证和可接受使用条款。