# Long Cot Training > Navigate Long CoT Degradation phenomenon when training small models on extended reasoning, understanding recovery dynamics and implementing strategies to maintain performance. - Skill: `adu2021/long-cot-training` (Agent Skill) - Install (CLI): `npx skillmds@latest add adu2021/long-cot-training` - Raw SKILL.md: https://api.skillmd.com/api/skills/adu2021/long-cot-training/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/long-cot-training --- # 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