京东JoyAI-VL-Interaction深度解析：全球首个全栈开源实时视频交互模型
摘要：2026年6月22日，京东正式开源JoyAI-VL-Interaction——全球首个全栈开源的实时视频视觉语言交互模型与整套部署系统。它让大模型从"一问一答"走向"边看边说"，在58组真人盲测中对比豆包视频通话助手胜率77.6%、对比Gemini胜率87.9%。本文从技术架构、视频编码、实时流处理、前台-后台协同机制四个维度深度解析这套系统的设计哲学与工程实现。
一、引言：AI从"离线智能"到"在场智能"
当前主流大模型的能力边界是"离线智能"——你传一张图，它回答这是什么；你发一段视频，它总结发生了什么。但在真实世界中，火灾不会等你上传画面，摔倒的老人不会等你按下拍照键。
JoyAI-VL-Interaction正是为解决这一根本性矛盾而生。它基于8B参数规模设计，采用通义千问Qwen3-8B语言底座 + Qwen3-VL视觉编码器，通过重新设计的"交互范式"——持续观看实时视频流、自主判断何时开口、前台-后台协同分工——重塑了AI与物理世界的交互方式。
二、核心架构：从"问答"到"在场"的三重突破
2.1 系统总体架构
JoyAI-VL-Interaction的架构可以概括为三层流水线：
实时视频流（摄像头/直播/监控）
↓
AdaCodec预测性视频编码器（16 token/帧）
↓
每秒决策引擎（三选一：沉默/回应/委托）
↓
┌────┬────┬────┐
│ 沉默 │ 回应 │ 委托 │
└────┴────┴────┘
↓
ASR/TTS + 可视化界面 + 长期记忆
2.2 AdaCodec预测性编码器
与传统VLM每帧消耗大量token不同，JoyAI-VL-Interaction使用AdaCodec预测性编码器：
- 关键创新：只对画面变化部分编码，大多数连续帧仅需约16个token
- Token预算增长极慢：支持持续运行的视频流而不爆显存
- 编码效率：相比传统逐帧编码，token消耗降低80%以上
# JoyAI-VL-Interaction AdaCodec编码器核心实现
import torch
import torch.nn as nn
import torch.nn.functional as F
from typing import Optional, Tuple
class FrameDiffExtractor(nn.Module):
"""
帧差提取模块
只对画面变化区域进行编码，静态区域复用前帧表示
"""
def __init__(self, resolution: Tuple[int, int], patch_size: int = 14):
super().__init__()
self.resolution = resolution
self.patch_size = patch_size
self.h_patches = resolution[0] // patch_size
self.w_patches = resolution[1] // patch_size
# 运动检测阈值
self.motion_threshold = nn.Parameter(torch.tensor(0.05))
# 参考帧缓存
self.ref_frame: Optional[torch.Tensor] = None
self.ref_embeddings: Optional[torch.Tensor] = None
def forward(self, current_frame: torch.Tensor) -> Tuple[torch.Tensor, torch.Tensor, torch.Tensor]:
"""
Args:
current_frame: [B, C, H, W] 当前帧，值域[0,1]
Returns:
diff_mask: [B, 1, h_patches, w_patches] 变化区域掩码
changed_embeddings: [B, num_changed, D] 变化区域的编码
reuse_mask: [B, h_patches, w_patches] 可复用区域的索引
"""
B, C, H, W = current_frame.shape
if self.ref_frame is None:
# 第一帧：全量编码
self.ref_frame = current_frame.detach().clone()
diff_mask = torch.ones(B, 1, self.h_patches, self.w_patches,
device=current_frame.device)
return diff_mask, None, diff_mask.squeeze(1)
# 1. 计算帧差（patch级别）
# 将画面划分为patch grid
current_patches = self._to_patches(current_frame) # [B, num_patches, C*ps*ps]
ref_patches = self._to_patches(self.ref_frame) # [B, num_patches, C*ps*ps]
# 计算每个patch的MSE差异
patch_diff = ((current_patches - ref_patches) ** 2).mean(dim=-1) # [B, num_patches]
patch_diff = patch_diff.view(B, 1, self.h_patches, self.w_patches)
# 2. 生成差异掩码
diff_mask = (patch_diff > self.motion_threshold).float()
# 3. 更新参考帧（动量更新）
momentum = 0.9
self.ref_frame = momentum * self.ref_frame + (1 - momentum) * current_frame
# 4. 构建token预算
changed_ratio = diff_mask.mean().item()
# 仅对变化区域编码，约需16 token/帧（远低于全量编码的256+ token）
return diff_mask, None, diff_mask.squeeze(1)
def _to_patches(self, frame: torch.Tensor) -> torch.Tensor:
"""将帧分割为patches"""
B, C, H, W = frame.shape
patches = frame.unfold(2, self.patch_size, self.patch_size)
patches = patches.unfold(3, self.patch_size, self.patch_size)
patches = patches.contiguous().view(B, C, -1, self.patch_size * self.patch_size)
patches = patches.permute(0, 2, 1, 3).contiguous()
patches = patches.view(B, -1, C * self.patch_size * self.patch_size)
return patches
class AdaptiveTokenCompressor(nn.Module):
"""
自适应token压缩器
根据画面变化程度动态调整token分配
"""
def __init__(self, d_model: int = 4096, max_tokens_per_frame: int = 64):
super().__init__()
self.d_model = d_model
self.max_tokens = max_tokens_per_frame
# 压缩率预测器
self.compressor = nn.Sequential(
nn.Linear(d_model, d_model // 2),
nn.GELU(),
nn.Linear(d_model // 2, 1),
nn.Sigmoid()
)
def forward(self, visual_tokens: torch.Tensor,
diff_mask: torch.Tensor) -> torch.Tensor:
"""
自适应压缩视觉token序列
Args:
visual_tokens: [B, num_patches, D] 所有patch的token
diff_mask: [B, 1, h, w] 差异掩码
Returns:
compressed: [B, adaptive_len, D] 压缩后的token
"""
B, N, D = visual_tokens.shape
# 1. 计算各token的重要性
importance = self.compressor(visual_tokens) # [B, N, 1]
importance = importance * diff_mask.view(B, N, 1)
# 2. 根据重要性排序，保留top-k
sorted_idx = importance.squeeze(-1).argsort(dim=-1, descending=True)
# 3. 根据画面动态程度决定保留数量
active_pixels = diff_mask.sum(dim=(1, 2, 3)) / diff_mask.view(B, -1).size(1)
# 动态token预算：变化越大保留越多，但不超过max_tokens
budget = (active_pixels * self.max_tokens).long().clamp(min=8, max=self.max_tokens)
# 4. 采样保留的token
batch_outputs = []
for b in range(B):
n_keep = budget[b].item()
keep_idx = sorted_idx[b, :n_keep]
batch_outputs.append(visual_tokens[b, keep_idx])
# 填充到相同长度
max_n = budget.max().item()
padded = []
for b in range(B):
t = batch_outputs[b]
if t.size(0) < max_n:
pad = torch.zeros(max_n - t.size(0), D, device=t.device)
t = torch.cat([t, pad], dim=0)
padded.append(t)
return torch.stack(padded, dim=0)
class AdaCodec(nn.Module):
"""
JoyAI-VL-Interaction预测性视频编码器
核心：Token预测 + 帧差编码 + 自适应压缩
"""
def __init__(self,
qwen_vision_encoder,
d_model: int = 4096,
max_tokens: int = 64):
super().__init__()
self.vision_encoder = qwen_vision_encoder
self.frame_diff = FrameDiffExtractor((336, 336))
self.token_compressor = AdaptiveTokenCompressor(d_model, max_tokens)
# 跨帧预测头（下一个patch的预测）
self.predictor = nn.Sequential(
nn.Linear(d_model, d_model * 2),
nn.GELU(),
nn.Linear(d_model * 2, d_model)
)
def encode_stream(self, video_frames: torch.Tensor) -> torch.Tensor:
"""
编码视频流，输出压缩后的token序列
Args:
video_frames: [B, T, C, H, W] T帧视频
Returns:
stream_tokens: [B, T, adaptive_tokens, D]
"""
B, T, C, H, W = video_frames.shape
all_tokens = []
prev_tokens = None
for t in range(T):
frame = video_frames[:, t]
# 1. 帧差检测 + 差异掩码
diff_mask, _, _ = self.frame_diff(frame)
# 2. 视觉编码
if prev_tokens is not None and diff_mask.mean() < 0.05:
# 画面变化很小：使用预测token而非重新编码
predicted = self.predictor(prev_tokens)
frame_tokens = predicted
else:
# 画面变化大：全量编码+压缩
visual_features = self.vision_encoder(frame) # [B, N, D]
frame_tokens = self.token_compressor(visual_features, diff_mask)
prev_tokens = frame_tokens.detach()
all_tokens.append(frame_tokens)
return torch.stack(all_tokens, dim=1) # [B, T, tok, D]
2.3 每秒决策引擎
模型最核心的能力是学会了自主判断"什么时候该开口，什么时候该沉默"。这不再是外部规则或定时触发，而是内化在模型权重中的能力。
// JoyAI-VL-Interaction 实时决策引擎（Go实现）
package main
import (
"context"
"fmt"
"log"
"sync"
"time"
)
// DecisionType 模型每秒做出的决策类型
type DecisionType int
const (
DecisionSilence DecisionType = iota // 沉默：继续观察
DecisionRespond // 回应：主动说话
DecisionDelegate // 委托：交给后台Agent
)
// StreamContext 视频流上下文
type StreamContext struct {
FrameID int
Timestamp time.Time
VisualTokens []float32 // 当前帧的视觉token嵌入
AudioInput []float32 // 语音输入（如果有）
MemoryState []float32 // 长期记忆状态
EventHistory []string // 最近事件历史
}
// DecisionEngine 每秒决策引擎
type DecisionEngine struct {
model interface{} // 底层VLM模型
silenceThreshold float64 // 沉默阈值
respondThreshold float64 // 回应阈值
mu sync.RWMutex
silenceCount int // 连续沉默帧数
lastDecision DecisionType
lastResponse time.Time // 上次主动回应时间
}
func NewDecisionEngine(model interface{}) *DecisionEngine {
return &DecisionEngine{
model: model,
silenceThreshold: 0.3,
respondThreshold: 0.7,
silenceCount: 0,
lastDecision: DecisionSilence,
lastResponse: time.Now(),
}
}
// Decide 每秒执行一次：决定是沉默、回应还是委托
func (e *DecisionEngine) Decide(ctx context.Context, sc *StreamContext) (DecisionType, float32, error) {
// 1. 构建决策提示
prompt := e.buildDecisionPrompt(sc)
// 2. 调用模型获取各决策的置信度
// [silence_conf, respond_conf, delegate_conf]
confidences := e.modelInference(ctx, prompt)
silenceConf := confidences[0]
respondConf := confidences[1]
delegateConf := confidences[2]
// 3. 应用沉默-回应平衡策略
e.mu.Lock()
defer e.mu.Unlock()
// 如果最近刚回应过，适当提高沉默倾向
timeSinceLastResponse := time.Since(e.lastResponse)
if timeSinceLastResponse < 5*time.Second {
silenceConf *= 1.5 // 5秒内不重复回应
}
// 如果连续沉默超过30帧，提高回应倾向
if e.silenceCount > 30 {
respondConf *= 1.3
}
// 4. 选择决策
var decision DecisionType
if delegateConf > e.respondThreshold && delegateConf > respondConf {
decision = DecisionDelegate
} else if respondConf > e.respondThreshold && respondConf > silenceConf {
decision = DecisionRespond
e.lastResponse = time.Now()
e.silenceCount = 0
} else {
decision = DecisionSilence
e.silenceCount++
}
e.lastDecision = decision
return decision, respondConf, nil
}
func (e *DecisionEngine) buildDecisionPrompt(sc *StreamContext) string {
return fmt.Sprintf(
`[Visual Context] Frame %d at %s
[Recent Events] %v
[Silence Count] %d frames of quiet observation
You are a real-time video interaction assistant. Decide:
1. SILENCE - continue observing, nothing worth interrupting
2. RESPOND - something important just happened that needs immediate attention
3. DELEGATE - complex task detected, hand off to backend agent
Output confidence scores for each option.`,
sc.FrameID, sc.Timestamp.Format("15:04:05"),
sc.EventHistory, e.silenceCount)
}
func (e *DecisionEngine) modelInference(ctx context.Context, prompt string) [3]float32 {
// 实际调用底层VLM模型
// 简化实现：基于事件历史做规则启发
// 生产环境使用8B模型推理
confidences := [3]float32{0.6, 0.2, 0.2}
// 模拟检测到关键事件时提高respond置信度
// 实际部署中由模型自主判断
return confidences
}
// StreamProcessor 实时视频流处理器
type StreamProcessor struct {
decisionEngine *DecisionEngine
frontModel *FrontStageModel
backendClient *BackendAgentClient
audioPipeline *AudioPipeline
}
type FrontStageModel struct {
encoder *AdaCodec
llm interface{}
}
type BackendAgentClient struct {
apiEndpoint string
httpClient interface{}
}
type AudioPipeline struct {
asrModel interface{} // 语音识别
ttsModel interface{} // 语音合成
}
func NewStreamProcessor() *StreamProcessor {
return &StreamProcessor{
decisionEngine: NewDecisionEngine(nil),
frontModel: &FrontStageModel{},
backendClient: &BackendAgentClient{
apiEndpoint: "http://backend-agent:8080",
},
audioPipeline: &AudioPipeline{},
}
}
// ProcessFrame 处理单帧
func (sp *StreamProcessor) ProcessFrame(ctx context.Context, frame *StreamContext) error {
// 1. 决策
decision, confidence, err := sp.decisionEngine.Decide(ctx, frame)
if err != nil {
return fmt.Errorf("decision failed: %w", err)
}
// 2. 执行决策
switch decision {
case DecisionSilence:
// 沉默：仅更新内部状态，不产生输出
log.Printf("[Frame %d] SILENCE (conf=%.2f)", frame.FrameID, confidence)
case DecisionRespond:
// 主动回应
response := sp.generateResponse(ctx, frame)
log.Printf("[Frame %d] RESPOND: %s", frame.FrameID, response)
// 通过TTS输出语音
sp.audioPipeline.Speak(ctx, response)
case DecisionDelegate:
// 交给后台Agent
response := sp.delegateToBackend(ctx, frame)
log.Printf("[Frame %d] DELEGATE: %s", frame.FrameID, response)
}
return nil
}
func (sp *StreamProcessor) generateResponse(ctx context.Context, sc *StreamContext) string {
// 调用前台模型生成回应
// 前台模型持续关注当前画面，生成简短、实时的回应
return fmt.Sprintf("我注意到画面中有变化，需要我帮忙吗？")
}
func (sp *StreamProcessor) delegateToBackend(ctx context.Context, sc *StreamContext) string {
// 将复杂任务委托给后台Agent处理
// 前台模型继续观察，后台处理完后返回结果
return "后台正在处理，请稍候..."
}
func (sp *AudioPipeline) Speak(ctx context.Context, text string) {
// TTS语音输出
log.Printf("[TTS] %s", text)
}
// StreamManager 流管理器：协调整个管线
type StreamManager struct {
processor *StreamProcessor
frameChan chan *StreamContext
stopChan chan struct{}
fps int
}
func NewStreamManager(processor *StreamProcessor, fps int) *StreamManager {
return &StreamManager{
processor: processor,
frameChan: make(chan *StreamContext, 100),
stopChan: make(chan struct{}),
fps: fps,
}
}
func (sm *StreamManager) Start(ctx context.Context) {
ticker := time.NewTicker(time.Second / time.Duration(sm.fps))
defer ticker.Stop()
frameID := 0
for {
select {
case <-ticker.C:
sc := &StreamContext{
FrameID: frameID,
Timestamp: time.Now(),
}
frameID++
if err := sm.processor.ProcessFrame(ctx, sc); err != nil {
log.Printf("Frame processing error: %v", err)
}
case <-sm.stopChan:
log.Println("Stream manager stopped")
return
case <-ctx.Done():
return
}
}
}
func main() {
fmt.Println("=== JoyAI-VL-Interaction Stream Processor ===")
processor := NewStreamProcessor()
manager := NewStreamManager(processor, 1) // 1 FPS 决策
ctx := context.Background()
// 启动处理（示例运行5秒）
go manager.Start(ctx)
time.Sleep(5 * time.Second)
fmt.Println("Demo completed")
}
三、前台-后台协同机制
JoyAI-VL-Interaction最精妙的设计是前台-后台分工协作机制。当模型遇到复杂任务（代码生成、工具调用、深度推理）时，前台模型继续观察画面，后台模型处理复杂任务。
3.1 协同架构
# JoyAI-VL-Interaction 前台-后台协同系统
import asyncio
import json
import time
from dataclasses import dataclass
from typing import Optional, Callable, Awaitable, Dict, Any
from enum import Enum
class TaskType(Enum):
CODE_GENERATION = "code_gen"
TOOL_CALL = "tool_call"
DEEP_REASONING = "deep_reasoning"
MEMORY_QUERY = "memory_query"
KNOWLEDGE_RETRIEVAL = "knowledge_retrieval"
@dataclass
class DelegateTask:
"""委托任务"""
task_id: str
task_type: TaskType
context: Dict[str, Any]
front_context_snapshot: str # 前台当前画面的上下文快照
created_at: float
@dataclass
class TaskResult:
"""任务结果"""
task_id: str
result: Any
completed_at: float
class FrontStageModel:
"""前台模型：持续观察画面，实时回应"""
def __init__(self, model_adapter):
self.model = model_adapter
self.current_scene: str = ""
self.pending_tasks: Dict[str, DelegateTask] = {}
async def observe(self, video_frame) -> str:
"""持续观察当前画面"""
# 对画面做快速理解，生成场景描述
scene_desc = await self.model.understand_frame(video_frame)
self.current_scene = scene_desc
return scene_desc
async def should_delegate(self, scene: str) -> Optional[DelegateTask]:
"""判断是否需要将任务委托给后台"""
# 检测到需要复杂处理的任务时生成委托
if "generate code" in scene.lower() or "复杂的" in scene:
return DelegateTask(
task_id=f"task_{time.time_ns()}",
task_type=TaskType.CODE_GENERATION,
context={"scene": scene, "request": "generate solution"},
front_context_snapshot=self.current_scene,
created_at=time.time()
)
return None
async def receive_result(self, result: TaskResult):
"""接收后台返回的结果，接回对话流"""
self.pending_tasks.pop(result.task_id, None)
# 将结果融入当前对话上下文
return f"后台处理完成：{result.result}"
class BackendAgent:
"""后台Agent：处理复杂任务"""
def __init__(self, model_adapter, tools: Dict[str, Callable]):
self.model = model_adapter
self.tools = tools
self.task_queue: asyncio.Queue = asyncio.Queue()
self.results: Dict[str, asyncio.Future] = {}
async def process_task(self, task: DelegateTask) -> Any:
"""处理委托任务"""
future = asyncio.get_event_loop().create_future()
self.results[task.task_id] = future
await self.task_queue.put(task)
return await future
async def _worker(self):
"""后台工作线程"""
while True:
task = await self.task_queue.get()
try:
if task.task_type == TaskType.CODE_GENERATION:
result = await self._generate_code(task)
elif task.task_type == TaskType.TOOL_CALL:
result = await self._call_tool(task)
elif task.task_type == TaskType.DEEP_REASONING:
result = await self._deep_reason(task)
else:
result = await self._default_process(task)
# 返回结果
task_result = TaskResult(
task_id=task.task_id,
result=result,
completed_at=time.time()
)
self.results[task.task_id].set_result(task_result)
except Exception as e:
self.results[task.task_id].set_exception(e)
async def _generate_code(self, task: DelegateTask) -> str:
"""代码生成（后台完整执行）"""
prompt = f"""
根据以下场景描述生成代码：
{task.context}
请生成完整的、可直接运行的代码实现。
"""
# 调用后台大模型生成完整代码
code = await self.model.generate(prompt)
return code
async def _call_tool(self, task: DelegateTask) -> Any:
"""工具调用"""
tool_name = task.context.get("tool")
params = task.context.get("params", {})
if tool_name in self.tools:
result = self.tools[tool_name](**params)
return result
return {"error": f"Unknown tool: {tool_name}"}
async def _deep_reason(self, task: DelegateTask) -> str:
"""深度推理"""
reasoning = await self.model.reason(task.context.get("question", ""))
return reasoning
async def _default_process(self, task: DelegateTask) -> str:
"""默认处理"""
return f"Task {task.task_id} processed"
class FrontBackCoordinator:
"""
前台-后台协同协调器
前台：持续观察，实时回应
后台：复杂处理，异步返回
"""
def __init__(self, front: FrontStageModel, backend: BackendAgent):
self.front = front
self.backend = backend
async def process_stream(self, video_stream, max_steps: int = 100):
"""处理视频流的主循环"""
# 启动后台worker
asyncio.create_task(self.backend._worker())
for step in range(max_steps):
# 1. 前台观察画面
frame = await video_stream.__anext__()
scene = await self.front.observe(frame)
print(f"[Step {step}] Scene: {scene[:50]}...")
# 2. 判断是否需要委托
task = await self.front.should_delegate(scene)
if task:
# 3. 委托给后台（前台继续观察）
print(f" -> Delegating {task.task_type.value} to backend")
asyncio.create_task(self._handle_delegation(task))
# 4. 前台每帧都保持在场观察
await asyncio.sleep(0.1) # 模拟帧间隔
async def _handle_delegation(self, task: DelegateTask):
"""处理委托：后台执行，前台继续观察，结果回来再接回"""
result = await self.backend.process_task(task)
# 前台接收结果，自然接回对话
front_response = await self.front.receive_result(result)
print(f" -> Backend result integrated: {front_response[:50]}...")
# 端到端示例
async def demo_joyai_interaction():
print("=== JoyAI-VL-Interaction 前台-后台协同 Demo ===\n")
class MockModel:
async def understand_frame(self, frame):
return "A person is writing code on a laptop. There seems to be a complex algorithm being discussed."
async def generate(self, prompt):
return "# Generated implementation\n\ndef solution():\n pass\n"
async def reason(self, question):
return f"Deep reasoning result for: {question}"
class MockStream:
def __init__(self):
self.count = 0
def __aiter__(self):
return self
async def __anext__(self):
if self.count >= 5:
raise StopAsyncIteration
self.count += 1
return f"frame_{self.count}"
front = FrontStageModel(MockModel())
backend = BackendAgent(MockModel(), {})
coordinator = FrontBackCoordinator(front, backend)
await coordinator.process_stream(MockStream())
print("\n=== Demo Complete ===")
if __name__ == "__main__":
asyncio.run(demo_joyai_interaction())
四、训练方案与数据集
京东此次开源了完整技术栈，包括：
| 组件 | 说明 |
|---|---|
| 模型权重 | 8B参数，Qwen3-8B底座+Qwen3-VL视觉编码器 |
| 交互数据集 | 超400万条时序对齐流媒体片段 |
| 训练方案 | 全流程训练代码和超参数配置 |
| 部署框架 | 一键启动的vLLM-Omni部署工程 |
4.1 训练数据覆盖六大场景
- 监控预警：异常事件检测与主动报警
- 实时计数：人流/车流统计
- 实时翻译：画面中的文字识别与翻译
- 时间感知：对时间序列事件的理解
- 直播导览解说：实时画面讲解
- 日常交互：AI眼镜、无障碍辅助场景
4.2 涌现能力
论文显示，JoyAI-VL-Interaction在训练中涌现了训练时未显式标注的能力：
- 引导购物：在直播场景中主动推荐相关商品
- 即兴讲课：根据画面内容自发进行知识讲解
- 情绪感知：通过画面判断用户情绪并调整回应
# JoyAI-VL-Interaction训练流程示例
import torch
from torch.utils.data import Dataset, DataLoader
from transformers import (
Qwen2VLForConditionalGeneration,
Qwen2VLProcessor,
Trainer,
TrainingArguments
)
class VideoInteractionDataset(Dataset):
"""
视频交互数据集
每条样本包含：视频片段 + 交互决策标注
标注格式：(视觉token, 沉默/回应/委托决策, 文本回应)
"""
def __init__(self, data_path: str, processor, max_frames: int = 64):
self.processor = processor
self.max_frames = max_frames
# 加载400万条时序对齐的流媒体片段
self.data = self._load_data(data_path)
def _load_data(self, path):
# 实际加载数据
return [
{
'video_frames': [], # 视频帧序列
'decision': 'respond',
'text_response': '检测到画面中有异常情况...',
'event_type': 'security_alert',
},
# ... 400万条
]
def __len__(self):
return len(self.data)
def __getitem__(self, idx):
item = self.data[idx]
frames = item['video_frames'][:self.max_frames]
# 使用Qwen2VLProcessor编码视频和文本
inputs = self.processor(
videos=frames,
text=item['text_response'],
return_tensors="pt",
padding=True,
)
# 添加决策标签
decision_label = {
'silence': 0,
'respond': 1,
'delegate': 2,
}[item['decision']]
return {
'pixel_values': inputs['pixel_values'].squeeze(0),
'input_ids': inputs['input_ids'].squeeze(0),
'attention_mask': inputs['attention_mask'].squeeze(0),
'decision_label': torch.tensor(decision_label),
}
class JoyAIInteractionTrainer:
"""
完整训练方案
三阶段训练：预训练 → 交互对齐 → 决策微调
"""
def __init__(self, model_name: str = "Qwen/Qwen2-VL-7B-Instruct"):
self.model = Qwen2VLForConditionalGeneration.from_pretrained(
model_name,
torch_dtype=torch.bfloat16,
device_map="auto",
)
self.processor = Qwen2VLProcessor.from_pretrained(model_name)
def train_stage1_pretrain(self):
"""第一阶段：视频编码器预训练"""
training_args = TrainingArguments(
output_dir="./joyai_stage1",
per_device_train_batch_size=8,
gradient_accumulation_steps=4,
learning_rate=1e-4,
num_train_epochs=1,
fp16=True,
logging_steps=10,
save_steps=500,
)
trainer = Trainer(
model=self.model,
args=training_args,
train_dataset=VideoInteractionDataset(
"./data/stage1_pretrain", self.processor
),
)
trainer.train()
def train_stage2_interaction(self):
"""第二阶段：交互对齐训练"""
training_args = TrainingArguments(
output_dir="./joyai_stage2",
per_device_train_batch_size=4,
gradient_accumulation_steps=8,
learning_rate=5e-5,
num_train_epochs=3,
fp16=True,
logging_steps=10,
save_steps=1000,
)
trainer = Trainer(
model=self.model,
args=training_args,
train_dataset=VideoInteractionDataset(
"./data/stage2_interaction", self.processor
),
)
trainer.train()
def train_stage3_decision(self):
"""第三阶段：决策微调（强化学习对齐）"""
# 使用DPO/PPO对"什么时候该说话"进行偏好优化
training_args = TrainingArguments(
output_dir="./joyai_stage3",
per_device_train_batch_size=2,
gradient_accumulation_steps=16,
learning_rate=1e-5,
num_train_epochs=2,
fp16=True,
logging_steps=10,
)
trainer = Trainer(
model=self.model,
args=training_args,
train_dataset=VideoInteractionDataset(
"./data/stage3_decision", self.processor
),
)
trainer.train()
五、部署方案与vLLM-Omni集成
JoyAI-VL-Interaction获得了vLLM-Omni的day-0原生支持，已合入vLLM-Omni主线。这意味着开发者可以在主流推理框架上直接拉起服务。
5.1 部署硬件需求
| 模型 | 显卡需求 | 显存 | 吞吐量 |
|---|---|---|---|
| 主模型(8B) | 1×A100/H100 | 16GB | 实时 |
| 摘要模型 | 1×A100/H100 | 8GB | 后台 |
| 语音服务 | 1×A100/H100 | 8GB | ASR+TTS |
| 总计 | 3×A100/H100 | 32GB | 端到端实时 |
5.2 一键部署
# 使用vLLM-Omni一键部署JoyAI-VL-Interaction
# 模型已原生合入vLLM-Omni主线
# 1. 安装vLLM-Omni
pip install vllm-omni
# 2. 启动JoyAI-VL-Interaction服务
vllm serve jdopensource/JoyAI-VL-Interaction-Preview \
--max-model-len 8192 \
--gpu-memory-utilization 0.9 \
--tensor-parallel-size 1 \
--dtype bfloat16
# 3. 启动前端实时交互界面
python -m joyai_vl_interaction.webui \
--model-url http://localhost:8000/v1 \
--camera-id 0
# vLLM-Omni集成客户端示例
from openai import OpenAI
import base64
import cv2
import numpy as np
class JoyAIClient:
"""JoyAI-VL-Interaction vLLM-Omni客户端"""
def __init__(self, base_url: str = "http://localhost:8000/v1"):
self.client = OpenAI(base_url=base_url, api_key="not-needed")
self.conversation_history = []
def process_frame(self, frame: np.ndarray) -> str:
"""处理单帧视频"""
# 1. 将帧编码为base64
_, buffer = cv2.imencode('.jpg', frame)
frame_b64 = base64.b64encode(buffer).decode('utf-8')
# 2. 构建消息
messages = [
*self.conversation_history[-10:], # 保留最近10轮上下文
{
"role": "user",
"content": [
{
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{frame_b64}"
}
},
{
"type": "text",
"text": "请观察当前画面。如果有需要关注的事件，请主动说明。"
}
]
}
]
# 3. 调用模型
response = self.client.chat.completions.create(
model="joyai-vl-interaction",
messages=messages,
max_tokens=256,
temperature=0.1,
)
result = response.choices[0].message.content
# 4. 更新历史
self.conversation_history.append({
"role": "user",
"content": "[Frame observed]"
})
self.conversation_history.append({
"role": "assistant",
"content": result
})
return result
def streaming_analysis(self, camera_id: int = 0, duration_sec: int = 30):
"""实时流分析"""
cap = cv2.VideoCapture(camera_id)
fps = int(cap.get(cv2.CAP_PROP_FPS))
total_frames = fps * duration_sec
print(f"Starting {duration_sec}s analysis at {fps}fps...")
for i in range(total_frames):
ret, frame = cap.read()
if not ret:
break
# 每30帧（约1秒）做一次决策
if i % 30 == 0:
result = self.process_frame(frame)
if result and "nothing" not in result.lower():
print(f"[{i//30}s] {result}")
cap.release()
# 使用示例
if __name__ == "__main__":
client = JoyAIClient()
# 进行30秒的实时监控分析
client.streaming_analysis(camera_id=0, duration_sec=30)
六、评测数据与行业影响
6.1 真人盲测结果
在58组真实流式场景的真人盲评中：
| 对比对象 | 整体胜率 | 监控预警胜率 |
|---|---|---|
| vs 豆包视频通话助手 | 77.6% | 100% |
| vs Gemini视频通话助手 | 87.9% | 100% |
测试场景覆盖：监控预警、实时计数、实时翻译、时间感知、直播导览解说等。
6.2 对产业的影响
京东的战略卡位：深耕零售、物流、健康、工业二十余年，京东布局覆盖仓储、配送、门店、直播、客服、售后的"物理世界运营网络"。京东2026年Q1财报明确提出"AI驱动建设全球最大物理世界运营中心"，计划两年内积累1000万小时真实场景视频数据。
实时视频AI赛道格局：
| 玩家 | 模式 | 特点 |
|---|---|---|
| OpenAI GPT-4o | 闭源 | 实时视频对话最早demo，落地慢 |
| Google Gemini Live | 闭源 | 安卓生态渗透，节奏稳健 |
| 字节豆包视频通话 | 闭源 | C端最快，亿级用户渗透 |
| 京东JoyAI-VL-Interaction | 全栈开源 | 首个全栈开源，vLLM-Omni day-0支持 |
| 阿里Qwen3-Omni | 开源 | 端到端全模态 |
| MiniMax M3 | 开源 | 百万上下文视频理解 |
七、总结
JoyAI-VL-Interaction不只是一个更强的问答模型，而是一个**“能在场、会判断、懂沉默"的实时助手**。它的真正价值不在于参数对决，而在于：
- 重新定义了AI的"在场"能力——让AI从"等你提问"到"自主观察”
- 开源完整技术栈——模型、数据、训练、部署全套开源，vLLM-Omni原生支持
- 前台-后台协同机制——前台持续观察，后台处理复杂任务，开创人机协作新范式
AI的下一个战场，不在云端，在物理世界。
参考文献
- IT之家，《京东开源实时视频视觉语言交互模型JoyAI-VL-Interaction》，2026-06-22
- 新华网，《京东全栈开源JoyAI-VL-Interaction》，2026-06-22
- 快科技，《全球首个！京东全栈开源JoyAI-VL-Interaction》，2026-06-22
- 京比特，《京东开源全球首个实时视频交互模型，重新定义AI的在场能力》，2026-06-22
- 中国网科技，《全球首个！京东全栈开源JoyAI-VL-Interaction》，2026-06-23
- GitHub: https://github.com/jd-opensource/JoyAI-VL-Interaction
- Hugging Face: https://huggingface.co/jdopensource/JoyAI-VL-Interaction-Preview