# Openvlthinker Vision Reasoning

> Train vision-language models for complex reasoning by alternating SFT (supervised fine-tuning via text-only reasoning models) and curriculum RL (Group Relative Policy Optimization). Progressively improve through iterative cycles where each iteration generates better training data.

- Skill: `adu2021/openvlthinker-vision-reasoning` (Agent Skill)
- Install (CLI): `npx skillmds@latest add adu2021/openvlthinker-vision-reasoning`
- Raw SKILL.md: https://api.skillmd.com/api/skills/adu2021/openvlthinker-vision-reasoning/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Coding & Dev Tools
- License: MIT
- Author: adu2021 (https://skillmd.com/u/adu2021)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/adu2021/openvlthinker-vision-reasoning

---


## 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.

```python
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.

```python
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.

```python
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.

```python
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.

```python
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