#!/usr/bin/env python3 """ Retrieval-provenance audit — runner. Asks four assistants the same questions, with web search enabled, and records every raw response to a JSONL file. It does not code the responses and it does not score them; coding is a separate, human step against the scheme in the protocol. Usage: export OPENAI_API_KEY=... ANTHROPIC_API_KEY=... GEMINI_API_KEY=... PERPLEXITY_API_KEY=... python3 run_audit.py --questions questions.json --out responses.jsonl --runs 3 python3 run_audit.py --systems openai,anthropic --questions questions.json --runs 1 python3 run_audit.py --dry-run # print what would be called, call nothing Every record written is the vendor's raw JSON plus the metadata needed to reproduce the call. Nothing is summarised at collection time — if you want to disagree with a coding decision, the untouched response is in the file. NOTE ON API SHAPES: these four vendors change their web-search parameters more often than they change their models. If a call fails with a 4xx, check the vendor's current documentation and adjust the builder function for that system; the rest of the harness does not care. `--dry-run` shows you the exact payloads. """ from __future__ import annotations import argparse import json import os import sys import time import urllib.error import urllib.request from datetime import datetime, timezone TIMEOUT = 180 # -------------------------------------------------------------------------------------- # Systems. `build` returns (url, headers, payload). `extract` pulls the answer text and # whatever citation list the vendor exposes, without interpreting either. # -------------------------------------------------------------------------------------- def _post(url: str, headers: dict, payload: dict) -> dict: data = json.dumps(payload).encode("utf-8") req = urllib.request.Request(url, data=data, headers=headers, method="POST") with urllib.request.urlopen(req, timeout=TIMEOUT) as resp: return json.loads(resp.read().decode("utf-8")) def build_openai(prompt: str, model: str) -> tuple[str, dict, dict]: return ( "https://api.openai.com/v1/responses", { "Authorization": f"Bearer {os.environ.get('OPENAI_API_KEY','')}", "Content-Type": "application/json", }, {"model": model, "input": prompt, "tools": [{"type": "web_search"}]}, ) def build_anthropic(prompt: str, model: str) -> tuple[str, dict, dict]: return ( "https://api.anthropic.com/v1/messages", { "x-api-key": os.environ.get('ANTHROPIC_API_KEY',''), "anthropic-version": "2023-06-01", "Content-Type": "application/json", }, { "model": model, "max_tokens": 2048, "messages": [{"role": "user", "content": prompt}], "tools": [{"type": "web_search_20250305", "name": "web_search", "max_uses": 8}], }, ) def build_gemini(prompt: str, model: str) -> tuple[str, dict, dict]: return ( f"https://generativelanguage.googleapis.com/v1beta/models/{model}:generateContent" f"?key={os.environ.get('GEMINI_API_KEY','')}", {"Content-Type": "application/json"}, { "contents": [{"parts": [{"text": prompt}]}], "tools": [{"google_search": {}}], }, ) def build_perplexity(prompt: str, model: str) -> tuple[str, dict, dict]: return ( "https://api.perplexity.ai/chat/completions", { "Authorization": f"Bearer {os.environ.get('PERPLEXITY_API_KEY','')}", "Content-Type": "application/json", }, {"model": model, "messages": [{"role": "user", "content": prompt}]}, ) def extract_openai(r: dict) -> dict: text, cites = [], [] for item in r.get("output", []): for part in item.get("content", []) or []: if part.get("type") in ("output_text", "text"): text.append(part.get("text", "")) for ann in part.get("annotations", []) or []: if ann.get("url"): cites.append({"title": ann.get("title"), "url": ann["url"]}) return {"text": "".join(text), "citations": cites} def extract_anthropic(r: dict) -> dict: text, cites = [], [] for block in r.get("content", []): if block.get("type") == "text": text.append(block.get("text", "")) for c in block.get("citations", []) or []: if c.get("url"): cites.append({"title": c.get("title"), "url": c["url"]}) return {"text": "".join(text), "citations": cites} def extract_gemini(r: dict) -> dict: text, cites = [], [] for cand in r.get("candidates", []): for part in cand.get("content", {}).get("parts", []) or []: if "text" in part: text.append(part["text"]) meta = cand.get("groundingMetadata", {}) or {} for chunk in meta.get("groundingChunks", []) or []: web = chunk.get("web", {}) or {} if web.get("uri"): cites.append({"title": web.get("title"), "url": web["uri"]}) return {"text": "".join(text), "citations": cites} def extract_perplexity(r: dict) -> dict: text = "".join(c.get("message", {}).get("content", "") for c in r.get("choices", [])) cites = [] for c in r.get("citations", []) or []: cites.append({"title": None, "url": c} if isinstance(c, str) else c) for c in r.get("search_results", []) or []: if c.get("url"): cites.append({"title": c.get("title"), "url": c["url"]}) return {"text": text, "citations": cites} SYSTEMS = { "openai": {"model": "gpt-4.1-mini", "key": "OPENAI_API_KEY", "build": build_openai, "extract": extract_openai}, "anthropic": {"model": "claude-sonnet-4-5", "key": "ANTHROPIC_API_KEY", "build": build_anthropic, "extract": extract_anthropic}, "gemini": {"model": "gemini-2.5-flash", "key": "GEMINI_API_KEY", "build": build_gemini, "extract": extract_gemini}, "perplexity": {"model": "sonar", "key": "PERPLEXITY_API_KEY", "build": build_perplexity, "extract": extract_perplexity}, } def _normalise(s: str) -> str: """Lowercase, fold 'percent'/'per cent' to '%', collapse whitespace. Applied identically to the answer text and to the discriminator string, so that 'Give the 40 percent figure' and the registered match '40%' meet in the middle. """ s = s.lower().replace(" ", " ").replace(" ", " ") s = s.replace(" per cent", "%").replace(" percent", "%").replace("percent", "%") return " ".join(s.split()) def flag_discriminators(text: str, discriminators: list[dict]) -> list[dict]: """Report which pre-registered strings are present. This is a lookup, not a judgement. The strings and what they diagnose were fixed in questions.json before the runs. A hit here is a prompt to a human coder, never a coding decision in itself: substring matching cannot tell '40%' in the study's result from '40%' in an unrelated aside, and it cannot see a figure that is described rather than stated. Every counted cell is coded by a human reading the full answer against section 6 of the protocol. """ norm = _normalise(text) return [d for d in discriminators if _normalise(d["match"]) in norm] def main() -> int: p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) p.add_argument("--questions", default="questions.json") p.add_argument("--out", default="responses.jsonl") p.add_argument("--runs", type=int, default=3, help="repeats per system per question") p.add_argument("--systems", default=",".join(SYSTEMS)) p.add_argument("--only", default="", help="comma-separated question ids, e.g. Q4,Q5") p.add_argument("--sleep", type=float, default=1.0) p.add_argument("--dry-run", action="store_true") a = p.parse_args() spec = json.load(open(a.questions, encoding="utf-8")) instruction = spec["instruction"] wanted = {q.strip() for q in a.only.split(",") if q.strip()} questions = [q for q in spec["questions"] if not wanted or q["id"] in wanted] systems = [s.strip() for s in a.systems.split(",") if s.strip()] for s in systems: if s not in SYSTEMS: print(f"unknown system: {s}. known: {', '.join(SYSTEMS)}", file=sys.stderr) return 2 if not a.dry_run and not os.environ.get(SYSTEMS[s]["key"]): print(f"missing environment variable {SYSTEMS[s]['key']} for {s}", file=sys.stderr) return 2 planned = len(questions) * len(systems) * a.runs print(f"{len(questions)} questions x {len(systems)} systems x {a.runs} runs = {planned} calls") if a.dry_run: print("dry run — nothing will be called\n") written = failed = 0 with open(a.out, "a", encoding="utf-8") as fh: for q in questions: prompt = f"{q['text']} {instruction}" for run_i in range(1, a.runs + 1): for name in systems: cfg = SYSTEMS[name] label = f"{q['id']} run {run_i} {name}" if a.dry_run: url, _, payload = cfg["build"](prompt, cfg["model"]) print(f"{label}\n POST {url.split('?')[0]}\n {json.dumps(payload)[:220]}\n") continue started = datetime.now(timezone.utc).isoformat() try: url, headers, payload = cfg["build"](prompt, cfg["model"]) raw = _post(url, headers, payload) parsed = cfg["extract"](raw) err = None except urllib.error.HTTPError as e: raw, parsed, err = None, None, f"HTTP {e.code}: {e.read()[:400]!r}" except Exception as e: # noqa: BLE001 raw, parsed, err = None, None, f"{type(e).__name__}: {e}" if err: failed += 1 print(f" FAIL {label} — {err}", file=sys.stderr) else: written += 1 hits = flag_discriminators(parsed["text"], q.get("discriminators", [])) note = " ".join(f"[{h['match']} -> {h['diagnoses']}]" for h in hits) print(f" ok {label} — {len(parsed['citations'])} citations {note}") fh.write(json.dumps({ "question_id": q["id"], "question_class": q["class"], "prompt": prompt, "system": name, "model_requested": cfg["model"], "run": run_i, "started_utc": started, "surface": "api", "error": err, "answer_text": parsed["text"] if parsed else None, "citations": parsed["citations"] if parsed else None, "discriminator_hits": flag_discriminators( parsed["text"], q.get("discriminators", [])) if parsed else None, "raw_response": raw, }, ensure_ascii=False) + "\n") fh.flush() time.sleep(a.sleep) if not a.dry_run: print(f"\n{written} written, {failed} failed -> {a.out}") print("Responses are uncoded. Code them by hand against section 6 of the protocol.") return 1 if failed and not written else 0 if __name__ == "__main__": raise SystemExit(main())