feat: isolate DeepSeek history boundary token
This commit is contained in:
@@ -319,3 +319,55 @@ in 24/24 layer×domain cells, with all 24 paired intervals below zero. See
|
||||
`research/DEEPSEEK_ROUTING_HISTORY_DISTANCE_CONTROL_AUDIT.md` for the full
|
||||
table, token-level route stability, BF16 batch-shape boundary, literature
|
||||
context, and non-claims.
|
||||
|
||||
## History-boundary single-token control
|
||||
|
||||
`v2_lite_routing_history_boundary_token_control.py` keeps the repeated-token
|
||||
history from the preceding probe and changes exactly one token ID at the
|
||||
completed assistant boundary:
|
||||
|
||||
```text
|
||||
system off/on × official EOS / x / period / newline
|
||||
```
|
||||
|
||||
The official template places EOS between the filler assistant content and the
|
||||
next `User:` marker. The three controls replace only that EOS ID after official
|
||||
tokenization. They are explicit counterfactual token sequences, not valid
|
||||
official chat serializations. All eight conditions preserve sequence length,
|
||||
target position, role markers, attention mask, and within-run batch shape.
|
||||
|
||||
```bash
|
||||
PYTHONPATH=/path/to/transformers-4.41.2-deps:/usr/lib/python3/dist-packages \
|
||||
python -B experiments/deepseek/v2_lite_routing_history_boundary_token_control.py \
|
||||
--artifact-dir /path/to/deepseek-v2-lite \
|
||||
--human-eval /path/to/HumanEval.jsonl.gz \
|
||||
--gsm8k /path/to/gsm8k/test.jsonl \
|
||||
--tnews /path/to/tnews/test.json \
|
||||
--tnews-archive /path/to/tnews_public.zip \
|
||||
--wikitext /path/to/wikitext-validation.parquet \
|
||||
--output src/data/deepseek-v2-lite-routing-history-boundary-token-control.json \
|
||||
--per-domain 32 \
|
||||
--content-tokens 23 \
|
||||
--batch-prompts 4 \
|
||||
--layers 7 \
|
||||
--bootstrap 2000 \
|
||||
--seed 20260729 \
|
||||
--captured-at 2026-07-29T11:12:00+00:00
|
||||
```
|
||||
|
||||
The eight cells add 2,044,224 real top-6 route selections. All 256
|
||||
source×system groups are equal-length and equal-position; all 768
|
||||
counterfactual cells differ from their official sequence at exactly one input
|
||||
ID. The committed run and independent rerun are byte-exact:
|
||||
|
||||
```text
|
||||
9bb93834ffd8536aeebe4325e45d2179ba590554c6ff6b6fcceaba2f499b9c37
|
||||
```
|
||||
|
||||
For exact target content under prompt-balanced aggregation, mean system-edge
|
||||
TV is `.0374 / .0539 / .0553 / .0492` for EOS / x / period / newline. X and
|
||||
period exceed EOS in 24/24 layer×domain cells, while newline does so in 23/24.
|
||||
See `research/DEEPSEEK_ROUTING_HISTORY_BOUNDARY_TOKEN_AUDIT.md` for all paired
|
||||
intervals, per-token route alignment, scope split, BF16 batch-shape audit,
|
||||
primary literature, and the boundary between an input-ID intervention and a
|
||||
chat-turn semantic claim.
|
||||
|
||||
@@ -0,0 +1,774 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Run a 2 x 4 history-boundary token control on DeepSeek-V2-Lite.
|
||||
|
||||
Every condition contains the same repeated-token user/assistant history and
|
||||
the same target user content. The system factor is off/on. The second factor
|
||||
changes exactly one token ID at the completed assistant-turn boundary:
|
||||
|
||||
eos: the official-template EOS token
|
||||
x: ordinary one-token content control
|
||||
period: ordinary one-token punctuation control
|
||||
newline: ordinary one-token formatting control
|
||||
|
||||
The three non-EOS conditions are explicit token-ID counterfactuals rather than
|
||||
official serialized chats. They preserve sequence length, target position,
|
||||
role-marker tokens, attention mask, and padded batch shape. This identifies
|
||||
the routing effect of replacing the learned EOS ID at this one boundary; it
|
||||
does not identify the effect of removing all turn-boundary information.
|
||||
|
||||
The official BF16 forward path is reused from the audited message-history
|
||||
runner. This file owns the eight-cell renderer, shared-bootstrap statistics,
|
||||
counterfactual contract, and final result schema.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import contextlib
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import numpy as np
|
||||
|
||||
import v2_lite_routing_history_distance_control as prior
|
||||
|
||||
|
||||
base = prior.base
|
||||
|
||||
SYSTEM_MESSAGE = prior.SYSTEM_MESSAGE
|
||||
FILLER_USER = prior.FILLER_USER
|
||||
FILLER_ASSISTANT = prior.FILLER_ASSISTANT
|
||||
|
||||
BOUNDARY_LEVELS = ("eos", "x", "period", "newline")
|
||||
BOUNDARY_TEXT = {
|
||||
"x": "x",
|
||||
"period": ".",
|
||||
"newline": "\n",
|
||||
}
|
||||
CONDITIONS = tuple(
|
||||
f"s{system}_{boundary}"
|
||||
for boundary in BOUNDARY_LEVELS
|
||||
for system in (0, 1)
|
||||
)
|
||||
FACTORS = {
|
||||
condition: {
|
||||
"system": int(condition[1]),
|
||||
"history": "filler",
|
||||
"boundary": condition.split("_", 1)[1],
|
||||
}
|
||||
for condition in CONDITIONS
|
||||
}
|
||||
SYSTEM_CELLS = {
|
||||
boundary: (f"s0_{boundary}", f"s1_{boundary}")
|
||||
for boundary in BOUNDARY_LEVELS
|
||||
}
|
||||
SYSTEM_EDGE_CONTRASTS = {
|
||||
f"{boundary}_minus_eos": ("eos", boundary)
|
||||
for boundary in BOUNDARY_LEVELS
|
||||
if boundary != "eos"
|
||||
}
|
||||
COMPARISONS = tuple(
|
||||
[
|
||||
(
|
||||
f"system_{boundary}",
|
||||
f"s0_{boundary}",
|
||||
f"s1_{boundary}",
|
||||
)
|
||||
for boundary in BOUNDARY_LEVELS
|
||||
]
|
||||
+ [
|
||||
(
|
||||
f"{boundary}_at_s{system}",
|
||||
f"s{system}_eos",
|
||||
f"s{system}_{boundary}",
|
||||
)
|
||||
for boundary in BOUNDARY_LEVELS
|
||||
if boundary != "eos"
|
||||
for system in (0, 1)
|
||||
]
|
||||
)
|
||||
ALIGNMENT_COMPARISONS = COMPARISONS
|
||||
|
||||
RENDER_AUDIT: list[dict[str, Any]] = []
|
||||
BOUNDARY_TOKEN_IDS: dict[str, int] = {}
|
||||
|
||||
|
||||
def condition_messages(
|
||||
content: str,
|
||||
condition: str,
|
||||
) -> list[dict[str, str]]:
|
||||
factors = FACTORS[condition]
|
||||
messages: list[dict[str, str]] = []
|
||||
if factors["system"]:
|
||||
messages.append({"role": "system", "content": SYSTEM_MESSAGE})
|
||||
messages.extend(
|
||||
[
|
||||
{"role": "user", "content": FILLER_USER},
|
||||
{"role": "assistant", "content": FILLER_ASSISTANT},
|
||||
{"role": "user", "content": content},
|
||||
]
|
||||
)
|
||||
return messages
|
||||
|
||||
|
||||
def boundary_token_ids(tokenizer: Any) -> dict[str, int]:
|
||||
if tokenizer.eos_token_id is None:
|
||||
raise RuntimeError("tokenizer has no EOS token")
|
||||
ids = {"eos": int(tokenizer.eos_token_id)}
|
||||
for name, text in BOUNDARY_TEXT.items():
|
||||
encoded = list(
|
||||
tokenizer(text, add_special_tokens=False).input_ids
|
||||
)
|
||||
if len(encoded) != 1:
|
||||
raise RuntimeError(
|
||||
f"{name} boundary control is not one token: {encoded}"
|
||||
)
|
||||
if encoded[0] in tokenizer.all_special_ids:
|
||||
raise RuntimeError(
|
||||
f"{name} boundary control unexpectedly uses a special token"
|
||||
)
|
||||
ids[name] = int(encoded[0])
|
||||
if len(set(ids.values())) != len(ids):
|
||||
raise RuntimeError(f"boundary token IDs are not distinct: {ids}")
|
||||
return ids
|
||||
|
||||
|
||||
def render_boundary_variant(
|
||||
tokenizer: Any,
|
||||
content: str,
|
||||
condition: str,
|
||||
) -> dict[str, Any]:
|
||||
messages = condition_messages(content, condition)
|
||||
rendered = tokenizer.apply_chat_template(
|
||||
messages,
|
||||
tokenize=False,
|
||||
add_generation_prompt=True,
|
||||
)
|
||||
official_ids = list(
|
||||
tokenizer.apply_chat_template(
|
||||
messages,
|
||||
tokenize=True,
|
||||
add_generation_prompt=True,
|
||||
)
|
||||
)
|
||||
token_ids, offsets = base.tokenize_with_offsets(
|
||||
tokenizer,
|
||||
rendered,
|
||||
add_special_tokens=False,
|
||||
)
|
||||
if token_ids != official_ids:
|
||||
raise RuntimeError(
|
||||
f"{condition} rendered token IDs differ from apply_chat_template"
|
||||
)
|
||||
|
||||
ids = boundary_token_ids(tokenizer)
|
||||
BOUNDARY_TOKEN_IDS.update(ids)
|
||||
boundary_positions = [
|
||||
index
|
||||
for index, token_id in enumerate(official_ids)
|
||||
if token_id == ids["eos"]
|
||||
]
|
||||
if len(boundary_positions) != 1:
|
||||
raise RuntimeError(
|
||||
f"{condition} expected one history EOS; got {boundary_positions}"
|
||||
)
|
||||
boundary_position = boundary_positions[0]
|
||||
boundary = FACTORS[condition]["boundary"]
|
||||
token_ids = list(token_ids)
|
||||
token_ids[boundary_position] = ids[boundary]
|
||||
differing_ids = sum(
|
||||
left != right
|
||||
for left, right in zip(official_ids, token_ids, strict=True)
|
||||
)
|
||||
expected_differences = 0 if boundary == "eos" else 1
|
||||
if differing_ids != expected_differences:
|
||||
raise RuntimeError(
|
||||
f"{condition} changed {differing_ids} IDs; "
|
||||
f"expected {expected_differences}"
|
||||
)
|
||||
|
||||
content_start = rendered.rfind(content)
|
||||
if content_start < 0:
|
||||
raise RuntimeError(
|
||||
f"{condition} target content is absent after rendering"
|
||||
)
|
||||
content_end = content_start + len(content)
|
||||
positions, records, crossing = base.content_positions(
|
||||
token_ids,
|
||||
offsets,
|
||||
content_start,
|
||||
content_end,
|
||||
)
|
||||
if not positions:
|
||||
raise RuntimeError(f"{condition} has no target-content tokens")
|
||||
if boundary_position >= min(positions):
|
||||
raise RuntimeError(
|
||||
f"{condition} history boundary is not before target content"
|
||||
)
|
||||
|
||||
decoded = tokenizer.decode(
|
||||
token_ids,
|
||||
skip_special_tokens=False,
|
||||
clean_up_tokenization_spaces=False,
|
||||
)
|
||||
RENDER_AUDIT.append(
|
||||
{
|
||||
"content_sha256": base.text_sha256(content),
|
||||
"condition": condition,
|
||||
"boundary": boundary,
|
||||
"boundary_position": boundary_position,
|
||||
"boundary_token_id": ids[boundary],
|
||||
"official_eos_token_id": ids["eos"],
|
||||
"input_tokens": len(token_ids),
|
||||
"target_first_position": min(positions),
|
||||
"changed_token_ids_vs_official": differing_ids,
|
||||
"counterfactual_decoded_sha256": base.text_sha256(decoded),
|
||||
}
|
||||
)
|
||||
return {
|
||||
"condition": condition,
|
||||
"messages_sha256": base.canonical_hash(messages),
|
||||
"rendered_sha256": base.text_sha256(rendered),
|
||||
"token_ids": token_ids,
|
||||
"tokens": len(token_ids),
|
||||
"token_ids_sha256": base.canonical_hash(token_ids),
|
||||
"content_positions": positions,
|
||||
"content_records": records,
|
||||
"content_tokens": len(positions),
|
||||
"wrapper_tokens": len(token_ids) - len(positions),
|
||||
"boundary_crossing_tokens": crossing,
|
||||
}
|
||||
|
||||
|
||||
def boundary_control_domain(
|
||||
loads: dict[str, np.ndarray],
|
||||
mode: str,
|
||||
replicates: int,
|
||||
seed: int,
|
||||
scope: str,
|
||||
) -> dict[str, Any]:
|
||||
"""Compute all eight cells with one shared source-bootstrap matrix."""
|
||||
shapes = {value.shape for value in loads.values()}
|
||||
if len(shapes) != 1:
|
||||
raise ValueError(f"boundary-control shape mismatch: {sorted(shapes)}")
|
||||
rows = next(iter(loads.values())).shape[0]
|
||||
rng = np.random.default_rng(base.scoped_seed(seed, scope))
|
||||
sampled = rng.integers(
|
||||
0,
|
||||
rows,
|
||||
size=(replicates, rows),
|
||||
endpoint=False,
|
||||
)
|
||||
point = {
|
||||
condition: base.distribution(value, mode)
|
||||
for condition, value in loads.items()
|
||||
}
|
||||
boot = {
|
||||
condition: base.bootstrap_distributions(value, mode, sampled)
|
||||
for condition, value in loads.items()
|
||||
}
|
||||
point_metrics = {
|
||||
condition: base.metric_vector(value)
|
||||
for condition, value in point.items()
|
||||
}
|
||||
boot_metrics = {
|
||||
condition: base.metric_vector(value)
|
||||
for condition, value in boot.items()
|
||||
}
|
||||
|
||||
def system_edges(
|
||||
values: dict[str, np.ndarray],
|
||||
) -> dict[str, np.ndarray]:
|
||||
return {
|
||||
boundary: values[after] - values[before]
|
||||
for boundary, (before, after) in SYSTEM_CELLS.items()
|
||||
}
|
||||
|
||||
def edge_contrasts(
|
||||
edges: dict[str, np.ndarray],
|
||||
) -> dict[str, np.ndarray]:
|
||||
return {
|
||||
name: edges[after] - edges[before]
|
||||
for name, (before, after) in SYSTEM_EDGE_CONTRASTS.items()
|
||||
}
|
||||
|
||||
metric_system_edges: dict[str, Any] = {}
|
||||
metric_system_edge_contrasts: dict[str, Any] = {}
|
||||
for metric in point_metrics["s0_eos"]:
|
||||
point_edges = system_edges(
|
||||
{
|
||||
condition: point_metrics[condition][metric]
|
||||
for condition in CONDITIONS
|
||||
}
|
||||
)
|
||||
boot_edges = system_edges(
|
||||
{
|
||||
condition: boot_metrics[condition][metric]
|
||||
for condition in CONDITIONS
|
||||
}
|
||||
)
|
||||
metric_system_edges[metric] = {
|
||||
boundary: {
|
||||
"point": float(point_edges[boundary][0]),
|
||||
"ci95": base.interval(boot_edges[boundary]),
|
||||
}
|
||||
for boundary in BOUNDARY_LEVELS
|
||||
}
|
||||
point_contrasts = edge_contrasts(point_edges)
|
||||
boot_contrasts = edge_contrasts(boot_edges)
|
||||
metric_system_edge_contrasts[metric] = {
|
||||
name: {
|
||||
"point": float(point_contrasts[name][0]),
|
||||
"ci95": base.interval(boot_contrasts[name]),
|
||||
}
|
||||
for name in SYSTEM_EDGE_CONTRASTS
|
||||
}
|
||||
|
||||
point_vectors = system_edges(point)
|
||||
boot_vectors = system_edges(boot)
|
||||
point_vector_contrasts = edge_contrasts(point_vectors)
|
||||
boot_vector_contrasts = edge_contrasts(boot_vectors)
|
||||
distribution_system_edge_contrasts = {}
|
||||
for name in SYSTEM_EDGE_CONTRASTS:
|
||||
point_magnitude = 0.5 * np.abs(
|
||||
point_vector_contrasts[name]
|
||||
).sum()
|
||||
boot_magnitude = 0.5 * np.abs(
|
||||
boot_vector_contrasts[name]
|
||||
).sum(axis=1)
|
||||
distribution_system_edge_contrasts[name] = {
|
||||
"half_l1_magnitude": {
|
||||
"point": float(point_magnitude),
|
||||
"ci95": base.interval(boot_magnitude),
|
||||
},
|
||||
"expert_share_difference_in_system_edges": (
|
||||
point_vector_contrasts[name].tolist()
|
||||
),
|
||||
"expert_share_difference_in_system_edges_ci95": (
|
||||
base.interval(boot_vector_contrasts[name])
|
||||
),
|
||||
}
|
||||
|
||||
point_tv: dict[str, float] = {}
|
||||
boot_tv: dict[str, np.ndarray] = {}
|
||||
point_jsd: dict[str, float] = {}
|
||||
boot_jsd: dict[str, np.ndarray] = {}
|
||||
system_edge_distances: dict[str, Any] = {}
|
||||
for boundary, (before, after) in SYSTEM_CELLS.items():
|
||||
point_delta = point[after] - point[before]
|
||||
point_tv[boundary] = float(0.5 * np.abs(point_delta).sum())
|
||||
boot_tv[boundary] = 0.5 * np.abs(
|
||||
boot[after] - boot[before]
|
||||
).sum(axis=1)
|
||||
point_jsd[boundary] = float(
|
||||
base.js_divergence(point[before], point[after])[0]
|
||||
)
|
||||
boot_jsd[boundary] = base.js_divergence(
|
||||
boot[before],
|
||||
boot[after],
|
||||
)
|
||||
system_edge_distances[boundary] = {
|
||||
"total_variation": {
|
||||
"point": point_tv[boundary],
|
||||
"ci95": base.interval(boot_tv[boundary]),
|
||||
},
|
||||
"js_divergence": {
|
||||
"point": point_jsd[boundary],
|
||||
"ci95": base.interval(boot_jsd[boundary]),
|
||||
"unit": "nats",
|
||||
},
|
||||
}
|
||||
|
||||
system_edge_distance_contrasts = {}
|
||||
for name, (before, after) in SYSTEM_EDGE_CONTRASTS.items():
|
||||
system_edge_distance_contrasts[name] = {
|
||||
"total_variation_delta": {
|
||||
"point": point_tv[after] - point_tv[before],
|
||||
"ci95": base.interval(boot_tv[after] - boot_tv[before]),
|
||||
},
|
||||
"js_divergence_delta": {
|
||||
"point": point_jsd[after] - point_jsd[before],
|
||||
"ci95": base.interval(boot_jsd[after] - boot_jsd[before]),
|
||||
"unit": "nats",
|
||||
},
|
||||
}
|
||||
|
||||
direct_substitutions = {}
|
||||
for boundary in BOUNDARY_LEVELS:
|
||||
if boundary == "eos":
|
||||
continue
|
||||
direct_substitutions[boundary] = {}
|
||||
direct_point_tv: dict[int, float] = {}
|
||||
direct_boot_tv: dict[int, np.ndarray] = {}
|
||||
direct_point_jsd: dict[int, float] = {}
|
||||
direct_boot_jsd: dict[int, np.ndarray] = {}
|
||||
for system in (0, 1):
|
||||
before = f"s{system}_eos"
|
||||
after = f"s{system}_{boundary}"
|
||||
point_delta = point[after] - point[before]
|
||||
tv_boot = 0.5 * np.abs(
|
||||
boot[after] - boot[before]
|
||||
).sum(axis=1)
|
||||
jsd_boot = base.js_divergence(boot[before], boot[after])
|
||||
direct_point_tv[system] = float(
|
||||
0.5 * np.abs(point_delta).sum()
|
||||
)
|
||||
direct_boot_tv[system] = tv_boot
|
||||
direct_point_jsd[system] = float(
|
||||
base.js_divergence(
|
||||
point[before],
|
||||
point[after],
|
||||
)[0]
|
||||
)
|
||||
direct_boot_jsd[system] = jsd_boot
|
||||
direct_substitutions[boundary][f"at_s{system}"] = {
|
||||
"total_variation": {
|
||||
"point": direct_point_tv[system],
|
||||
"ci95": base.interval(tv_boot),
|
||||
},
|
||||
"js_divergence": {
|
||||
"point": direct_point_jsd[system],
|
||||
"ci95": base.interval(jsd_boot),
|
||||
"unit": "nats",
|
||||
},
|
||||
}
|
||||
direct_substitutions[boundary]["s1_minus_s0"] = {
|
||||
"total_variation_delta": {
|
||||
"point": direct_point_tv[1] - direct_point_tv[0],
|
||||
"ci95": base.interval(
|
||||
direct_boot_tv[1] - direct_boot_tv[0]
|
||||
),
|
||||
},
|
||||
"js_divergence_delta": {
|
||||
"point": direct_point_jsd[1] - direct_point_jsd[0],
|
||||
"ci95": base.interval(
|
||||
direct_boot_jsd[1] - direct_boot_jsd[0]
|
||||
),
|
||||
"unit": "nats",
|
||||
},
|
||||
}
|
||||
|
||||
return {
|
||||
"metric_system_edges": metric_system_edges,
|
||||
"metric_system_edge_contrasts": metric_system_edge_contrasts,
|
||||
"system_edge_distances": system_edge_distances,
|
||||
"system_edge_distance_contrasts": (
|
||||
system_edge_distance_contrasts
|
||||
),
|
||||
"distribution_system_edge_contrasts": (
|
||||
distribution_system_edge_contrasts
|
||||
),
|
||||
"direct_substitutions": direct_substitutions,
|
||||
}
|
||||
|
||||
|
||||
def layer_statistics(
|
||||
prompt_rows: list[dict[str, Any]],
|
||||
replicates: int,
|
||||
seed: int,
|
||||
layer_index: int,
|
||||
) -> dict[str, Any]:
|
||||
scopes = {}
|
||||
for load_scope, load_key in (
|
||||
("full_input", "full_load"),
|
||||
("target_content", "content_load"),
|
||||
):
|
||||
modes = {}
|
||||
for mode in ("token_weighted", "prompt_balanced"):
|
||||
conditions: dict[str, Any] = {}
|
||||
domain_loads: dict[str, dict[str, np.ndarray]] = {}
|
||||
for domain in base.DOMAIN_ORDER:
|
||||
rows = [
|
||||
row
|
||||
for row in prompt_rows
|
||||
if row["domain"] == domain
|
||||
]
|
||||
domain_loads[domain] = {
|
||||
condition: np.asarray(
|
||||
[
|
||||
row["conditions"][condition][load_key]
|
||||
for row in rows
|
||||
],
|
||||
dtype=np.int64,
|
||||
)
|
||||
for condition in CONDITIONS
|
||||
}
|
||||
for condition in CONDITIONS:
|
||||
conditions.setdefault(condition, {})[domain] = (
|
||||
base.bootstrap_domain(
|
||||
domain_loads[domain][condition],
|
||||
mode,
|
||||
replicates,
|
||||
seed,
|
||||
(
|
||||
f"layer={layer_index}|scope={load_scope}|"
|
||||
f"mode={mode}|condition={condition}|"
|
||||
f"domain={domain}"
|
||||
),
|
||||
)
|
||||
)
|
||||
|
||||
comparisons: dict[str, Any] = {}
|
||||
for comparison, before, after in COMPARISONS:
|
||||
comparisons[comparison] = {}
|
||||
for domain in base.DOMAIN_ORDER:
|
||||
comparisons[comparison][domain] = base.paired_domain(
|
||||
domain_loads[domain][before],
|
||||
domain_loads[domain][after],
|
||||
mode,
|
||||
replicates,
|
||||
seed,
|
||||
(
|
||||
f"layer={layer_index}|scope={load_scope}|"
|
||||
f"mode={mode}|comparison={comparison}|"
|
||||
f"domain={domain}"
|
||||
),
|
||||
)
|
||||
|
||||
control = {
|
||||
domain: boundary_control_domain(
|
||||
domain_loads[domain],
|
||||
mode,
|
||||
replicates,
|
||||
seed,
|
||||
(
|
||||
f"layer={layer_index}|scope={load_scope}|"
|
||||
f"mode={mode}|boundary_control|domain={domain}"
|
||||
),
|
||||
)
|
||||
for domain in base.DOMAIN_ORDER
|
||||
}
|
||||
modes[mode] = {
|
||||
"conditions": conditions,
|
||||
"comparisons": comparisons,
|
||||
"boundary_control": control,
|
||||
}
|
||||
scopes[load_scope] = {"modes": modes}
|
||||
return scopes
|
||||
|
||||
|
||||
def install_control_contract() -> None:
|
||||
"""Install the eight-cell renderer/statistics into the audited runner."""
|
||||
base.CONDITIONS = CONDITIONS
|
||||
base.FACTORS = FACTORS
|
||||
base.COMPARISONS = COMPARISONS
|
||||
base.ALIGNMENT_COMPARISONS = ALIGNMENT_COMPARISONS
|
||||
base.condition_messages = condition_messages
|
||||
base.render_variant = render_boundary_variant
|
||||
base.layer_statistics = layer_statistics
|
||||
|
||||
|
||||
def output_path_from_argv() -> Path:
|
||||
try:
|
||||
return Path(sys.argv[sys.argv.index("--output") + 1])
|
||||
except (ValueError, IndexError) as error:
|
||||
raise ValueError("--output is required") from error
|
||||
|
||||
|
||||
def render_contract_summary() -> dict[str, Any]:
|
||||
if not RENDER_AUDIT:
|
||||
raise RuntimeError("render audit is empty")
|
||||
by_content: dict[str, list[dict[str, Any]]] = {}
|
||||
for row in RENDER_AUDIT:
|
||||
by_content.setdefault(row["content_sha256"], []).append(row)
|
||||
if len(by_content) != 128 and "--per-domain" not in sys.argv:
|
||||
raise RuntimeError(
|
||||
f"expected 128 selected contents; observed {len(by_content)}"
|
||||
)
|
||||
|
||||
equal_lengths = 0
|
||||
equal_target_positions = 0
|
||||
one_id_replacements = 0
|
||||
official_eos_exact = 0
|
||||
for rows in by_content.values():
|
||||
for system in (0, 1):
|
||||
cells = [
|
||||
row
|
||||
for row in rows
|
||||
if FACTORS[row["condition"]]["system"] == system
|
||||
]
|
||||
if len(cells) != len(BOUNDARY_LEVELS):
|
||||
raise RuntimeError(
|
||||
"render audit lacks one or more boundary cells"
|
||||
)
|
||||
if len({row["input_tokens"] for row in cells}) == 1:
|
||||
equal_lengths += 1
|
||||
if len({row["target_first_position"] for row in cells}) == 1:
|
||||
equal_target_positions += 1
|
||||
official_eos_exact += sum(
|
||||
row["boundary"] == "eos"
|
||||
and row["changed_token_ids_vs_official"] == 0
|
||||
for row in cells
|
||||
)
|
||||
one_id_replacements += sum(
|
||||
row["boundary"] != "eos"
|
||||
and row["changed_token_ids_vs_official"] == 1
|
||||
for row in cells
|
||||
)
|
||||
|
||||
systems = 2 * len(by_content)
|
||||
replacements = 3 * systems
|
||||
return {
|
||||
"selected_contents": len(by_content),
|
||||
"system_groups": systems,
|
||||
"equal_input_length_groups": equal_lengths,
|
||||
"equal_target_position_groups": equal_target_positions,
|
||||
"official_eos_cells_exact": official_eos_exact,
|
||||
"one_id_counterfactual_cells_exact": one_id_replacements,
|
||||
"expected_one_id_counterfactual_cells": replacements,
|
||||
"all_group_lengths_equal": equal_lengths == systems,
|
||||
"all_target_positions_equal": equal_target_positions == systems,
|
||||
"all_counterfactuals_change_exactly_one_id": (
|
||||
one_id_replacements == replacements
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
def finalize_result(path: Path) -> dict[str, Any]:
|
||||
result = json.loads(path.read_text(encoding="utf-8"))
|
||||
result["schema_version"] = 3
|
||||
result["evidence_identity"] = (
|
||||
"X / official BF16 weights and tokenizer; official EOS sequence plus "
|
||||
"three paired one-token boundary-ID counterfactuals on local "
|
||||
"truncated forward"
|
||||
)
|
||||
boundary = result["boundary"]
|
||||
boundary.pop("factorial_claim", None)
|
||||
boundary.update(
|
||||
{
|
||||
"boundary_token_identity_control": True,
|
||||
"official_serialization_by_boundary": {
|
||||
"eos": True,
|
||||
"x": False,
|
||||
"period": False,
|
||||
"newline": False,
|
||||
},
|
||||
"single_input_id_intervention": True,
|
||||
"turn_boundary_removed": False,
|
||||
"role_markers_held_fixed": True,
|
||||
"target_position_held_fixed": True,
|
||||
"task_performance": False,
|
||||
"causal_boundary": (
|
||||
"EOS-to-control comparisons causally intervene on exactly "
|
||||
"one prior input token ID within this fixed forward contract; "
|
||||
"they do not remove the following User role marker, establish "
|
||||
"general EOS semantics, or measure answer quality"
|
||||
),
|
||||
}
|
||||
)
|
||||
|
||||
old_contract = result.pop("message_history_contract")
|
||||
render_validation = render_contract_summary()
|
||||
result["history_boundary_token_contract"] = {
|
||||
"chat_template_revision": base.MODEL_REVISION,
|
||||
"chat_template": old_contract["chat_template"],
|
||||
"chat_template_sha256": old_contract["chat_template_sha256"],
|
||||
"official_assistant_boundary": (
|
||||
"Assistant: {content} + eos_token + User:"
|
||||
),
|
||||
"system_message": SYSTEM_MESSAGE,
|
||||
"system_message_sha256": base.text_sha256(SYSTEM_MESSAGE),
|
||||
"filler_user": FILLER_USER,
|
||||
"filler_user_sha256": base.text_sha256(FILLER_USER),
|
||||
"filler_assistant": FILLER_ASSISTANT,
|
||||
"filler_assistant_sha256": base.text_sha256(FILLER_ASSISTANT),
|
||||
"boundary_levels": list(BOUNDARY_LEVELS),
|
||||
"boundary_token_ids": BOUNDARY_TOKEN_IDS,
|
||||
"boundary_text_controls": BOUNDARY_TEXT,
|
||||
"conditions": FACTORS,
|
||||
"comparisons": [
|
||||
{"name": name, "before": before, "after": after}
|
||||
for name, before, after in COMPARISONS
|
||||
],
|
||||
"system_edge_contrasts": {
|
||||
name: {
|
||||
"before_boundary": before,
|
||||
"after_boundary": after,
|
||||
"definition": (
|
||||
f"system edge at {after} minus system edge at {before}"
|
||||
),
|
||||
}
|
||||
for name, (before, after) in (
|
||||
SYSTEM_EDGE_CONTRASTS.items()
|
||||
)
|
||||
},
|
||||
"render_validation": render_validation,
|
||||
"target_role": "user",
|
||||
"add_generation_prompt": True,
|
||||
"scope_split": {
|
||||
"full_input": (
|
||||
"all rendered or counterfactually edited BOS, system/history, "
|
||||
"target, newline, and generation-prompt token IDs"
|
||||
),
|
||||
"target_content": (
|
||||
"exact intersection of (relative character span, token ID) "
|
||||
"inside target user content across all eight conditions"
|
||||
),
|
||||
},
|
||||
}
|
||||
|
||||
inference = result["inference_contract"]
|
||||
inference["batch_grouping"] = (
|
||||
"all eight boundary variants of one source prompt execute in the "
|
||||
"same right-padded batch"
|
||||
)
|
||||
statistical = result["statistical_contract"]
|
||||
statistical["paired_indices"] = (
|
||||
"one sampled source-prompt index matrix is reused across all eight "
|
||||
"cells for every boundary-token contrast within each "
|
||||
"domain/layer/scope/mode"
|
||||
)
|
||||
statistical.pop("interaction_distribution_magnitude", None)
|
||||
statistical["system_edge_contrast_distribution_magnitude"] = (
|
||||
"0.5 * L1 norm of the signed difference between two system-edge "
|
||||
"expert-share vectors; this is not labeled standard TV"
|
||||
)
|
||||
|
||||
path.write_text(
|
||||
json.dumps(result, indent=2, ensure_ascii=False) + "\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
return result
|
||||
|
||||
|
||||
def main() -> None:
|
||||
install_control_contract()
|
||||
output = output_path_from_argv()
|
||||
with open(os.devnull, "w", encoding="utf-8") as sink:
|
||||
with contextlib.redirect_stdout(sink):
|
||||
base.main()
|
||||
result = finalize_result(output)
|
||||
payload = output.read_bytes()
|
||||
print(
|
||||
json.dumps(
|
||||
{
|
||||
"output": str(output),
|
||||
"sha256": hashlib.sha256(payload).hexdigest(),
|
||||
"bytes": len(payload),
|
||||
"source_prompts": result["inference_contract"][
|
||||
"total_source_prompts"
|
||||
],
|
||||
"prompt_variants": result["inference_contract"][
|
||||
"total_prompt_variants"
|
||||
],
|
||||
"input_tokens_by_condition": result[
|
||||
"inference_contract"
|
||||
]["input_tokens_by_condition"],
|
||||
"total_routes": result["inference_contract"][
|
||||
"total_routes_all_conditions_all_moe_layers"
|
||||
],
|
||||
"render_validation": result[
|
||||
"history_boundary_token_contract"
|
||||
]["render_validation"],
|
||||
},
|
||||
indent=2,
|
||||
ensure_ascii=False,
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user