TransformerLens pe GPT-2: activation patching pentru head-urile care schimbă predicția unui token

Două propoziții aproape identice ajung la GPT-2, dar numele care ar trebui continuat este diferit. În loc să inspectăm atenția și să presupunem că un head „pare important”, vom muta activarea lui din rularea corectă în rularea coruptă și vom măsura dacă preferința modelului pentru următorul token se schimbă.

Rezultatul este un laborator local cu TransformerLens 4.0.0 și GPT-2: generează 12 perechi controlate, separă opt exemple pentru descoperirea head-urilor și patru pentru test, evaluează toate cele 144 de head-uri și salvează clasamentul în CSV. Head-ul câștigător este apoi retestat pe setul nevăzut și expus printr-un endpoint FastAPI local.

Acesta este un experiment de intervenție cauzală limitat, nu o demonstrație că un head are singur o funcție semantică stabilă. Un rezultat pozitiv arată că, pentru prompturile și metrica alese, înlocuirea acelei activări modifică diferența dintre doi logit-uri. Pentru context despre schimbările de API din ecosistem, este util și auditul migrării la Transformers v5.

Fixăm o țintă măsurabilă: alegerea dintre două nume

Fiecare exemplu are o versiune clean, o versiune corrupt și două răspunsuri candidate. În propoziția curată, persoana care primește obiectul este primul nume; în propoziția coruptă, ordinea rolurilor este inversată.

Metrica este diferența de logit:

logit_diff = logit(răspuns_clean) - logit(răspuns_corrupt)

Calculăm trei valori:

• clean_logit_diff, pe promptul curat;
• corrupt_logit_diff, pe promptul corupt;
• patched_logit_diff, după ce activarea unui head din rularea curată este introdusă în rularea coruptă.

Scorul normalizat al intervenției este:

recovery = (patched_logit_diff - corrupt_logit_diff)
           / (clean_logit_diff - corrupt_logit_diff)

Un scor pozitiv indică deplasarea în direcția comportamentului curat. Un scor apropiat de 1 înseamnă recuperarea aproximativă a întregului efect măsurat, iar un scor negativ arată că intervenția a împins predicția în direcția opusă. Valori peste 1 sunt posibile; metrica nu este o probabilitate.

Pe lângă recovery, raportăm pairwise_flips: câte exemple trec de la preferința pentru răspunsul corupt la preferința pentru răspunsul curat. Nu confundăm această comparație între două tokenuri cu schimbarea tokenului top-1 din întregul vocabular.

Laboratorul rulează pe CPU și păstrează exact checkpointul GPT-2

Configurația folosită este Ubuntu 24.04 LTS, Python 3.11, TransformerLens 4.0.0 și GPT-2 de la openai-community/gpt2, fixat la revizia Hugging Face 607a30d783dfa663caf39e06633721c8d4cfcd7e. Modelul are 12 straturi și 12 head-uri pe strat, deci experimentul execută 144 de intervenții pentru splitul de descoperire.

Sunt suficienți 8 GB RAM și aproximativ 3 GB liberi pe disc. Pe CPU, rularea completă poate dura zeci de minute; timpul depinde de procesor. Nu folosim GPU, tocmai pentru a păstra o singură cale de execuție. Experimentul necesită acces la internet numai la prima descărcare a pachetelor și a checkpointului.

Verifică gazda înainte de instalare:

cat /etc/os-release
python3 --version
python3 -c 'import platform; print(platform.machine())'
free -h
df -h .

Exemplu de structură a rezultatului; valorile de memorie și disc diferă:

PRETTY_NAME="Ubuntu 24.04 LTS"
Python 3.11.x
x86_64
Mem: cel puțin 8 GiB disponibili pe gazdă
Filesystem: cel puțin 3 GiB liberi

TransformerLens 4.0 a eliminat vechiul HookedTransformer. Proiectul folosește TransformerBridge.boot_transformers() și activează explicit modul de compatibilitate, care înregistrează aliasurile clasice de hook și aplică procesarea numerică echivalentă vechilor fluxuri TransformerLens.

Instalăm stack-ul și descărcăm checkpointul imuabil

Creează directorul de lucru și mediul virtual. Fișierul de dependențe fixează pachetele apelate direct de proiect; pip check validează rezolvarea dependențelor tranzitive.

mkdir -p tl-gpt2-patching
cd tl-gpt2-patching
python3 -m venv .venv
source .venv/bin/activate
python -m pip install --upgrade pip
mkdir -p src data artifacts models

Fișierul requirements.txt:

transformer-lens==4.0.0
torch==2.8.0
huggingface-hub==0.34.4
fastapi==0.116.1
uvicorn==0.35.0
pydantic==2.11.7

Instalează și verifică versiunile:

python -m pip install -r requirements.txt
python -m pip check
python -c 'import torch, transformer_lens; print(torch.__version__); print(transformer_lens.__version__)'

Outputul de versiune trebuie să înceapă astfel:

2.8.0
4.0.0

Descarcă revizia fixată într-un director local. local_dir permite codului ulterior să lucreze numai cu fișierele din proiect:

python -c 'from huggingface_hub import snapshot_download; snapshot_download(repo_id="openai-community/gpt2", revision="607a30d783dfa663caf39e06633721c8d4cfcd7e", local_dir="models/gpt2")'

Punct de control: models/gpt2/config.json, fișierele tokenizerului și greutățile modelului trebuie să existe. Dacă revizia nu poate fi descărcată, nu înlocui hash-ul cu main; verifică accesul la Hugging Face și comanda copiată.

Cele patru componente separă datele, intervenția și servirea

Arborele final este:

tl-gpt2-patching/
├── requirements.txt          # versiunile instalate
├── data/
│   └── ioi_pairs.json        # cele 12 perechi generate local
├── models/
│   └── gpt2/                 # checkpointul fixat
├── artifacts/
│   ├── head_scores.csv       # toate cele 144 de intervenții
│   └── report.json           # splituri, baseline și test final
└── src/
    ├── prepare_data.py       # construiește datasetul controlat
    ├── core.py               # încărcare, metrici și patching
    ├── analyze.py            # clasifică head-urile și testează câștigătorul
    └── serve.py              # endpoint local pentru o pereche nouă

Datasetul se numește în proiect IOI Pairs RO Tutorial v1. Este un set sintetic de 12 perechi, generat integral de scriptul următor și dedicat prin CC0-1.0. Nu este corpusul de antrenare al GPT-2 și nu pretinde că reprezintă limbajul natural în ansamblu. Dimensiunea mică îl face potrivit pentru un test controlat, nu pentru generalizarea unei concluzii despre model.

Pasul 1 — Generăm perechile și înghețăm splitul înainte de analiză

Scopul este să evităm alegerea head-ului și raportarea performanței pe exact aceleași exemple. Seed-ul 47 produce un split reproductibil cu opt cazuri de descoperire și patru de test.

Fișierul src/prepare_data.py:

import json
import random
from pathlib import Path

SEED = 47
OUTPUT = Path("data/ioi_pairs.json")

NAME_PAIRS = [
    ("John", "Mary"),
    ("Alice", "Bob"),
    ("Sarah", "David"),
    ("Michael", "Emma"),
    ("James", "Susan"),
    ("Robert", "Linda"),
    ("Thomas", "Lisa"),
    ("Daniel", "Karen"),
    ("George", "Helen"),
    ("Peter", "Laura"),
    ("Mark", "Julia"),
    ("Paul", "Anne"),
]

TEMPLATE = (
    "When {first} and {second} went to the office, "
    "{second} handed the report to"
)


def make_example(index: int, first: str, second: str) -> dict:
    clean = TEMPLATE.format(first=first, second=second)
    corrupt = TEMPLATE.format(first=second, second=first)
    return {
        "id": f"pair-{index:02d}",
        "clean_prompt": clean,
        "corrupt_prompt": corrupt,
        "clean_answer": f" {first}",
        "corrupt_answer": f" {second}",
    }


def main() -> None:
    examples = [
        make_example(index, first, second)
        for index, (first, second) in enumerate(NAME_PAIRS, start=1)
    ]
    ids = [item["id"] for item in examples]
    random.Random(SEED).shuffle(ids)
    discovery_ids = sorted(ids[:8])
    test_ids = sorted(ids[8:])

    payload = {
        "name": "IOI Pairs RO Tutorial v1",
        "license": "CC0-1.0",
        "seed": SEED,
        "examples": examples,
        "splits": {
            "discovery": discovery_ids,
            "test": test_ids,
        },
    }

    OUTPUT.parent.mkdir(parents=True, exist_ok=True)
    OUTPUT.write_text(
        json.dumps(payload, indent=2, ensure_ascii=False) + "n",
        encoding="utf-8",
    )
    print(f"saved={OUTPUT}")
    print(f"examples={len(examples)}")
    print(f"discovery={len(discovery_ids)} test={len(test_ids)}")


if __name__ == "__main__":
    main()

Rulează:

python src/prepare_data.py

Trebuie să vezi:

saved=data/ioi_pairs.json
examples=12
discovery=8 test=4

Parametrul SEED controlează numai împărțirea datasetului. Inference-ul GPT-2 rulează în mod de evaluare, fără sampling.

Punct de control: dacă numărul exemplelor nu este 12, oprește experimentul și verifică lista NAME_PAIRS; toate metricile ulterioare presupun opt exemple discovery și patru test.

Pasul 2 — Implementăm baseline-ul și transplantul unui singur head

Codul central validează trei proprietăți înainte de intervenție: răspunsurile trebuie să fie câte un singur token GPT-2, prompturile clean și corrupt trebuie să aibă aceeași lungime, iar cache-ul trebuie să conțină activarea hook_z cerută.

Patch-ul înlocuiește ieșirea unui head la toate pozițiile. Astfel evităm alegerea post-hoc a poziției care produce scorul cel mai convenabil.

Fișierul src/core.py:

import json
from pathlib import Path
from typing import Iterable

import torch
from transformer_lens import TransformerBridge

MODEL_PATH = "models/gpt2"
N_LAYERS = 12
N_HEADS = 12


def load_model() -> TransformerBridge:
    torch.manual_seed(47)
    torch.set_num_threads(4)
    model = TransformerBridge.boot_transformers(MODEL_PATH)
    model.enable_compatibility_mode()
    model.eval()
    return model


def extract_logits(output: object) -> torch.Tensor:
    if isinstance(output, torch.Tensor):
        return output
    if hasattr(output, "logits"):
        return output.logits
    raise TypeError(f"Output fără logits: {type(output)!r}")


def load_dataset(path: str = "data/ioi_pairs.json") -> dict:
    payload = json.loads(Path(path).read_text(encoding="utf-8"))
    if len(payload["examples"]) != 12:
        raise ValueError("Datasetul trebuie să conțină exact 12 exemple.")
    return payload


def select_split(payload: dict, split: str) -> list[dict]:
    wanted = set(payload["splits"][split])
    selected = [item for item in payload["examples"] if item["id"] in wanted]
    if len(selected) != len(wanted):
        raise ValueError(f"Split incomplet: {split}")
    return selected


def answer_token_id(model: TransformerBridge, answer: str) -> int:
    token_ids = model.tokenizer.encode(answer, add_special_tokens=False)
    if len(token_ids) != 1:
        raise ValueError(
            f"Răspunsul {answer!r} are {len(token_ids)} tokenuri, nu unul."
        )
    return int(token_ids[0])


def prepare_batch(
    model: TransformerBridge,
    examples: Iterable[dict],
) -> tuple[torch.Tensor, torch.Tensor, torch.Tensor, torch.Tensor]:
    rows = list(examples)
    clean_prompts = [row["clean_prompt"] for row in rows]
    corrupt_prompts = [row["corrupt_prompt"] for row in rows]

    clean_tokens = model.to_tokens(clean_prompts, prepend_bos=True)
    corrupt_tokens = model.to_tokens(corrupt_prompts, prepend_bos=True)

    if clean_tokens.shape != corrupt_tokens.shape:
        raise ValueError(
            "Prompturile clean și corrupt nu au aceeași formă tokenizată: "
            f"{tuple(clean_tokens.shape)} versus {tuple(corrupt_tokens.shape)}"
        )

    clean_answers = torch.tensor(
        [answer_token_id(model, row["clean_answer"]) for row in rows],
        dtype=torch.long,
        device=clean_tokens.device,
    )
    corrupt_answers = torch.tensor(
        [answer_token_id(model, row["corrupt_answer"]) for row in rows],
        dtype=torch.long,
        device=clean_tokens.device,
    )
    return clean_tokens, corrupt_tokens, clean_answers, corrupt_answers


def logit_differences(
    logits: torch.Tensor,
    clean_answers: torch.Tensor,
    corrupt_answers: torch.Tensor,
) -> torch.Tensor:
    final_logits = logits[:, -1, :]
    batch = torch.arange(final_logits.shape[0], device=final_logits.device)
    return (
        final_logits[batch, clean_answers]
        - final_logits[batch, corrupt_answers]
    )


def run_baselines(
    model: TransformerBridge,
    clean_tokens: torch.Tensor,
    corrupt_tokens: torch.Tensor,
    clean_answers: torch.Tensor,
    corrupt_answers: torch.Tensor,
) -> dict:
    with torch.inference_mode():
        clean_logits = extract_logits(model(clean_tokens))
        corrupt_logits = extract_logits(model(corrupt_tokens))

    clean_diff = logit_differences(
        clean_logits, clean_answers, corrupt_answers
    )
    corrupt_diff = logit_differences(
        corrupt_logits, clean_answers, corrupt_answers
    )
    return {
        "clean_per_example": clean_diff.cpu(),
        "corrupt_per_example": corrupt_diff.cpu(),
        "clean_mean": float(clean_diff.mean().item()),
        "corrupt_mean": float(corrupt_diff.mean().item()),
    }


def cache_clean_heads(
    model: TransformerBridge,
    clean_tokens: torch.Tensor,
) -> dict:
    names = {
        f"blocks.{layer}.attn.hook_z"
        for layer in range(N_LAYERS)
    }
    with torch.inference_mode():
        _, cache = model.run_with_cache(
            clean_tokens,
            names_filter=lambda name: name in names,
        )
    missing = sorted(names.difference(cache.keys()))
    if missing:
        raise KeyError(f"Hook-uri lipsă din cache: {missing}")
    return cache


def run_patched_head(
    model: TransformerBridge,
    corrupt_tokens: torch.Tensor,
    clean_cache: dict,
    clean_answers: torch.Tensor,
    corrupt_answers: torch.Tensor,
    layer: int,
    head: int,
) -> torch.Tensor:
    hook_name = f"blocks.{layer}.attn.hook_z"
    clean_z = clean_cache[hook_name]

    def patch_head(activation: torch.Tensor, hook: object) -> torch.Tensor:
        if activation.ndim != 4:
            raise ValueError(
                f"{hook_name} are forma neașteptată {tuple(activation.shape)}"
            )
        patched = activation.clone()
        patched[:, :, head, :] = clean_z[:, :, head, :]
        return patched

    with torch.inference_mode():
        output = model.run_with_hooks(
            corrupt_tokens,
            fwd_hooks=[(hook_name, patch_head)],
        )
    logits = extract_logits(output)
    return logit_differences(logits, clean_answers, corrupt_answers)


def score_intervention(
    clean_diff: torch.Tensor,
    corrupt_diff: torch.Tensor,
    patched_diff: torch.Tensor,
) -> dict:
    clean_mean = float(clean_diff.mean().item())
    corrupt_mean = float(corrupt_diff.mean().item())
    patched_mean = float(patched_diff.mean().item())
    denominator = clean_mean - corrupt_mean

    if abs(denominator) < 1e-8:
        raise ZeroDivisionError(
            "Baseline-urile clean și corrupt au aceeași medie."
        )

    recovery = (patched_mean - corrupt_mean) / denominator
    pairwise_flips = int(
        ((corrupt_diff <= 0) & (patched_diff > 0)).sum().item()
    )
    return {
        "clean_logit_diff": clean_mean,
        "corrupt_logit_diff": corrupt_mean,
        "patched_logit_diff": patched_mean,
        "recovery": recovery,
        "pairwise_flips": pairwise_flips,
    }

hook_z are forma [batch, position, head, d_head] în convenția folosită aici. Indexarea [:, :, head, :] înlocuiește numai head-ul testat, pentru toate exemplele și pozițiile.

Punct de control: mesajul „Hook-uri lipsă din cache” indică de obicei că modul de compatibilitate nu a fost activat sau că versiunea instalată nu este cea fixată. Nu adapta numele hook-ului prin presupunere; verifică model.list_hooks() și documentația versiunii instalate.

Pasul 3 — Clasificăm toate cele 144 de intervenții fără acces la test

Scriptul următor folosește exclusiv splitul discovery pentru clasament. CSV-ul conține câte un rând pentru fiecare combinație strat–head. Abia după selectarea primului rând, head-ul câștigător este evaluat o singură dată pe splitul test.

Fișierul src/analyze.py:

import csv
import json
import math
from pathlib import Path

from core import (
    N_HEADS,
    N_LAYERS,
    cache_clean_heads,
    load_dataset,
    load_model,
    prepare_batch,
    run_baselines,
    run_patched_head,
    score_intervention,
    select_split,
)

ARTIFACTS = Path("artifacts")


def evaluate_head(model, examples, layer: int, head: int) -> dict:
    clean_tokens, corrupt_tokens, clean_answers, corrupt_answers = (
        prepare_batch(model, examples)
    )
    baseline = run_baselines(
        model,
        clean_tokens,
        corrupt_tokens,
        clean_answers,
        corrupt_answers,
    )
    clean_cache = cache_clean_heads(model, clean_tokens)
    patched = run_patched_head(
        model,
        corrupt_tokens,
        clean_cache,
        clean_answers,
        corrupt_answers,
        layer,
        head,
    )
    return score_intervention(
        baseline["clean_per_example"],
        baseline["corrupt_per_example"],
        patched.cpu(),
    )


def main() -> None:
    ARTIFACTS.mkdir(parents=True, exist_ok=True)
    payload = load_dataset()
    discovery = select_split(payload, "discovery")
    test = select_split(payload, "test")
    model = load_model()

    clean_tokens, corrupt_tokens, clean_answers, corrupt_answers = (
        prepare_batch(model, discovery)
    )
    baseline = run_baselines(
        model,
        clean_tokens,
        corrupt_tokens,
        clean_answers,
        corrupt_answers,
    )
    clean_cache = cache_clean_heads(model, clean_tokens)

    rows = []
    for layer in range(N_LAYERS):
        for head in range(N_HEADS):
            patched = run_patched_head(
                model,
                corrupt_tokens,
                clean_cache,
                clean_answers,
                corrupt_answers,
                layer,
                head,
            )
            score = score_intervention(
                baseline["clean_per_example"],
                baseline["corrupt_per_example"],
                patched.cpu(),
            )
            row = {"layer": layer, "head": head, **score}
            if not all(
                math.isfinite(float(row[key]))
                for key in (
                    "clean_logit_diff",
                    "corrupt_logit_diff",
                    "patched_logit_diff",
                    "recovery",
                )
            ):
                raise ValueError(f"Scor ne-finit pentru {layer}.{head}")
            rows.append(row)
            print(
                f"layer={layer:02d} head={head:02d} "
                f"recovery={row['recovery']:.6f} "
                f"flips={row['pairwise_flips']}"
            )

    rows.sort(
        key=lambda row: (row["recovery"], row["pairwise_flips"]),
        reverse=True,
    )

    csv_path = ARTIFACTS / "head_scores.csv"
    with csv_path.open("w", newline="", encoding="utf-8") as handle:
        writer = csv.DictWriter(handle, fieldnames=list(rows[0].keys()))
        writer.writeheader()
        writer.writerows(rows)

    winner = rows[0]
    test_score = evaluate_head(
        model,
        test,
        int(winner["layer"]),
        int(winner["head"]),
    )

    report = {
        "model": "openai-community/gpt2",
        "revision": "607a30d783dfa663caf39e06633721c8d4cfcd7e",
        "transformer_lens": "4.0.0",
        "discovery_ids": payload["splits"]["discovery"],
        "test_ids": payload["splits"]["test"],
        "heads_evaluated": len(rows),
        "selected_head": {
            "layer": int(winner["layer"]),
            "head": int(winner["head"]),
        },
        "discovery_score": winner,
        "test_score": test_score,
    }
    report_path = ARTIFACTS / "report.json"
    report_path.write_text(
        json.dumps(report, indent=2, ensure_ascii=False) + "n",
        encoding="utf-8",
    )

    print(f"csv={csv_path} rows={len(rows)}")
    print(f"report={report_path}")
    print(
        "selected="
        f"L{report['selected_head']['layer']}H"
        f"{report['selected_head']['head']}"
    )
    print(f"test_recovery={test_score['recovery']:.6f}")
    print(f"test_pairwise_flips={test_score['pairwise_flips']}")


if __name__ == "__main__":
    main()

Rulează din rădăcina proiectului:

PYTHONPATH=src python src/analyze.py

Valorile numerice și head-ul câștigător trebuie citite din execuția locală, nu copiate dintr-un exemplu editorial. Finalul trebuie să confirme această structură:

csv=artifacts/head_scores.csv rows=144
report=artifacts/report.json
selected=L<strat>H<head>
test_recovery=<valoare calculată>
test_pairwise_flips=<număr între 0 și 4>

CSV-ul are 145 de linii: antetul și cele 144 de head-uri. report.json păstrează revizia modelului, spliturile, selecția și scorul de test. Această separare este echivalentul unei evaluări holdout; nu este antrenare sau fine-tuning. Pentru un proiect în care parametrii chiar sunt optimizați, fluxul diferă de distilarea GPT-2 pe WikiText-2 cu KLDivLoss.

Punct de control: dacă test_recovery este mic sau negativ, nu schimba splitul până obții un rezultat favorabil. Concluzia corectă este că head-ul selectat pe cele opt exemple nu a generalizat intervenția la cele patru exemple nevăzute.

Pasul 4 — Servim head-ul selectat printr-un API local

Endpointul primește o pereche nouă, validează tokenizarea și compară baseline-ul corupt cu aceeași rulare după intervenția head-ului salvat. Serverul ascultă numai pe 127.0.0.1; nu este un serviciu de producție și nu trebuie expus direct pe internet.

Fișierul src/serve.py:

import json
from pathlib import Path

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field

from core import (
    cache_clean_heads,
    load_model,
    prepare_batch,
    run_baselines,
    run_patched_head,
    score_intervention,
)

REPORT_PATH = Path("artifacts/report.json")

if not REPORT_PATH.exists():
    raise RuntimeError(
        "Lipsește artifacts/report.json. Rulează mai întâi analyze.py."
    )

REPORT = json.loads(REPORT_PATH.read_text(encoding="utf-8"))
MODEL = load_model()
LAYER = int(REPORT["selected_head"]["layer"])
HEAD = int(REPORT["selected_head"]["head"])

app = FastAPI(title="GPT-2 Head Patching", version="1.0.0")


class PatchRequest(BaseModel):
    clean_prompt: str = Field(min_length=1, max_length=500)
    corrupt_prompt: str = Field(min_length=1, max_length=500)
    clean_answer: str = Field(min_length=1, max_length=50)
    corrupt_answer: str = Field(min_length=1, max_length=50)


@app.get("/health")
def health() -> dict:
    return {
        "status": "ok",
        "selected_layer": LAYER,
        "selected_head": HEAD,
    }


@app.post("/patch")
def patch(request: PatchRequest) -> dict:
    example = request.model_dump()
    example["id"] = "api-request"

    try:
        clean_tokens, corrupt_tokens, clean_answers, corrupt_answers = (
            prepare_batch(MODEL, [example])
        )
        baseline = run_baselines(
            MODEL,
            clean_tokens,
            corrupt_tokens,
            clean_answers,
            corrupt_answers,
        )
        clean_cache = cache_clean_heads(MODEL, clean_tokens)
        patched = run_patched_head(
            MODEL,
            corrupt_tokens,
            clean_cache,
            clean_answers,
            corrupt_answers,
            LAYER,
            HEAD,
        )
        score = score_intervention(
            baseline["clean_per_example"],
            baseline["corrupt_per_example"],
            patched.cpu(),
        )
    except (KeyError, TypeError, ValueError, ZeroDivisionError) as error:
        raise HTTPException(status_code=422, detail=str(error)) from error

    return {
        "selected_head": {"layer": LAYER, "head": HEAD},
        "baseline_preference": (
            "clean_answer"
            if score["corrupt_logit_diff"] > 0
            else "corrupt_answer"
        ),
        "patched_preference": (
            "clean_answer"
            if score["patched_logit_diff"] > 0
            else "corrupt_answer"
        ),
        **score,
    }

Pornește serverul:

PYTHONPATH=src uvicorn serve:app --host 127.0.0.1 --port 8000

În alt terminal, din același director și cu mediul virtual activat:

curl --fail --silent http://127.0.0.1:8000/health

Răspunsul trebuie să conțină head-ul ales în report.json:

{"status":"ok","selected_layer":<strat>,"selected_head":<head>}

Testează o pereche:

curl --fail --silent 
  -H 'Content-Type: application/json' 
  -d '{
    "clean_prompt":"When John and Mary went to the office, Mary handed the report to",
    "corrupt_prompt":"When Mary and John went to the office, John handed the report to",
    "clean_answer":" John",
    "corrupt_answer":" Mary"
  }' 
  http://127.0.0.1:8000/patch

Răspunsul include corrupt_logit_diff, patched_logit_diff, recovery și preferința dintre cele două răspunsuri. Valorile se calculează pe gazda cititorului; nu există o valoare editorială prestabilită.

Punct de control: un răspuns HTTP 422 este o validare utilă, nu un motiv pentru a elimina verificările. Detaliul indică de regulă un răspuns format din mai multe tokenuri sau prompturi cu lungimi tokenizate diferite.

Acceptăm rezultatul numai dacă artefactele și spliturile trec testul de capăt

Oprește serverul cu Ctrl+C, apoi execută verificarea:

python - <<'PY'
import csv
import json
import math
from pathlib import Path

csv_path = Path("artifacts/head_scores.csv")
report_path = Path("artifacts/report.json")

assert csv_path.is_file(), "Lipsește head_scores.csv"
assert report_path.is_file(), "Lipsește report.json"

with csv_path.open(encoding="utf-8", newline="") as handle:
    rows = list(csv.DictReader(handle))

report = json.loads(report_path.read_text(encoding="utf-8"))
discovery = set(report["discovery_ids"])
test = set(report["test_ids"])

assert len(rows) == 144, len(rows)
assert report["heads_evaluated"] == 144
assert len(discovery) == 8
assert len(test) == 4
assert discovery.isdisjoint(test)
assert len(discovery | test) == 12
assert 0 <= report["selected_head"]["layer"] < 12
assert 0 <= report["selected_head"]["head"] < 12
assert math.isfinite(float(report["test_score"]["recovery"]))
assert 0 <= int(report["test_score"]["pairwise_flips"]) <= 4

print("OK: 144 head-uri, split 8/4, artefacte și metrici valide")
PY

Rezultatul de acceptanță este:

OK: 144 head-uri, split 8/4, artefacte și metrici valide

Testul confirmă integritatea pipeline-ului, nu garantează că intervenția generalizează. Pentru o afirmație mai puternică sunt necesare mai multe șabloane, mai multe seed-uri și validare pe distribuții distincte.

Cinci defecțiuni specifice acestui experiment

ImportError: cannot import name 'HookedTransformer' from 'transformer_lens'

Cauza este folosirea unui exemplu anterior TransformerLens 4.0. Importul corect în acest proiect este from transformer_lens import TransformerBridge. Reinstalează mediul fixat:

python -m pip install --force-reinstall -r requirements.txt

KeyError: Hook-uri lipsă din cache: [...]

Aliasurile blocks.N.attn.hook_z nu au fost înregistrate. Confirmă că model.enable_compatibility_mode() este apelat imediat după boot_transformers() și verifică versiunea:

python -c 'import transformer_lens; print(transformer_lens.__version__)'

Răspunsul ' ...' are 2 tokenuri, nu unul.

Tokenul candidat nu respectă ipoteza metricii. Inspectează tokenizarea fără a continua experimentul:

PYTHONPATH=src python - <<'PY'
from core import load_model
model = load_model()
for text in [" John", " Mary"]:
    print(repr(text), model.tokenizer.encode(text, add_special_tokens=False))
PY

Folosește doar exemplele deja validate de prepare_batch; schimbarea numelor cere regenerarea și rerularea întregului clasament.

Prompturile clean și corrupt nu au aceeași formă tokenizată

Patching-ul poziție-la-poziție nu mai are o aliniere validă. Nu completa cu padding și nu trunchia după rezultat. Păstrează același șablon și nume care produc aceeași lungime tokenizată, apoi rulează din nou prepare_data.py.

Baseline-urile clean și corrupt au aceeași medie.

Numitorul recovery este zero sau numeric prea apropiat de zero, deci scorul normalizat nu este definit. Inspectează diferențele brute din baseline și schimbă ipoteza experimentală înainte de selectarea head-urilor; nu adăuga un epsilon doar pentru a obține un clasament.

Curățăm modelul, mediul și toate rezultatele derivate

Serverul trebuie oprit înainte de ștergere. Comenzile următoare elimină integral laboratorul, inclusiv checkpointul descărcat și artefactele:

deactivate 2>/dev/null || true
cd ..
rm -rf tl-gpt2-patching

Dacă vrei să refaci numai analiza, păstrând modelul și codul, șterge exclusiv datele derivate:

rm -f artifacts/head_scores.csv artifacts/report.json
python src/prepare_data.py
PYTHONPATH=src python src/analyze.py

Nu există configurații de sistem, servicii permanente sau porturi deschise după oprirea procesului Uvicorn.

Bifele care decid dacă experimentul poate fi citat

• ☐ TransformerLens raportează versiunea 4.0.0.
• ☐ Checkpointul GPT-2 provine din revizia fixată în proiect.
• ☐ Datasetul conține exact 12 perechi.
• ☐ Spliturile discovery și test au 8, respectiv 4 exemple și nu se suprapun.
• ☐ Fiecare răspuns candidat este un singur token GPT-2.
• ☐ Prompturile clean și corrupt au aceeași formă tokenizată.
• ☐ head_scores.csv conține 144 de rânduri de date.
• ☐ Head-ul este ales numai din splitul discovery.
• ☐ report.json păstrează separat scorul pe test.
• ☐ API-ul ascultă numai pe 127.0.0.1.

Următorul experiment logic este patching-ul pe poziții individuale pentru head-ul deja selectat, cu selecția poziției făcută pe discovery și verificată separat pe test. A doua extensie este compararea hook_z cu fluxul rezidual, fără a interpreta scorurile celor două tipuri de activări ca fiind direct echivalente. Pentru studii mai mari, costul memoriei cache-ului poate fi controlat cu tehnici precum Selective Activation Checkpointing în PyTorch, dar acesta rezolvă o problemă diferită de intervenția cauzală folosită aici.

Întrebări frecvente

Experimentul antrenează sau modifică permanent GPT-2?

Nu. Parametrii modelului rămân neschimbați; activation patching înlocuiește temporar o activare în timpul unui forward pass.

Un recovery mare dovedește funcția semantică a head-ului?

Nu. Arată doar că intervenția asupra acelui head influențează metrica pentru prompturile testate. Sunt necesare șabloane și distribuții suplimentare înaintea unei concluzii generale.

De ce sunt separate exemplele discovery și test?

Head-ul este selectat pe cele opt exemple discovery, iar efectul său este măsurat ulterior pe patru exemple nevăzute. Astfel, scorul de test nu este folosit pentru alegerea head-ului.

API-ul schimbă tokenul top-1 al GPT-2?

Endpointul raportează preferința dintre cele două răspunsuri candidate. Aceasta nu este automat aceeași cu schimbarea tokenului top-1 din întregul vocabular.

Surse

• TransformerLens 4.0 release notes
• Migrating to TransformerLens 4.0
• TransformerBridge Compatibility Mode
• TransformerBridge Model Structure
• TransformerLens patching module
• GPT-2 model repository