Run archives and reanalyze¶
Business context: Model calls are the expensive part of an explanation. An archive keeps every coalition’s generated text next to the run’s configuration, so a reviewer can reproduce the attribution months later, or score the same generations with another value function, without paying for a single model call.
Goal: Cover the archive workflow end to end: the archive.path template, the folder layout, manifest version 4 with the recorded embedding source, the SQLite file in WAL mode, on_exists, ExplainRunner.reanalyze reproducing explain, rescoring with the text and embedding value functions and every normalizer, and the guards (ArchiveExistsError, ReanalyzeMismatchError, ArchiveNotFoundError, ArchiveIncompleteError, strict=True).
Prerequisites: mllm-shap, Python 3.12. A rule-based stand-in plays the model (see explain_pipeline.ipynb); archives are written to a temporary folder.
Expected output: An archive folder with three files; reanalyze returning the explain values exactly with zero backend calls; a value-function and normalizer comparison; one line per refused archive.
import json
import re
import shutil
import sqlite3
import tempfile
from contextlib import closing
from pathlib import Path
from mllm_shap import (
ArchiveCorruptError,
ArchiveExistsError,
ArchiveIncompleteError,
ArchiveNotFoundError,
CoalitionMask,
ConversationSnapshot,
DeletePolicy,
ExplainConfig,
ExplainRunner,
GenerationRecord,
ManifestVersionError,
Message,
ReanalyzeMismatchError,
Role,
Turn,
configure_logging,
explain_config_from_yaml,
)
from mllm_shap.domain.config import GenerationConfig
from mllm_shap.runtime import RunArchive
from mllm_shap.value.generation_records import generation_record_from_text
Stand-in model, conversation and configuration¶
Technical: The bot answers from keywords. It can also fail after a number of calls (to leave an unfinished archive) and attach its own embedding vectors when the pipeline asks for embeddings (generation.compute_embeddings), as the transformers backend does. The archive path is a template: {run_id} is replaced by the run’s id, a digest of the snapshot, the configuration (without the archive section) and the model id.
class RefundBot:
"""Rule-based stand-in for a support model."""
model_id = "refund-bot"
def __init__(self, fail_after: int | None = None) -> None:
"""Optionally raise ``ConnectionError`` after ``fail_after`` calls."""
self.calls = 0
self.fail_after = fail_after
async def generate(
self, snapshot: ConversationSnapshot, generation: GenerationConfig
) -> GenerationRecord:
"""Answer from keywords; attach topic embeddings when asked to."""
self.calls += 1
if self.fail_after is not None and self.calls > self.fail_after:
msg = "model server went away"
raise ConnectionError(msg)
prompt = " ".join(message.text for message in snapshot.messages())
words = set(re.findall(r"[a-z]+", prompt.lower()))
topics = (
"refund" in words,
"receipt" in words,
bool(words & {"thirty", "days"}),
)
phrases = ("refund approved", "receipt checked", "deadline met")
reply = " ".join(p for p, on in zip(phrases, topics, strict=True) if on)
record = generation_record_from_text(reply or "please clarify")
if not generation.compute_embeddings:
return record
vector = (1.0, *(float(on) for on in topics))
return GenerationRecord(
text=record.text, text_token_rows=record.text_token_rows, embedding=vector
)
SYSTEM = (
"You are a support assistant for a store. Answer only from the question, in one "
'short sentence. If it does not give enough information, say "Not enough '
'information." Never invent policy details and never ask a question.'
)
snapshot = ConversationSnapshot.from_turns((
Turn(
messages=(
Message(role=Role.SYSTEM, text=SYSTEM),
Message(role=Role.USER, text="Refund within thirty days with receipt?"),
)
),
))
WORKDIR = Path(tempfile.mkdtemp(prefix="mllm-shap-archives-"))
config = explain_config_from_yaml(f"""
backend:
kind: mock
model_id: refund-bot
explanation:
estimator: complementary
budget:
fraction: 0.5
value_fn: tfidf_cosine
absence_policy: delete
archive:
path: {WORKDIR}/runs/{{run_id}}
on_exists: error
seed: 0
plugin_options:
tokens: {{roles: [user]}}
""")
def variant(**explanation: object) -> ExplainConfig:
"""Return ``config`` with some explanation keys replaced (validated)."""
payload = config.model_dump(mode="json")
payload["explanation"].update(explanation)
return ExplainConfig.model_validate(payload)
Write an archive¶
Business: The folder is self-describing: an auditor can open the manifest without mllm-shap installed and see which estimator, estimand, value function, model and package version produced the numbers.
Technical: An archived run holds manifest.json (schema version 4: run and snapshot ids, plugin names, player ids, the full configuration, and embedding_source, which records whether embedding value functions scored none, hash or backend embeddings), base_generation.txt and archive.sqlite with one row per evaluated coalition (its mask, generated text and token logprobs when the backend returns them). The database uses WAL journaling and user_version 4; RunArchive.load opens it read-only.
output = await ExplainRunner(config, backend=RefundBot()).explain(snapshot)
archive_root = WORKDIR / "runs" / output.run_id
print("run_id:", output.run_id)
print("files:", sorted(path.name for path in archive_root.iterdir()))
manifest = json.loads((archive_root / "manifest.json").read_text())
print({
key: value for key, value in manifest.items() if key not in ("config", "player_ids")
})
with closing(
sqlite3.connect(f"file:{archive_root / 'archive.sqlite'}?mode=ro", uri=True)
) as db:
print("journal_mode:", db.execute("PRAGMA journal_mode").fetchone()[0])
print("user_version:", db.execute("PRAGMA user_version").fetchone()[0])
with RunArchive.load(archive_root) as archive:
rows = list(archive.history_lazy())
print("coalition rows:", len(rows), " requested:", output.metrics.requested)
print("first row:", rows[0].generation_text, "| cache hit:", rows[0].cache_hit)
run_id: 84caf96b080cfbe6-5866a88b5be6
files: ['archive.sqlite', 'base_generation.txt', 'manifest.json']
{'absence_policy': 'delete', 'backend_kind': 'mock', 'embedding_model': '', 'embedding_source': 'none', 'estimand': 'shapley', 'estimator': 'complementary', 'mllm_shap_version': '0.6.0', 'model_id': 'refund-bot', 'normalizer': 'identity', 'players': 'tokens', 'run_id': '84caf96b080cfbe6-5866a88b5be6', 'schema_version': 4, 'snapshot_id': '84caf96b080cfbe6', 'value_fn': 'tfidf_cosine'}
journal_mode: wal
user_version: 4
coalition rows: 32 requested: 32
first row: refund approved receipt checked deadline met | cache hit: True
What the model answers. The archive stores one answer per sampled coalition. For orientation, here are the full prompt and the prompt with each user word deleted, as the stand-in answers them.
backend = RefundBot()
full = await backend.generate(snapshot, config.generation)
print(f"{'full prompt':>16} -> {full.text}")
for index, span in enumerate(output.player_set.spans):
mask = CoalitionMask.from_sequence(
tuple(other != index for other in range(output.player_set.num_players))
)
absent = DeletePolicy().apply(snapshot, output.player_set, mask)
answer = await backend.generate(absent, config.generation)
print(f"{'without ' + span.text:>16} -> {answer.text}")
full prompt -> refund approved receipt checked deadline met
without Refund -> receipt checked deadline met
without within -> refund approved receipt checked deadline met
without thirty -> refund approved receipt checked deadline met
without days -> refund approved receipt checked deadline met
without with -> refund approved receipt checked deadline met
without receipt? -> refund approved deadline met
Run the same explanation again¶
Business: An archive must never be silently mixed with a second run, and overwriting one should be an explicit decision.
Technical: The same configuration gives the same run_id and therefore the same folder. With archive.on_exists: error (the default) a second run raises ArchiveExistsError before any model call; on_exists: overwrite replaces the archive. The run id ignores the archive section, so switching on_exists keeps the folder.
try:
await ExplainRunner(config, backend=RefundBot()).explain(snapshot)
except ArchiveExistsError as exc:
print("ArchiveExistsError:", str(exc).replace(str(WORKDIR), "<workdir>"))
overwrite = variant(
archive={"path": config.explanation.archive.path, "on_exists": "overwrite"}
)
rerun = await ExplainRunner(overwrite, backend=RefundBot()).explain(snapshot)
print("same run_id:", rerun.run_id == output.run_id)
print("same values:", rerun.result.values == output.result.values)
ArchiveExistsError: run archive already exists at <workdir>/runs/84caf96b080cfbe6-5866a88b5be6; delete it or set explanation.archive.on_exists to 'overwrite'
same run_id: True
same values: True
reanalyze reproduces explain¶
Business: Reproducing a reported attribution from stored generations, with no model access, is the strongest audit evidence the package can give.
Technical: ExplainRunner(config).reanalyze(archive_root) reads the archive, fits and runs the value function on the archived text and lets the archived estimator re-aggregate the same coalitions. No backend is built: the metrics show zero executed calls and every coalition served from the archive. The result has no player_set (archives store player ids, not text), so its JSON labels carry ids only.
replayed = ExplainRunner(config).reanalyze(archive_root)
print("values identical to explain:", replayed.result.values == output.result.values)
print(replayed.metrics)
print("label 0:", replayed.to_dict()["labels"][0])
print("explain label 0:", output.to_dict()["labels"][0])
values identical to explain: True
SchedulerMetrics(requested=32, executed=0, deduplicated=0, cache_hits=32, kv_cache_hits=0, kv_cache_reused_tokens=0)
label 0: {'player_id': '84caf96b080cfbe6:0:1:0', 'text': None, 'message_index': None, 'start': None, 'end': None, 'speaker': None, 'label': None}
explain label 0: {'player_id': '84caf96b080cfbe6:0:1:0', 'text': 'Refund', 'message_index': 1, 'start': 0, 'end': 6, 'speaker': None, 'label': 'Refund'}
Rescore with another value function or normalizer¶
Business: choosing a value function is a modelling decision; comparing candidates on archived generations costs nothing.
Technical: reanalyze uses the runner’s current value_fn, normalizer and plugin_options, so one archive can be scored many ways. The archived run sampled half of the 64 coalitions with the complementary estimator, so even the tfidf_cosine row is an estimate: within and with get small non-zero values (exact enumeration gives 0). Archives store text only, so the embedding value functions score hash embeddings; those rows show that the path runs, not a meaningful attribution. Normalizers rescale for display: abs_sum divides by the sum of absolute values; min_max and power_shift (with its power option) shift to non-negative values that sum to one.
def rescored(**explanation: object) -> tuple[float, ...]:
"""Reanalyze the archive with some explanation keys replaced."""
replay = ExplainRunner(variant(**explanation)).reanalyze(archive_root)
return tuple(round(value, 3) for value in replay.result.values)
labels = [span.text for span in output.player_set.spans]
print(f"{'':22}", " ".join(f"{label:>8}" for label in labels))
for name, settings in (
("tfidf_cosine", {"value_fn": "tfidf_cosine"}),
("embedding_cosine", {"value_fn": "embedding_cosine"}),
("embedding_euclidean", {"value_fn": "embedding_euclidean"}),
("abs_sum", {"normalizer": "abs_sum"}),
("min_max", {"normalizer": "min_max"}),
(
"power_shift (power 2)",
{
"normalizer": "power_shift",
"plugin_options": {"power_shift": {"power": 2.0}},
},
),
):
print(f"{name:22}", " ".join(f"{value:8.3f}" for value in rescored(**settings)))
Refund within thirty days with receipt?
tfidf_cosine 0.273 -0.036 0.181 0.226 0.024 0.362
embedding_cosine 0.882 0.026 0.017 0.103 0.197 0.528
embedding_euclidean 0.297 0.033 0.012 0.090 0.028 0.319
abs_sum 0.248 -0.033 0.164 0.205 0.022 0.329
min_max 0.248 0.000 0.174 0.210 0.048 0.319
power_shift (power 2) 0.257 0.000 0.126 0.184 0.010 0.424
Archives reanalyze refuses¶
Business: A replay that silently mixes settings, or reads a half-written run, would produce numbers nobody can defend.
Technical: A different estimator, absence policy or partition than the archived run raises ReanalyzeMismatchError, and so does value_fn: logprob on rows without token logprobs (the bot returns none); a folder without manifest.json or archive.sqlite raises ArchiveNotFoundError (a FileNotFoundError); an unreadable manifest raises ArchiveCorruptError, and a manifest of another schema version (archives written by 0.1.18 or earlier) raises ManifestVersionError; a run that failed part-way leaves an archive marked unfinished, and reanalyze raises ArchiveIncompleteError. Running explain again replaces an unfinished archive without on_exists: overwrite.
for mismatch in ({"estimator": "mc"}, {"value_fn": "logprob"}):
try:
ExplainRunner(variant(**mismatch)).reanalyze(archive_root)
except ReanalyzeMismatchError as exc:
print("ReanalyzeMismatchError:", str(exc).replace(str(WORKDIR), "<workdir>"))
try:
ExplainRunner(config).reanalyze(WORKDIR / "missing")
except ArchiveNotFoundError as exc:
print("ArchiveNotFoundError:", str(exc).replace(str(WORKDIR), "<workdir>"))
damaged = WORKDIR / "damaged"
shutil.copytree(archive_root, damaged)
(damaged / "manifest.json").write_text("{not json")
try:
ExplainRunner(config).reanalyze(damaged)
except ArchiveCorruptError as exc:
print("ArchiveCorruptError:", str(exc).replace(str(WORKDIR), "<workdir>")[:110])
old = WORKDIR / "old_schema"
shutil.copytree(archive_root, old)
(old / "manifest.json").write_text(json.dumps({**manifest, "schema_version": 2}))
try:
ExplainRunner(config).reanalyze(old)
except ManifestVersionError as exc:
print("ManifestVersionError:", exc)
ReanalyzeMismatchError: estimator 'mc' differs from archived 'complementary'
ReanalyzeMismatchError: archive <workdir>/runs/84caf96b080cfbe6-5866a88b5be6 has coalition rows without token logprobs (backend 'mock'); value_fn 'logprob' cannot rescore it
ArchiveNotFoundError: archive manifest missing: <workdir>/missing/manifest.json
ArchiveCorruptError: cannot parse archive manifest <workdir>/damaged/manifest.json: Expecting property name enclosed in double quot
ManifestVersionError: unsupported manifest schema_version 2; expected 4. Re-run explain to rebuild the archive.
failing = variant(seed=1)
runs_before = {path.name for path in (WORKDIR / "runs").iterdir()}
try:
await ExplainRunner(failing, backend=RefundBot(fail_after=5)).explain(snapshot)
except ConnectionError as exc:
print("explain failed:", exc)
(new_run,) = {path.name for path in (WORKDIR / "runs").iterdir()} - runs_before
unfinished = WORKDIR / "runs" / new_run
try:
ExplainRunner(failing).reanalyze(unfinished)
except ArchiveIncompleteError as exc:
print("ArchiveIncompleteError:", str(exc).replace(str(WORKDIR), "<workdir>"))
retried = await ExplainRunner(failing, backend=RefundBot()).explain(snapshot)
replay = ExplainRunner(failing).reanalyze(unfinished)
print("retry replaced the unfinished archive:", replay.result == retried.result)
explain failed: model server went away
ArchiveIncompleteError: archive <workdir>/runs/84caf96b080cfbe6-eff16602657b belongs to a run that did not finish; re-run explain to rebuild it
retry replaced the unfinished archive: True
Backend embeddings and strict¶
Business: When an attribution depends on model embeddings that the archive does not store, a replay cannot reproduce it; the package says so instead of returning different numbers under the same name.
Technical: With value_fn: embedding_cosine the pipeline asks the backend for embeddings. The bot returns its own topic vectors, so the manifest records embedding_source: backend and the model id in embedding_model. reanalyze then scores hash embeddings of the archived text and logs a WARNING (shown here as one JSON line through configure_logging(level="WARNING", json=True)); with strict=True (mllm-shap reanalyze --strict) it raises ReanalyzeMismatchError instead.
configure_logging(level="WARNING", json=True)
embedded = variant(value_fn="embedding_cosine")
embedded_output = await ExplainRunner(embedded, backend=RefundBot()).explain(snapshot)
embedded_root = WORKDIR / "runs" / embedded_output.run_id
print(
"embedding_source:",
embedded_output.manifest.embedding_source.value,
"| embedding_model:",
embedded_output.manifest.embedding_model,
)
print("explain: ", [round(value, 3) for value in embedded_output.result.values])
loose = ExplainRunner(embedded).reanalyze(embedded_root)
print("reanalyze:", [round(value, 3) for value in loose.result.values])
try:
ExplainRunner(embedded).reanalyze(embedded_root, strict=True)
except ReanalyzeMismatchError as exc:
print("strict:", str(exc).replace(str(WORKDIR), "<workdir>"))
embedding_source: backend | embedding_model: refund-bot
explain: [0.119, -0.014, 0.096, 0.123, 0.009, 0.181]
reanalyze: [0.882, 0.026, 0.017, 0.103, 0.197, 0.528]
strict: archive <workdir>/runs/84caf96b080cfbe6-28fc7cfecd2f was scored with embeddings of backend model 'refund-bot'; reanalyze scores hash embeddings of the archived text, so 'embedding_cosine' values differ from explain
{"time": "2026-10-13 09:16:51,220", "level": "WARNING", "logger": "mllm_shap.pipeline.reanalyze", "message": "archive /var/folders/44/35090ghj57v176gf4tjwsvcr0000gn/T/mllm-shap-archives-qqgiscxn/runs/84caf96b080cfbe6-28fc7cfecd2f was scored with embeddings of backend model 'refund-bot'; reanalyze scores hash embeddings of the archived text, so 'embedding_cosine' values differ from explain"}