Imported from ADu2021/skillXiv (skills/skillxiv-v0.0.2-claude-opus-4.6/kv-cache-steering-reasoning/SKILL.md). Install upstream with npx skills add ADu2021/skillXiv --skill kv-cache-steering-reasoning. Copyright stays with the author (MIT).

KV Cache Steering: Inference-Time Reasoning Induction Without Model Changes

Language models can reason through problems, but they don't always choose to—they often generate superficial answers. Traditional activation steering applies per-token interventions throughout generation, causing effects to compound and amplify. KV Cache Steering sidesteps this by modifying the key-value cache once at the prefilling stage, inserting a reasoning signal that propagates cleanly through subsequent generation without cascading amplification. This single intervention at the right architectural layer dramatically improves reasoning (5-15% accuracy gains on GSM8K) while adding negligible overhead.

The key insight is that the KV cache is the information bottleneck in transformers. By shifting cached representations toward a reasoning-aligned direction before decoding begins, you guide the entire generation trajectory toward step-by-step reasoning without per-token interference or model weight changes.

Core Concept

KV Cache Steering operates through four steps:

1. Steering Vector Extraction: Compute mean difference between KV cache from two prompts—one demonstrating desired reasoning, another without it
2. Cache Modification: Add the steering vector to cached K and V representations using scalar coefficients
3. Unmodified Generation: Proceed with standard autoregressive generation using the modified cache
4. Single Intervention Point: Unlike per-token steering, this modifies the cache once, allowing clean information flow

The steering vector captures the latent direction toward reasoning; applying it shifts the model's internal state toward step-by-step problem-solving before generation even begins.

Architecture Overview

• Contrastive Prompt Pair: Positive example (with explicit reasoning) and negative example (direct answer)
• Frozen Base Model: Standard LLM (no weights change)
• KV Cache Extractor: Captures key and value representations after prefilling
• Steering Vector Computation: Mean-of-differences across all layers and positions
• Cache Modifier: Linear addition of steering vectors to K and V with learned scalar coefficients (per-layer)
• Standard Decoder: Unchanged generation using modified cache
• Hyperparameter Interface: Scale coefficient α controlling steering strength

Implementation

The following demonstrates steering vector extraction and KV cache modification:

import torch
import torch.nn as nn
import torch.nn.functional as F
from typing import List, Tuple, Dict

class SteeringVectorExtractor(nn.Module):
    """Extract steering vectors from contrastive prompt pairs."""
    def __init__(self, model: nn.Module, num_layers: int = 32):
        super().__init__()
        self.model = model
        self.num_layers = num_layers

        # Hooks to capture KV cache at each layer
        self.cache_hooks = []
        self._register_hooks()

    def _register_hooks(self):
        """Register hooks to capture KV cache during forward pass."""
        self.positive_cache = {}
        self.negative_cache = {}

        def positive_hook(layer_idx):
            def hook(module, input, output):
                # Capture KV cache from this layer
                if hasattr(output, 'past_key_values'):
                    self.positive_cache[layer_idx] = output.past_key_values
            return hook

        def negative_hook(layer_idx):
            def hook(module, input, output):
                if hasattr(output, 'past_key_values'):
                    self.negative_cache[layer_idx] = output.past_key_values
            return hook

        # Register hooks on each transformer layer (simplified)
        for layer_idx in range(self.num_layers):
            layer = self.model.transformer.h[layer_idx]
            positive_handle = layer.register_forward_hook(positive_hook(layer_idx))
            negative_handle = layer.register_forward_hook(negative_hook(layer_idx))
            self.cache_hooks.append((positive_handle, negative_handle))

    def extract_steering_vectors(self, positive_prompt: str, negative_prompt: str,
                                 tokenizer, max_length: int = 100) -> Dict[int, Tuple[torch.Tensor, torch.Tensor]]:
        """
        Extract steering vectors from contrastive prompt pairs.

        Args:
            positive_prompt: Prompt with explicit reasoning steps ("Let me think step by step...")
            negative_prompt: Prompt with direct answer ("The answer is...")
            tokenizer: Tokenizer for the model
            max_length: Context length for encoding

        Returns:
            steering_vectors: Dict mapping layer_idx → (K_direction, V_direction)
        """
        # Tokenize prompts
        positive_ids = tokenizer.encode(positive_prompt, max_length=max_length, return_tensors='pt')
        negative_ids = tokenizer.encode(negative_prompt, max_length=max_length, return_tensors='pt')

        # Forward pass on positive prompt (captures reasoning)
        with torch.no_grad():
            _ = self.model(positive_ids)
            positive_cache = {k: v.clone() for k, v in self.positive_cache.items()}

        # Forward pass on negative prompt (no reasoning)
        with torch.no_grad():
            _ = self.model(negative_ids)
            negative_cache = {k: v.clone() for k, v in self.negative_cache.items()}

        # Compute steering vectors as mean difference
        steering_vectors = {}
        for layer_idx in range(self.num_layers):
            if layer_idx in positive_cache and layer_idx in negative_cache:
                pos_k, pos_v = positive_cache[layer_idx]
                neg_k, neg_v = negative_cache[layer_idx]

                # Direction toward reasoning
                k_direction = (pos_k - neg_k).mean(dim=(0, 1, 2))  # Average across batch, sequence, heads
                v_direction = (pos_v - neg_v).mean(dim=(0, 1, 2))

                steering_vectors[layer_idx] = (k_direction, v_direction)

        return steering_vectors

class KVCacheModifier(nn.Module):
    """Apply steering vectors to KV cache for inference-time reasoning control."""
    def __init__(self, model: nn.Module, num_layers: int = 32, hidden_dim: int = 768):
        super().__init__()
        self.model = model
        self.num_layers = num_layers
        self.hidden_dim = hidden_dim

        # Learnable steering coefficients per layer (initialize to small values)
        self.steering_coefficients = nn.Parameter(torch.ones(num_layers) * 0.1)

    def forward(self, input_ids: torch.Tensor, steering_vectors: Dict[int, Tuple[torch.Tensor, torch.Tensor]],
                alpha: float = 1.0) -> Tuple[torch.Tensor, Dict]:
        """
        Generate with steering applied to KV cache.

        Args:
            input_ids: (batch, seq_len) input tokens
            steering_vectors: Dict mapping layer_idx → (K_direction, V_direction)
            alpha: Scalar multiplier for steering strength (0=no steering, 1=full steering)

        Returns:
            output: Generated tokens
            info: Dict with steering application details
        """
        batch_size, seq_len = input_ids.shape

        # Prefill: encode input and get initial KV cache
        with torch.no_grad():
            outputs = self.model(input_ids, output_hidden_states=True, return_dict=True)
            initial_cache = outputs.past_key_values

        # Modify KV cache: apply steering vectors
        modified_cache = []
        for layer_idx in range(self.num_layers):
            if layer_idx in steering_vectors and initial_cache is not None:
                k, v = initial_cache[layer_idx]
                k_direction, v_direction = steering_vectors[layer_idx]

                # Apply steering with learned coefficient and global alpha
                steering_strength = self.steering_coefficients[layer_idx] * alpha
                k_modified = k + steering_strength * k_direction.to(k.device).unsqueeze(0).unsqueeze(0)
                v_modified = v + steering_strength * v_direction.to(v.device).unsqueeze(0).unsqueeze(0)

                modified_cache.append((k_modified, v_modified))
            else:
                modified_cache.append(initial_cache[layer_idx] if initial_cache is not None else None)

        # Generate with modified cache (clean propagation, no per-token interventions)
        generated_tokens = []
        current_cache = tuple(modified_cache)

        for step in range(256):  # Max generation length
            # Single-token forward pass with modified cache
            with torch.no_grad():
                outputs = self.model(
                    input_ids[:, -1:],  # Only last token (prefill already done)
                    past_key_values=current_cache,
                    return_dict=True
                )

            logits = outputs.logits[:, -1, :]
            next_token = logits.argmax(dim=-1, keepdim=True)

            generated_tokens.append(next_token)
            current_cache = outputs.past_key_values

            # Early stopping condition (EOS token)
            if next_token.item() == 2:  # Assuming EOS=2
                break

        # Stack generated tokens
        generated_sequence = torch.cat(generated_tokens, dim=1)

        info = {
            'steering_applied': True,
            'steering_strength': [self.steering_coefficients[i].item() for i in range(self.num_layers)],
            'generated_length': generated_sequence.shape[1]
        }

        return generated_sequence, info

class ContrastiveSteeringTrainer(nn.Module):
    """Train steering coefficients on downstream tasks (optional fine-tuning)."""
    def __init__(self, cache_modifier: KVCacheModifier):
        super().__init__()
        self.cache_modifier = cache_modifier
        self.criterion = nn.CrossEntropyLoss()

    def train_step(self, batch: Dict, steering_vectors: Dict,
                   optimizer: torch.optim.Optimizer) -> float:
        """
        Fine-tune steering coefficients on task-specific examples.

        Args:
            batch: Dict with 'input_ids', 'target_ids' (ground truth outputs)
            steering_vectors: Pre-computed directions for this task
            optimizer: Optimizer for steering coefficients
        """
        optimizer.zero_grad()

        input_ids = batch['input_ids']
        target_ids = batch['target_ids']

        # Generate with steering
        generated, info = self.cache_modifier(input_ids, steering_vectors, alpha=1.0)

        # Compute loss against target (e.g., accuracy on reasoning benchmarks)
        # Simplified: compare first token match
        loss = self.criterion(
            generated[:, 0, :].unsqueeze(1),
            target_ids[:, 0].unsqueeze(1)
        )

        loss.backward()
        torch.nn.utils.clip_grad_norm_(self.cache_modifier.steering_coefficients, 1.0)
        optimizer.step()

        return loss.item()

class AdaptiveSteeringSchedule:
    """Dynamically adjust steering strength based on task difficulty."""
    def __init__(self, initial_alpha: float = 1.0):
        self.alpha = initial_alpha

    def update(self, accuracy: float, threshold: float = 0.5):
        """Adjust alpha based on task performance."""
        if accuracy < threshold:
            # Difficulty detected: increase steering
            self.alpha = min(self.alpha * 1.2, 2.0)
        else:
            # Task easy: reduce steering
            self.alpha = max(self.alpha * 0.9, 0.1)

        return self.alpha

def apply_kv_cache_steering(model, prompt: str, reasoning_prompt: str,
                           tokenizer, alpha: float = 1.0,
                           max_gen_length: int = 256) -> str:
    """
    Simplified API for applying KV cache steering.

    Args:
        model: Frozen LLM
        prompt: Input prompt
        reasoning_prompt: Example of reasoning for steering vector extraction
        tokenizer: Model tokenizer
        alpha: Steering strength multiplier
        max_gen_length: Maximum tokens to generate

    Returns:
        generated_text: Model output with steering applied
    """
    # Extract steering vectors
    extractor = SteeringVectorExtractor(model)
    steering_vectors = extractor.extract_steering_vectors(reasoning_prompt, prompt, tokenizer)

    # Initialize modifier and apply steering
    modifier = KVCacheModifier(model)
    input_ids = tokenizer.encode(prompt, return_tensors='pt')

    generated, info = modifier(input_ids, steering_vectors, alpha=alpha)

    # Decode generated tokens
    generated_text = tokenizer.decode(generated[0])

    return generated_text, info

This implementation demonstrates single-point cache intervention with negligible overhead compared to per-token steering.

Practical Guidance

When to Use KV Cache Steering

• Frozen model deployment: Improve reasoning without fine-tuning weights or retraining
• Inference-time control: Adjust steering strength dynamically based on query complexity
• Resource-constrained environments: Negligible overhead (10ms/token); suitable for edge devices
• Reasoning benchmarks: GSM8K, ARC-Challenge, CommonsenseQA—all show consistent gains
• Multi-task adaptation: One model, different steering vectors for different tasks
• Latency-critical systems: Single cache modification vs. per-token interventions is vastly faster

When NOT to Use

• Creative or open-ended generation: Steering toward reasoning can suppress creative diversity
• Fine-grained control: If you need token-level control, per-token steering is necessary (at computational cost)
• Models with frozen cache: Some architectures don't expose KV cache; fall back to activation steering
• Knowledge-heavy tasks: Steering toward reasoning helps primarily with logical tasks; limited impact on factual recall
• Extremely short sequences (<20 tokens): Cache modification overhead outweighs benefits; direct prompting is simpler

Common Pitfalls

1. Contrastive Pair Mismatch: If positive and negative prompts are too similar, steering vectors are noisy. Ensure clear contrast (explicit reasoning vs. direct answer).
2. Alpha Too Large: Setting α > 2.0 causes output incoherence. Start at 1.0, tune down.
3. Wrong Cache Layer: Modifying layers before transformers don't capture reasoning signals. Confirm steering vectors extracted from middle/late layers (12-24 for 32L model).
4. Single Steering Vector: Using one contrastive pair overfits. Extract vectors from multiple examples, average them.
5. Ignoring Generalization: Steering vectors trained on GSM8K may not transfer to ARC. Re-extract for new domains.

Reference

Hong, Z., Zhang, L., et al. (2025). KV Cache Steering for Inducing Reasoning in Small Language Models. arXiv preprint arXiv:2507.08799.

Available at: https://arxiv.org/abs/2507.08799