菜单
Enhance multimodal models through cascade RL for reasoning improvement and visual resolution routing for inference efficiency, achieving 16% reasoning gains and 4.05x speedup.
默认使用会先检查来源的 Prompt；你也可以切换为直接命令，或下载本地副本。
决定是否安装前，请先阅读 SKILL.md，以及 SkillsMP 当前展示的配套文件。
用 Codex 或 Claude 帮你安装 复制这段 Prompt，粘贴到 Codex、Claude 或其他助手里，让它检查 Skill 页面并帮你完成安装。
直接命令不会经过审查 Prompt；运行前请先检查来源。
npx skills add https://github.com/ADu2021/skillXiv --skill internvl3-5-multimodal-cascade-rl命令会保持在同一行。复制前请横向滚动并检查完整内容。
想先保存到本地？可下载 SkillsMP 当前能够提供的文件。
基于 SOC 职业分类
正在显示 SKILL.md
| name | internvl3-5-multimodal-cascade-rl |
| title | InternVL3.5: Cascade RL and Visual Resolution Router for Multimodal Efficiency |
| version | 0.0.2 |
| engine | skillxiv-v0.0.2-claude-opus-4.6 |
| license | MIT |
| url | https://arxiv.org/abs/2508.18265 |
| keywords | ["multimodal-models","cascade-reinforcement-learning","visual-efficiency","reasoning","inference-optimization"] |
| description | Enhance multimodal models through cascade RL for reasoning improvement and visual resolution routing for inference efficiency, achieving 16% reasoning gains and 4.05x speedup. |
InternVL3.5 advances multimodal AI through two complementary techniques: Cascade RL for enhanced reasoning and Visual Resolution Router (ViR) for adaptive efficiency. Cascade RL combines offline RL for stable convergence with online RL for refined alignment, improving performance on visual reasoning benchmarks. ViR dynamically adjusts visual token resolution based on image complexity, maintaining performance while reducing computation. Together, these achieve 16% reasoning improvement and 4.05x inference speedup.
Two-stage training for stable and effective policy learning:
import torch
import torch.nn as nn
from typing import Dict, List, Tuple
class CascadeRLTrainer:
"""Two-stage RL: offline stable convergence + online refinement."""
def __init__(
self,
model: "MultimodalLLM",
offline_steps: int = 1000,
online_steps: int = 500,
offline_lr: float = 1e-5,
online_lr: float = 5e-6
):
self.model = model
self.offline_steps = offline_steps
self.online_steps = online_steps
.offline_optimizer = torch.optim.Adam(model.parameters(), lr=offline_lr)
.online_optimizer = torch.optim.Adam(model.parameters(), lr=online_lr)
.training_logs = {: [], : []}
() -> [, ]:
()
metrics = {: [], : []}
step (.offline_steps):
batch = ._sample_batch(dataset, batch_size=)
outputs = .model(batch[], batch[])
logits = outputs.logits
rewards = ._compute_correctness_reward(
outputs.sequences,
batch[]
)
loss = ._compute_offline_loss(logits, rewards)
.offline_optimizer.zero_grad()
loss.backward()
torch.nn.utils.clip_grad_norm_(.model.parameters(), )
.offline_optimizer.step()
metrics[].append(loss.item())
metrics[].append(rewards.mean().item())
(step + ) % == :
avg_loss = (metrics[][-:]) /
avg_reward = (metrics[][-:]) /
()
.training_logs[] = metrics
{: (metrics[]) / (metrics[]),
: (metrics[]) / (metrics[])}
() -> [, ]:
()
metrics = {: [], : []}
episode (num_episodes):
state = env.reset()
episode_reward =
episode_steps =
step (.online_steps):
action = .model.generate(
state[],
state[],
max_tokens=
)
next_state, reward, done = env.step(action)
._store_transition(state, action, reward, next_state, done)
loss = ._compute_online_loss(state, action, reward)
.online_optimizer.zero_grad()
loss.backward()
.online_optimizer.step()
episode_reward += reward
episode_steps +=
state = next_state
done:
metrics[].append(episode_reward)
metrics[].append(episode_steps)
(episode + ) % == :
avg_reward = (metrics[][-:]) /
()
.training_logs[] = metrics
{: (metrics[]) / (metrics[]),
: (metrics[]) / (metrics[])}
() -> torch.Tensor:
log_probs = torch.log_softmax(logits, dim=-)
policy_loss = -(log_probs.mean(dim=) * rewards).mean()
regularization =
policy_loss + regularization
() -> torch.Tensor:
outputs = .model(state[], state[])
logits = outputs.logits
action_log_prob = ._get_log_prob_of_action(action, logits)
loss = -action_log_prob * reward
loss
() -> torch.Tensor:
rewards = []
pred, target (predictions, targets):
correct = pred.strip() == target.strip()
reward = correct -
rewards.append(reward)
torch.tensor(rewards)
():
random
batch_samples = random.sample(dataset, (batch_size, (dataset)))
{
: [s[] s batch_samples],
: [s[] s batch_samples],
: [s[] s batch_samples]
}
():
() -> torch.Tensor:
Dynamically adjust image resolution based on complexity:
class VisualResolutionRouter(nn.Module):
"""Adaptively select image resolution for efficiency."""
def __init__(self, model: "MultimodalLLM", resolution_options: List[int] = [224, 336, 448, 672]):
super().__init__()
self.model = model
self.resolution_options = sorted(resolution_options)
# Router network: image complexity → resolution
self.complexity_encoder = nn.Sequential(
nn.Conv2d(3, 64, kernel_size=3, padding=1),
nn.ReLU(),
nn.AdaptiveAvgPool2d((1, 1)),
nn.Flatten(),
nn.Linear(64, 32),
nn.ReLU(),
nn.Linear(32, len(resolution_options))
)
def forward(
self,
images: torch.Tensor, # (batch, 3, h, w)
return_complexity: bool = False
) -> Tuple[torch.Tensor, List[int]]:
"""
Select resolution for each image based on complexity.
"""
batch_size = images.shape[0]
# Normalize image sizes
normalized_images = torch.nn.functional.interpolate(
images,
size=(224, 224), # Compute complexity at fixed size
mode='bilinear',
align_corners=False
)
# Compute complexity scores
complexity_logits = self.complexity_encoder(normalized_images)
complexity_probs = torch.softmax(complexity_logits, dim=-1)
complexity_scores = torch.max(complexity_probs, dim=-1)[1] # Argmax
# Select resolution for each image
selected_resolutions = [
self.resolution_options[score.item()] for score in complexity_scores
]
# Resize images to selected resolutions (batch-wise is inefficient, so do sequentially)
resized_images = []
for img, res in zip(images, selected_resolutions):
resized = torch.nn.functional.interpolate(
img.unsqueeze(0),
size=(res, res),
mode='bilinear',
align_corners=False
).squeeze(0)
resized_images.append(resized)
return resized_images, selected_resolutions
def train_router(
self,
train_images: List[torch.Tensor],
train_quality: List[float],
num_epochs: int = 10
):
"""
Train router to balance quality and efficiency.
"""
optimizer = torch.optim.Adam(self.complexity_encoder.parameters(), lr=1e-4)
for epoch in range(num_epochs):
total_loss = 0.0
for image, target_quality in zip(train_images, train_quality):
# Normalize to 224x224 for complexity encoding
img_normalized = torch.nn.functional.interpolate(
image.unsqueeze(0),
size=(224, 224)
)
# Get complexity prediction
logits = self.complexity_encoder(img_normalized)
complexity_score = torch.softmax(logits, dim=-1)
# Actual quality at each resolution (could be pre-computed)
# For simplicity: assume quality increases with resolution
expected_quality = complexity_score * torch.tensor([0.7, 0.8, 0.9, 1.0])
# Loss: match predicted complexity to target
loss = torch.nn.functional.mse_loss(expected_quality.sum(), torch.tensor(target_quality))
optimizer.zero_grad()
loss.backward()
optimizer.step()
total_loss += loss.item()
if (epoch + 1) % 2 == 0:
print(f"Router Training Epoch {epoch+1}: Loss={total_loss/len(train_images):.4f}")
Separate vision and language models for distributed inference:
class DecoupledInference:
"""Vision and language models on separate GPUs for efficiency."""
def __init__(
self,
vision_model: nn.Module,
language_model: nn.Module,
vision_device: str = "cuda:0",
language_device: str = "cuda:1"
):
self.vision_model = vision_model.to(vision_device)
self.language_model = language_model.to(language_device)
self.vision_device = vision_device
self.language_device = language_device
def forward(
self,
images: torch.Tensor,
questions: List[str]
) -> List[str]:
"""
Distributed forward pass: vision on GPU0, language on GPU1.
"""
# Vision processing on vision GPU
images = images.to(self.vision_device)
with torch.no_grad():
vision_embeddings = self.vision_model(images)
# Transfer to language GPU
vision_embeddings = vision_embeddings.to(self.language_device)
# Language processing on language GPU
language_inputs = self.language_model.encode_multimodal(
vision_embeddings,
questions
)
# Generate responses
responses = self.language_model.generate(language_inputs, max_length=100)
return responses
def measure_throughput(
self,
num_examples: int = 100,
batch_size: int = 8
) -> Dict[str, float]:
"""
Measure inference throughput with decoupled deployment.
"""
import time
start_time = time.time()
for _ in range(num_examples // batch_size):
# Generate dummy batch
dummy_images = torch.randn(batch_size, 3, 224, 224)
dummy_questions = ["What is in the image?"] * batch_size
# Forward pass
_ = self.forward(dummy_images, dummy_questions)
elapsed = time.time() - start_time
throughput = num_examples / elapsed
return {
"throughput_examples_per_second": throughput,
"latency_per_example_ms": (elapsed / num_examples) * 1000,
"total_time_seconds": elapsed
}
Researchers. (2024). InternVL3.5: Advancing Open-Source Multimodal Models. arXiv preprint arXiv:2508.18265.