| """Classification example on synthetic data: Claude fixes a logistic model's residuals. |
| |
| Each example belongs to the class with the largest score, with 2% label noise: |
| |
| s0 = -2 x2 x5 |
| s1 = 2 x2 x5 |
| s2 = 1.5 (x4^2 - 1) + 0.8 x1 - 0.6 x3 + 0.5 |
| |
| Here x ~ N(0, I_6), N = 1000 (600 / 200 / 200), and x6 is irrelevant. Logistic regression can use |
| the linear x1 and x3 terms, but it cannot represent the x2 x5 interaction or the even x4 term. |
| |
| Protocol (paper Sec. 3.5): residuals, the high-residual pool, critiques and p_k use the training |
| split; the retained iterate m_k* and the blend beta are chosen on validation; test is used only |
| for final metrics. Predictions follow Eq. 2. |
| |
| Usage: |
| python classification.py --seed 0 # real run (needs Anthropic credentials) |
| python classification.py --seed 0 --mock --T 3 # offline pipeline check, no API calls |
| """ |
| from __future__ import annotations |
|
|
| import argparse |
| import contextlib |
| import json |
| import os |
| import random |
| import sys |
| import time |
|
|
| import numpy as np |
| from sklearn.linear_model import LogisticRegression |
| from sklearn.metrics import accuracy_score, f1_score, log_loss |
| from xgboost import XGBClassifier |
|
|
| from claude_backend import ClaudeLLM, install_import_shims |
|
|
| install_import_shims() |
| import maricl_core as mc |
| from maricl_headline import maricl_predict_proba, select_beta, select_classification |
| from regression import FEATURE_DESCS, FEATURES, load_dotenv, term_flags |
|
|
| DOMAIN = ("Synthetic 3-class classification benchmark with standardised inputs. Feature semantics " |
| "are unknown and no domain knowledge applies.") |
| N, D, C = 1000, 6, 3 |
|
|
|
|
| def true_scores(X): |
| x1, x2, x3, x4, x5 = (X[:, i] for i in range(5)) |
| return np.stack([-2.0 * x2 * x5, 2.0 * x2 * x5, |
| 1.5 * (x4**2 - 1.0) + 0.8 * x1 - 0.6 * x3 + 0.5], axis=1) |
|
|
|
|
| def make_data(seed): |
| rng = np.random.default_rng(seed) |
| X = rng.standard_normal((N, D)) |
| clean = true_scores(X).argmax(axis=1) |
| y = clean.copy() |
| flip = rng.random(N) < 0.02 |
| y[flip] = rng.integers(0, C, size=int(flip.sum())) |
| idx = np.random.RandomState(seed).permutation(N) |
| split = {"train": idx[:600], "val": idx[600:800], "test": idx[800:]} |
| return X, y, clean, split |
|
|
|
|
| def metrics(y, proba): |
| pred = proba.argmax(axis=1) |
| return {"accuracy": float(accuracy_score(y, pred)), |
| "macro_f1": float(f1_score(y, pred, average="macro")), |
| "log_loss": float(log_loss(y, proba, labels=list(range(C))))} |
|
|
|
|
| def label_metrics(y, pred): |
| return {"accuracy": float(accuracy_score(y, pred)), |
| "macro_f1": float(f1_score(y, pred, average="macro"))} |
|
|
|
|
| def recovered_structure(model, correction, X): |
| """How closely one correction's class scores track the two true nonlinear terms.""" |
| delta = model._apply_single(correction, X, FEATURES) |
| x2, x4, x5 = X[:, 1], X[:, 3], X[:, 4] |
| with np.errstate(divide="ignore", invalid="ignore"): |
| inter = np.corrcoef(delta[:, 1] - delta[:, 0], x2 * x5)[0, 1] |
| even = np.corrcoef(delta[:, 2] - 0.5 * (delta[:, 0] + delta[:, 1]), x4**2)[0, 1] |
| return {"r_interaction_x2x5": float(np.nan_to_num(inter)), "r_even_x4": float(np.nan_to_num(even))} |
|
|
|
|
| |
| class _MockRunnable: |
| def __init__(self, schema, rng): |
| self.schema, self.rng = schema, rng |
|
|
| def invoke(self, messages): |
| name = self.schema.__name__ |
| if name == "Hypothesis": |
| return self.schema(hypothesised_pattern="mock", implicated_features=["x2", "x5", "x4"], |
| functional_form_guess="[-a*x2*x5, a*x2*x5, b*(x4**2 - 1)]", |
| rationale="offline mock") |
| if name == "Correction": |
| a, b = self.rng.uniform(0.5, 2.0), self.rng.uniform(0.2, 1.5) |
| return self.schema(T_k="mock", f_k=f"[-{a:.3f}*x2*x5, {a:.3f}*x2*x5, {b:.3f}*(x4**2 - 1)]") |
| return self.schema(diagnosis="mock", structural_or_coefficient="coefficient", suggested_change="mock") |
|
|
|
|
| class MockLLM: |
| def __init__(self, seed): |
| self.rng = random.Random(seed) |
|
|
| def with_structured_output(self, schema): |
| return _MockRunnable(schema, self.rng) |
|
|
|
|
| |
| def run_experiment(seed, T, llm, llm_creative, out_dir, tag, calls_path=None, fit_log=None): |
| """Run one seed end to end; returns the results dict (also written to out_dir/tag.json).""" |
| X, y, clean, sp = make_data(seed) |
| tr, va, te = sp["train"], sp["val"], sp["test"] |
| base = LogisticRegression(max_iter=1000).fit(X[tr], y[tr]) |
| P_tr, P_va, P_te = base.predict_proba(X[tr]), base.predict_proba(X[va]), base.predict_proba(X[te]) |
| mc.configure_llms(llm, llm_creative) |
| model = mc.MARICL(task="classification", K=2, kappa=0.3, T=T, B=10, p_min=0.1, |
| gamma=2.0, tau_fail=0.5, verbose=True) |
| t0 = time.time() |
| sink = open(fit_log, "w") if fit_log else None |
| try: |
| with contextlib.redirect_stdout(sink) if sink else contextlib.nullcontext(): |
| model.fit(X[tr], y[tr], base_model=base, feature_names=FEATURES, |
| feature_descs=FEATURE_DESCS, domain_context=DOMAIN, n_classes=C) |
| finally: |
| if sink: |
| sink.close() |
| fit_seconds = time.time() - t0 |
| xgb = XGBClassifier(random_state=seed).fit(X[tr], y[tr]) |
|
|
| def maricl(t_max): |
| chosen, scores, ts = select_classification(model, X[tr], y[tr], X[va], y[va], P_tr, P_va, t_max) |
| beta = select_beta(model, X[va], y[va], P_va, chosen, scores) |
| return chosen, scores, ts, beta, maricl_predict_proba(model, X[te], P_te, chosen, scores, beta) |
|
|
| chosen, scores, ts, beta, proba = maricl(T) |
| res = { |
| "seed": seed, "T": T, "fit_seconds": round(fit_seconds, 1), "class_counts": np.bincount(y).tolist(), |
| "logistic": metrics(y[te], P_te), "xgboost": metrics(y[te], xgb.predict_proba(X[te])), |
| "oracle": label_metrics(y[te], clean[te]), "maricl": metrics(y[te], proba), "beta": beta, |
| "corrections": [], "budget_curve": [], |
| } |
| for k, (s, c, p, t) in enumerate(zip(model.states_, chosen, scores, ts)): |
| res["corrections"].append({ |
| "k": k + 1, "p_k": p, "retained_t": t, |
| "hypothesis": s.hypothesis.hypothesised_pattern, |
| "form_guess": s.hypothesis.functional_form_guess, |
| "T_k": c.T_k if c else None, "f_k": c.f_k if c else None, |
| "flags": term_flags(c.f_k) if c else None, |
| "structure": recovered_structure(model, c, X[te]) if c else None, |
| "history": [{"t": e["t"], "f_k": e["correction"].f_k, "train_log_loss": e["loss"]} |
| for e in s.history], |
| }) |
| for t_max in range(T + 1): |
| res["budget_curve"].append({"T": t_max, **metrics(y[te], maricl(t_max)[4])}) |
| if calls_path and os.path.exists(calls_path): |
| calls = [json.loads(line) for line in open(calls_path)] |
| res["calls"] = {"n": len(calls), "input_tokens": sum(c["input_tokens"] for c in calls), |
| "output_tokens": sum(c["output_tokens"] for c in calls)} |
| os.makedirs(out_dir, exist_ok=True) |
| with open(os.path.join(out_dir, f"{tag}.json"), "w") as fh: |
| json.dump(res, fh, indent=2) |
| return res |
|
|
|
|
| def main(): |
| load_dotenv(os.path.join(os.path.dirname(os.path.abspath(__file__)), ".env")) |
| ap = argparse.ArgumentParser() |
| ap.add_argument("--seed", type=int, required=True) |
| ap.add_argument("--T", type=int, default=10) |
| ap.add_argument("--model", default="claude-opus-5-5") |
| ap.add_argument("--effort", default="medium") |
| ap.add_argument("--mock", action="store_true") |
| ap.add_argument("--out", default="results_api/classification") |
| args = ap.parse_args() |
| os.makedirs(args.out, exist_ok=True) |
| tag = f"{'mock_' if args.mock else ''}seed{args.seed}" |
| calls_path = os.path.join(args.out, f"calls_{tag}.jsonl") |
| if os.path.exists(calls_path): |
| os.remove(calls_path) |
| if args.mock: |
| llm = llm_creative = MockLLM(args.seed) |
| else: |
| llm = ClaudeLLM(args.model, args.effort, calls_path, role="base") |
| llm_creative = ClaudeLLM(args.model, args.effort, calls_path, role="encoder") |
| res = run_experiment(args.seed, args.T, llm, llm_creative, args.out, tag, calls_path) |
| print(json.dumps({k: res[k] for k in ("logistic", "xgboost", "maricl", "oracle")})) |
|
|
|
|
| if __name__ == "__main__": |
| sys.exit(main()) |
|
|
