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。请同时查阅上游模型、第三方库以及与本检查点搭配使用的任何数据的许可证和可接受使用条款。
Realtime-Venus supports proactive audio-visual interaction, asynchronous delegation, and interruption-aware full-duplex dialogue.
1. 🧭 Overview
This repository hosts two checkpoints of the Realtime-Venus system:
- Realtime-Venus-Omni (
Realtime-Venus-Omni/): the 9B audio-visual interaction model. It continuously watches and listens, decides whether and when to respond, and generates text and speech on a shared causal timeline. Adapted from MiniCPM-o 4.5, it supports proactive interaction, semantic interruption handling, and training-free long-video memory. - Realtime-Venus-Audio (
Realtime-Venus-Audio/): the audio-focused checkpoint on the same streaming backbone, for audio understanding and audio-driven conversation with text or speech output.
Both directories contain model weights and custom Hugging Face Transformers code. The asynchronous Realtime-Venus-Harness and its external tool integrations live in the GitHub repository.
2. ✨ Highlights
- Native full-duplex conversation: keeps perceiving while speaking and distinguishes backchannels, interruptions, corrections, and redirections.
- Omni-Proactive interaction: continuously processes temporally aligned video and audio, and initiates a response when an event warrants it — without waiting for a user prompt.
- Delegation: emits in-stream
<delegate>requests on the shared causal timeline and consumes asynchronous backend results the same way, so external tasks never block the ongoing conversation. (Executing requests requires the Realtime-Venus-Harness runtime, available in the GitHub repository.) - Training-free long-video Memory: archives visually informative moments, retrieves query-relevant and non-redundant evidence, and reassembles the corresponding audio-visual context — no additional training required.
- Text and speech output: generates response text together with native speech through the bundled Token2wav resources and a reference voice.
3. 📋 Model Details
| Item | Realtime-Venus-Omni | Realtime-Venus-Audio |
|---|---|---|
| Parameters | 9B | 9B |
| Base architecture | MiniCPM-o 4.5 / Omni-Flow | MiniCPM-o 4.5 / Omni-Flow |
| Visual encoder | SigLIP2 | not used at inference |
| Audio encoder | Whisper-Medium | Whisper-Medium |
| Language backbone | Qwen3-8B | Qwen3-8B |
| Speech generation | Discrete S3 speech tokens with a streaming flow-matching decoder | same decoder, enabled in full-duplex mode |
| Inputs | Video/images, audio, and text | Audio and text |
| Outputs | Text and optional speech waveform | Text and speech waveform |
| Context length | 40,960 tokens | 40,960 tokens |
| Weight dtype | BF16 | BF16 |
4. 📊 Evaluation
All values are reported in the Realtime-Venus technical report.
Figure 1. Video and audio understanding results from the paper.
Figure 2. Full-duplex interaction results from the paper.
5. 🗂️ Repository Layout
.
├── 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
The examples below write generated media to output/. Use a new filename or a new output directory when repeating an experiment.
6. 🛠️ Installation
Requires Python 3.10, CUDA, and FFmpeg. Download the repository (the two checkpoints live in its sub-directories) and install the Python dependencies:
huggingface-cli download inclusionAI/Realtime-Venus --local-dir .
python -m pip install -r Realtime-Venus-Omni/requirements.txt
All example paths below are relative to this repository's root directory. from_pretrained does not address sub-directories of a Hub repo, so the examples point at the local Realtime-Venus-Omni/ and Realtime-Venus-Audio/ paths after the download.
7. 🎙️ Realtime-Venus-Omni Usages
Runnable standalone versions of these examples live in the Omni cookbook on GitHub.
7.1 🧱 Model Initialization
The examples below share the following model initialization; run each example in a fresh Python process. Chat and Duplex automatically load the default reference voice.
Click to show Omni model loading code.
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 🔊 Duplex Omni Mode
model = model.as_duplex() switches the model to full-duplex streaming: prepare() initializes the session, then each second of input is handled by one streaming_prefill() + streaming_generate() pair, and as_simplex() switches back to offline mode. Set MAX_NUM_FRAMES before importing minicpmo.utils, otherwise videos longer than 64 seconds are truncated to the default frame cap.
Subtitle font note: Duplex examples burn the response text into the output video through FFmpeg/libass, which resolves fonts via fontconfig. Rendering non-Latin responses (e.g. Chinese) requires a CJK-capable font on the system, otherwise those glyphs show up as empty boxes. On any Linux distribution, install one without root and refresh the font cache:
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
Package-manager equivalents: apt install -y fonts-noto-cjk (Debian/Ubuntu) or yum install -y cjkuni-ukai-fonts cjkuni-uming-fonts (RHEL/Alibaba Cloud Linux). No code changes are needed.
7.2.1 Duplex Chat
Stream the demo video second by second and inject text questions at the seconds given by question_times (paired with questions). The model listens continuously and speaks when it answers.
Click to show the Duplex Chat code.
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
Same as above, except the question is spoken and already mixed into the video's audio track (at ~3 s, asking for an alert when the water boils), so no text is injected — the model must hear it.
Click to show the Speech-In Duplex Chat code.
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) enables the long-video Memory before entering duplex mode.
Click to show the Memory Duplex Chat code.
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 💬 Half-Duplex Omni Mode
model.chat(...) answers one turn at a time over the whole video. model.init_tts() enables speech output.
7.3.1 Offline Chat
Sampled frames, per-second audio, and the question go into a single chat() call. The 128-frame cap (MAX_NUM_FRAMES) limits the visual load, while max_inp_length=32768 sets the input-token budget. Full audio is still retained, so very long videos can exceed that budget even with frame sampling.
Click to show the Offline Chat code.
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 Memory Offline Chat
model.use_memory() enables Memory before the chat call; retrieval selects up to 96 historical frames plus 4 recent frames, each with ±1 s of audio.
Click to show the Memory Offline Chat code.
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 Usages
Runnable standalone versions of these examples live in the Audio cookbook on GitHub.
The Audio checkpoint runs audio-only inference in two ways: turn-based model.chat (text response) and the full-duplex streaming API (spoken response). Inputs are decoded as 16 kHz mono audio from any audio or video file.
8.1 🧱 Model Initialization
Speech output is enabled with init_tts=True so the same model serves both examples; use init_tts=False for text-only chat to load faster.
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 💭 Offline Chat
One deterministic turn over the full audio input: the audio (plus an optional text instruction) goes into a single model.chat() call.
Click to show the Offline Chat code.
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 🎙️ Duplex Chat
model.as_duplex(generate_audio=True) switches to full-duplex streaming: audio is fed second by second, the model listens continuously and speaks when it answers. The example appends 10 s of trailing silence so the model can finish its response after the input ends, and writes the generated speech to output/audio_full_duplex.wav.
Click to show the Duplex Chat code.
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. 📝 Citation
If you find Realtime-Venus useful, please cite the technical report:
@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. 📄 License
This repository includes an Apache License 2.0. Please also review the licenses and acceptable-use terms of the upstream model, third-party libraries, and any data used with this checkpoint.