# Temporal Diffusion Lm

> Leverages temporal dynamics in diffusion models by aggregating predictions across denoising steps for improved inference quality without retraining.

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

---


## Time Is a Feature: Temporal Dynamics in Diffusion Language Models

### Core Concept

Diffusion language models (dLLMs) exhibit temporal oscillation where correct answers often emerge during intermediate denoising steps but are overwritten in later iterations. Time Is a Feature exploits this temporal dimension by aggregating predictions across denoising steps rather than relying on final outputs, achieving substantial improvements through training-free temporal self-consistency voting and post-training temporal consistency reinforcement.

### Architecture Overview

- **Temporal Oscillation Detection**: Identify when correct outputs emerge during denoising
- **Self-Consistency Voting Across Steps**: Aggregate predictions from multiple denoising phases
- **Temporal Semantic Entropy**: Measure semantic stability across denoising iterations
- **Temporal Reinforcement Learning**: Train models to maintain consistency over time
- **Multi-Timestep Decoding**: Leverage intermediate predictions as features

### Implementation Steps

**Step 1: Analyze Temporal Dynamics**

Understand when correct predictions emerge:

```python
# Pseudocode for temporal analysis
class TemporalDynamicsAnalyzer:
    def __init__(self, diffusion_model):
        super().__init__()
        self.diffusion_model = diffusion_model

    def extract_timestep_predictions(self, prompt, num_steps=50):
        """
        Extract model predictions at each denoising step.

        Args:
            prompt: Input prompt
            num_steps: Number of denoising steps

        Returns:
            predictions_by_step: Dict mapping timestep to predictions
        """
        predictions_by_step = {}

        # Initialize noise
        x_t = torch.randn(1, 768)  # Latent dimension

        # Reverse diffusion with checkpointing
        for t in range(num_steps - 1, -1, -1):
            # Denoise step
            with torch.no_grad():
                x_t = self.diffusion_model.denoise_step(x_t, t, prompt)

            # Extract predictions at this step
            logits = self.diffusion_model.decode_to_logits(x_t)
            predictions = F.softmax(logits, dim=-1)

            predictions_by_step[t] = {
                'logits': logits,
                'probs': predictions,
                'top_token': torch.argmax(predictions, dim=-1)
            }

        return predictions_by_step

    def analyze_correctness_trajectory(self, predictions_by_step, ground_truth_token):
        """
        Track how correctness changes across timesteps.

        Args:
            predictions_by_step: Predictions at each step
            ground_truth_token: Correct token ID

        Returns:
            correctness_trajectory: When true answer emerges and fades
        """
        trajectory = []

        for step in sorted(predictions_by_step.keys()):
            pred = predictions_by_step[step]
            top_token = pred['top_token'].item()

            is_correct = (top_token == ground_truth_token)
            confidence = pred['probs'][0, ground_truth_token].item()

            trajectory.append({
                'step': step,
                'is_correct': is_correct,
                'confidence': confidence,
                'top_token': top_token
            })

        # Find emergence point
        correct_steps = [t for t in trajectory if t['is_correct']]
        if correct_steps:
            emergence_step = min(t['step'] for t in correct_steps)
        else:
            emergence_step = None

        return {
            'trajectory': trajectory,
            'emergence_step': emergence_step,
            'peak_confidence': max(t['confidence'] for t in trajectory)
        }

    def compute_temporal_statistics(self, prompts, ground_truths):
        """
        Analyze temporal dynamics across multiple examples.
        """
        all_emergences = []
        all_confidences = []

        for prompt, truth in zip(prompts, ground_truths):
            preds = self.extract_timestep_predictions(prompt)
            analysis = self.analyze_correctness_trajectory(preds, truth)

            if analysis['emergence_step'] is not None:
                all_emergences.append(analysis['emergence_step'])
                all_confidences.append(analysis['peak_confidence'])

        return {
            'avg_emergence_step': np.mean(all_emergences) if all_emergences else None,
            'median_emergence_step': np.median(all_emergences) if all_emergences else None,
            'avg_peak_confidence': np.mean(all_confidences),
            'emergence_rate': len(all_emergences) / len(prompts)
        }
```

**Step 2: Implement Temporal Self-Consistency Voting**

Aggregate predictions across timesteps:

```python
# Pseudocode for temporal voting
class TemporalSelfConsistencyVoting:
    def __init__(self, diffusion_model, num_samples=3):
        super().__init__()
        self.diffusion_model = diffusion_model
        self.num_samples = num_samples

    def temporal_voting_decode(self, prompt, max_length=100, num_timesteps=50):
        """
        Generate text using temporal voting across denoising steps.

        Args:
            prompt: Input prompt
            max_length: Maximum sequence length
            num_timesteps: Number of denoising timesteps to consider

        Returns:
            best_sequence: Highest-voted sequence
            confidence_scores: Confidence by position
        """
        generated_tokens = []
        confidence_scores = []

        for position in range(max_length):
            # Collect predictions from multiple denoising runs
            timestep_votes = []

            for sample_idx in range(self.num_samples):
                # Run diffusion from scratch (or from checkpoint)
                predictions_by_step = self._diffuse_and_predict(
                    prompt + ''.join(generated_tokens),
                    num_timesteps
                )

                # Vote on token at this position
                votes = self._extract_position_votes(predictions_by_step, position)
                timestep_votes.append(votes)

            # Aggregate votes across timesteps and samples
            best_token, confidence = self._aggregate_votes(timestep_votes)

            if best_token is None or confidence < 0.3:
                break

            generated_tokens.append(best_token)
            confidence_scores.append(confidence)

        return ''.join(generated_tokens), confidence_scores

    def _diffuse_and_predict(self, prefix, num_timesteps):
        """
        Run diffusion and collect predictions at each step.
        """
        x_t = torch.randn(1, 768)
        predictions_by_step = {}

        for t in range(num_timesteps - 1, -1, -1):
            with torch.no_grad():
                x_t = self.diffusion_model.denoise_step(x_t, t, prefix)

            logits = self.diffusion_model.decode_to_logits(x_t)
            predictions_by_step[t] = F.softmax(logits, dim=-1)

        return predictions_by_step

    def _extract_position_votes(self, predictions_by_step, position):
        """
        Extract votes for a specific position from all timesteps.
        """
        votes = {}

        for step, probs in predictions_by_step.items():
            # Top-k tokens at this step
            top_k = torch.topk(probs[0], k=5)

            for token_id, prob in zip(top_k.indices.tolist(), top_k.values.tolist()):
                if token_id not in votes:
                    votes[token_id] = []
                votes[token_id].append((step, prob))

        return votes

    def _aggregate_votes(self, timestep_votes):
        """
        Aggregate votes across samples and timesteps.
        """
        token_scores = {}

        for sample_votes in timestep_votes:
            for token_id, vote_list in sample_votes.items():
                if token_id not in token_scores:
                    token_scores[token_id] = 0

                # Average probability across steps for this sample
                avg_prob = np.mean([v[1] for v in vote_list])
                token_scores[token_id] += avg_prob

        if not token_scores:
            return None, 0.0

        # Select token with highest aggregate score
        best_token = max(token_scores.keys(), key=lambda x: token_scores[x])
        confidence = token_scores[best_token] / len(timestep_votes)

        return best_token, confidence
```

**Step 3: Compute Temporal Semantic Entropy**

Measure semantic stability across denoising:

```python
# Pseudocode for temporal semantic entropy
class TemporalSemanticEntropy:
    def __init__(self, semantic_model):
        super().__init__()
        self.semantic_model = semantic_model

    def compute_semantic_entropy(self, predictions_by_step, decode_fn):
        """
        Measure semantic stability across denoising steps.

        Args:
            predictions_by_step: Predictions at each denoising step
            decode_fn: Function to decode predictions to text

        Returns:
            semantic_entropy: Measure of semantic variability (lower = more stable)
        """
        texts_by_step = []

        for step in sorted(predictions_by_step.keys()):
            probs = predictions_by_step[step]

            # Generate text from this timestep
            text = decode_fn(probs)
            texts_by_step.append(text)

        # Compute semantic embeddings
        embeddings = []
        for text in texts_by_step:
            embedding = self.semantic_model.encode(text)
            embeddings.append(embedding)

        embeddings = torch.stack(embeddings)

        # Compute pairwise similarities
        similarities = torch.mm(
            F.normalize(embeddings, dim=-1),
            F.normalize(embeddings, dim=-1).T
        )

        # Entropy: how variable are similarities?
        # Low variability = high consistency = lower entropy
        consistency = similarities.mean().item()
        entropy = 1.0 - consistency  # Higher consistency = lower entropy

        return entropy

    def compute_temporal_consistency_score(self, predictions_by_step, decode_fn):
        """
        Compute how consistent predictions remain through denoising.

        Returns:
            tse_score: Temporal Semantic Entropy score (higher = more inconsistent)
        """
        entropy = self.compute_semantic_entropy(predictions_by_step, decode_fn)

        return entropy
```

**Step 4: Implement Temporal Consistency Reinforcement Learning**

Train models to maintain semantic consistency:

```python
# Pseudocode for temporal RL training
class TemporalConsistencyRL:
    def __init__(self, diffusion_model):
        super().__init__()
        self.diffusion_model = diffusion_model
        self.semantic_model = SentenceTransformer('all-MiniLM-L6-v2')
        self.tse_computer = TemporalSemanticEntropy(self.semantic_model)

    def compute_temporal_reward(self, predictions_by_step, target_text, decode_fn):
        """
        Compute reward based on temporal consistency.

        Args:
            predictions_by_step: Predictions at each timestep
            target_text: Ground truth target
            decode_fn: Function to decode predictions

        Returns:
            reward: Combined reward signal
        """
        # Component 1: Temporal consistency (stability across timesteps)
        tse_score = self.tse_computer.compute_temporal_consistency_score(
            predictions_by_step,
            decode_fn
        )
        consistency_reward = 1.0 / (1.0 + tse_score)  # Higher consistency = higher reward

        # Component 2: Final accuracy
        final_text = decode_fn(predictions_by_step[0])  # Final step
        accuracy = self._compute_similarity(final_text, target_text)

        # Component 3: Early emergence bonus
        early_correct = 0
        for step in sorted(predictions_by_step.keys(), reverse=True):
            step_text = decode_fn(predictions_by_step[step])
            if self._compute_similarity(step_text, target_text) > 0.8:
                early_correct += 1

        emergence_bonus = min(early_correct / 10.0, 1.0)

        # Combined reward
        total_reward = (
            0.5 * consistency_reward +
            0.3 * accuracy +
            0.2 * emergence_bonus
        )

        return total_reward

    def _compute_similarity(self, text1, text2):
        """
        Compute semantic similarity between texts.
        """
        emb1 = self.semantic_model.encode(text1)
        emb2 = self.semantic_model.encode(text2)

        return torch.cosine_similarity(
            torch.tensor(emb1).unsqueeze(0),
            torch.tensor(emb2).unsqueeze(0)
        ).item()

    def train_temporal_consistency(self, training_data, num_epochs=3):
        """
        Train model to maintain temporal consistency.

        Args:
            training_data: Pairs of (prompts, targets)
            num_epochs: Training epochs

        Returns:
            trained_model: Model with improved temporal consistency
        """
        optimizer = AdamW(self.diffusion_model.parameters(), lr=1e-5)

        for epoch in range(num_epochs):
            total_loss = 0

            for prompt, target in training_data:
                # Get predictions across timesteps
                predictions_by_step = self._get_timestep_predictions(prompt)

                # Compute reward
                reward = self.compute_temporal_reward(
                    predictions_by_step,
                    target,
                    self._simple_decode
                )

                # Policy gradient loss
                log_probs = self._compute_log_probs(predictions_by_step)
                loss = -log_probs * reward

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

                total_loss += loss.item()

            print(f"Epoch {epoch+1}: Loss = {total_loss / len(training_data):.4f}")

        return self.diffusion_model

    def _get_timestep_predictions(self, prompt):
        """
        Extract predictions at multiple timesteps.
        """
        # Simplified version of earlier analysis
        return {}

    def _simple_decode(self, probs):
        """
        Simple decoding from probability distribution.
        """
        token_id = torch.argmax(probs).item()
        return f"token_{token_id}"

    def _compute_log_probs(self, predictions_by_step):
        """
        Compute log probabilities for taken actions.
        """
        return torch.tensor(0.1)
```

### Practical Guidance

**Hyperparameters and Configuration**:
- Number of denoising timesteps: 50-100
- Voting samples for self-consistency: 3-5
- Temporal consistency threshold: 0.3-0.5
- RL training learning rate: 1e-5 to 5e-5
- Consistency/accuracy/emergence weights: 0.5/0.3/0.2

**When to Use Time Is a Feature**:
- Diffusion language models with temporal oscillation
- Scenarios where intermediate predictions are meaningful
- Tasks where inference quality matters more than speed
- Models exhibiting correctness emergence at intermediate steps

**When NOT to Use**:
- Autoregressive language models (no temporal dynamics)
- Real-time inference with strict latency constraints
- Systems where final-step predictions are inherently stable
- Memory-constrained environments (requires storing multiple predictions)

**Implementation Notes**:
- Temporal voting is training-free and can be applied immediately
- Semantic entropy requires representation model (BERT-like)
- Monitor emergence step distribution to understand model dynamics
- Consider adaptive timestep selection (skip stable steps)
- Combine temporal voting with other decoding strategies (temperature, top-k)

### Reference

Paper: Time Is a Feature: Temporal Dynamics in Diffusion Language Models
ArXiv: 2508.09138
Performance: 24.7% average gain on Countdown dataset, up to 6.6% improvement on mathematics benchmarks when combined with accuracy rewards