"""Export the replay data behind the browser demo on the project page. For each bundled scenario and each failing case this runs the real pipeline under every coalition of fixes (the same BlameGame the package uses), and writes, per coalition: v(S), and every step's output in the reference world (what the page shows on the step cards). It also stores the package's own Shapley / fix-alone / leave-one-out / last-touch results, so the page's JavaScript can be checked against them. python examples/export_web_demo.py # writes demo/data.json """ from __future__ import annotations import json import os import sys HERE = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, os.path.dirname(HERE)) sys.path.insert(0, HERE) from agentblame import NoiseSource, analyze_case, blame_batch # noqa: E402 from agentblame.attribution import make_game # noqa: E402 from agentblame.pipeline import _fingerprint # noqa: E402 from agentblame.demos import eight_step, refund_policy, stochastic_rag # noqa: E402 OUT = os.path.join(os.path.dirname(HERE), "demo", "data.json") class Strings: """Interned string table, so repeated outputs are stored once.""" def __init__(self): self.items: list[str] = [] self.index: dict[str, int] = {} def __call__(self, s: str) -> int: if s not in self.index: self.index[s] = len(self.items) self.items.append(s) return self.index[s] def export_game(game, show, S, py): """All 2^n coalitions: v(S), shown output and fingerprint of every step.""" n = game.n steps = game.pipeline.steps v, out, fp = [], [], [] for m in range(1 << n): v.append(game.value(m)) tr = game.trace(m) out.append([S(show(s.name, tr.outputs[s.output])) for s in steps]) fp.append([_fingerprint(tr.outputs[s.output]) for s in steps]) # Last-touch compares fingerprints (exact output equality). Ship them only # for steps where the shown text is not a faithful stand-in for them. need = [any((out[a][j] == out[b][j]) != (fp[a][j] == fp[b][j]) for a in range(1 << n) for b in range(a)) for j in range(len(steps))] fps = {steps[j].name: [S(fp[m][j]) for m in range(1 << n)] for j in range(len(steps)) if need[j]} sh = py.shapley res = {"v": v, "out": out, "py": {"shapley": [sh.values[p] for p in game.players], "fix_alone": [py.fix_alone[p] for p in game.players], "loo": [py.leave_one_out[p] for p in game.players], "last_touch": py.last_touch}} if fps: res["fp"] = fps return res # --- how each scenario's step outputs are shown on the cards ---------------- def show_refund(step, x): if step == "planner": return "search: refund policy" + (f", effective {x['year']}" if x["year"] else "") if step == "retriever": return f"{x['id']}: “{x['text']}”" if step == "writer": c = x["claims"][0] return c["text"] + (f" [cites {c['cite']}]" if c["cite"] else " [no citation]") return x def show_rag(step, x): return str(x) def show_eight(step, x): if x is None: return "nothing (searched the wrong index)" if step != "intake" else "-" if step == "planner": return f"metric travel, year {x['fiscal_year'] or 'not set'}" if step == "search": return f"{x} table" if step == "extractor": return f"{x['fy']}: before {x['prev']}, after {x['cur']}" + (" (unit error)" if x["cur"] > 10000 else "") if step == "calculator": return f"{x}%" if step == "drafter": return x["text"] + (" (cited)" if x["cited"] else " (uncited)") return str(x) def main(): S = Strings() scenarios = [] # (a) refund policy: one deterministic case, two choices of reviewer fix pipe = refund_policy.build_pipeline() case = refund_policy.CASE games = [] for key, fixes, label in (("A", refund_policy.FIXES_A, "checks cited claims"), ("B", refund_policy.FIXES_B, "checks all claims")): py = analyze_case(pipe, case, fixes=fixes, method="exact") g = make_game(pipe, case, fixes=fixes) games.append(dict(export_game(g, show_refund, S, py), id=key, label=label)) scenarios.append({"id": "refund", "players": g.players, "steps": [s.name for s in pipe.steps], "games": games}) # (b) stochastic RAG: every failing case of the 200, three noise modes pipe = stochastic_rag.build_pipeline() cases = stochastic_rag.make_cases(200) games, batch_py = [], {} modes = (("recorded", 1), ("posterior", 8), ("prior", 8)) for mode, nw in modes: b = blame_batch(pipe, cases, mode=mode, n_worlds=nw) batch_py[mode] = {s.name: {"shapley": s.shapley, "fix_alone": s.fix_alone, "share": s.share, "last_touch": s.last_touch} for s in b.steps} for cb in b.cases: c = next(c for c in cases if c.id == cb.case_id) g = make_game(pipe, c, mode=mode, n_worlds=nw) e = export_game(g, show_rag, S, cb) assert all(abs(a - b_) == 0 for a, b_ in zip(e["v"], cb.coalition_values.values())) games.append(dict(e, id=c.id, mode=mode, difficulty=c.inputs["difficulty"])) scenarios.append({"id": "rag", "players": g.players, "steps": [s.name for s in pipe.steps], "games": games, "batch_py": batch_py, "n_cases": len(cases)}) # (c) eight-step finance agent: the first 12 cases (all of them fail as run) pipe = eight_step.build_pipeline() games = [] for c in eight_step.make_cases(12): if pipe.run(c.inputs, noise=NoiseSource(c.seed)).score >= 1: continue py = analyze_case(pipe, c, method="exact") g = make_game(pipe, c) games.append(dict(export_game(g, show_eight, S, py), id=c.id)) scenarios.append({"id": "eight", "players": g.players, "steps": [s.name for s in pipe.steps], "games": games}) # (d) quickstart: 3 steps; the "format" fix changes nothing (a null player) import quickstart pipe = quickstart.pipe games = [] for c in quickstart.cases: if pipe.run(c.inputs, noise=NoiseSource(c.seed)).score >= 1: continue py = analyze_case(pipe, c, method="exact") g = make_game(pipe, c) games.append(dict(export_game(g, lambda s, x: "None" if x is None else str(x), S, py), id=c.id, question=list(c.inputs["question"]))) scenarios.append({"id": "quick", "players": g.players, "steps": [s.name for s in pipe.steps], "games": games}) os.makedirs(os.path.dirname(OUT), exist_ok=True) with open(OUT, "w") as f: json.dump({"strings": S.items, "scenarios": scenarios}, f, separators=(",", ":")) print(f"wrote {OUT}: {os.path.getsize(OUT) / 1024:.0f} KB, " + ", ".join(f"{s['id']} {len(s['games'])} games" for s in scenarios)) if __name__ == "__main__": main()