"""Week 7a toolkit: data selection and quality, standard library only. Nothing here calls a model, an API or the network. Every world is made up: the tasks, the "models", the annotators, the traces. The code shows how the pieces work, not how real data behaves. Parts: 1 informativeness 2 diversity and dedupe 3 gradient similarity (LESS-like) proxy 4 label noise and data maps 5 agreement 6 IRT and tiny benchmarks 7 EvalGen-style alignment 8 trace-to-dataset pipeline 9 demo """ import hashlib import json import math import random import re from dataclasses import dataclass, field # ---------------------------------------------------------------- shared helpers _WORD = re.compile(r"[a-z0-9_]+") def words(text): return _WORD.findall(str(text).lower()) def stable_hash(s, bits=64): h = hashlib.blake2b(str(s).encode("utf-8"), digest_size=8).hexdigest() return int(h[: bits // 4], 16) def mean(xs): xs = list(xs) return sum(xs) / len(xs) if xs else 0.0 def stdev(xs): xs = list(xs) if not xs: return 0.0 m = mean(xs) return math.sqrt(sum((x - m) ** 2 for x in xs) / len(xs)) def sigmoid(x): if x >= 0: return 1.0 / (1.0 + math.exp(-x)) e = math.exp(x) return e / (1.0 + e) def logit(p, eps=1e-6): p = min(max(p, eps), 1 - eps) return math.log(p / (1 - p)) def dot(a, b): return sum(x * y for x, y in zip(a, b)) def norm(a): return math.sqrt(dot(a, a)) def cosine(a, b): na, nb = norm(a), norm(b) return 0.0 if na == 0 or nb == 0 else dot(a, b) / (na * nb) def euclid(a, b): return math.sqrt(sum((x - y) ** 2 for x, y in zip(a, b))) def overlap(a, b): """Share of the first selection that the second selection also has.""" a, b = set(a), set(b) return len(a & b) / len(a) if a else 0.0 # ---------------------------------------------------------------- 1. informativeness def entropy(probs): """Shannon entropy in bits of a probability list. High means the model is unsure.""" return -sum(p * math.log2(p) for p in probs if p > 0) def least_confidence(probs): return 1.0 - max(probs) def margin(probs): """Top probability minus the second. Small means a close call.""" s = sorted(probs, reverse=True) return s[0] - (s[1] if len(s) > 1 else 0.0) def vote_entropy(votes): """Entropy (bits) of the labels that a committee of models voted.""" counts = {} for v in votes: counts[v] = counts.get(v, 0) + 1 n = len(votes) return entropy([c / n for c in counts.values()]) if n else 0.0 def disagreement(votes): """0 when all votes agree. Otherwise one minus the share of the most common vote.""" if not votes: return 0.0 top = max(votes.count(v) for v in set(votes)) return 1.0 - top / len(votes) def pass_rate(outcomes): return mean(1.0 if o else 0.0 for o in outcomes) def pass_variance(p): """p(1-p). Largest at p = 0.5. A task that is always passed or always failed carries no signal.""" return p * (1.0 - p) def difficulty_from_passes(passes, k): return 1.0 - passes / k def sometimes_passes(passes, k): """The week 6b difficulty filter: keep a task when 0 < passes < k.""" return 0 < passes < k def rank_ids(scores, top=None, largest=True): """Ids sorted by score, ties broken by id. Pass top to keep only the first top ids.""" ordered = sorted(scores, key=lambda i: ((-scores[i]) if largest else scores[i], i)) return ordered if top is None else ordered[:top] def cap_per_group(items, key, cap): """Keep at most cap items per group, in the order given. SWE-smith caps one task at 3 trajectories.""" seen, out = {}, [] for it in items: g = key(it) if seen.get(g, 0) < cap: seen[g] = seen.get(g, 0) + 1 out.append(it) return out def make_task_pool(n_tasks, ability=0.0, seed=0): """Made-up tasks. Task i has a hidden difficulty d. The current model passes it with p = sigmoid(ability - d).""" rng = random.Random(seed) return {"ability": ability, "difficulty": {i: rng.gauss(0.0, 1.8) for i in range(n_tasks)}} def pass_probability(tasks, i): return sigmoid(tasks["ability"] - tasks["difficulty"][i]) def sample_passes(tasks, k, seed=0): """Attempt every task k times. Returns {task id: passes}. tasks comes from make_task_pool.""" rng = random.Random(seed) out = {} for i in sorted(tasks["difficulty"]): p = pass_probability(tasks, i) out[i] = sum(rng.random() < p for _ in range(k)) return out def true_signal(tasks): """The hidden learning signal of each task in this toy: p(1-p) at the true pass probability.""" return {i: pass_variance(pass_probability(tasks, i)) for i in tasks["difficulty"]} def informativeness_demo(n_tasks=300, top=50, ks=(1, 4, 16, 64), seed=0): """How well does the pass-rate score from k attempts find the tasks with the most true signal?""" tasks = make_task_pool(n_tasks, ability=0.0, seed=seed) truth = rank_ids(true_signal(tasks), top) rows = [] for k in ks: passes = sample_passes(tasks, k, seed=seed + 1) est = {i: pass_variance(passes[i] / k) for i in passes} # ties are broken by id, so break them by a seeded random key instead for a fair comparison rng = random.Random(seed + 2) jitter = {i: est[i] + rng.random() * 1e-9 for i in est} sel = rank_ids(jitter, top) kept = sum(sometimes_passes(passes[i], k) for i in passes) rows.append({"k": k, "overlap_with_truth": round(overlap(truth, sel), 3), "share_0_lt_passes_lt_k": round(kept / n_tasks, 3)}) return rows # ---------------------------------------------------------------- 2. diversity, dedupe, importance resampling def k_center_greedy(points, k, start=None): """Farthest-first traversal. points is {id: vector}. Each new center is the point farthest from all centers.""" ids = sorted(points) if k >= len(ids): return ids if k <= 0: return [] centers = [min(ids) if start is None else start] dmin = {i: euclid(points[i], points[centers[0]]) for i in ids} while len(centers) < k: nxt = min(ids, key=lambda i: (-dmin[i], i)) centers.append(nxt) for i in ids: d = euclid(points[i], points[nxt]) if d < dmin[i]: dmin[i] = d return centers def covering_radius(points, centers): """The largest distance from any point to its nearest center. Smaller means the centers cover the data better.""" return max(min(euclid(points[i], points[c]) for c in centers) for i in points) def score_first_diverse(items, budget, min_dist): """Take items best score first. Skip one that is closer than min_dist to something already taken. items is a list of (id, score, vector). This is the shape of a score-first, diversity-aware filter.""" chosen = [] vecs = {} for iid, _score, vec in sorted(items, key=lambda t: (-t[1], t[0])): if len(chosen) >= budget: break if all(euclid(vec, v) >= min_dist for v in vecs.values()): chosen.append(iid) vecs[iid] = vec return chosen def shingles(text, k=3): w = words(text) if len(w) < k: return {" ".join(w)} if w else set() return {" ".join(w[i:i + k]) for i in range(len(w) - k + 1)} def jaccard(a, b): if not a and not b: return 1.0 return len(a & b) / len(a | b) _PRIME = (1 << 61) - 1 def minhash(sh, num_perm=64, seed=0): rng = random.Random(seed) params = [(rng.randrange(1, _PRIME), rng.randrange(0, _PRIME)) for _ in range(num_perm)] hs = [stable_hash(s) for s in sh] if not hs: return tuple([_PRIME] * num_perm) return tuple(min((a * h + b) % _PRIME for h in hs) for a, b in params) def lsh_candidates(texts, k=3, num_perm=64, bands=16, seed=0): """Pairs whose MinHash signatures agree on every row of at least one band. Cheap. A pair with Jaccard similarity s is found with probability 1 - (1 - s^rows)^bands, so close pairs are usually found and some are missed.""" if num_perm % bands: raise ValueError("bands must divide num_perm") rows = num_perm // bands sig = {i: minhash(shingles(t, k), num_perm, seed) for i, t in texts.items()} buckets = {} for i in sorted(texts): for b in range(bands): buckets.setdefault((b, sig[i][b * rows:(b + 1) * rows]), []).append(i) cands = set() for ids in buckets.values(): for x in range(len(ids)): for y in range(x + 1, len(ids)): cands.add((ids[x], ids[y])) return sorted(cands) def near_duplicates(texts, threshold=0.8, k=3, num_perm=64, bands=16, seed=0): """Pairs with shingle Jaccard at least threshold. MinHash bands pick candidates and exact Jaccard decides.""" sh = {i: shingles(t, k) for i, t in texts.items()} out = [] for a, b in lsh_candidates(texts, k, num_perm, bands, seed): j = jaccard(sh[a], sh[b]) if j >= threshold: out.append((a, b, round(j, 4))) return out def duplicate_groups(texts, threshold=0.8, **kw): """Groups of ids linked by near-duplicate pairs (union-find). Only groups of two or more are returned.""" parent = {i: i for i in texts} def find(x): while parent[x] != x: parent[x] = parent[parent[x]] x = parent[x] return x for a, b, _j in near_duplicates(texts, threshold, **kw): ra, rb = find(a), find(b) if ra != rb: parent[max(ra, rb)] = min(ra, rb) groups = {} for i in texts: groups.setdefault(find(i), []).append(i) return sorted(sorted(g) for g in groups.values() if len(g) > 1) def dedupe_keep_first(texts, threshold=0.8, **kw): """Keep the smallest id of every near-duplicate group. Returns (kept ids, {dropped id: kept id}).""" dropped = {} for g in duplicate_groups(texts, threshold, **kw): for i in g[1:]: dropped[i] = g[0] return [i for i in sorted(texts) if i not in dropped], dropped def hashed_features(text, buckets=256): """Hashed unigram and bigram counts, as a {bucket: count} dict.""" w = words(text) feats = w + [a + " " + b for a, b in zip(w, w[1:])] out = {} for f in feats: b = stable_hash(f) % buckets out[b] = out.get(b, 0) + 1 return out def bucket_probs(texts, buckets=256, alpha=1.0): counts = [0.0] * buckets for t in texts: for b, c in hashed_features(t, buckets).items(): counts[b] += c total = sum(counts) + alpha * buckets return [(c + alpha) / total for c in counts] def dsir_log_weights(raw, target_texts, buckets=256, alpha=1.0): """For each raw document, the sum over its features of log(target prob / raw prob). raw is {id: text}. A simplified version of importance resampling with hashed n-gram features.""" pt = bucket_probs(target_texts, buckets, alpha) pr = bucket_probs(list(raw.values()), buckets, alpha) out = {} for i, t in raw.items(): out[i] = sum(c * math.log(pt[b] / pr[b]) for b, c in hashed_features(t, buckets).items()) return out def gumbel_top_k(log_weights, k, seed=0): """Sample k ids without replacement with probability proportional to exp(log weight).""" rng = random.Random(seed) keys = {} for i in sorted(log_weights): u = min(max(rng.random(), 1e-12), 1 - 1e-12) keys[i] = log_weights[i] - math.log(-math.log(u)) return rank_ids(keys, k) TOPICS = { "code": ["function", "compile", "variable", "return", "loop", "array", "debug", "parser", "module", "thread"], "food": ["simmer", "garlic", "butter", "recipe", "oven", "pepper", "dough", "sauce", "spoon", "broth"], "sport": ["match", "goal", "season", "coach", "league", "score", "pitch", "referee", "trophy", "sprint"], } def make_topic_docs(n_per_topic=100, doc_len=14, seed=0, topics=None): """Made-up documents. Each is random words from one topic's word list plus some common words. Returns {id: (topic, text)}.""" rng = random.Random(seed) common = ["the", "a", "of", "and", "to", "in", "is", "with"] docs = {} for t in sorted(topics or TOPICS): for _ in range(n_per_topic): ws = [rng.choice(TOPICS[t]) if rng.random() < 0.6 else rng.choice(common) for _ in range(doc_len)] docs[len(docs)] = (t, " ".join(ws)) return docs def selection_share(docs, selected, topic): return mean(1.0 if docs[i][0] == topic else 0.0 for i in selected) # ---------------------------------------------------------------- 3. gradient similarity (a LESS-like proxy) def logistic_p(w, x): return sigmoid(dot(w, x)) def example_grad(w, x, y): """Gradient of the logistic loss for one example: (p - y) * x.""" p = logistic_p(w, x) return [(p - y) * xi for xi in x] def train_logistic(data, steps=200, lr=0.5, w0=None, dim=None): """Full-batch gradient descent on (x, y) pairs. Deterministic.""" d = dim or len(data[0][0]) w = list(w0) if w0 is not None else [0.0] * d n = len(data) for _ in range(steps): g = [0.0] * d for x, y in data: gi = example_grad(w, x, y) for j in range(d): g[j] += gi[j] w = [wj - lr * gj / n for wj, gj in zip(w, g)] return w def accuracy(w, data): return mean(1.0 if (logistic_p(w, x) >= 0.5) == (y == 1) else 0.0 for x, y in data) def _unit(v): n = norm(v) return [x / n for x in v] def make_less_world(n_per_task=150, dim=8, seed=0, big_frac=0.2, big_scale=4.0, label_noise=0.05, n_support=8, n_test=300): """Three made-up tasks share one linear model. Task A is the target. Task B is unrelated. Task C partly contradicts A. A fraction of examples has a large feature norm (a stand-in for long text).""" rng = random.Random(seed) wA = _unit([rng.gauss(0, 1) for _ in range(dim)]) wB = _unit([rng.gauss(0, 1) for _ in range(dim)]) wC = _unit([-0.8 * a + 0.6 * b for a, b in zip(wA, wB)]) dirs = {"A": wA, "B": wB, "C": wC} def draw(task, scale_ok): x = [rng.gauss(0, 1) for _ in range(dim)] big = scale_ok and rng.random() < big_frac if big: x = [big_scale * v for v in x] y = 1 if dot(dirs[task], x) > 0 else 0 if scale_ok and rng.random() < label_noise: y = 1 - y return x, y, big pool = [] for task in "ABC": for _ in range(n_per_task): x, y, big = draw(task, True) pool.append({"id": len(pool), "task": task, "x": x, "y": y, "big": big}) support = [draw("A", False)[:2] for _ in range(n_support)] test = [draw("A", False)[:2] for _ in range(n_test)] return {"pool": pool, "support": support, "test": test, "dirs": dirs} def make_projection(d_in, d_out, seed=0): """A Rademacher random projection matrix (entries +1 or -1, scaled). The JL lemma says it roughly keeps dot products.""" rng = random.Random(seed) s = 1.0 / math.sqrt(d_out) return [[s if rng.random() < 0.5 else -s for _ in range(d_in)] for _ in range(d_out)] def project(mat, v): return [dot(row, v) for row in mat] def less_scores(pool, support, w, proj=None, use_cosine=True): """Score each pool example by how well its gradient lines up with the mean gradient of the target examples. use_cosine=False uses the raw dot product instead, which favours large gradients.""" gt = [0.0] * len(w) for x, y in support: for j, v in enumerate(example_grad(w, x, y)): gt[j] += v / len(support) if proj is not None: gt = project(proj, gt) out = {} for ex in pool: g = example_grad(w, ex["x"], ex["y"]) if proj is not None: g = project(proj, g) out[ex["id"]] = cosine(g, gt) if use_cosine else dot(g, gt) return out def less_demo(seed=0, frac=0.2, warmup_frac=0.05, warmup_steps=20, proj_dim=None): """Train on a random, a gradient-selected (cosine), a dot-product-selected and the full pool. Report test accuracy on target task A and how many selected examples came from each task.""" world = make_less_world(seed=seed) pool = world["pool"] rng = random.Random(seed + 100) k = int(len(pool) * frac) warm = rng.sample(pool, max(2, int(len(pool) * warmup_frac))) dim = len(pool[0]["x"]) w0 = train_logistic([(e["x"], e["y"]) for e in warm], steps=warmup_steps, dim=dim) proj = make_projection(dim, proj_dim, seed + 1) if proj_dim else None def fit(ids): by_id = {e["id"]: e for e in pool} return train_logistic([(by_id[i]["x"], by_id[i]["y"]) for i in ids], dim=dim) picks = { "random": sorted(e["id"] for e in rng.sample(pool, k)), "less_cosine": rank_ids(less_scores(pool, world["support"], w0, proj, True), k), "dot_product": rank_ids(less_scores(pool, world["support"], w0, proj, False), k), "full_pool": [e["id"] for e in pool], } by_id = {e["id"]: e for e in pool} rows = {} for name, ids in picks.items(): w = fit(ids) rows[name] = {"n": len(ids), "acc_A": round(accuracy(w, world["test"]), 3), "from_A": sum(by_id[i]["task"] == "A" for i in ids), "from_C": sum(by_id[i]["task"] == "C" for i in ids), "big": sum(by_id[i]["big"] for i in ids)} return rows # ---------------------------------------------------------------- 4. label noise and data maps def make_blobs(n_per_class, k=3, dim=2, sep=3.0, seed=0, spreads=None): """k Gaussian clusters on a circle. spreads gives each cluster its own standard deviation. Returns (points, true labels).""" rng = random.Random(seed) spreads = spreads or [1.0] * k xs, ys = [], [] for c in range(k): ang = 2 * math.pi * c / k centre = [sep * math.cos(ang), sep * math.sin(ang)] + [0.0] * (dim - 2) for _ in range(n_per_class): xs.append([m + rng.gauss(0, spreads[c]) for m in centre]) ys.append(c) return xs, ys def flip_labels(labels, rate, k, seed=0): """Make some labels wrong. A flipped label moves to the next class (c + 1) mod k, so the noise is class-conditional. Returns (noisy labels, sorted ids that were flipped).""" rng = random.Random(seed) noisy, flipped = list(labels), [] for i, y in enumerate(labels): if rng.random() < rate: noisy[i] = (y + 1) % k flipped.append(i) return noisy, flipped def centroid_probs(train_x, train_y, test_x, k, tau=1.5): """A simple classifier: softmax of minus squared distance to each class mean, scaled by tau.""" cent = [] for c in range(k): pts = [x for x, y in zip(train_x, train_y) if y == c] d = len(train_x[0]) cent.append([mean(p[j] for p in pts) if pts else 0.0 for j in range(d)]) out = [] for x in test_x: logits = [-sum((a - b) ** 2 for a, b in zip(x, c)) / (2 * tau * tau) for c in cent] m = max(logits) e = [math.exp(v - m) for v in logits] s = sum(e) out.append([v / s for v in e]) return out def cross_val_probs(xs, labels, k, folds=5, seed=0, tau=1.5): """Out-of-sample predicted probabilities: every example is scored by a model that did not train on it.""" n = len(xs) order = list(range(n)) random.Random(seed).shuffle(order) fold_of = {i: pos % folds for pos, i in enumerate(order)} probs = [None] * n for f in range(folds): tr = [i for i in range(n) if fold_of[i] != f] te = [i for i in range(n) if fold_of[i] == f] p = centroid_probs([xs[i] for i in tr], [labels[i] for i in tr], [xs[i] for i in te], k, tau) for i, pi in zip(te, p): probs[i] = pi return probs def class_thresholds(probs, labels, k): """t_j: the mean predicted probability of class j over the examples labelled j.""" t = [] for j in range(k): ps = [p[j] for p, y in zip(probs, labels) if y == j] t.append(mean(ps)) return t def confident_joint(probs, labels, k): """C[i][j]: examples labelled i that are confidently class j. Confidently means p_j >= t_j, and among the classes that pass, take the one with the largest probability. Returns (matrix, {example id: j}).""" t = class_thresholds(probs, labels, k) C = [[0] * k for _ in range(k)] assigned = {} for idx, (p, y) in enumerate(zip(probs, labels)): ok = [j for j in range(k) if p[j] >= t[j]] if not ok: continue j = max(ok, key=lambda c: (p[c], -c)) C[y][j] += 1 assigned[idx] = j return C, assigned def find_label_issues(probs, labels, k): """Ids whose confident class differs from the given label, worst first (lowest self-confidence).""" _C, assigned = confident_joint(probs, labels, k) bad = [i for i, j in assigned.items() if j != labels[i]] return sorted(bad, key=lambda i: (probs[i][labels[i]], i)) def self_confidence_rank(probs, labels): """All ids, least confident in the given label first. The simplest baseline.""" return sorted(range(len(labels)), key=lambda i: (probs[i][labels[i]], i)) def precision_recall(flagged, truth): flagged, truth = set(flagged), set(truth) tp = len(flagged & truth) return (tp / len(flagged) if flagged else 0.0), (tp / len(truth) if truth else 0.0) def label_noise_demo(n_per_class=100, rate=0.1, seed=0, spreads=(0.6, 0.9, 2.2)): xs, ys = make_blobs(n_per_class, 3, 2, 3.0, seed, list(spreads)) noisy, flipped = flip_labels(ys, rate, 3, seed + 1) probs = cross_val_probs(xs, noisy, 3, seed=seed + 2) cl = find_label_issues(probs, noisy, 3) base = self_confidence_rank(probs, noisy)[:len(cl)] p_cl, r_cl = precision_recall(cl, flipped) p_b, r_b = precision_recall(base, flipped) return {"n": len(xs), "flipped": len(flipped), "cl_flagged": len(cl), "cl_precision": round(p_cl, 3), "cl_recall": round(r_cl, 3), "base_precision": round(p_b, 3), "base_recall": round(r_b, 3), "unfixed_error_rate": round(mean(1.0 if probs[i].index(max(probs[i])) != noisy[i] else 0.0 for i in range(len(xs))), 3)} def label_noise_study(seeds=8, **kw): """Average label_noise_demo over several seeds.""" rows = [label_noise_demo(seed=sd, **kw) for sd in range(seeds)] keys = ["flipped", "cl_flagged", "cl_precision", "cl_recall", "base_precision", "base_recall"] return {k: round(mean(r[k] for r in rows), 3) for k in keys} def training_dynamics(p_true_by_epoch, correct_by_epoch): """Data-map statistics for one example. confidence is the mean probability of the given label across epochs, variability is the standard deviation of that probability, and correctness is the share of epochs predicted right.""" return {"confidence": mean(p_true_by_epoch), "variability": stdev(p_true_by_epoch), "correctness": mean(1.0 if c else 0.0 for c in correct_by_epoch)} def train_logistic_history(data, epochs=10, lr=0.1, seed=0): """One-example-at-a-time SGD for a binary task. After each epoch, record p(given label) and whether it is right for every example. Returns (p_hist[i][e], correct_hist[i][e]).""" rng = random.Random(seed) d = len(data[0][0]) w = [0.0] * d order = list(range(len(data))) p_hist = [[] for _ in data] c_hist = [[] for _ in data] for _ in range(epochs): rng.shuffle(order) for i in order: x, y = data[i] g = example_grad(w, x, y) w = [wj - lr * gj for wj, gj in zip(w, g)] for i, (x, y) in enumerate(data): p1 = logistic_p(w, x) p_hist[i].append(p1 if y == 1 else 1 - p1) c_hist[i].append((p1 >= 0.5) == (y == 1)) return p_hist, c_hist def data_map_regions(stats, fraction=1 / 3): """Split ids into the three regions of a data map: hard (lowest confidence), easy (highest confidence) and ambiguous (highest variability). An id can be in more than one set, so ambiguous is taken first.""" ids = sorted(stats) n = max(1, int(len(ids) * fraction)) amb = rank_ids({i: stats[i]["variability"] for i in ids}, n) rest = [i for i in ids if i not in set(amb)] hard = rank_ids({i: stats[i]["confidence"] for i in rest}, n, largest=False) easy = rank_ids({i: stats[i]["confidence"] for i in rest if i not in set(hard)}, n) return {"ambiguous": sorted(amb), "hard": sorted(hard), "easy": sorted(easy)} def data_map_demo(n=300, noise=0.08, seed=0, fraction=0.15): rng = random.Random(seed) xs = [[rng.gauss(0, 1), rng.gauss(0, 1), 1.0] for _ in range(n)] clean = [1 if x[0] + 0.5 * x[1] > 0 else 0 for x in xs] noisy, flipped = [], [] for i, y in enumerate(clean): if rng.random() < noise: noisy.append(1 - y) flipped.append(i) else: noisy.append(y) ph, ch = train_logistic_history(list(zip(xs, noisy)), epochs=10, lr=0.1, seed=seed) stats = {i: training_dynamics(ph[i], ch[i]) for i in range(n)} reg = data_map_regions(stats, fraction) f = set(flipped) return {"n": n, "flipped": len(f), "share_flipped_in_hard": round(len(f & set(reg["hard"])) / max(1, len(reg["hard"])), 3), "share_flipped_in_easy": round(len(f & set(reg["easy"])) / max(1, len(reg["easy"])), 3), "share_flipped_in_ambiguous": round(len(f & set(reg["ambiguous"])) / max(1, len(reg["ambiguous"])), 3), "base_rate": round(len(f) / n, 3)} # ---------------------------------------------------------------- 5. agreement among annotators def percent_agreement(a, b): return mean(1.0 if x == y else 0.0 for x, y in zip(a, b)) def _marginals(vals): n = len(vals) out = {} for v in vals: out[v] = out.get(v, 0) + 1 return {k: c / n for k, c in out.items()} def cohen_kappa(a, b): """Two coders. Chance agreement uses each coder's own label shares.""" po = percent_agreement(a, b) ma, mb = _marginals(a), _marginals(b) pe = sum(ma.get(c, 0.0) * mb.get(c, 0.0) for c in set(ma) | set(mb)) return 1.0 if pe >= 1.0 else (po - pe) / (1 - pe) def scott_pi(a, b): """Two coders. Chance agreement uses the pooled label shares of both coders.""" po = percent_agreement(a, b) pooled = _marginals(list(a) + list(b)) pe = sum(p * p for p in pooled.values()) return 1.0 if pe >= 1.0 else (po - pe) / (1 - pe) def weighted_kappa(a, b, dist): """Cohen's kappa with partial credit. dist(x, y) is 0 for equal labels and 1 for maximal disagreement.""" do = mean(dist(x, y) for x, y in zip(a, b)) ma, mb = _marginals(a), _marginals(b) de = sum(pa * pb * dist(x, y) for x, pa in ma.items() for y, pb in mb.items()) return 1.0 if de == 0 else 1 - do / de def fleiss_multi_pi(ratings): """Fleiss's coefficient for any number of raters, with the same number r of raters on every item. It generalises Scott's pi (pooled shares), not Cohen's kappa, although it is often called Fleiss's kappa. ratings is a list with one list of labels per item.""" r = len(ratings[0]) if r < 2 or any(len(x) != r for x in ratings): raise ValueError("every item needs the same number of raters, at least two") ao = 0.0 for item in ratings: counts = {} for v in item: counts[v] = counts.get(v, 0) + 1 ao += sum(c * (c - 1) for c in counts.values()) / (r * (r - 1)) ao /= len(ratings) pooled = _marginals([v for item in ratings for v in item]) pe = sum(p * p for p in pooled.values()) return 1.0 if pe >= 1.0 else (ao - pe) / (1 - pe) def nominal_distance(a, b): return 0.0 if a == b else 1.0 def squared_distance(a, b): return float((a - b) ** 2) def krippendorff_alpha(units, dist=nominal_distance): """Krippendorff's alpha. units is a list with one list of values per item. None means a missing rating, and the number of raters may differ per item. Items with fewer than two values are ignored.""" vals = [[v for v in u if v is not None] for u in units] vals = [u for u in vals if len(u) >= 2] coinc = {} for u in vals: m = len(u) for i in range(m): for j in range(m): if i != j: key = (u[i], u[j]) coinc[key] = coinc.get(key, 0.0) + 1.0 / (m - 1) n_c = {} for (c, _k), v in coinc.items(): n_c[c] = n_c.get(c, 0.0) + v n = sum(n_c.values()) if n <= 1: return 1.0 do = sum(v * dist(c, k) for (c, k), v in coinc.items()) / n de = sum(n_c[c] * n_c[k] * dist(c, k) for c in n_c for k in n_c) / (n * (n - 1)) return 1.0 if de == 0 else 1 - do / de def ensemble_max(labels): """The most severe label wins. This is how the SWE-bench Verified post ensembles three annotators.""" return max(labels) def ensemble_median(labels): s = sorted(labels) return s[len(s) // 2] def keep_by_severity(annotations, ensemble=ensemble_max, threshold=2): """annotations is {id: [[labels of criterion 1 by annotator], [labels of criterion 2 by annotator], ...]}. An item is kept only when every criterion's ensemble label is below threshold.""" return sorted(i for i, crits in annotations.items() if all(ensemble(c) < threshold for c in crits)) def simulate_annotation(n_items=500, n_annotators=3, accuracy=0.7, seed=0, severity_probs=(0.45, 0.25, 0.2, 0.1)): """Made-up severity labels 0 to 3 for two criteria. Each annotator reports the true label with probability accuracy and otherwise one step off. Returns (annotations, ids whose true severity is below 2 on both).""" rng = random.Random(seed) ann, ok = {}, [] for i in range(n_items): truth = [rng.choices(range(4), severity_probs)[0] for _ in range(2)] crits = [] for t in truth: labs = [] for _ in range(n_annotators): if rng.random() < accuracy: labs.append(t) else: step = rng.choice([-1, 1]) labs.append(min(3, max(0, t + step))) crits.append(labs) ann[i] = crits if all(t < 2 for t in truth): ok.append(i) return ann, ok def selection_quality(kept, truly_ok, n_total): """Precision of the kept set (how many are truly fine), recall of the fine items, and the share kept.""" kept, truly_ok = set(kept), set(truly_ok) tp = len(kept & truly_ok) return {"kept_share": round(len(kept) / n_total, 3), "precision": round(tp / len(kept), 3) if kept else 0.0, "recall": round(tp / len(truly_ok), 3) if truly_ok else 0.0} # ---------------------------------------------------------------- 6. item response theory and tiny benchmarks def kendall_tau_b(x, y): """Kendall's tau-b rank correlation, which handles ties. Returns 0.0 when one side is constant.""" n = len(x) conc = disc = tx = ty = 0 for i in range(n): for j in range(i + 1, n): dx, dy = x[i] - x[j], y[i] - y[j] if dx == 0 and dy == 0: continue if dx == 0: tx += 1 elif dy == 0: ty += 1 elif (dx > 0) == (dy > 0): conc += 1 else: disc += 1 den = math.sqrt((conc + disc + tx) * (conc + disc + ty)) return 0.0 if den == 0 else (conc - disc) / den def pearson(x, y): mx, my = mean(x), mean(y) sx, sy = stdev(x), stdev(y) if sx == 0 or sy == 0: return 0.0 return mean((a - mx) * (b - my) for a, b in zip(x, y)) / (sx * sy) def make_irt_world(n_models, n_items, seed=0, dims=1): """Made-up response matrix. The chance that model m gets item i right is sigmoid(alpha_i . theta_m - beta_i). With dims=2 the abilities are two correlated skills and each item loads on them in its own mix.""" rng = random.Random(seed) theta, alpha, beta = [], [], [] for _ in range(n_models): t0 = rng.gauss(0, 1) theta.append([t0] if dims == 1 else [t0, 0.5 * t0 + math.sqrt(0.75) * rng.gauss(0, 1)]) for _ in range(n_items): a = math.exp(rng.gauss(0, 0.35)) if dims == 1: alpha.append([a]) else: phi = rng.random() * math.pi / 2 alpha.append([a * math.cos(phi), a * math.sin(phi)]) beta.append(rng.gauss(0, 1.2)) resp = [[1 if rng.random() < sigmoid(dot(alpha[i], theta[m]) - beta[i]) else 0 for i in range(n_items)] for m in range(n_models)] return {"resp": resp, "theta": theta, "alpha": alpha, "beta": beta} def fit_irt(resp, model="2pl", epochs=150, lr=2.0): """Fit a one-dimensional IRT model by gradient ascent with weak priors. 1pl fixes every alpha at 1 (the Rasch model). Returns (theta list, alpha list, beta list). Priors: theta ~ N(0,1), alpha pulled toward 1, beta lightly shrunk.""" M, N = len(resp), len(resp[0]) acc_m = [mean(r) for r in resp] acc_i = [mean(resp[m][i] for m in range(M)) for i in range(N)] theta = [logit(a * 0.98 + 0.01) / 1.5 for a in acc_m] beta = [-logit(a * 0.98 + 0.01) for a in acc_i] alpha = [1.0] * N free_alpha = model == "2pl" for _ in range(epochs): gt = [0.0] * M ga = [0.0] * N gb = [0.0] * N for m in range(M): tm = theta[m] row = resp[m] for i in range(N): e = row[i] - sigmoid(alpha[i] * tm - beta[i]) gt[m] += alpha[i] * e ga[i] += tm * e gb[i] -= e for m in range(M): theta[m] += lr * (gt[m] - theta[m]) / N for i in range(N): beta[i] += lr * (gb[i] - 0.1 * beta[i]) / M if free_alpha: alpha[i] += lr * (ga[i] - (alpha[i] - 1.0)) / M alpha[i] = min(max(alpha[i], 0.05), 5.0) return theta, alpha, beta def fit_theta(alpha, beta, items, y): """MAP ability of one new model from its answers y[i] on a subset of items, with the item parameters fixed.""" t = 0.0 for _ in range(30): g = -t h = 1.0 for i in items: p = sigmoid(alpha[i] * t - beta[i]) g += alpha[i] * (y[i] - p) h += alpha[i] ** 2 * p * (1 - p) t += g / h return t def estimate_pirt(alpha, beta, items, y): """Predicted accuracy on all items from a model's answers on a few: fit its ability, then average the predicted chance.""" t = fit_theta(alpha, beta, items, y) return mean(sigmoid(alpha[i] * t - beta[i]) for i in range(len(alpha))) def fisher_information(alpha_i, beta_i, theta): p = sigmoid(alpha_i * theta - beta_i) return alpha_i ** 2 * p * (1 - p) def select_random(n_items, k, seed=0): return sorted(random.Random(seed).sample(range(n_items), k)) def select_stratified(item_acc, k): """Sort items by how often the training models got them right, and take k evenly spaced ones.""" order = sorted(range(len(item_acc)), key=lambda i: (item_acc[i], i)) n = len(order) return sorted(order[int((j + 0.5) * n / k)] for j in range(k)) def select_irt_grid(alpha, beta, thetas, k): """For ability levels spread over the fitted models, pick the item with the most Fisher information at that level (taking levels in turn) until k items are chosen.""" ts = sorted(thetas) n = len(ts) grid = [ts[int((j + 0.5) * n / k)] for j in range(k)] chosen = [] for g in grid: best = max((i for i in range(len(alpha)) if i not in chosen), key=lambda i: (fisher_information(alpha[i], beta[i], g), -i)) chosen.append(best) return sorted(chosen) def select_anchors(resp_train, k): """Greedy k-center on the columns of the response matrix (an item is a vector of right/wrong over models). Returns (anchor ids, weights) where an anchor's weight is the share of items closest to it.""" M, N = len(resp_train), len(resp_train[0]) cols = [tuple(resp_train[m][i] for m in range(M)) for i in range(N)] def d(a, b): return sum(x != y for x, y in zip(cols[a], cols[b])) / M var = [pass_variance(mean(c)) for c in cols] start = min(range(N), key=lambda i: (-var[i], i)) anchors = [start] dmin = [d(i, start) for i in range(N)] while len(anchors) < k: nxt = min(range(N), key=lambda i: (-dmin[i], i)) anchors.append(nxt) for i in range(N): dmin[i] = min(dmin[i], d(i, nxt)) owner = [min(range(len(anchors)), key=lambda a: (d(i, anchors[a]), a)) for i in range(N)] w = [owner.count(a) / N for a in range(len(anchors))] return anchors, w def weighted_accuracy(items, weights, y): return sum(w * y[i] for i, w in zip(items, weights)) / sum(weights) def tiny_benchmark_study(seed=0, n_train=40, n_test=24, n_items=200, k=24, dims=1, epochs=120): """Fit IRT on train models, choose k items, then estimate each held-out model's full-benchmark accuracy from those k items. Report Kendall tau (ranking) and mean absolute error (accuracy) for several recipes.""" world = make_irt_world(n_train + n_test, n_items, seed, dims) R = world["resp"] train, test = R[:n_train], R[n_train:] theta, alpha, beta = fit_irt(train, "2pl", epochs) _t1, _a1, b1 = fit_irt(train, "1pl", epochs) a1 = [1.0] * n_items item_acc = [mean(train[m][i] for m in range(n_train)) for i in range(n_items)] full = [mean(r) for r in test] rnd = select_random(n_items, k, seed + 7) strat = select_stratified(item_acc, k) irt_sel = select_irt_grid(alpha, beta, theta, k) anc, anc_w = select_anchors(train, k) recipes = { "random, mean": lambda y: mean(y[i] for i in rnd), "stratified by difficulty, mean": lambda y: mean(y[i] for i in strat), "anchor points, weighted mean": lambda y: weighted_accuracy(anc, anc_w, y), "random, 2PL ability": lambda y: estimate_pirt(alpha, beta, rnd, y), "IRT-chosen, 2PL ability": lambda y: estimate_pirt(alpha, beta, irt_sel, y), "IRT-chosen, 1PL ability": lambda y: estimate_pirt(a1, b1, select_irt_grid(a1, b1, theta, k), y), } out = {} for name, f in recipes.items(): est = [f(y) for y in test] out[name] = {"tau": round(kendall_tau_b(est, full), 3), "mae": round(mean(abs(e - t) for e, t in zip(est, full)), 4)} return out def tiny_benchmark_average(seeds=20, **kw): """Average tiny_benchmark_study over several made-up worlds. Returns {recipe: {tau, tau_sd, mae}}.""" rows = [tiny_benchmark_study(seed=s, **kw) for s in range(seeds)] out = {} for name in rows[0]: taus = [r[name]["tau"] for r in rows] out[name] = {"tau": round(mean(taus), 3), "tau_sd": round(stdev(taus), 3), "mae": round(mean(r[name]["mae"] for r in rows), 4)} return out # ---------------------------------------------------------------- 7. EvalGen-style alignment and criteria drift @dataclass(frozen=True) class Assertion: """A yes/no check on one output. fn(output) is True when the output passes.""" name: str criterion: str fn: object = field(compare=False, hash=False, repr=False) def coverage(fails, good): """Of the outputs the user graded bad, the share that the assertion set fails. fails[i] is True when some assertion fails output i.""" bad = [i for i in range(len(good)) if not good[i]] return mean(1.0 if fails[i] else 0.0 for i in bad) if bad else 0.0 def false_failure_rate(fails, good): """Of the outputs the user graded good, the share that the assertion set fails anyway.""" ok = [i for i in range(len(good)) if good[i]] return mean(1.0 if fails[i] else 0.0 for i in ok) if ok else 0.0 def alignment(cov, ffr): """Harmonic mean of coverage and (1 - false failure rate).""" a, b = cov, 1.0 - ffr return 0.0 if a + b == 0 else 2 * a * b / (a + b) def set_fails(assertions, outputs): return [any(not a.fn(o) for a in assertions) for o in outputs] def selectivity(assertion, outputs): """The pass rate of an assertion over a set of outputs.""" return mean(1.0 if assertion.fn(o) else 0.0 for o in outputs) def failure_score(output, assertions, sel): """Sum of the selectivity of the assertions that fail this output. A high score means a rarely-failing assertion failed it, so it is probably bad. The paper's formula and its prose disagree on the indicator. This follows the prose.""" return sum(sel[a.name] for a in assertions if not a.fn(output)) def grading_order(outputs, assertions, policy, seed=0): """The order in which to ask a person to grade outputs. policy is random, highest, lowest or alternating.""" ids = [o["id"] for o in outputs] if policy == "random": ids = list(ids) random.Random(seed).shuffle(ids) return ids sel = {a.name: selectivity(a, outputs) for a in assertions} score = {o["id"]: failure_score(o, assertions, sel) for o in outputs} hi = rank_ids(score) if policy == "highest": return hi lo = rank_ids(score, largest=False) if policy == "lowest": return lo if policy == "alternating": order, seen = [], set() for a, b in zip(hi, lo): for x in (a, b): if x not in seen: seen.add(x) order.append(x) return order raise ValueError("unknown policy " + policy) def select_per_criterion(assertions, outputs, graded_ids, good_by_id, ffr_max=0.2): """For each criterion pick the assertion with the best alignment on the graded outputs, among those whose false failure rate is at most ffr_max. Returns {criterion: Assertion or None}.""" sub = [o for o in outputs if o["id"] in graded_ids] good = [good_by_id[o["id"]] for o in sub] chosen = {} for crit in sorted({a.criterion for a in assertions}): best, best_score = None, -1.0 for a in sorted((x for x in assertions if x.criterion == crit), key=lambda x: x.name): f = set_fails([a], sub) ffr = false_failure_rate(f, good) if ffr > ffr_max: continue s = alignment(coverage(f, good), ffr) if s > best_score: best, best_score = a, s chosen[crit] = best return chosen def make_outputs(n=100, seed=0): """Made-up outputs of an entity-extraction prompt: the share of extracted entities that are proper nouns, the word count, and whether a hashtag was copied.""" rng = random.Random(seed) out = [] for i in range(n): out.append({"id": i, "proper_frac": rng.choices([0.0, 1 / 3, 0.5, 2 / 3, 1.0], [10, 15, 15, 20, 40])[0], "words": rng.randint(20, 90), "hashtag": rng.random() < 0.3}) return out def user_grade_strict(o): """Version 1 of the made-up user's rule: every entity is a proper noun, and at most 60 words.""" return o["proper_frac"] >= 1.0 and o["words"] <= 60 def user_grade_relaxed(o): """Version 2, after the user has seen outputs: most entities are proper nouns is enough.""" return o["proper_frac"] >= 0.5 and o["words"] <= 60 def candidate_assertions(): out = [] for t in (1.0, 0.6, 0.5, 0.3): out.append(Assertion(f"proper>={t}", "proper_nouns", lambda o, t=t: o["proper_frac"] >= t - 1e-9)) for cap in (40, 50, 60, 80): out.append(Assertion(f"words<={cap}", "length", lambda o, cap=cap: o["words"] <= cap)) out.append(Assertion("no_hashtag", "hashtags", lambda o: not o["hashtag"])) return out def set_alignment(assertions, outputs, good): f = set_fails(assertions, outputs) ffr = false_failure_rate(f, good) return alignment(coverage(f, good), ffr) def criteria_drift_demo(n=200, n_grade=20, seed=0, ffr_max=0.2, policy="random"): """A user grades n_grade outputs with the strict rule. The assertions picked from that look good on those grades. Then the user's rule relaxes. The old pick scores worse against the new grades. Regrading the same outputs under the new rule and picking again recovers.""" outs = make_outputs(n, seed) cands = candidate_assertions() graded = set(grading_order(outs, cands, policy, seed)[:n_grade]) g1 = {o["id"]: user_grade_strict(o) for o in outs} g2 = {o["id"]: user_grade_relaxed(o) for o in outs} pick1 = select_per_criterion(cands, outs, graded, g1, ffr_max) pick2 = select_per_criterion(cands, outs, graded, g2, ffr_max) set1 = [a for a in pick1.values() if a] set2 = [a for a in pick2.values() if a] good1 = [g1[o["id"]] for o in outs] good2 = [g2[o["id"]] for o in outs] return {"graded": n_grade, "picked_early": sorted(a.name for a in set1), "picked_after_drift": sorted(a.name for a in set2), "early_set_vs_strict_grades": round(set_alignment(set1, outs, good1), 3), "early_set_vs_relaxed_grades": round(set_alignment(set1, outs, good2), 3), "regraded_set_vs_relaxed_grades": round(set_alignment(set2, outs, good2), 3)} def sampling_study(policies=("random", "highest", "lowest", "alternating"), worlds=10, n=100, n_grade=16): """Alignment against the strict grades on all outputs, when the user grades only n_grade outputs chosen by each policy. Returns {policy: {mean, sd}} over several made-up worlds.""" cands = candidate_assertions() res = {p: [] for p in policies} for w in range(worlds): outs = make_outputs(n, w) g = {o["id"]: user_grade_strict(o) for o in outs} good = [g[o["id"]] for o in outs] for p in policies: graded = set(grading_order(outs, cands, p, seed=w)[:n_grade]) picks = [a for a in select_per_criterion(cands, outs, graded, g).values() if a] res[p].append(set_alignment(picks, outs, good)) return {p: {"mean": round(mean(v), 3), "sd": round(stdev(v), 3)} for p, v in res.items()} def paper_table_alignment(): """Alignment recomputed from the coverage and false failure rates printed in the EvalGen paper's Table 1.""" rows = {"medical EvalGen and SPADE": (0.33, 0.10), "product EvalGen": (0.73, 0.39), "product SPADE": (0.49, 0.39)} return {k: round(100 * alignment(c, f), 2) for k, (c, f) in rows.items()} # ---------------------------------------------------------------- 8. trace-to-dataset pipeline with quality gates @dataclass class Trace: id: int task_id: str repo: str prompt: str actions: list passed: bool consent_train: bool = True defect: str = "" # ground truth for the lab only: what was injected, "" for none REPOS = ["libalpha", "libbeta", "libgamma", "libdelta", "libomega", "libsigma"] NOUNS = ["date", "row", "path", "unit", "cfg", "key", "vec", "text", "blob", "id", "grid", "tok"] ACTS = ["parse", "merge", "split", "norm", "read", "join", "scale", "trim", "load", "sort", "pad", "count"] FUNCS = [f"{a}_{n}" for a in ACTS for n in NOUNS] SYMPTOMS = ["returns an empty list for repeated keys.", "raises a type error when the input is a tuple.", "drops the last item of an odd-length input.", "ignores the timezone argument.", "loops forever on an empty string.", "mixes up rows and columns on wide input.", "gives a wrong total after the second call.", "fails on paths that end with a slash.", "returns none when the key is missing.", "rounds half values the wrong way.", "skips blank lines at the top.", "keeps stale values between calls."] VERBS = ["view", "grep", "edit", "run"] EMAIL_TAIL = " Please email jo.lee@example.com when it is done." def make_prompt(repo, fn, symptom, rng=None): """A task prompt. The example numbers make each prompt unique. Without rng they are fixed.""" r = rng or random.Random(0) a, b, c, d = [r.randint(100, 999) for _ in range(4)] return (f"In {repo} the function {fn} {symptom} Example: {fn} of {a} and {b} gives {c} but should give {d}. " "Fix the bug and make the failing tests pass.") def _reword(prompt, rng): """Change one word in the middle of a prompt: a near copy, not an exact one.""" w = prompt.split() i = rng.randrange(3, len(w) - 6) w[i] = w[i] + "x" return " ".join(w) def make_eval_prompts(n=12, seed=0): rng = random.Random(seed) seen, out = set(), [] while len(out) < n: p = make_prompt(rng.choice(REPOS), rng.choice(FUNCS), rng.choice(SYMPTOMS), rng) if p not in seen: seen.add(p) out.append(p) return out def make_trace_pool(n_tasks=150, seed=0, eval_prompts=None): """A made-up pool of agent traces with known injected defects. Defect names: loop, tamper (edits the tests), pii, leak (prompt is a near copy of an eval prompt), near_dup (task is a near copy of an earlier task), truncated (no submit step), weak_tests (passes, but only because the tests are weak, and nothing can see it).""" rng = random.Random(seed) eval_prompts = eval_prompts or make_eval_prompts(12, seed + 1) tasks, used = [], set(eval_prompts) for t in range(n_tasks): kind = rng.choices(["plain", "near_dup", "leak"], [86, 8, 6])[0] if t >= 3 else "plain" if kind == "near_dup": src = rng.choice([x for x in tasks if x["kind"] == "plain"]) prompt, repo = _reword(src["prompt"], rng), src["repo"] elif kind == "leak": prompt = _reword(rng.choice(eval_prompts), rng) repo = rng.choice(REPOS) else: while True: repo = rng.choice(REPOS) prompt = make_prompt(repo, rng.choice(FUNCS), rng.choice(SYMPTOMS), rng) if prompt not in used: used.add(prompt) break p = rng.choice([0.1, 0.3, 0.5, 0.7, 0.85, 0.97]) tasks.append({"id": f"t{t:03d}", "repo": repo, "prompt": prompt, "kind": kind, "p": p, "weak": rng.random() < 0.05}) traces = [] for t in tasks: for _ in range(rng.randint(3, 10)): acts, last = [], "" for _ in range(rng.randint(8, 20)): v = rng.choice([x for x in VERBS if x != last or rng.random() < 0.3]) acts.append(f"{v} src/{rng.choice(FUNCS)}.py") last = v acts.append("submit") passed = rng.random() < t["p"] defect = {"near_dup": "near_dup", "leak": "leak"}.get(t["kind"], "") prompt = t["prompt"] r = rng.random() if r < 0.06: k = rng.randrange(1, len(acts)) acts = acts[:k] + ["view src/core.py"] * 12 + acts[k:] defect, passed = "loop", rng.random() < 0.4 elif r < 0.11: acts.insert(len(acts) - 1, "edit tests/test_core.py") defect, passed = "tamper", True elif r < 0.15: prompt += EMAIL_TAIL defect = "pii" elif r < 0.19: acts = acts[:-1] defect = "truncated" elif t["weak"] and rng.random() < 0.5: defect, passed = "weak_tests", True traces.append(Trace(len(traces), t["id"], t["repo"], prompt, acts, passed, rng.random() > 0.12, defect)) return traces def longest_run(actions): """Length of the longest run of the same first word (the base command, without arguments).""" best = run = 0 prev = None for a in actions: w = a.split()[0] if a.split() else "" run = run + 1 if w == prev else 1 prev = w best = max(best, run) return best _EMAIL = re.compile(r"[\w.+-]+@[\w-]+\.[\w.-]+") _PHONE = re.compile(r"\b\d{3}[-. ]\d{4}\b") def has_pii(text): return bool(_EMAIL.search(text) or _PHONE.search(text)) def gate_consent(traces, **_): return [t for t in traces if t.consent_train] def gate_resolved(traces, **_): return [t for t in traces if t.passed] def gate_complete(traces, **_): return [t for t in traces if t.actions and t.actions[-1] == "submit"] def gate_loops(traces, max_run=10, **_): return [t for t in traces if longest_run(t.actions) < max_run] def gate_tamper(traces, **_): return [t for t in traces if not any(a.startswith("edit tests/") or "sys.exit(0)" in a or "conftest" in a for a in t.actions)] def gate_pii(traces, **_): return [t for t in traces if not has_pii(t.prompt)] def gate_leak(traces, eval_prompts=(), leak_threshold=0.6, **_): texts = {("e", i): p for i, p in enumerate(eval_prompts)} texts.update({("t", t.id): t.prompt for t in traces}) bad = set() for a, b, _j in near_duplicates(texts, leak_threshold): if a[0] != b[0]: bad.add(a[1] if a[0] == "t" else b[1]) return [t for t in traces if t.id not in bad] def gate_dedupe_tasks(traces, dup_threshold=0.6, **_): """Drop every trace of a task whose prompt is a near copy of an earlier task's prompt.""" prompts = {} for t in sorted(traces, key=lambda x: x.id): prompts.setdefault(t.task_id, t.prompt) _kept, dropped = dedupe_keep_first(prompts, dup_threshold) return [t for t in traces if t.task_id not in dropped] def gate_not_trivial(traces, rates=None, max_pass_rate=0.9, **_): """Drop tasks that the agent almost always solves. rates is {task id: pass rate over all attempts}.""" rates = rates or {} return [t for t in traces if rates.get(t.task_id, 0.0) <= max_pass_rate] def gate_cap(traces, cap=3, **_): return cap_per_group(sorted(traces, key=lambda t: t.id), lambda t: t.task_id, cap) GATES = [("consent", gate_consent), ("resolved", gate_resolved), ("complete", gate_complete), ("loops", gate_loops), ("tamper", gate_tamper), ("pii", gate_pii), ("leak", gate_leak), ("dedupe tasks", gate_dedupe_tasks), ("not trivial", gate_not_trivial), ("cap per task", gate_cap)] def task_pass_rates(traces): by = {} for t in traces: by.setdefault(t.task_id, []).append(1.0 if t.passed else 0.0) return {k: mean(v) for k, v in by.items()} def run_pipeline(traces, eval_prompts=(), skip=(), **kw): """Run the gates in order. Returns {kept, funnel: [(gate, n_in, n_out)]}. skip names gates to leave out.""" rates = task_pass_rates(traces) cur = list(traces) funnel = [("pool", len(cur), len(cur))] for name, gate in GATES: if name in skip: continue nxt = gate(cur, eval_prompts=eval_prompts, rates=rates, **kw) funnel.append((name, len(cur), len(nxt))) cur = nxt return {"kept": cur, "funnel": funnel} def defect_report(pool, kept): """For each injected defect: how many were in the pool and how many are still in the kept set. Also how many clean, passing traces the gates removed. Ground truth is available only because the lab made the data.""" kept_ids = {t.id for t in kept} out = {} for d in sorted({t.defect for t in pool if t.defect}): ids = [t.id for t in pool if t.defect == d] out[d] = {"in_pool": len(ids), "kept": sum(i in kept_ids for i in ids)} good = [t.id for t in pool if not t.defect and t.passed and t.consent_train] out["clean_passing"] = {"in_pool": len(good), "kept": sum(i in kept_ids for i in good)} return out def wilson_interval(k, n, z=1.96): """A 95 percent Wilson interval for a share k out of n.""" if n == 0: return (0.0, 1.0) p = k / n d = 1 + z * z / n c = (p + z * z / (2 * n)) / d h = z * math.sqrt(p * (1 - p) / n + z * z / (4 * n * n)) / d return (max(0.0, c - h), min(1.0, c + h)) def audit_sample(kept, n, seed=0): """A sample for human review: round robin over repos so that no repo is missed.""" rng = random.Random(seed) by = {} for t in kept: by.setdefault(t.repo, []).append(t) for v in by.values(): rng.shuffle(v) out = [] while len(out) < min(n, len(kept)): for r in sorted(by): if by[r] and len(out) < n: out.append(by[r].pop()) return sorted(out, key=lambda t: t.id) def dataset_card(kept, funnel, eval_prompts=()): return {"traces": len(kept), "tasks": len({t.task_id for t in kept}), "repos": sorted({t.repo for t in kept}), "gates": [f[0] for f in funnel[1:]], "eval_prompts_checked": len(eval_prompts), "known_gaps": ["weak tests are not detected by any gate", "gates only catch what they were written for", "all traces and defects here are made up"]} def artstein_example(): """The two-coder example from Artstein and Poesio (2008), Table 4: 100 utterances, three labels. Returns (coder A, coder B).""" pairs = ([("stat", "stat")] * 46 + [("ireq", "stat")] * 6 + [("ireq", "ireq")] * 32 + [("ireq", "chck")] * 6 + [("chck", "chck")] * 10) return [p[0] for p in pairs], [p[1] for p in pairs] def artstein_distance(a, b): """The weights used in that paper: check is half way between statement and info request.""" if a == b: return 0.0 return 0.5 if "chck" in (a, b) else 1.0 def prevalence_demo(): """Ninety percent agreement and a kappa below zero, because almost every item has the same label.""" a = ["yes"] * 90 + ["yes"] * 5 + ["no"] * 5 b = ["yes"] * 90 + ["no"] * 5 + ["yes"] * 5 return {"agreement": round(percent_agreement(a, b), 3), "kappa": round(cohen_kappa(a, b), 3), "alpha": round(krippendorff_alpha([[x, y] for x, y in zip(a, b)]), 3)} def main(): print("== 1. informativeness: find the tasks with the most signal (made-up pass probabilities)") for r in informativeness_demo(): print(" attempts per task %3d overlap with true top 50: %.2f share with 0 %4d" % (g, n_in, n_out)) print(" defects in pool and still kept:", {k: (v["in_pool"], v["kept"]) for k, v in defect_report(pool, res["kept"]).items()}) skip = run_pipeline(pool, ev, skip=("complete", "tamper")) print(" without the complete and tamper gates:", {k: (v["in_pool"], v["kept"]) for k, v in defect_report(pool, skip["kept"]).items() if k in ("truncated", "tamper")}) sample = audit_sample(res["kept"], 40, seed=2) weak = sum(t.defect == "weak_tests" for t in sample) lo, hi = wilson_interval(weak, len(sample)) print(" audit of %d kept traces: %d with weak tests, 95 percent interval %.3f to %.3f; true rate %.3f" % (len(sample), weak, lo, hi, sum(t.defect == "weak_tests" for t in res["kept"]) / len(res["kept"]))) print("== 9. data card") print(json.dumps(dataset_card(res["kept"], res["funnel"], ev), indent=1, sort_keys=True)) if __name__ == "__main__": main()