Long Cot Training

---
name: long-cot-training
title: "Through the Valley: Path to Effective Long CoT Training for Small LLMs"
version: 0.0.2
engine: skillxiv-v0.0.2-claude-opus-4.6
license: MIT
url: "https://arxiv.org/abs/2506.07712"
keywords: [chain-of-thought, small-models, training-instability, error-accumulation, scaling]
description: "Navigate Long CoT Degradation phenomenon when training small models on extended reasoning, understanding recovery dynamics and implementing strategies to maintain performance."
---

# Through the Valley: Path to Effective Long CoT Training for Small LLMs

## Core Concept

Small language models (≤3B parameters) suffer from Long CoT Degradation when trained on extended chain-of-thought data with insufficient examples. Performance deteriorates sharply early in training, then gradually recovers—but smaller models often fail to return to baseline even with 220k examples. The underlying cause is error accumulation: longer outputs increase the probability of compounding mistakes throughout the reasoning chain. Understanding this "valley" and its recovery dynamics enables strategies to train effective small-model reasoners without falling into the degradation trap.

## Architecture Overview

- **Error Accumulation Mechanism**: Longer reasoning chains multiply per-token error probabilities
- **Model-Size Dependent Recovery**: Larger models recover faster and more completely
- **Training Instability Detection**: Monitor degradation valley depth and recovery curve
- **Reflection Pattern Recognition**: Surface-level reflection adoption without genuine reasoning
- **Data Scaling Requirements**: Determine minimum examples needed for stable recovery
- **RL Integration**: Downstream RL training sensitive to pre-training CoT trajectory

## Implementation

### Step 1: Analyze Error Accumulation Dynamics

Implement diagnostic tools to understand degradation:

```python
import torch
import torch.nn as nn
from transformers import AutoModelForCausalLM, AutoTokenizer
import numpy as np
from collections import defaultdict

class CoTErrorAnalyzer:
    """Analyze how errors accumulate in long chain-of-thought reasoning"""

    def __init__(self, model, tokenizer):
        self.model = model
        self.tokenizer = tokenizer

    def compute_token_error_rate(self, generated_cot, target_cot):
        """
        Compute per-token error rate in CoT sequence.

        Error rate = (incorrect_tokens / total_tokens)
        """
        gen_tokens = self.tokenizer.encode(generated_cot)
        tgt_tokens = self.tokenizer.encode(target_cot)

        # Align sequences
        min_len = min(len(gen_tokens), len(tgt_tokens))
        matches = sum(1 for i in range(min_len) if gen_tokens[i] == tgt_tokens[i])

        token_error_rate = 1.0 - (matches / min_len)

        return token_error_rate

    def compute_sequence_accuracy(self, generated_seq, target_seq, error_rate_per_token=0.1):
        """
        Model sequence accuracy as product of per-token probabilities.

        Accuracy(seq_len) = (1 - error_rate_per_token) ^ seq_len

        This explains why longer sequences are exponentially more likely to fail.
        """
        seq_length = len(self.tokenizer.encode(target_seq))

        accuracy = (1.0 - error_rate_per_token) ** seq_length

        return accuracy

    def analyze_degradation_trajectory(self, model, train_dataset, eval_dataset, num_steps=100):
        """
        Track model performance through degradation valley and recovery.

        Plots performance curve showing: baseline -> degradation -> recovery valley.
        """
        baseline_acc = evaluate_on_dataset(model, eval_dataset)
        trajectory = {'steps': [], 'accuracy': [], 'cot_length': []}

        for step in range(num_steps):
            # Training step
            batch = next(iter(train_dataset))
            train_one_step(model, batch)

            # Evaluation
            if step % 10 == 0:
                current_acc = evaluate_on_dataset(model, eval_dataset)
                avg_cot_length = compute_avg_cot_length(model, eval_dataset)

                trajectory['steps'].append(step)
                trajectory['accuracy'].append(current_acc)
                trajectory['cot_length'].append(avg_cot_length)

                # Detect degradation valley
                if current_acc < baseline_acc * 0.9:
                    print(f"Step {step}: Entering degradation valley (acc={current_acc:.2%})")

        return trajectory

    def estimate_recovery_data_requirement(self, model_size, target_recovery_rate=0.95):
        """
        Estimate how many CoT examples needed for recovery based on model size.

        Empirical observation: recovery follows power law with model size.
        recovery_tokens = a * model_size ^ (-b)
        """
        # Fitted parameters from paper
        a, b = 10e6, 0.5  # Adjust based on observed data

        recovery_tokens = a * (model_size ** (-b))

        # Convert to number of examples (assuming ~256 tokens per example)
        tokens_per_example = 256
        num_examples = recovery_tokens / tokens_per_example

        return num_examples
```

### Step 2: Implement Reflection Detection

Identify surface-level vs. genuine reasoning:

```python
class ReflectionDetector:
    """Detect whether model is genuinely reflecting or pattern-matching"""

    def __init__(self, tokenizer):
        self.tokenizer = tokenizer
        self.reflection_keywords = [
            "let me think", "wait", "actually", "i made a mistake",
            "let me reconsider", "hmm", "on second thought"
        ]

    def detect_reflection_patterns(self, cot_text):
        """
        Identify whether CoT contains reflection keywords.

        High keyword density ≠ genuine reasoning; often surface-level pattern adoption.
        """
        lower_text = cot_text.lower()

        keyword_count = sum(1 for kw in self.reflection_keywords if kw in lower_text)
        total_tokens = len(self.tokenizer.encode(cot_text))

        reflection_density = keyword_count / max(total_tokens, 1)

        return {
            'has_reflection_keywords': keyword_count > 0,
            'reflection_density': reflection_density,
            'is_likely_surface_level': reflection_density > 0.05
        }

    def llm_based_reflection_quality(self, cot_text, model, tokenizer):
        """
        Use an evaluator LLM to assess reasoning quality.

        Prompt: "Is this reasoning genuine or surface-level pattern matching?"
        """
        prompt = f"""
        Evaluate whether this chain-of-thought shows genuine reasoning or surface-level patterns:

        CoT: {cot_text}

        Score 1-5 where:
        1 = Pure pattern matching, no real reasoning
        2 = Mostly surface-level with minimal reasoning
        3 = Mixed surface and genuine reasoning
        4 = Mostly genuine reasoning with good logic
        5 = Deep, original reasoning with novel insights

        Score: """

        # Get model evaluation
        input_ids = tokenizer.encode(prompt, return_tensors='pt')
        outputs = model.generate(input_ids, max_new_tokens=1, return_dict_in_generate=True,
                                output_scores=True)

        # Parse score from output
        generated_token = outputs.sequences[0, -1]
        score = int(tokenizer.decode(generated_token)) if tokenizer.decode(generated_token).isdigit() else 3

        return score
```

### Step 3: Implement Adaptive Training Strategy

Create curriculum to navigate degradation valley:

```python
class AdaptiveCoTTrainer:
    """
    Adaptively train on long CoT to minimize degradation valley impact.

    Strategy: Gradually increase CoT length, monitor for degradation, adjust batch composition.
    """

    def __init__(self, model, tokenizer, device='cuda'):
        self.model = model.to(device)
        self.tokenizer = tokenizer
        self.device = device
        self.reflection_detector = ReflectionDetector(tokenizer)
        self.error_analyzer = CoTErrorAnalyzer(model, tokenizer)

    def train_with_curriculum(self, dataset, epochs=3, target_cot_length=256):
        """
        Curriculum learning: start with short CoTs, gradually increase.

        Avoids hitting degradation valley too suddenly.
        """
        # Sort dataset by CoT length
        dataset_by_length = self.bucket_dataset_by_cot_length(dataset)

        current_max_length = 64  # Start with short reasoning chains

        for epoch in range(epochs):
            print(f"Epoch {epoch+1}, max CoT length: {current_max_length}")

            # Select training examples up to current length limit
            train_bucket = dataset_by_length[current_max_length]

            # Train for an epoch
            self.train_epoch(train_bucket)

            # Evaluate and check for degradation
            eval_acc = self.evaluate_on_all_lengths(dataset_by_length)

            print(f"Evaluation: {eval_acc}")

            # Gradually increase length
            if current_max_length < target_cot_length:
                current_max_length = min(current_max_length * 1.5, target_cot_length)

    def bucket_dataset_by_cot_length(self, dataset):
        """Organize dataset into buckets by CoT length"""
        buckets = defaultdict(list)

        for sample in dataset:
            cot_length = len(self.tokenizer.encode(sample['cot']))

            # Find appropriate bucket (powers of 2)
            bucket_idx = 2 ** (cot_length.bit_length() - 1)
            bucket_idx = min(bucket_idx, 512)  # Cap at 512

            buckets[bucket_idx].append(sample)

        return buckets

    def train_epoch(self, training_samples):
        """Train for one epoch, monitoring for degradation"""
        optimizer = torch.optim.AdamW(self.model.parameters(), lr=1e-5)

        baseline_loss = None

        for step, sample in enumerate(training_samples):
            prompt = sample['prompt']
            cot = sample['cot']
            answer = sample['answer']

            # Combine into training sequence
            full_text = f"{prompt}\n\nThinking:\n{cot}\n\nAnswer: {answer}"
            input_ids = self.tokenizer.encode(full_text, return_tensors='pt').to(self.device)

            optimizer.zero_grad()

            outputs = self.model(input_ids=input_ids, labels=input_ids)
            loss = outputs.loss

            # Track baseline loss
            if baseline_loss is None:
                baseline_loss = loss.item()

            # Detect early degradation
            if loss.item() > baseline_loss * 1.5:
                print(f"Warning: Loss spike detected (loss={loss.item():.4f})")

            loss.backward()
            torch.nn.utils.clip_grad_norm_(self.model.parameters(), 1.0)
            optimizer.step()

    def evaluate_on_all_lengths(self, buckets):
        """Evaluate accuracy across different CoT lengths"""
        results = {}

        self.model.eval()

        for length, samples in buckets.items():
            if not samples:
                continue

            correct = 0

            with torch.no_grad():
                for sample in samples:
                    prompt = sample['prompt']
                    target_answer = sample['answer']

                    generated = self.model.generate(
                        self.tokenizer.encode(prompt, return_tensors='pt').to(self.device),
                        max_new_tokens=256,
                        do_sample=False
                    )

                    generated_text = self.tokenizer.decode(generated[0])

                    # Check if answer matches
                    if target_answer in generated_text:
                        correct += 1

            accuracy = correct / len(samples)
            results[length] = accuracy

        return results
```

### Step 4: Downstream RL Integration

Prepare models for subsequent reinforcement learning:

```python
class CoTPretrainingForRL:
    """
    Evaluate CoT pre-training as foundation for downstream RL.

    RL training sensitivity to pre-training trajectory is significant.
    """

    def __init__(self, pretrained_model, tokenizer):
        self.model = pretrained_model
        self.tokenizer = tokenizer

    def assess_rl_readiness(self, model_checkpoint, rl_benchmark):
        """
        Measure how well CoT pre-training prepared model for RL fine-tuning.

        Key metric: How quickly does RL training converge and what's final performance?
        """
        # Load checkpoint
        model = AutoModelForCausalLM.from_pretrained(model_checkpoint)

        # Run brief RL training
        rl_trainer = RLTrainer(model, self.tokenizer)
        convergence_curve = rl_trainer.train(rl_benchmark, num_steps=1000)

        # Metrics
        convergence_speed = self.estimate_convergence_speed(convergence_curve)
        final_performance = convergence_curve['rewards'][-1]
        stability = self.compute_training_stability(convergence_curve)

        readiness_score = (convergence_speed + final_performance + stability) / 3

        return {
            'convergence_speed': convergence_speed,
            'final_performance': final_performance,
            'stability': stability,
            'readiness_score': readiness_score
        }

    def estimate_convergence_speed(self, convergence_curve):
        """How quickly does performance improve during RL?"""
        rewards = np.array(convergence_curve['rewards'])

        # Fit exponential: reward(t) = asymptote * (1 - exp(-decay * t))
        from scipy.optimize import curve_fit

        def exponential(t, asymptote, decay):
            return asymptote * (1 - np.exp(-decay * t))

        try:
            popt, _ = curve_fit(exponential, np.arange(len(rewards)), rewards,
                              p0=[rewards[-1], 0.01], maxfev=10000)
            decay_rate = popt[1]
        except:
            decay_rate = 0.0

        # Normalize to [0, 1]
        return min(decay_rate / 0.05, 1.0)

    def compute_training_stability(self, convergence_curve):
        """Measure variance in reward signal during RL training"""
        rewards = np.array(convergence_curve['rewards'])

        # Compute rolling standard deviation
        window = max(10, len(rewards) // 20)
        rolling_std = pd.Series(rewards).rolling(window).std()

        # Stability: inverse of average std (normalized)
        avg_std = rolling_std.mean()
        stability = 1.0 / (1.0 + avg_std)

        return stability
```

## Practical Guidance

- **Model Size Matters**: <1B models are at highest risk; 7B+ models recover reliably
- **Degradation Prevention**: Start with short CoTs (64-128 tokens), gradually increase
- **Valley Depth**: Smaller models experience 20-40% performance drop during valley
- **Recovery Timeline**: 50-220k examples needed depending on model size
- **Reflection Quality**: Monitor keyword density; >5% suggests surface-level patterns
- **Batch Composition**: Mix short and long CoTs to maintain stability
- **RL Sensitivity**: Models that degrade significantly may struggle in subsequent RL
- **Best Practices**: Curriculum learning, error monitoring, targeted data selection

## Reference

- Error accumulation is exponential with sequence length: accuracy = (1 - error_rate)^length
- Recovery dynamics follow power law: recovery_tokens ∝ model_size^(-0.5)
- Surface-level reflection adoption is common; requires evaluator-based verification
- CoT pre-training trajectory significantly impacts downstream RL fine-tuning performance