#!/usr/bin/env python3 # ============================================================================= # OTAP: one-run experiment suite for # "OTAP: Structure-Aware Optimal Transport for Evaluating the Planning and # Execution Trajectories of LLM Agents" (ICTAI 2026 submission) # # WHAT THIS PRODUCES (single run): # Table A : perturbation meta-evaluation -- pairwise ranking accuracy (PRA) # per perturbation family P1..P5, severity ordering (P4 > P5), # and valid-vs-invalid AUROC, for every baseline metric and OTAP. # Table B : correlation (Spearman rho / Kendall tau) of every metric with # a constructed graded-quality ladder, overall and damaged-only. # (Optionally with REAL human ratings via --human-ratings.) # Table C : ablations of OTAP components (structure, sign, unbalancedness, # transport vs Hungarian, fielded cost, tool term, #references, # soft-min temperature, causal mass, edge extraction). # Runtime : mean wall-clock per evaluation for every metric. # # All results are printed as markdown AND written to otap_results.json. # Run: # pip install numpy scipy sentence-transformers # torch pulled in # python otap_experiments.py # full run # python otap_experiments.py --quick # fast smoke test # python otap_experiments.py --embedder hash # no-download debug # ANTHROPIC_API_KEY=... python otap_experiments.py --judge # + LLM judge # python otap_experiments.py --human-ratings ratings.csv # real Part B # # The corpus is a curated multi-domain plan corpus (20 tasks, 6 steps each, # realistic tools and dependency DAGs). The perturbation generator and all # metric implementations are below and fully deterministic (fixed seed). # ============================================================================= import argparse, hashlib, itertools, json, math, os, random, re, sys, time from collections import defaultdict from dataclasses import dataclass, field, replace import numpy as np from scipy.optimize import linear_sum_assignment from scipy.stats import spearmanr, kendalltau SEED = 20260706 # ============================================================================= # 1. DATA MODEL # ============================================================================= @dataclass class Step: action: str tool: str # tool name or "none" for pure reasoning context: str effect: str uid: frozenset = frozenset() # artifact tokens this step PRODUCES consumes: frozenset = frozenset() # artifact tokens this step CONSUMES def text(self): return f"{self.action}. {self.context}. -> {self.effect}" @dataclass class Plan: task_id: str goal: str steps: list label: str = "original" # perturbation family / reference name def n(self): return len(self.steps) # ---- graph extraction (artifact tracing) ------------------------------------ def trace_edges(plan): """Edge i->k iff step i (earlier in the executed sequence) produces an artifact that step k consumes. If a producer appears AFTER its consumer (causal inversion) the edge is simply absent: the dependency was not satisfiable at execution time.""" edges = [] for k, sk in enumerate(plan.steps): for i in range(k): if plan.steps[i].uid & sk.consumes: edges.append((i, k)) return edges def total_order_edges(plan): """Conservative fallback: chain of execution order.""" return [(i, i + 1) for i in range(plan.n() - 1)] def signed_dep_matrix(plan, edge_mode="trace"): """Signed, height-normalised dependency-distance matrix D in [-1,1]. D[i,k] = +hop(i,k)/H if i is an ancestor of k, -hop(k,i)/H if descendant, 0 if incomparable.""" n = plan.n() edges = trace_edges(plan) if edge_mode == "trace" else total_order_edges(plan) INF = 10 ** 9 hop = np.full((n, n), INF, dtype=float) for i in range(n): hop[i, i] = 0 for (i, k) in edges: hop[i, k] = min(hop[i, k], 1) for m in range(n): # Floyd-Warshall (n<=15) hop = np.minimum(hop, hop[:, m:m + 1] + hop[m:m + 1, :]) finite = hop[hop < INF] H = max(finite.max(), 1.0) D = np.zeros((n, n)) for i in range(n): for k in range(n): if i == k: continue if hop[i, k] < INF: D[i, k] = hop[i, k] / H elif hop[k, i] < INF: D[i, k] = -hop[k, i] / H return D def critical_mask(plan): """crit(j)=1 if removing step j disconnects the final step from all source steps (graph-theoretic proxy for counterfactual necessity).""" n = plan.n() edges = trace_edges(plan) goal = n - 1 def reachable_goal(removed): adj = defaultdict(list) for (i, k) in edges: if i != removed and k != removed: adj[i].append(k) indeg_zero = [i for i in range(n) if i != removed and not any(e[1] == i and e[0] != removed for e in edges)] seen = set() stack = list(indeg_zero) while stack: u = stack.pop() if u in seen: continue seen.add(u) stack.extend(adj[u]) return goal in seen crit = np.zeros(n) for j in range(n - 1): if not reachable_goal(j): crit[j] = 1.0 crit[goal] = 1.0 return crit # ============================================================================= # 2. CORPUS (20 tasks x 6 steps; deps as lists of producing-step indices) # ============================================================================= def _mk(task_id, goal, rows): steps = [] for idx, (a, t, c, e, deps) in enumerate(rows): steps.append(Step(a, t, c, e, uid=frozenset({f"{task_id}#{idx}"}), consumes=frozenset({f"{task_id}#{d}" for d in deps}))) return Plan(task_id, goal, steps, "original") CORPUS = [ _mk("kyoto_hotel", "Book a hotel in Kyoto under $150 per night near Kyoto Station.", [ ("Search hotels", "web_search", "hotels in Kyoto for target dates", "candidate hotel list", []), ("Filter by price", "browser", "under $150 per night", "price-filtered list", [0]), ("Filter by location", "browser", "near Kyoto Station", "location-filtered list", [0]), ("Rank remaining options", "browser", "by rating and review count", "ranked shortlist", [1, 2]), ("Select the top hotel", "browser", "best ranked option", "chosen hotel", [3]), ("Book the room", "browser", "dates and payment details", "confirmed reservation", [4])]), _mk("flight_search", "Find and book the cheapest flight from NYC to London next month.", [ ("Search direct flights", "web_search", "NYC to London on target dates", "direct flight options", []), ("Search flexible dates", "web_search", "plus or minus two days", "flexible flight options", []), ("Merge and compare results", "python", "combine both search results", "comparison table", [0, 1]), ("Remove expensive options", "python", "fares above budget threshold", "affordable shortlist", [2]), ("Select the cheapest flight", "browser", "lowest total fare with fees", "selected flight", [3]), ("Book the ticket", "browser", "passenger and payment details", "booking confirmation", [4])]), _mk("sales_report", "Produce a quarterly sales report with charts from the raw CSV.", [ ("Download the sales data", "filesystem", "Q3 raw CSV export", "raw dataset", []), ("Clean and deduplicate records", "python", "remove nulls and duplicate rows", "clean dataset", [0]), ("Aggregate revenue by region", "python", "group by region and month", "summary table", [1]), ("Create revenue charts", "python", "bar and line charts", "chart images", [2]), ("Draft the report narrative", "none", "key findings and trends", "report draft", [2, 3]), ("Compile the final report", "filesystem", "narrative plus charts", "final report file", [3, 4])]), _mk("lit_review", "Write a short literature review on graph neural networks for drug discovery.", [ ("Search recent papers", "web_search", "graph neural networks for drug discovery", "paper list", []), ("Filter by relevance and venue", "none", "top venues in the last three years", "paper shortlist", [0]), ("Read abstracts and take notes", "browser", "shortlisted papers", "reading notes", [1]), ("Group papers by method", "none", "message passing versus transformers", "thematic groups", [2]), ("Summarize each group", "none", "key contributions and open gaps", "section summaries", [3]), ("Write the review", "filesystem", "assemble sections with citations", "review document", [4])]), _mk("bug_fix", "Diagnose and fix the failing unit test in the payments module.", [ ("Reproduce the failure", "python", "run the failing test locally", "error trace", []), ("Read the stack trace", "none", "identify the failing assertion", "fault location", [0]), ("Inspect recent commits", "filesystem", "changes touching the payments module", "suspect change", [1]), ("Write a regression test", "python", "capture the expected behavior", "new test case", [1]), ("Patch the defect", "filesystem", "correct the rounding logic", "code fix", [2, 3]), ("Run the full test suite", "python", "verify no regressions", "passing tests", [4])]), _mk("grocery_budget", "Plan a week of dinners under a $70 grocery budget.", [ ("Choose seven dinner recipes", "web_search", "quick weeknight meals", "recipe list", []), ("Extract and consolidate ingredients", "none", "merge lists across recipes", "ingredient list", [0]), ("Check pantry stock", "none", "remove items already owned", "shopping list", [1]), ("Price items at the local store", "browser", "current shelf prices", "priced list", [2]), ("Adjust recipes to fit the budget", "calculator", "swap costly ingredients", "final meal plan", [3]), ("Compile the shopping list", "filesystem", "organized by store aisle", "final shopping list", [4])]), _mk("db_migration", "Migrate the users table to the new schema without downtime.", [ ("Snapshot the current database", "sql", "backup of the users table", "backup snapshot", []), ("Create the new schema", "sql", "users_v2 with new columns", "empty new table", []), ("Write the migration script", "python", "transform and copy rows", "migration script", [1]), ("Dry-run on staging", "python", "validate row counts and types", "staging validation", [0, 2]), ("Run the migration in production", "sql", "execute with monitoring", "migrated table", [3]), ("Verify integrity and switch traffic", "sql", "checksums and cutover", "completed migration", [4])]), _mk("conference_trip", "Arrange travel and lodging for the Boston conference.", [ ("Check the conference dates", "browser", "official conference site", "confirmed dates", []), ("Book the flights", "browser", "arrival before the keynote", "flight booking", [0]), ("Book a hotel near the venue", "browser", "walking distance to venue", "hotel booking", [0]), ("Register for the conference", "browser", "early-bird rate", "registration receipt", [0]), ("Add the itinerary to the calendar", "calendar", "flights hotel and sessions", "calendar entries", [1, 2, 3]), ("Submit expense preapproval", "email", "estimated costs to manager", "approval request", [1, 2, 3])]), _mk("ab_test", "Analyze the A/B test and recommend whether to ship variant B.", [ ("Pull the experiment data", "sql", "conversions by variant", "raw experiment data", []), ("Check sample ratio and health", "python", "guardrail metrics", "sanity report", [0]), ("Compute lift and confidence interval", "python", "conversion rate delta", "statistical results", [1]), ("Segment results by platform", "python", "mobile versus desktop", "segment breakdown", [1]), ("Interpret the findings", "none", "practical significance", "recommendation draft", [2, 3]), ("Write the recommendation memo", "email", "ship or hold decision", "final memo", [4])]), _mk("onboarding_doc", "Create an onboarding guide for new engineers.", [ ("Interview two senior engineers", "email", "common newcomer questions", "interview notes", []), ("Collect existing wiki pages", "filesystem", "setup and process docs", "doc inventory", []), ("Identify gaps and outdated content", "none", "compare notes against docs", "gap list", [0, 1]), ("Draft the guide structure", "none", "sections and ordering", "guide outline", [2]), ("Write and edit the sections", "filesystem", "fill the outline with content", "draft guide", [3]), ("Review with a new hire", "email", "usability feedback", "revised guide", [4])]), _mk("tax_prep", "Prepare documents for the annual tax filing.", [ ("Gather income statements", "filesystem", "W-2 and 1099 forms", "income documents", []), ("Collect deduction receipts", "filesystem", "charitable and medical receipts", "deduction records", []), ("Total income and deductions", "calculator", "sum all categories", "computed totals", [0, 1]), ("Check eligibility for credits", "web_search", "current year credit rules", "applicable credits", [2]), ("Fill the tax forms", "filesystem", "enter totals and credits", "completed forms", [3]), ("Review and submit the filing", "browser", "e-file with confirmation", "submission receipt", [4])]), _mk("launch_email", "Draft and send the product launch announcement.", [ ("Summarize the key features", "none", "top three new capabilities", "feature summary", []), ("Draft the announcement copy", "none", "subject line and body", "email draft", [0]), ("Add pricing and availability", "none", "tiers and launch date", "complete draft", [1]), ("Get legal and brand approval", "email", "compliance review", "approved copy", [2]), ("Build the recipient segment", "sql", "active customers last 90 days", "recipient list", []), ("Schedule and send the campaign", "email", "launch morning send time", "sent campaign", [3, 4])]), _mk("house_search", "Shortlist three apartments to visit this weekend.", [ ("Define requirements and budget", "none", "two bedrooms under $2500", "search criteria", []), ("Search the listing sites", "web_search", "listings matching the criteria", "candidate listings", [0]), ("Filter by commute time", "browser", "under 30 minutes to the office", "commute-filtered list", [1]), ("Filter by amenities", "browser", "in-unit laundry and parking", "amenity-filtered list", [1]), ("Cross-check building reviews", "web_search", "management company ratings", "vetted list", [2, 3]), ("Schedule three viewings", "email", "weekend time slots", "confirmed viewings", [4])]), _mk("data_pipeline", "Set up a nightly pipeline to sync orders into the warehouse.", [ ("Inspect the source API", "api_call", "orders endpoint schema", "API schema notes", []), ("Design the warehouse table", "sql", "orders fact table schema", "table definition", [0]), ("Write the extraction job", "python", "paginate and fetch orders", "extraction script", [0]), ("Write the load and transform step", "python", "upsert into the warehouse", "load script", [1, 2]), ("Schedule the nightly run", "api_call", "2am cron trigger", "scheduled job", [3]), ("Add failure alerts", "api_call", "notify the team on error", "alerting rule", [3])]), _mk("billing_ticket", "Resolve the customer's billing dispute ticket.", [ ("Read the ticket and history", "browser", "prior contacts and invoices", "case summary", []), ("Pull the billing records", "sql", "charges for the disputed period", "billing data", [0]), ("Identify the discrepancy", "none", "locate the duplicate charge", "root cause", [1]), ("Issue the refund", "api_call", "refund the duplicate charge", "refund confirmation", [2]), ("Update the ticket notes", "browser", "document the resolution", "updated ticket", [3]), ("Reply to the customer", "email", "apology and confirmation", "sent reply", [3])]), _mk("blog_publish", "Publish a technical blog post about the caching redesign.", [ ("Outline the post", "none", "problem approach and results", "post outline", []), ("Write the first draft", "filesystem", "expand the outline", "draft post", [0]), ("Create the architecture diagram", "python", "before and after design", "diagram image", [0]), ("Run the technical review", "email", "accuracy check by the team", "reviewed draft", [1, 2]), ("Edit for clarity and style", "none", "tighten the prose", "final draft", [3]), ("Publish and share the post", "browser", "company blog and social", "published post", [4])]), _mk("restock", "Decide restock quantities for the top 20 products.", [ ("Export sales velocity data", "sql", "last 90 days by SKU", "velocity data", []), ("Export current stock levels", "sql", "on-hand inventory", "stock data", []), ("Forecast demand per SKU", "python", "next 30 days", "demand forecast", [0]), ("Compute reorder quantities", "calculator", "forecast minus stock and lead time", "reorder amounts", [1, 2]), ("Check supplier minimums and prices", "browser", "order constraints", "adjusted quantities", [3]), ("Create the purchase orders", "spreadsheet", "final quantities per supplier", "purchase orders", [4])]), _mk("wedding_venue", "Choose and reserve a wedding venue for next June.", [ ("Set the guest count and budget", "none", "120 guests under $8000", "planning constraints", []), ("Search venues in the area", "web_search", "outdoor venues next June", "venue list", [0]), ("Check availability for target dates", "email", "June weekend openings", "available venues", [1]), ("Compare packages and prices", "spreadsheet", "catering and rentals", "comparison sheet", [2]), ("Visit the top two venues", "calendar", "schedule walkthroughs", "visit notes", [3]), ("Reserve the chosen venue", "email", "deposit and contract", "signed reservation", [4])]), _mk("security_audit", "Audit the web app for common vulnerabilities before release.", [ ("Enumerate the app endpoints", "browser", "public routes and forms", "endpoint inventory", []), ("Run a dependency vulnerability scan", "python", "known CVEs in packages", "dependency report", []), ("Review the authentication flows", "none", "session and password handling", "auth findings", [0]), ("Test input validation", "browser", "form field handling", "validation findings", [0]), ("Prioritize issues by severity", "none", "rank all findings", "prioritized list", [1, 2, 3]), ("File the remediation tickets", "api_call", "one ticket per issue", "tracked issues", [4])]), _mk("meeting_summary", "Produce a summary document of the recorded all-hands meeting.", [ ("Transcribe the recording", "python", "speech to text", "raw transcript", []), ("Clean the transcript", "python", "remove filler and fix names", "clean transcript", [0]), ("Extract decisions and action items", "none", "scan for commitments", "action item list", [1]), ("Summarize each agenda topic", "none", "three sentences per topic", "topic summaries", [1]), ("Compile the summary document", "filesystem", "summaries plus action items", "summary document", [2, 3]), ("Distribute to the team", "email", "send with the recording link", "sent summary", [4])]), ] DISTRACTORS = [ ("Check the social media feed", "browser", "unrelated browsing", "no useful output"), ("Reorganize desktop icons", "filesystem", "cosmetic cleanup", "no useful output"), ("Search for celebrity news", "web_search", "trending gossip", "irrelevant results"), ("Compute digits of pi", "calculator", "first fifty digits", "irrelevant number"), ("Play a music playlist", "browser", "background music", "no useful output"), ("Query the weather in another city", "api_call", "unrelated forecast", "irrelevant data"), ("Draft a note to self", "none", "random reminder", "unrelated note"), ("Refresh the inbox repeatedly", "email", "no new messages", "no useful output"), ("Open and close the calendar", "calendar", "no changes made", "no useful output"), ("List files in the home directory", "filesystem", "unrelated directory", "irrelevant listing"), ] TOOLS = ["web_search", "browser", "python", "sql", "calculator", "spreadsheet", "email", "calendar", "filesystem", "api_call", "none"] INCOMPATIBLE_SWAP = {"browser": "sql", "web_search": "calculator", "python": "email", "sql": "browser", "calculator": "web_search", "email": "python", "calendar": "sql", "filesystem": "browser", "api_call": "calculator", "spreadsheet": "email", "none": "sql"} SUBSTITUTABLE = {frozenset({"python", "calculator"}): 0.3, frozenset({"web_search", "browser"}): 0.3, frozenset({"sql", "spreadsheet"}): 0.35, frozenset({"python", "spreadsheet"}): 0.5} def tool_dist(t1, t2): if t1 == t2: return 0.0 if "none" in (t1, t2): return 0.5 return SUBSTITUTABLE.get(frozenset({t1, t2}), 1.0) # ============================================================================= # 3. PERTURBATION GENERATOR # ============================================================================= SYN = {"search": "look up", "find": "locate", "filter": "narrow down", "book": "reserve", "compile": "assemble", "compare": "contrast", "select": "choose", "download": "fetch", "clean": "tidy up", "verify": "confirm", "summarize": "condense", "extract": "pull out", "rank": "order", "merge": "combine", "analyze": "examine", "draft": "compose", "send": "dispatch", "review": "inspect", "identify": "pinpoint", "calculate": "compute", "check": "confirm", "gather": "collect", "create": "produce", "update": "refresh", "schedule": "arrange", "remove": "drop", "read": "go through", "write": "author", "run": "execute", "pull": "retrieve", "collect": "gather up", "inspect": "look over", "test": "probe", "choose": "pick", "adjust": "tune", "add": "attach", "prioritize": "triage", "distribute": "circulate", "register": "sign up", "reproduce": "replicate", "visit": "tour", "issue": "process", "reply": "respond", "publish": "post", "transcribe": "convert to text", "enumerate": "list out", "export": "dump", "forecast": "project", "price": "cost out", "define": "specify", "set": "fix", "interview": "talk to", "snapshot": "back up", "design": "lay out", "fill": "complete", "get": "obtain", "build": "construct", "resolve": "settle", "total": "add up", "group": "cluster", "segment": "break down", "interpret": "make sense of", "cross-check": "double check", "dry-run": "trial run", "edit": "polish", "outline": "sketch"} def _syn(text, rng): words = text.split() out = [] for w in words: key = re.sub(r"[^a-z-]", "", w.lower()) if key in SYN and rng.random() < 0.9: rep = SYN[key] out.append(rep if w[:1].islower() else rep.capitalize()) else: out.append(w) return " ".join(out) def perturb_paraphrase(plan, rng): # P1 (benign) steps = [replace(s, action=_syn(s.action, rng), context=_syn(s.context, rng), effect=_syn(s.effect, rng)) for s in plan.steps] return Plan(plan.task_id, plan.goal, steps, "P1_paraphrase") def _topo_orders(plan, rng, tries=200): """Sample a valid topological order different from identity, if one exists.""" n = plan.n() edges = trace_edges(plan) preds = defaultdict(set) for (i, k) in edges: preds[k].add(i) for _ in range(tries): remaining = set(range(n)) placed, order = set(), [] while remaining: ready = [i for i in remaining if preds[i] <= placed] i = rng.choice(ready) order.append(i); placed.add(i); remaining.remove(i) if order != list(range(n)): return order return None def perturb_reorder(plan, rng): # P2 (benign) order = _topo_orders(plan, rng) if order is None: return None return Plan(plan.task_id, plan.goal, [plan.steps[i] for i in order], "P2_reorder") def perturb_merge_split(plan, rng): # P3 (benign) steps = list(plan.steps) if rng.random() < 0.5: # SPLIT a step whose action contains " and " into two chained steps cands = [i for i, s in enumerate(steps) if " and " in s.action.lower()] if cands: i = rng.choice(cands) s = steps[i] parts = re.split(r"\s+and\s+", s.action, maxsplit=1, flags=re.I) tok = frozenset({next(iter(s.uid)) + ".a"}) s_a = Step(parts[0].strip().capitalize(), s.tool, s.context, "intermediate result", uid=tok, consumes=s.consumes) s_b = Step(parts[1].strip().capitalize(), s.tool, s.context, s.effect, uid=s.uid, consumes=tok) steps = steps[:i] + [s_a, s_b] + steps[i + 1:] return Plan(plan.task_id, plan.goal, steps, "P3_split") # MERGE two adjacent dependent steps edges = set(trace_edges(plan)) cands = [i for i in range(len(steps) - 1) if (i, i + 1) in edges] if not cands: return None i = rng.choice(cands) a, b = steps[i], steps[i + 1] merged = Step(f"{a.action} and {b.action.lower()}", a.tool if a.tool != "none" else b.tool, f"{a.context}; {b.context}", b.effect, uid=a.uid | b.uid, consumes=(a.consumes | b.consumes) - a.uid) steps = steps[:i] + [merged] + steps[i + 2:] return Plan(plan.task_id, plan.goal, steps, "P3_merge") def perturb_causal_inversion(plan, rng): # P4 (order violation) edges = trace_edges(plan) if not edges: return None (i, k) = edges[rng.randrange(len(edges))] steps = list(plan.steps) steps[i], steps[k] = steps[k], steps[i] # consumer now precedes producer return Plan(plan.task_id, plan.goal, steps, "P4_inversion") def perturb_corruption(plan, rng): # P5 (corruption) mode = rng.choice(["delete", "insert", "toolswap", "delete+insert"]) steps = list(plan.steps) label = "P5_" + mode if "delete" in mode: crit = critical_mask(plan) cands = [j for j in range(len(steps) - 1) if crit[j] > 0] or list(range(len(steps) - 1)) del steps[rng.choice(cands)] if "insert" in mode: for _ in range(rng.randint(1, 3)): a, t, c, e = DISTRACTORS[rng.randrange(len(DISTRACTORS))] d = Step(a, t, c, e, uid=frozenset({f"noise{rng.random()}"}), consumes=frozenset()) steps.insert(rng.randrange(len(steps) + 1), d) if mode == "toolswap": for j in rng.sample(range(len(steps)), k=min(2, len(steps))): s = steps[j] steps[j] = replace(s, tool=INCOMPATIBLE_SWAP.get(s.tool, "sql")) return Plan(plan.task_id, plan.goal, steps, label) # --- graded-quality ladder for Part B --------------------------------------- def graded_variant(plan, level, rng): """quality level 5 (best) .. 1 (worst); returns a stacked perturbation.""" p = perturb_paraphrase(plan, rng) if level >= 5: r = perturb_reorder(p, rng) return replace(r or p, label="Q5") if level == 4: p = perturb_causal_inversion(p, rng) or p return replace(p, label="Q4") if level == 3: p = perturb_causal_inversion(p, rng) or p p = perturb_causal_inversion(p, rng) or p return replace(p, label="Q3") if level == 2: rng2 = random.Random(rng.random()) q = perturb_corruption(p, rng2) while "delete" not in q.label: q = perturb_corruption(p, rng2) return replace(q, label="Q2") # level 1: delete + insert + toolswap + inversion rng2 = random.Random(rng.random()) q = perturb_corruption(p, rng2) while q.label != "P5_delete+insert": q = perturb_corruption(p, rng2) for j in rng.sample(range(q.n()), k=min(2, q.n())): q.steps[j] = replace(q.steps[j], tool=INCOMPATIBLE_SWAP.get(q.steps[j].tool, "sql")) q = perturb_causal_inversion(q, rng) or q return replace(q, label="Q1") # --- reference set ----------------------------------------------------------- def make_references(plan, K, rng): """R1 = canonical; R2 = paraphrased+reordered valid variant; R3 = coarser-granularity (merged) variant. Distinct RNG stream from the evaluation perturbations.""" refs = [replace(plan, label="ref_canonical")] if K >= 2: r = perturb_paraphrase(plan, rng) r = perturb_reorder(r, rng) or r refs.append(replace(r, label="ref_variant")) if K >= 3: r = perturb_merge_split(plan, rng) or perturb_paraphrase(plan, rng) refs.append(replace(r, label="ref_coarse")) return refs[:K] # ============================================================================= # 4. EMBEDDERS # ============================================================================= class HashEmbedder: """Deterministic no-download fallback (word + char-3gram hashing). Use ONLY for smoke tests; run the real thing with sentence-transformers.""" name = "hash-512" def __init__(self, dim=512): self.dim = dim self.cache = {} def encode(self, texts): out = np.zeros((len(texts), self.dim)) for r, t in enumerate(texts): if t in self.cache: out[r] = self.cache[t]; continue v = np.zeros(self.dim) toks = re.findall(r"[a-z0-9$]+", t.lower()) grams = toks + [t.lower()[i:i + 3] for i in range(max(0, len(t) - 2))] for g in grams: h = int(hashlib.md5(g.encode()).hexdigest(), 16) v[h % self.dim] += (1 if (h >> 20) % 2 else -1) n = np.linalg.norm(v) v = v / n if n > 0 else v self.cache[t] = v out[r] = v return out class STEmbedder: name = "all-MiniLM-L6-v2" def __init__(self): from sentence_transformers import SentenceTransformer self.model = SentenceTransformer("sentence-transformers/all-MiniLM-L6-v2") self.cache = {} def encode(self, texts): missing = [t for t in texts if t not in self.cache] if missing: embs = self.model.encode(missing, normalize_embeddings=True, batch_size=64, show_progress_bar=False) for t, e in zip(missing, embs): self.cache[t] = np.asarray(e, dtype=float) return np.stack([self.cache[t] for t in texts]) EMB = None def emb(texts): return EMB.encode(list(texts)) def cosd(A, B): """Cosine distance matrix mapped to [0,1] (rows of A vs rows of B).""" S = A @ B.T return np.clip((1.0 - S) / 2.0, 0.0, 1.0) # ============================================================================= # 5. METRICS # ============================================================================= # ---- 5.1 baselines ---------------------------------------------------------- def m_exact(P, refs): def one(R): m = min(P.n(), R.n()) hits = sum(1 for i in range(m) if P.steps[i].text().lower() == R.steps[i].text().lower()) return hits / max(P.n(), R.n()) return max(one(R) for R in refs) def _ngrams(toks, n): return [tuple(toks[i:i + n]) for i in range(len(toks) - n + 1)] def m_bleu(P, refs): hyp = re.findall(r"[a-z0-9$]+", " ".join(s.text() for s in P.steps).lower()) def one(R): ref = re.findall(r"[a-z0-9$]+", " ".join(s.text() for s in R.steps).lower()) logs = 0.0 for n in range(1, 5): h, r = _ngrams(hyp, n), _ngrams(ref, n) if not h: return 0.0 rc = defaultdict(int) for g in r: rc[g] += 1 match = 0 hc = defaultdict(int) for g in h: hc[g] += 1 for g, c in hc.items(): match += min(c, rc.get(g, 0)) logs += math.log((match + 1e-9) / len(h)) bp = min(1.0, math.exp(1 - len(ref) / max(len(hyp), 1))) return bp * math.exp(logs / 4) return max(one(R) for R in refs) def m_rougeL(P, refs): hyp = re.findall(r"[a-z0-9$]+", " ".join(s.text() for s in P.steps).lower()) def one(R): ref = re.findall(r"[a-z0-9$]+", " ".join(s.text() for s in R.steps).lower()) la, lb = len(hyp), len(ref) dp = np.zeros((la + 1, lb + 1), dtype=int) for i in range(la): for j in range(lb): dp[i + 1][j + 1] = dp[i][j] + 1 if hyp[i] == ref[j] \ else max(dp[i][j + 1], dp[i + 1][j]) lcs = dp[la][lb] if lcs == 0: return 0.0 p, r = lcs / la, lcs / lb return 2 * p * r / (p + r) return max(one(R) for R in refs) def _step_embs(plan): return emb([s.text() for s in plan.steps]) def m_embed_greedy(P, refs): """BERTScore-style greedy soft alignment at step level (F1 of best sims).""" Ep = _step_embs(P) def one(R): Er = _step_embs(R) S = Ep @ Er.T p = float(np.mean(np.max(S, axis=1))) r = float(np.mean(np.max(S, axis=0))) return max(0.0, 2 * p * r / (p + r + 1e-9)) return max(one(R) for R in refs) def m_embed_hungarian(P, refs, dummy_cost=0.5): """Whole-step embedding cost + one-to-one Hungarian with dummy padding.""" Ep = _step_embs(P) def one(R): Er = _step_embs(R) C = cosd(Ep, Er) n, m = C.shape N = max(n, m) Cp = np.full((N, N), dummy_cost) Cp[:n, :m] = C ri, ci = linear_sum_assignment(Cp) return 1.0 - float(Cp[ri, ci].mean()) return max(one(R) for R in refs) # ---- 5.2 optional LLM judge -------------------------------------------------- JUDGE_CACHE_FILE = "judge_cache.json" _judge_cache = None def m_llm_judge(P, refs, model="claude-sonnet-4-6"): global _judge_cache import anthropic if _judge_cache is None: _judge_cache = json.load(open(JUDGE_CACHE_FILE)) if os.path.exists(JUDGE_CACHE_FILE) else {} R = refs[0] def fmt(pl): return "\n".join(f"{i+1}. [{s.tool}] {s.text()}" for i, s in enumerate(pl.steps)) prompt = (f"Task: {P.goal}\n\nReference plan:\n{fmt(R)}\n\n" f"Candidate agent trajectory:\n{fmt(P)}\n\n" "Rate the candidate's PROCESS quality (decomposition, ordering, " "tool use, no missing or irrelevant steps) against the task. " "Different but valid plans deserve high scores. " "Respond with ONLY an integer 0-100.") key = hashlib.md5((model + prompt).encode()).hexdigest() if key in _judge_cache: return _judge_cache[key] client = anthropic.Anthropic() msg = client.messages.create(model=model, max_tokens=8, messages=[{"role": "user", "content": prompt}]) m = re.search(r"\d+", msg.content[0].text) score = (int(m.group()) if m else 50) / 100.0 _judge_cache[key] = score json.dump(_judge_cache, open(JUDGE_CACHE_FILE, "w")) return score # ---- 5.3 OTAP ---------------------------------------------------------------- @dataclass class OtapCfg: alpha: float = 0.35 # action beta: float = 0.20 # context gamma: float = 0.25 # effect delta: float = 0.20 # tool theta: float = 0.35 # structure weight eps: float = 0.05 # entropic reg lam: float = 1.0 # unbalanced KL strength (both marginals) softT: float = 0.05 # soft-min temperature over references K: int = 3 # references used signed: bool = True # signed dependency distances unbalanced: bool = True transport: str = "ot" # "ot" | "hungarian" fielded: bool = True # fielded node cost vs whole-step embedding causal_mass: bool = True kappa: float = 1.0 edge_mode: str = "trace" # "trace" | "total_order" outer_iters: int = 12 inner_iters: int = 300 def node_cost(P, R, cfg): if cfg.fielded: Ca = cosd(emb([s.action for s in P.steps]), emb([s.action for s in R.steps])) Cc = cosd(emb([s.context for s in P.steps]), emb([s.context for s in R.steps])) Ce = cosd(emb([s.effect for s in P.steps]), emb([s.effect for s in R.steps])) Ct = np.array([[tool_dist(a.tool, b.tool) for b in R.steps] for a in P.steps]) w = cfg.alpha + cfg.beta + cfg.gamma + cfg.delta return (cfg.alpha * Ca + cfg.beta * Cc + cfg.gamma * Ce + cfg.delta * Ct) / w return cosd(_step_embs(P), _step_embs(R)) def unbalanced_sinkhorn(M, mu, nu, eps, lam, iters): K = np.exp(-M / eps) u = np.ones_like(mu); v = np.ones_like(nu) fe = lam / (lam + eps) for _ in range(iters): u = (mu / np.maximum(K @ v, 1e-300)) ** fe v = (nu / np.maximum(K.T @ u, 1e-300)) ** fe return u[:, None] * K * v[None, :] def gkl(a, b): a = np.maximum(a, 1e-300) return float(np.sum(a * np.log(a / np.maximum(b, 1e-300)) - a + b)) def _gw_grad_and_energy(Dp, Dr, T): p = T.sum(axis=1); q = T.sum(axis=0) G = (Dp ** 2) @ p[:, None] + ((Dr ** 2) @ q)[None, :] - 2 * Dp @ T @ Dr.T G = G / 4.0 # (Dp-Dr)^2 <= 4 -> [0,1] return G, float(np.sum(G * T)) def otap_loss_single(P, R, cfg): C = node_cost(P, R, cfg) Dp = signed_dep_matrix(P, cfg.edge_mode) Dr = signed_dep_matrix(R, cfg.edge_mode) if not cfg.signed: Dp, Dr = np.abs(Dp), np.abs(Dr) n, m = C.shape mu = np.ones(n) / n nu = np.ones(m) / m if cfg.causal_mass: cm = 1.0 + cfg.kappa * critical_mask(R) nu = cm / cm.sum() if cfg.transport == "hungarian": N = max(n, m) Cp = np.full((N, N), 0.5) Cp[:n, :m] = C ri, ci = linear_sum_assignment(Cp) L = float(Cp[ri, ci].mean()) matched = sum(1 for a, b in zip(ri, ci) if a < n and b < m) return L, matched / n, matched / m lam = cfg.lam if cfg.unbalanced else 1e4 T = np.outer(mu, nu) for _ in range(cfg.outer_iters): if cfg.theta > 0: G, _ = _gw_grad_and_energy(Dp, Dr, T) M = (1 - cfg.theta) * C + cfg.theta * 2.0 * G else: M = C T = unbalanced_sinkhorn(M, mu, nu, cfg.eps, lam, cfg.inner_iters) _, Egw = _gw_grad_and_energy(Dp, Dr, T) feat = float(np.sum(C * T)) kl = cfg.lam * (gkl(T.sum(axis=1), mu) + gkl(T.sum(axis=0), nu)) L = (1 - cfg.theta) * feat + cfg.theta * Egw + kl pi = float(T.sum() / mu.sum()); rho = float(T.sum() / nu.sum()) return L, pi, rho def m_otap(P, refs, cfg=OtapCfg()): refs = refs[:cfg.K] losses = [otap_loss_single(P, R, cfg)[0] for R in refs] Tt = cfg.softT L = -Tt * math.log(np.mean(np.exp(-(np.array(losses)) / Tt)) + 1e-300) return math.exp(-L) # ============================================================================= # 6. EXPERIMENT DRIVERS # ============================================================================= def build_eval_set(quick=False): rng_ref = random.Random(SEED + 1) rng_pert = random.Random(SEED + 2) data = [] n_var = 1 if quick else 3 for plan in CORPUS: refs = make_references(plan, 3, rng_ref) variants = [replace(plan, label="original")] for _ in range(n_var): variants.append(perturb_paraphrase(plan, rng_pert)) r = perturb_reorder(plan, rng_pert) if r: variants.append(r) r = perturb_merge_split(plan, rng_pert) if r: variants.append(r) r = perturb_causal_inversion(plan, rng_pert) if r: variants.append(r) variants.append(perturb_corruption(plan, rng_pert)) graded = [] n_g = 1 if quick else 2 for level in (5, 4, 3, 2, 1): for _ in range(n_g): graded.append(graded_variant(plan, level, rng_pert)) data.append({"plan": plan, "refs": refs, "variants": variants, "graded": graded}) return data def fam(label): if label.startswith("P1"): return "P1" if label.startswith("P2"): return "P2" if label.startswith("P3"): return "P3" if label.startswith("P4"): return "P4" if label.startswith("P5"): return "P5" return "orig" def score_all(data, metrics): """scores[metric][task_idx] = list of (label, score); also runtime.""" scores = {m: [] for m in metrics} runtime = {m: [] for m in metrics} for ti, d in enumerate(data): for m, fn in metrics.items(): row = [] for v in d["variants"] + d["graded"]: t0 = time.perf_counter() s = fn(v, d["refs"]) runtime[m].append(time.perf_counter() - t0) row.append((v.label, s)) scores[m].append(row) print(f" scored task {ti+1}/{len(data)}: {d['plan'].task_id}", flush=True) return scores, runtime def pra_tables(scores, data): """Pairwise ranking accuracy per family + severity ordering + AUROC. Definitions (also used verbatim when populating the paper): PRA(Pb) for benign b in {P1,P2,P3}: over same-task pairs (x,y) with x a Pb variant and y any damaging (P4 or P5) variant, fraction with score(x) > score(y); ties count 0.5. PRA(P4): pairs (x,y), x in {original,P1,P2,P3}, y in P4. PRA(P5): pairs (x,y), x in {original,P1,P2,P3}, y in P5. SEV : pairs (x,y), x in P4, y in P5, fraction score(x) > score(y). AUROC : per-task AUROC separating {original,P1,P2,P3} (positive) from {P4,P5} (negative), averaged over tasks.""" out = {} for m, per_task in scores.items(): acc = defaultdict(lambda: [0.0, 0]) aurocs = [] for row in per_task: groups = defaultdict(list) for (lab, s) in row: if lab.startswith(("orig", "P")): groups[fam(lab)].append(s) damaging = groups["P4"] + groups["P5"] benign_all = groups["orig"] + groups["P1"] + groups["P2"] + groups["P3"] for b in ("P1", "P2", "P3"): for x in groups[b]: for y in damaging: acc[b][0] += 1.0 if x > y else (0.5 if x == y else 0.0) acc[b][1] += 1 for dmg in ("P4", "P5"): for x in benign_all: for y in groups[dmg]: acc[dmg][0] += 1.0 if x > y else (0.5 if x == y else 0.0) acc[dmg][1] += 1 for x in groups["P4"]: for y in groups["P5"]: acc["SEV"][0] += 1.0 if x > y else (0.5 if x == y else 0.0) acc["SEV"][1] += 1 if benign_all and damaging: c = t = 0 for x in benign_all: for y in damaging: c += 1.0 if x > y else (0.5 if x == y else 0.0) t += 1 aurocs.append(c / t) out[m] = {k: (100 * v[0] / v[1] if v[1] else float("nan")) for k, v in acc.items()} out[m]["AUROC"] = float(np.mean(aurocs)) if aurocs else float("nan") return out def partB_tables(scores, data, human=None): """Correlation of metric with graded quality Q5..Q1 (2 samples/level/task). Spearman/Kendall computed per task then averaged. 'Damaged-only' restricts to Q1..Q4 (analogue of failed-trajectory analysis). If a human-ratings CSV is provided (task_id,label,rating), correlations are computed against human means instead.""" out = {} for m, per_task in scores.items(): rs, ts, rs_d, ts_d = [], [], [], [] for ti, row in enumerate(per_task): q, s = [], [] for (lab, sc) in row: if lab.startswith("Q"): tid = data[ti]["plan"].task_id truth = (human.get((tid, lab)) if human else int(lab[1])) if truth is None: continue q.append(truth); s.append(sc) if len(set(q)) > 1: r, _ = spearmanr(q, s); t, _ = kendalltau(q, s) rs.append(r); ts.append(t) qd = [x for x, y in zip(q, s) if x <= 4] sd = [y for x, y in zip(q, s) if x <= 4] if len(set(qd)) > 1: r, _ = spearmanr(qd, sd); t, _ = kendalltau(qd, sd) rs_d.append(r); ts_d.append(t) out[m] = {"rho": float(np.nanmean(rs)), "tau": float(np.nanmean(ts)), "rho_dmg": float(np.nanmean(rs_d)), "tau_dmg": float(np.nanmean(ts_d)), "rho_sd": float(np.nanstd(rs))} return out ABLATIONS = { "OTAP (full)": OtapCfg(), "(i) no structure": OtapCfg(theta=0.0), "(ii) unsigned D": OtapCfg(signed=False), "(iii) balanced": OtapCfg(unbalanced=False), "(iv) Hungarian": OtapCfg(transport="hungarian"), "(v) whole-step emb": OtapCfg(fielded=False), "(vi) no tool term": OtapCfg(delta=0.0), "(vii) K=1 reference": OtapCfg(K=1), "(vii) soft-min T=1.0": OtapCfg(softT=1.0), "(viii) uniform mass": OtapCfg(causal_mass=False), "(ix) total-order edges": OtapCfg(edge_mode="total_order"), } def fmt_table(rows, headers): w = [max(len(str(r[i])) for r in [headers] + rows) for i in range(len(headers))] line = lambda r: "| " + " | ".join(str(x).ljust(w[i]) for i, x in enumerate(r)) + " |" sep = "|" + "|".join("-" * (x + 2) for x in w) + "|" return "\n".join([line(headers), sep] + [line(r) for r in rows]) def load_human(path): human = {} with open(path) as f: for ln in f: parts = [p.strip() for p in ln.strip().split(",")] if len(parts) >= 3 and parts[0] != "task_id": human[(parts[0], parts[1])] = float(parts[2]) return human # ============================================================================= # 7. MAIN # ============================================================================= def main(): global EMB ap = argparse.ArgumentParser() ap.add_argument("--embedder", choices=["st", "hash"], default="st") ap.add_argument("--quick", action="store_true") ap.add_argument("--judge", action="store_true", help="include LLM judge (needs ANTHROPIC_API_KEY)") ap.add_argument("--human-ratings", default=None, help="CSV task_id,label,rating to replace synthetic Part-B quality") ap.add_argument("--out", default="otap_results.json") args = ap.parse_args() random.seed(SEED); np.random.seed(SEED) if args.embedder == "st": try: EMB = STEmbedder() except Exception as e: print(f"[warn] sentence-transformers unavailable ({e}); " f"falling back to hash embedder. Results with the hash " f"embedder are NOT paper-grade.", file=sys.stderr) EMB = HashEmbedder() else: EMB = HashEmbedder() print(f"Embedder: {EMB.name}") print("Building evaluation set ...") data = build_eval_set(quick=args.quick) n_traj = sum(len(d["variants"]) + len(d["graded"]) for d in data) print(f" {len(data)} tasks, {n_traj} trajectories, K=3 references each.") metrics = { "Exact step match": m_exact, "BLEU": m_bleu, "ROUGE-L": m_rougeL, "Embed greedy (BERTScore-style)": m_embed_greedy, "Embed + Hungarian": m_embed_hungarian, "OTAP (ours)": lambda P, R: m_otap(P, R, OtapCfg()), } if args.judge: metrics["LLM-as-a-judge"] = m_llm_judge print("\n== Scoring all metrics (Part A + Part B trajectories) ==") scores, runtime = score_all(data, metrics) print("\n== Part A: perturbation meta-evaluation ==") A = pra_tables(scores, data) rowsA = [[m] + [f"{A[m][k]:.1f}" for k in ("P1", "P2", "P3", "P4", "P5", "SEV")] + [f"{100*A[m]['AUROC']:.1f}"] for m in metrics] tabA = fmt_table(rowsA, ["Metric", "P1", "P2", "P3", "P4", "P5", "P4>P5", "AUROC"]) print(tabA) print("\n== Part B: correlation with graded quality ==") human = load_human(args.human_ratings) if args.human_ratings else None if human: print(f" using HUMAN ratings from {args.human_ratings} ({len(human)} rows)") B = partB_tables(scores, data, human) rowsB = [[m, f"{B[m]['rho']:.3f}", f"{B[m]['tau']:.3f}", f"{B[m]['rho_dmg']:.3f}", f"{B[m]['tau_dmg']:.3f}"] for m in metrics] tabB = fmt_table(rowsB, ["Metric", "rho(all)", "tau(all)", "rho(dmg)", "tau(dmg)"]) print(tabB) print("\n== Ablations (OTAP variants) ==") abl_scores = {} for name, cfg in ABLATIONS.items(): fn = (lambda c: (lambda P, R: m_otap(P, R, c)))(cfg) s, _ = score_all(data, {name: fn}) abl_scores[name] = s[name] print(f" done: {name}") Aab = pra_tables(abl_scores, data) Bab = partB_tables(abl_scores, data, human) rowsC = [[n, f"{np.mean([Aab[n]['P1'],Aab[n]['P2'],Aab[n]['P3']]):.1f}", f"{Aab[n]['P4']:.1f}", f"{Aab[n]['P5']:.1f}", f"{100*Aab[n]['AUROC']:.1f}", f"{Bab[n]['rho']:.3f}"] for n in ABLATIONS] tabC = fmt_table(rowsC, ["Variant", "benign(avg)", "P4", "P5", "AUROC", "rho(B)"]) print(tabC) print("\n== Runtime (mean ms per evaluation, excludes embedding cache warm-up) ==") rowsR = [[m, f"{1000*np.mean(runtime[m]):.1f}"] for m in metrics] print(fmt_table(rowsR, ["Metric", "ms/eval"])) results = {"config": {"seed": SEED, "embedder": EMB.name, "quick": args.quick, "n_tasks": len(data), "n_trajectories": n_traj, "otap_defaults": vars(OtapCfg()), "human_ratings": bool(human)}, "definitions": pra_tables.__doc__ + "\n" + partB_tables.__doc__, "partA": A, "partB": B, "ablations_partA": Aab, "ablations_partB": Bab, "runtime_ms": {m: 1000 * float(np.mean(runtime[m])) for m in metrics}, "tables_markdown": {"A": tabA, "B": tabB, "C": tabC}} with open(args.out, "w") as f: json.dump(results, f, indent=2) print(f"\nAll results written to {args.out} -- send this file back to " f"populate the paper.") if __name__ == "__main__": main()