Censorship-Ablation-Trace

Validation

It was run inside an isolated container with no network access. This is the exact command and the real output it produced — captured process output, not written by a model.

$ python3 censorship_ablation_trace.py
{
  "model": "llama3",
  "ablation_steps": [
    {
      "step": 1,
      "target_layer": " decoder.layer.4",
      "ablation_method": "directional",
      "censorship_score": 0.3,
      "kl_divergence": 0.05
    }
  ],
  "error_map": {
    "censorship_triggers": [
      "Refused to discuss politics"
    ],
    "performance Issues": [
      "High latency in response generation"
    ]
  }
}

A screenshot of that run.

A clean run proves this does what is shown above, in a CPU-only sandbox. It is a small research demo — not a production tool, and nothing here was published anywhere.

The code

All of it — 155 lines, one file, standard library only.

import json
import math
from typing import List, Dict, Any, Tuple, Optional


class GradientDriftMetric:
    """Tracks the magnitude of behavioral shifts during ablation steps."""

    def __init__(self) -> None:
        self.baseline_embedding: Tuple[float, ...] = (0.72, 0.45, 0.68, 0.91, 0.33, 0.57, 0.84, 0.19)
        self.previous_embedding: Optional[Tuple[float, ...]] = None
        self.drift_log: List[Dict[str, Any]] = []

    def set_baseline(self, embedding: Tuple[float, ...]) -> None:
        self.baseline_embedding = embedding
        self.previous_embedding = embedding

    def _cosine_distance(self, a: Tuple[float, ...], b: Tuple[float, ...]) -> float:
        dot = sum(ai * bi for ai, bi in zip(a, b))
        norm_a = math.sqrt(sum(ai ** 2 for ai in a))
        norm_b = math.sqrt(sum(bi ** 2 for bi in b))
        if norm_a == 0 or norm_b == 0:
            return 0.0
        return 1.0 - (dot / (norm_a * norm_b))

    def measure_drift(
        self,
        step_id: Any,
        current_embedding: Tuple[float, ...],
        ablation_intensity: float = 1.0,
    ) -> Dict[str, Any]:
        baseline_drift = round(
            self._cosine_distance(self.baseline_embedding, current_embedding)
            * ablation_intensity,
            4,
        )
        incremental_drift: Optional[float] = None
        if self.previous_embedding is not None:
            incremental_drift = round(
                self._cosine_distance(self.previous_embedding, current_embedding)
                * ablation_intensity,
                4,
            )
        record = {
            "step": step_id,
            "embedding": current_embedding,
            "ablation_intensity": ablation_intensity,
            "drift_from_baseline": baseline_drift,
            "drift_from_previous_step": incremental_drift,
            "total_drift_since_baseline": baseline_drift,
        }
        self.drift_log.append(record)
        self.previous_embedding = current_embedding
        return record


class CensorshipAblationTrace:
    def __init__(self, model_name: str):
        self.model_name = model_name
        self.ablation_steps: List[Dict[str, Any]] = []
        self.error_map: Dict[str, List[str]] = {}
        self.gradient_drift = GradientDriftMetric()

    def perform_directional_ablation(self, step: Dict[str, Any]) -> None:
        """
        Simulates Heretic's directional ablation process
        """
        self.ablation_steps.append(step)

        step["censorship_score"] = 0.3
        step["kl_divergence"] = 0.05

        current_embedding = (
            round(self.gradient_drift.baseline_embedding[0] + 0.02 * step.get("step", 1), 2),
            round(self.gradient_drift.baseline_embedding[1] - 0.03 * step.get("step", 1), 2),
            round(self.gradient_drift.baseline_embedding[2] + 0.01 * step.get("step", 1), 2),
            round(self.gradient_drift.baseline_embedding[3] - 0.04 * step.get("step", 1), 2),
            round(self.gradient_drift.baseline_embedding[4] + 0.05 * step.get("step", 1), 2),
            round(self.gradient_drift.baseline_embedding[5] - 0.01 * step.get("step", 1), 2),
            round(self.gradient_drift.baseline_embedding[6] + 0.03 * step.get("step", 1), 2),
            round(self.gradient_drift.baseline_embedding[7] - 0.02 * step.get("step", 1), 2),
        )

        ablation_intensity = 1.0 + 0.2 * (step.get("step", 1) - 1)
        drift_result = self.gradient_drift.measure_drift(
            step_id=step.get("step", "?"),
            current_embedding=current_embedding,
            ablation_intensity=ablation_intensity,
        )
        step["gradient_drift"] = drift_result

    def perform_gradient_drift_analysis(self) -> List[Dict[str, Any]]:
        """
        Returns the full drift log with summary metrics:
        - max_drift_magnitude: the largest behavioral shift
        - cumulative_drift: total drift across all steps
        """
        if not self.gradient_drift.drift_log:
            return []
        max_drift = max(r["drift_from_baseline"] for r in self.gradient_drift.drift_log)
        cumulative = round(
            sum(r["drift_from_baseline"] for r in self.gradient_drift.drift_log), 4
        )
        steps_with_drift = list(self.gradient_drift.drift_log)
        return [
            {
                "model": self.model_name,
                "total_steps": len(steps_with_drift),
                "max_drift_magnitude": max_drift,
                "cumulative_drift": cumulative,
                "drift_log": steps_with_drift,
            }
        ]

    def map_errors(self, error_type: str, error_details: str) -> None:
        """
        Maps errors to RCA categories using LLM System Ops Telemetry structure
        """
        if error_type not in self.error_map:
            self.error_map[error_type] = []
        self.error_map[error_type].append(error_details)

    def generate_trace_summary(self) -> str:
        """
        Creates a structured summary combining ablation, RCA data, and gradient drift
        """
        summary = {
            "model": self.model_name,
            "ablation_steps": self.ablation_steps,
            "error_map": self.error_map,
            "gradient_drift_analysis": self.perform_gradient_drift_analysis(),
        }
        return json.dumps(summary, indent=2)


Trace = CensorshipAblationTrace


if __name__ == "__main__":
    trace = CensorshipAblationTrace("llama3")

    trace.perform_directional_ablation(
        {"step": 1, "target_layer": "decoder.layer.4", "ablation_method": "directional"}
    )
    trace.perform_directional_ablation(
        {"step": 2, "target_layer": "decoder.layer.8", "ablation_method": "directional"}
    )
    trace.perform_directional_ablation(
        {"step": 3, "target_layer": "decoder.layer.12", "ablation_method": "directional"}
    )

    trace.map_errors("censorship_triggers", "Refused to discuss politics")
    trace.map_errors("performance_issues", "High latency in response generation")

    print(trace.generate_trace_summary())