adu2021 ReVisual-R1: Advancing Multimodal Reasoning
Core Concept
ReVisual-R1 demonstrates that unlocking multimodal reasoning capabilities requires a carefully designed three-stage curriculum: text-only cold-start with complex reasoning examples, multimodal RL with prioritized advantage distillation to prevent gradient stagnation, and text-only refinement to consolidate linguistic fluency. This approach outperforms simultaneous mixed-modality training and achieves state-of-the-art performance among 3B/7B open-source models.
Architecture Overview
- Stage 1: Text-Centric Cold Start - 283K curated high-difficulty reasoning examples (language-only)
- Stage 2: Multimodal RL with PAD - GRPO enhanced by Prioritized Advantage Distillation
- Stage 3: Text-Only RL Refinement - Polish linguistic quality and reasoning consistency
- Prioritized Advantage Distillation (PAD): Filters zero-advantage samples and resamples informative trajectories
- Challenge Addressed: Gradient stagnation in multimodal GRPO settings where standard reward signals provide weak gradients
Implementation
Step 1: Prepare Cold-Start Text-Only Dataset
import json
from typing import List, Dict
class ColdStartDatasetBuilder:
def __init__(self, num_samples=283000):
self.target_size = num_samples
self.difficulty_threshold = "high"
def collect_complex_reasoning_examples(self):
"""Gather high-difficulty text reasoning from multiple sources"""
sources = {
'math_olympiad': self.source_olympiad_problems(),
'theoretical_physics': self.source_physics_proofs(),
'algorithm_design': self.source_algorithm_challenges(),
'logical_deduction': self.source_logic_puzzles(),
}
dataset = []
for source_name, examples in sources.items():
for example in examples:
if example['complexity'] == 'high':
dataset.append({
'question': example['prompt'],
'reasoning_chain': example['solution'],
'source': source_name,
'difficulty_score': example['difficulty'],
'reasoning_depth': example['step_count']
})
return dataset[:self.target_size]
def augment_with_synthetic_reasoning(self, examples):
"""Generate additional reasoning chains via CoT prompting"""
augmented = []
for example in examples:
# Rewrite reasoning with structured steps
structured_chain = self.reformat_chain_of_thought(
example['reasoning_chain']
)
augmented.append({
**example,
'structured_reasoning': structured_chain
})
return augmented
def reformat_chain_of_thought(self, reasoning_text):
"""Structure reasoning into explicit intermediate steps"""
steps = reasoning_text.split('.')
formatted = []
for i, step in enumerate(steps, 1):
formatted.append(f"Step {i}: {step.strip()}")
return " ".join(formatted)
def validate_reasoning_quality(self, example):
"""Check that reasoning is coherent and correct"""
# Verify each step logically follows from previous
# Check final answer matches stated solution
# Ensure no logical gaps
return True
# Build 283K cold-start examples
builder = ColdStartDatasetBuilder(num_samples=283000)
cold_start_data = builder.collect_complex_reasoning_examples()
cold_start_data = builder.augment_with_synthetic_reasoning(cold_start_data)
Step 2: Implement Prioritized Advantage Distillation
import torch
import numpy as np
class PrioritizedAdvantageDistillation:
def __init__(self, temperature=1.0):
self.temperature = temperature
def compute_advantages(self, rewards, baseline_values):
"""Calculate advantage for each trajectory"""
advantages = rewards - baseline_values
# Normalize for stability
mean_adv = np.mean(advantages)
std_adv = np.std(advantages)
normalized_advantages = (advantages - mean_adv) / (std_adv + 1e-8)
return normalized_advantages
def filter_zero_advantage_samples(self, trajectories, advantages,
threshold=0.01):
"""Remove non-informative samples where advantage ≈ 0"""
filtered = []
valid_indices = []
for idx, (traj, adv) in enumerate(zip(trajectories, advantages)):
if abs(adv) > threshold: # Keep only informative samples
filtered.append(traj)
valid_indices.append(idx)
print(f"Filtered {len(trajectories) - len(filtered)} zero-advantage samples")
print(f"Retained {len(filtered)}/{len(trajectories)} informative samples")
return filtered, valid_indices
def prioritized_resampling(self, advantages, trajectories,
temperature=1.0):
"""Resample trajectories weighted by advantage magnitude"""
# Convert advantages to probability distribution via softmax
# Using temperature to control concentration
exp_advantages = np.exp(advantages / temperature)
probabilities = exp_advantages / np.sum(exp_advantages)
# Resample with replacement using advantage weights
num_samples = len(trajectories)
resampled_indices = np.random.choice(
num_samples,
size=num_samples,
p=probabilities,
replace=True
)
resampled = [trajectories[i] for i in resampled_indices]
return resampled, resampled_indices
class GRPOWithPAD:
def __init__(self, model, pad_module):
self.model = model
self.pad = pad_module
def multimodal_grpo_step(self, batch_text, batch_multimodal,
reward_fn, optimizer):
"""GRPO training with prioritized advantage distillation"""
# Generate responses (both text and multimodal)
responses_text = self.model.generate(batch_text)
responses_mm = self.model.generate(batch_multimodal)
# Compute rewards
rewards_text = [reward_fn(r) for r in responses_text]
rewards_mm = [reward_fn(r) for r in responses_mm]
# Estimate baseline
baseline_text = self.estimate_baseline(batch_text)
baseline_mm = self.estimate_baseline(batch_multimodal)
# Compute advantages
advantages_text = self.pad.compute_advantages(
np.array(rewards_text),
baseline_text
)
advantages_mm = self.pad.compute_advantages(
np.array(rewards_mm),
baseline_mm
)
# Key innovation: Filter and resample with PAD
responses_filtered, _ = self.pad.filter_zero_advantage_samples(
responses_mm, advantages_mm
)
responses_resampled, _ = self.pad.prioritized_resampling(
advantages_mm[_], responses_mm, temperature=1.0
)
# Standard GRPO on resampled high-advantage data
policy_loss = self.compute_policy_loss(
responses_resampled,
advantages_mm
)
optimizer.zero_grad()
policy_loss.backward()
optimizer.step()
return policy_loss.item(), len(responses_filtered), len(responses_resampled)
Step 3: Implement Three-Stage Training Pipeline
class ThreeStageTrainingPipeline:
def __init__(self, model, cold_start_data):
self.model = model
self.cold_start_data = cold_start_data
self.pad = PrioritizedAdvantageDistillation()
def stage_1_text_coldstart(self, num_epochs=3):
"""
Train exclusively on text-only complex reasoning.
Duration: Initialize model reasoning capability.
"""
print("=== Stage 1: Text-Only Cold Start ===")
print(f"Dataset size: {len(self.cold_start_data)} examples")
print(f"Focus: Complex reasoning chains, logical coherence")
optimizer = self.get_optimizer(lr=1e-4)
for epoch in range(num_epochs):
total_loss = 0
for batch in self.create_batches(self.cold_start_data, bs=32):
# Standard supervised training on reasoning chains
loss = self.model.compute_language_loss(
batch['question'],
batch['structured_reasoning']
)
optimizer.zero_grad()
loss.backward()
optimizer.step()
total_loss += loss.item()
print(f"Epoch {epoch+1}: Loss = {total_loss:.4f}")
def stage_2_multimodal_rl(self, multimodal_data, num_iterations=5000):
"""
Apply GRPO with Prioritized Advantage Distillation.
Duration: Learn joint vision-language reasoning.
Key innovation: PAD prevents gradient stagnation.
"""
print("=== Stage 2: Multimodal RL with PAD ===")
print(f"Dataset size: {len(multimodal_data)} examples")
print(f"Method: GRPO + Prioritized Advantage Distillation")
optimizer = self.get_optimizer(lr=5e-5)
for iteration in range(num_iterations):
batch = self.sample_multimodal_batch(multimodal_data, bs=16)
loss, filtered_count, resampled_count = (
GRPOWithPAD(self.model, self.pad).multimodal_grpo_step(
batch['text'],
batch['images'],
reward_fn=self.compute_multimodal_reward,
optimizer=optimizer
)
)
if (iteration + 1) % 500 == 0:
print(f"Iter {iteration+1}: Loss={loss:.4f}, "
f"Filtered={filtered_count}, Resampled={resampled_count}")
def stage_3_text_refinement(self, text_data, num_epochs=2):
"""
Final text-only RL to polish linguistic quality.
Duration: Consolidate reasoning and coherence.
"""
print("=== Stage 3: Text-Only RL Refinement ===")
print(f"Dataset size: {len(text_data)} examples")
print(f"Focus: Linguistic fluency and reasoning clarity")
optimizer = self.get_optimizer(lr=1e-5)
for epoch in range(num_epochs):
for batch in self.create_batches(text_data, bs=32):
# RL with text-only rewards (fluency, coherence)
text_output = self.model.generate(batch['input'])
# Compute text quality rewards
fluency_reward = self.evaluate_fluency(text_output)
coherence_reward = self.evaluate_coherence(text_output)
combined_reward = 0.5 * fluency_reward + 0.5 * coherence_reward
# Policy gradient step
log_probs = self.model.log_probability(text_output)
loss = -log_probs * (combined_reward - 0.5)
optimizer.zero_grad()
loss.backward()
optimizer.step()
Practical Guidance
Text-First Initialization: Begin with 283K+ high-difficulty text-only examples before introducing visual data. This establishes reasoning foundations that transfer to multimodal tasks.
Detect Gradient Stagnation: Monitor loss plateaus during multimodal GRPO training. If loss stops decreasing despite correct implementation, PAD is likely needed.
PAD Implementation Details: Filter out advantages near zero (threshold ≈ 0.01), then resample using temperature-controlled softmax distribution over remaining samples. This concentrates training on informative trajectories.
Curriculum Progression: Don't skip stages or overlap them. The three-stage design ensures: (1) reasoning foundation, (2) multimodal alignment, (3) linguistic polish. Simultaneous mixed training fails empirically.
Evaluation Protocol: Test performance across text-only, vision-only, and vision-language tasks. Quality improvements should generalize across modalities, indicating robust reasoning rather than overfitting to training distribution.
Reference
- Paper: Advancing Multimodal Reasoning (2506.04207)
- Cold Start: 283K curated complex reasoning examples
- Method: GRPO + Prioritized Advantage Distillation
- Results: State-of-the-art for 3B/7B open-source models on multimodal reasoning