#!/usr/bin/env python3 """Run prompt regression cases against the Anthropic Messages API. Usage: python3 scripts/prompt_regress.py evals/cases.jsonl [--out results.jsonl] [--only ID ...] [--max-tokens N] Environment variables: ANTHROPIC_API_KEY required. Read by the anthropic SDK; never pass it on the command line. ANTHROPIC_MODEL required. A current model ID from Anthropic's models page. There is no default, so a model change is always an explicit diff. Requirements: Python 3.8+ and the official SDK: pip install anthropic Case file format (JSON Lines, one case per line): {"id": "refund-policy", required, unique "input": "Customer message text", required, sent as the user turn "system": "System prompt text", optional "system_file": "prompts/support.md", optional, path relative to the case file "max_tokens": 512, optional, default from --max-tokens "checks": { "contains": ["refund"], substrings that must appear (case-insensitive) "not_contains": ["guarantee"], substrings that must not appear (case-insensitive) "regex": ["\\b30 days\\b"], Python regexes that must match somewhere "max_chars": 800, upper bound on reply length "json": true, reply must parse as JSON (code fences are stripped) "json_keys": ["category"] top-level keys the JSON object must contain }} Blank lines and lines starting with // are ignored. Exit codes: 0 every case passed 1 one or more cases failed a check 2 usage or configuration error (bad case file, missing env var, SDK missing) 3 API error (authentication, rate limit after retries, server error) """ import argparse import json import os import re import sys import time def load_cases(path): cases, seen = [], set() base = os.path.dirname(os.path.abspath(path)) with open(path, encoding="utf-8") as handle: for number, line in enumerate(handle, start=1): stripped = line.strip() if not stripped or stripped.startswith("//"): continue try: case = json.loads(stripped) except json.JSONDecodeError as exc: raise ValueError("line %d: invalid JSON: %s" % (number, exc.msg)) if not isinstance(case, dict) or not case.get("id") or not case.get("input"): raise ValueError("line %d: each case needs non-empty 'id' and 'input'" % number) if case["id"] in seen: raise ValueError("line %d: duplicate id %r" % (number, case["id"])) seen.add(case["id"]) if case.get("system_file"): with open(os.path.join(base, case["system_file"]), encoding="utf-8") as sys_handle: case["system"] = sys_handle.read() for pattern in case.get("checks", {}).get("regex", []): re.compile(pattern) cases.append(case) if not cases: raise ValueError("no cases found in %s" % path) return cases def strip_fences(text): match = re.match(r"^\s*```[a-zA-Z0-9_-]*\s*\n(.*?)\n\s*```\s*$", text, re.DOTALL) return match.group(1) if match else text def run_checks(text, checks): failures = [] lower = text.lower() for needle in checks.get("contains", []): if needle.lower() not in lower: failures.append("missing text %r" % needle) for needle in checks.get("not_contains", []): if needle.lower() in lower: failures.append("forbidden text %r present" % needle) for pattern in checks.get("regex", []): if not re.search(pattern, text): failures.append("regex %r did not match" % pattern) if "max_chars" in checks and len(text) > checks["max_chars"]: failures.append("reply is %d chars (max %d)" % (len(text), checks["max_chars"])) if checks.get("json") or checks.get("json_keys"): try: parsed = json.loads(strip_fences(text)) except json.JSONDecodeError as exc: failures.append("reply is not valid JSON: %s" % exc.msg) else: missing = [k for k in checks.get("json_keys", []) if not isinstance(parsed, dict) or k not in parsed] if missing: failures.append("JSON missing keys: %s" % ", ".join(missing)) return failures def main(): parser = argparse.ArgumentParser(description="Prompt regression runner (Anthropic API).") parser.add_argument("cases") parser.add_argument("--out", help="write one JSON result per case to this file") parser.add_argument("--only", action="append", default=[], metavar="ID") parser.add_argument("--max-tokens", type=int, default=1024) args = parser.parse_args() model = os.environ.get("ANTHROPIC_MODEL", "").strip() if not model: print("error: set ANTHROPIC_MODEL to the model ID you want to test", file=sys.stderr) return 2 if not os.environ.get("ANTHROPIC_API_KEY"): print("error: ANTHROPIC_API_KEY is not set", file=sys.stderr) return 2 try: import anthropic except ImportError: print("error: the anthropic package is not installed (pip install anthropic)", file=sys.stderr) return 2 try: cases = load_cases(args.cases) except (OSError, ValueError, re.error) as exc: print("error: %s" % exc, file=sys.stderr) return 2 if args.only: cases = [c for c in cases if c["id"] in set(args.only)] if not cases: print("error: no cases match --only", file=sys.stderr) return 2 client = anthropic.Anthropic(max_retries=3) results, failed = [], 0 out = open(args.out, "w", encoding="utf-8") if args.out else None try: for case in cases: request = { "model": model, "max_tokens": case.get("max_tokens", args.max_tokens), "messages": [{"role": "user", "content": case["input"]}], } if case.get("system"): request["system"] = case["system"] started = time.monotonic() try: response = client.messages.create(**request) except anthropic.APIError as exc: print("error: API call failed on case %r: %s" % (case["id"], exc), file=sys.stderr) return 3 text = "".join(block.text for block in response.content if block.type == "text") failures = run_checks(text, case.get("checks", {})) if response.stop_reason == "max_tokens": failures.append("reply was cut off at max_tokens") status = "FAIL" if failures else "PASS" failed += bool(failures) print("%s %s (%.1fs, %d in / %d out tokens)" % ( status, case["id"], time.monotonic() - started, response.usage.input_tokens, response.usage.output_tokens)) for failure in failures: print(" - %s" % failure) record = {"id": case["id"], "status": status, "failures": failures, "model": model, "stop_reason": response.stop_reason, "output": text, "input_tokens": response.usage.input_tokens, "output_tokens": response.usage.output_tokens} results.append(record) if out: out.write(json.dumps(record, ensure_ascii=False) + "\n") finally: if out: out.close() print("\n%d/%d passed on %s" % (len(results) - failed, len(results), model)) return 1 if failed else 0 if __name__ == "__main__": sys.exit(main())