adu2021 Core Concept
Vision-language models struggle with complex reasoning tasks because distilling reasoning from text-only models like DeepSeek-R1 often degrades visual understanding. OpenVLThinker solves this through iterative SFT-RL cycles: alternate between lightweight SFT (using text models to generate reasoning demonstrations) and curriculum RL (applying Group Relative Policy Optimization with difficulty-based scheduling). Each cycle improves training data quality, enabling self-improvement with minimal total examples.
Architecture Overview
- SFT Phase: Text-only reasoning models (QwQ-32B) generate chain-of-thought demos from image captions
- RL Phase: Group Relative Policy Optimization (GRPO) with difficulty-based curriculum learning
- Curriculum Scheduler: Progressively increase difficulty: medium → hard → hardest data
- Iterative Refinement: Each cycle's improved model generates better training data for next cycle
- Minimal Data Requirement: Only 12K total examples across three cycles
Implementation Steps
Step 1: Prepare Training Data with Text-Only Reasoning Models
Use a capable text-only model to generate chain-of-thought demonstrations from image captions and questions.
import torch
from typing import List, Dict
class ReasoningDataGeneration:
"""
Generate chain-of-thought reasoning using a text-only model
(e.g., QwQ-32B) from image captions and task questions.
"""
def __init__(self, text_model, device='cuda'):
self.text_model = text_model
self.device = device
def generate_reasoning(
self,
image_caption: str,
question: str,
max_length: int = 512
) -> str:
"""
Generate step-by-step reasoning for a VQA task.
image_caption: Textual description of image
question: Question about the image
Returns: Chain-of-thought reasoning text
"""
prompt = f"""Given the following image description and question,
provide step-by-step reasoning to answer the question.
Image: {image_caption}
Question: {question}
Let me think through this step by step:"""
# Generate with text model
inputs = self.text_model.tokenize(prompt)
with torch.no_grad():
outputs = self.text_model.generate(
inputs,
max_length=max_length,
temperature=0.7,
top_p=0.9
)
reasoning = self.text_model.decode(outputs[0])
return reasoning
def create_sft_examples(
self,
captions: List[str],
questions: List[str],
answers: List[str]
) -> List[Dict]:
"""
Create SFT training examples with reasoning.
Returns list of (instruction, reasoning+answer) pairs.
"""
sft_examples = []
for caption, question, answer in zip(captions, questions, answers):
reasoning = self.generate_reasoning(caption, question)
example = {
'image_caption': caption,
'instruction': question,
'reasoning': reasoning,
'answer': answer,
'full_response': f"{reasoning}\n\nAnswer: {answer}"
}
sft_examples.append(example)
return sft_examples
Step 2: Identify Reasoning Keywords and SFT Efficacy
Analyze whether SFT successfully induces reasoning behaviors by examining keyword frequencies in model outputs.
import re
from collections import Counter
def analyze_reasoning_keywords(model_outputs: List[str]) -> Dict[str, float]:
"""
Count reasoning-related keywords to verify SFT effectiveness.
Keywords indicate reflection, step-by-step thinking, and uncertainty.
"""
reasoning_keywords = [
'let me', 'think', 'step', 'first', 'next', 'then',
'because', 'thus', 'therefore', 'however', 'although',
'observe', 'notice', 'identify', 'reason', 'conclusion'
]
keyword_counts = Counter()
for output in model_outputs:
output_lower = output.lower()
for keyword in reasoning_keywords:
count = len(re.findall(r'\b' + keyword + r'\b', output_lower))
keyword_counts[keyword] += count
total_keywords = sum(keyword_counts.values())
keyword_frequencies = {
k: v / max(total_keywords, 1) for k, v in keyword_counts.items()
}
return keyword_frequencies
def sft_efficacy_score(
base_model_outputs: List[str],
sft_model_outputs: List[str]
) -> float:
"""
Measure SFT effectiveness by comparing keyword distribution.
Higher score indicates SFT successfully induced reasoning.
"""
base_keywords = analyze_reasoning_keywords(base_model_outputs)
sft_keywords = analyze_reasoning_keywords(sft_model_outputs)
# Compute keyword improvement ratio
improvements = [
sft_keywords.get(k, 0) / max(base_keywords.get(k, 0), 1e-6)
for k in base_keywords.keys()
]
avg_improvement = sum(improvements) / len(improvements) if improvements else 1.0
return min(avg_improvement, 2.0) # Cap at 2x improvement
Step 3: Implement Curriculum RL with GRPO
Apply Group Relative Policy Optimization with progressive difficulty scheduling.
import torch.nn.functional as F
class CurriculumGRPO:
"""
Group Relative Policy Optimization with curriculum learning.
Progressively train on harder examples: medium → hard → hardest.
"""
def __init__(self, model, reference_model, device='cuda'):
self.model = model
self.reference_model = reference_model
self.device = device
def compute_difficulty_score(
self,
base_model_accuracy: float,
gpt4_rating: float
) -> float:
"""
Composite difficulty score combining two signals:
- Base model error rate (1 - accuracy)
- GPT-4 quality rating (normalized to [0,1])
"""
error_rate = 1.0 - base_model_accuracy
difficulty = (error_rate + (1 - gpt4_rating)) / 2
return difficulty
def group_by_difficulty(
self,
examples: List[Dict],
base_model_accuracy: List[float],
gpt4_ratings: List[float]
) -> Dict[str, List[Dict]]:
"""
Partition examples into difficulty tiers.
Returns: {'easy': [...], 'medium': [...], 'hard': [...]}
"""
difficulties = [
self.compute_difficulty_score(acc, rating)
for acc, rating in zip(base_model_accuracy, gpt4_ratings)
]
# Sort by difficulty and partition into thirds
sorted_indices = sorted(range(len(difficulties)), key=lambda i: difficulties[i])
n = len(sorted_indices)
easy_idx = sorted_indices[:n // 3]
medium_idx = sorted_indices[n // 3:2 * n // 3]
hard_idx = sorted_indices[2 * n // 3:]
return {
'easy': [examples[i] for i in easy_idx],
'medium': [examples[i] for i in medium_idx],
'hard': [examples[i] for i in hard_idx]
}
def grpo_loss(
self,
log_probs: torch.Tensor,
ref_log_probs: torch.Tensor,
rewards: torch.Tensor,
beta: float = 0.5
) -> torch.Tensor:
"""
GRPO loss: scales log probability ratios by reward advantage.
log_probs: current model log probabilities (batch,)
ref_log_probs: reference model log probabilities (batch,)
rewards: reward signal (batch,) - higher is better
"""
log_ratio = log_probs - ref_log_probs
# Normalize rewards for stability
rewards = (rewards - rewards.mean()) / (rewards.std() + 1e-8)
# GRPO: maximize log(pi/pi_ref) * advantage
advantage = rewards
loss = -torch.mean(log_ratio * advantage)
# KL regularization to prevent divergence from reference
kl_div = (log_probs - ref_log_probs).mean()
total_loss = loss + beta * kl_div
return total_loss
def train_curriculum_rl(
self,
training_examples: List[Dict],
base_accuracies: List[float],
gpt4_ratings: List[float],
num_epochs: int = 3,
batch_size: int = 8
):
"""
Train with curriculum: easy → medium → hard.
Each stage trains for one epoch on increasingly difficult examples.
"""
grouped = self.group_by_difficulty(
training_examples, base_accuracies, gpt4_ratings
)
curriculum_order = ['medium', 'hard'] # Start with medium, skip easy
for stage_idx, difficulty in enumerate(curriculum_order):
print(f"Training on {difficulty} examples (stage {stage_idx + 1})")
examples = grouped[difficulty]
for epoch in range(num_epochs):
total_loss = 0
num_batches = 0
for i in range(0, len(examples), batch_size):
batch = examples[i:i + batch_size]
# Prepare batch
prompts = [ex['instruction'] for ex in batch]
responses = [ex['full_response'] for ex in batch]
rewards = torch.tensor(
[ex.get('reward', 1.0) for ex in batch],
dtype=torch.float32,
device=self.device
)
# Forward pass
log_probs = self._compute_log_probs(prompts, responses)
with torch.no_grad():
ref_log_probs = self._compute_log_probs_ref(prompts, responses)
# Compute loss
loss = self.grpo_loss(log_probs, ref_log_probs, rewards)
# Backward pass
self.model.zero_grad()
loss.backward()
torch.nn.utils.clip_grad_norm_(self.model.parameters(), 1.0)
self.model.optimizer.step()
total_loss += loss.item()
num_batches += 1
avg_loss = total_loss / max(num_batches, 1)
print(f" Epoch {epoch + 1} - Loss: {avg_loss:.4f}")
def _compute_log_probs(self, prompts: List[str], responses: List[str]) -> torch.Tensor:
"""Compute log probabilities from current model."""
# Placeholder: actual implementation tokenizes and computes log probs
return torch.randn(len(prompts), device=self.device)
def _compute_log_probs_ref(self, prompts: List[str], responses: List[str]) -> torch.Tensor:
"""Compute log probabilities from reference model."""
with torch.no_grad():
return torch.randn(len(prompts), device=self.device)
Step 4: Iterative Cycle Management
Orchestrate multiple SFT-RL cycles where each iteration improves training data and model quality.
class IterativeSelfImprovement:
"""
Manage iterative SFT-RL cycles: cycle 1, 2, 3.
Each cycle generates better training data from improved model.
"""
def __init__(self, vlm_model, text_model, num_cycles: int = 3):
self.vlm_model = vlm_model
self.text_model = text_model
self.num_cycles = num_cycles
self.cycle_metrics = []
def run_full_pipeline(
self,
initial_examples: List[Dict],
validation_set: List[Dict]
):
"""
Execute full iterative pipeline: SFT → RL → evaluate → repeat.
initial_examples: Starting training data (images + captions + questions)
validation_set: Hold-out validation benchmarks
"""
training_data = initial_examples
for cycle in range(self.num_cycles):
print(f"\n===== Cycle {cycle + 1} / {self.num_cycles} =====")
# Phase 1: SFT
print(f"Phase 1: Supervised Fine-Tuning on {len(training_data)} examples")
self.run_sft(training_data)
# Evaluate after SFT
sft_metrics = self.evaluate(validation_set)
print(f" SFT Metrics: {sft_metrics}")
# Phase 2: RL
print(f"Phase 2: Curriculum RL")
self.run_curriculum_rl(training_data)
# Evaluate after RL
rl_metrics = self.evaluate(validation_set)
print(f" RL Metrics: {rl_metrics}")
# Record cycle metrics
self.cycle_metrics.append({
'cycle': cycle + 1,
'sft_metrics': sft_metrics,
'rl_metrics': rl_metrics
})
# Generate new training data for next cycle (if not last)
if cycle < self.num_cycles - 1:
print(f"Phase 3: Generating training data for cycle {cycle + 2}")
training_data = self.generate_next_cycle_data(
training_data, rl_metrics
)
def run_sft(self, training_data: List[Dict]):
"""Supervised fine-tuning phase."""
# Tokenize and train standard language model loss
pass
def run_curriculum_rl(self, training_data: List[Dict]):
"""Curriculum RL phase."""
grpo = CurriculumGRPO(self.vlm_model, self.vlm_model)
# Implement curriculum training
pass
def evaluate(self, validation_set: List[Dict]) -> Dict:
"""Evaluate on validation benchmarks."""
accuracies = []
for example in validation_set:
# Generate answer
output = self.vlm_model.generate(
image=example['image'],
prompt=example['question']
)
# Compare with ground truth
correct = self._match_answer(output, example['answer'])
accuracies.append(float(correct))
return {
'accuracy': sum(accuracies) / len(accuracies) if accuracies else 0.0,
'count': len(validation_set)
}
def generate_next_cycle_data(
self,
current_data: List[Dict],
metrics: Dict
) -> List[Dict]:
"""
Generate training data for next cycle using improved model.
Reuse successful examples and add new harder examples.
"""
# Keep ~70% of current data, generate ~30% new data
num_new = len(current_data) // 3
new_data = current_data[:2 * len(current_data) // 3]
# Generate new harder examples
data_gen = ReasoningDataGeneration(self.text_model)
new_examples = data_gen.create_sft_examples(
captions=[ex['caption'] for ex in current_data[-num_new:]],
questions=[ex['question'] for ex in current_data[-num_new:]],
answers=[ex['answer'] for ex in current_data[-num_new:]]
)
new_data.extend(new_examples)
return new_data
def _match_answer(self, output: str, ground_truth: str) -> bool:
"""Simple string matching for answer evaluation."""
return output.strip().lower() == ground_truth.strip().lower()
Step 5: Inference Integration
Combine trained components for efficient inference on reasoning tasks.
class OpenVLThinkerInference:
"""
Production inference pipeline combining trained VLM with optional
reasoning generation for transparent answer production.
"""
def __init__(self, vlm_model, generate_reasoning: bool = True):
self.vlm_model = vlm_model
self.generate_reasoning = generate_reasoning
def answer_question(
self,
image: torch.Tensor,
question: str,
return_reasoning: bool = False
) -> Dict:
"""
Answer visual question with optional reasoning.
image: PIL Image or tensor (batch, 3, H, W)
question: Text question about image
"""
if self.generate_reasoning:
# Generate reasoning-based answer
prompt = f"Question: {question}\n\nLet me think through this step by step:"
else:
prompt = f"Question: {question}\nAnswer:"
# Prepare input
inputs = self.vlm_model.process(image, prompt)
# Generate response
with torch.no_grad():
outputs = self.vlm_model.generate(
**inputs,
max_length=256,
temperature=0.7,
top_p=0.9
)
response = self.vlm_model.decode(outputs[0])
# Parse answer from response
if self.generate_reasoning:
parts = response.split('Answer:')
reasoning = parts[0] if len(parts) > 0 else ""
answer = parts[1].strip() if len(parts) > 1 else response.strip()
else:
reasoning = ""
answer = response.strip()
result = {
'answer': answer,
'full_response': response
}
if return_reasoning:
result['reasoning'] = reasoning
return result
Practical Guidance
When to Use:
- Training vision-language models for complex reasoning tasks (MathVista, visual reasoning)
- Scenarios with limited labeled reasoning data (few thousand examples)
- Cases where reasoning transparency is important (education, scientific analysis)
- Projects where iterative self-improvement through synthetic data generation is viable
When NOT to Use:
- Simple visual classification or object detection (overkill)
- Scenarios where text-only reasoning models aren't available or high quality
- Real-time inference systems where chain-of-thought verbosity is unacceptable
- Domain-specific reasoning where text models haven't seen your specific domain
Hyperparameter Tuning:
- num_cycles: 3 standard; 2 for fast iteration, 4-5 for maximum quality if data is abundant
- difficulty_ratio: 70% easy / medium, 30% hard; adjust based on validation plateau
- SFT_epochs: 1-2; use 1 for computational efficiency, 2 for better convergence
- RL_beta (KL weight): 0.5 standard; increase to 1.0 if model diverges from reference
- curriculum_order: ['medium', 'hard']; skip 'easy' unless validation accuracy is very low
Common Pitfalls:
- Weak text-only reasoning model produces garbage reasoning demonstrations; use QwQ-32B or DeepSeek-R1
- Insufficient validation data; GPT-4 ratings required for accurate difficulty scoring
- Over-training on narrow difficulty tier; rotate through curriculum stages gradually
- RL divergence: monitor KL divergence between model and reference; increase beta if KL > 0.2
References
- arXiv:2503.17352 - OpenVLThinker paper
- https://arxiv.org/abs/2412.15729 - Group Relative Policy Optimization (GRPO)
- https://arxiv.org/abs/2103.01734 - Chain-of-Thought Prompting
- QwQ-32B model: https://huggingface.co/Qwen/QwQ-32B