"""Example (a): the "refund policy" pipeline. Planner -> Retriever -> Writer -> Reviewer. The user asks about the refund window. The knowledge base holds a stale 2023 policy (90 days) and the current 2026 policy (30 days). As run, the pipeline answers "90 days". * Planner fix: adds an "effective 2026" constraint to the query, so even the as-run keyword retriever finds the current doc. * Retriever fix: drops superseded documents, whatever the query. * Writer fix: cites the source document for every claim. * Reviewer fix (mechanism A, "cited-only"): re-checks each *cited* claim against the live policy registry and corrects it. It cannot check uncited claims, so it only helps together with the Writer fix. * Reviewer fix (mechanism B, "checks-all"): checks every claim about refund windows against the registry, cited or not. Under A: v(S) = 1 iff Planner in S or Retriever in S or {Writer, Reviewer} in S, whose Shapley values are 5/12, 5/12, 1/12, 1/12. Under B the Reviewer alone suffices and the Writer becomes a null player: 1/3, 1/3, 0, 1/3. Run: ``python -m agentblame.demos.refund_policy [--out DIR]`` """ from __future__ import annotations import argparse import os import re from agentblame import (Case, Pipeline, analyze_case, format_case, step, write_html) DOCS = [ {"id": "policy-2023", "year": 2023, "superseded": True, "text": "Refund policy: refund policy allows refunds within 90 days of purchase."}, {"id": "policy-2026", "year": 2026, "superseded": False, "text": "Refunds are accepted within 30 days of purchase (effective 2026)."}, {"id": "shipping-2026", "year": 2026, "superseded": False, "text": "Shipping: orders ship within 2 business days."}, ] REGISTRY = {"refund_window_days": 30, "source": "policy-2026"} # source of truth def _days(text: str) -> int | None: m = re.search(r"(\d+) days", text) return int(m.group(1)) if m else None # --- Planner --------------------------------------------------------------- @step("planner") def planner(question): return {"keywords": ["refund", "policy"], "year": None} @planner.variant("adds_date_constraint") def planner_fixed(question): return {"keywords": ["refund", "policy"], "year": 2026} # --- Retriever ------------------------------------------------------------- def _keyword_score(doc, plan): t = doc["text"].lower() return sum(t.count(k) for k in plan["keywords"]) @step("retriever") def retriever(planner): docs = DOCS if planner["year"] is None else [d for d in DOCS if d["year"] == planner["year"]] return max(docs, key=lambda d: _keyword_score(d, planner)) @retriever.variant("filters_superseded") def retriever_fixed(planner): docs = [d for d in DOCS if not d["superseded"]] return max(docs, key=lambda d: _keyword_score(d, planner)) # --- Writer ---------------------------------------------------------------- @step("writer") def writer(question, retriever): return {"claims": [{"text": f"Refunds are accepted within {_days(retriever['text'])} days.", "cite": None}]} @writer.variant("cites_sources") def writer_fixed(question, retriever): return {"claims": [{"text": f"Refunds are accepted within {_days(retriever['text'])} days.", "cite": retriever["id"]}]} # --- Reviewer -------------------------------------------------------------- @step("reviewer") def reviewer(writer): return " ".join(c["text"] for c in writer["claims"]) # rubber stamp def _correct(claim): days = _days(claim["text"]) if days is not None and days != REGISTRY["refund_window_days"]: return {"text": claim["text"].replace(f"{days} days", f"{REGISTRY['refund_window_days']} days"), "cite": REGISTRY["source"]} return claim @reviewer.variant("checks_cited_claims") def reviewer_cited(writer): claims = [_correct(c) if c["cite"] else c for c in writer["claims"]] return " ".join(c["text"] for c in claims) @reviewer.variant("checks_all_claims", fix=False) def reviewer_all(writer): return " ".join(_correct(c)["text"] for c in writer["claims"]) def metric(out) -> float: return 1.0 if _days(out["reviewer"]) == 30 else 0.0 def build_pipeline() -> Pipeline: return Pipeline([planner, retriever, writer, reviewer], metric, name="refund_policy") CASE = Case("refund-q1", {"question": "How many days do I have to request a refund?"}, seed=0) FIXES_A = {"planner": "adds_date_constraint", "retriever": "filters_superseded", "writer": "cites_sources", "reviewer": "checks_cited_claims"} FIXES_B = dict(FIXES_A, reviewer="checks_all_claims") def main(argv=None): ap = argparse.ArgumentParser(description=__doc__.splitlines()[0]) ap.add_argument("--out", default="agentblame_out") args = ap.parse_args(argv) pipe = build_pipeline() fact = pipe.run(CASE.inputs) print(f"Factual answer: {fact.outputs['reviewer']!r} (score {fact.score:g})\n") a = analyze_case(pipe, CASE, fixes=FIXES_A, method="exact") print("Mechanism A -- Reviewer only checks *cited* claims") print(format_case(a)) print(" expected: planner 5/12=0.417, retriever 5/12, writer 1/12=0.083, reviewer 1/12\n") b = analyze_case(pipe, CASE, fixes=FIXES_B, method="exact") print("Mechanism B -- Reviewer checks *all* claims") print(format_case(b)) print(" expected: planner 1/3, retriever 1/3, writer 0, reviewer 1/3\n") os.makedirs(args.out, exist_ok=True) path = write_html( os.path.join(args.out, "refund_policy.html"), "Refund-policy pipeline: who caused the stale answer?", [("Mechanism A: Reviewer checks only cited claims", a), ("Mechanism B: Reviewer checks all claims", b)], intro="Planner → Retriever → Writer → Reviewer. The as-run pipeline answers '90 days' " "from a superseded policy; the correct answer is 30 days.", notes=["Planner and Retriever are each sufficient on their own (over-determination), so " "leave-one-out blames neither; fix-alone blames only them; last-step-to-touch blames " "the Reviewer. Shapley gives the redundant causes 5/12 each and splits the remaining " "1/6 between Writer and Reviewer, which only work together.", "Changing how the Reviewer works (mechanism B) changes the blame: Writer becomes a " "null player and the Reviewer joins the set of sufficient fixes."]) print(f"HTML report: {path}") return a, b if __name__ == "__main__": main()