Dans cet tutoriel, nous implémentons RRSI (Regularized Recursive Self-Improvement), une méthode qui permet à un agent LLM de réécrire son propre harnais, ses prompts, ses outils, sa mémoire, son flux de contrôle et ses sous-agents autour d’un modèle figé, sans que le harnais ne se surfeux sur les tâches qu’il développe. Le cycle complet de RRSI génère des modifications avec Claude Opus sur Vertex AI et les évalue dans des benchmarks Docker, ce qui n’est pas possible avec un notebook gratuit. La partie de RRSI qui incarne véritablement l’idée du paper, les règles qui déterminent quelles modifications proposées sont conservées, est en Python simple, et c’est ce que nous pilotons directement. Nous installons le package depuis le répoir officiel, examinons son estimateur, sa bande de bruit calibrée, les deux branches de son algorithme de sélection, son budget d’édition annealed, son écran de fuite déterministe et son historique d’édition, puis nous connectons un agent simulé à l’interface Domain de RRSI. Comme nous avons construit l’environnement simulé nous-mêmes, nous connaissons l’effet réel de chaque modification, ce qui nous permet de vérifier les décisions de RRSI par rapport à la vérité brute et de les comparer à une recherche non régularisée qui ne garde simplement que les scores les plus élevés.
Copier le codeimport os
import sys
import json
import math
import copy
import random
import tempfile
import textwrap
import traceback
import subprocess
import statistics as st
from pathlib import Path
RESULTS = {}
def banner(title):
print("\n" + "=" * 78)
print(title)
print("=" * 78)
def section(name):
def wrap(fn):
def run(*a, **kw):
banner(name)
try:
out = fn(*a, **kw)
RESULTS[name] = out if isinstance(out, str) else "ok"
return out
except Exception as e:
RESULTS[name] = f"SKIPPED / FAILED -> {type(e).__name__}: {e}"
print(f"\n[!] {name} did not complete: {type(e).__name__}: {e}")
traceback.print_exc(limit=3)
return None
return run
return wrap
banner("1. Install RRSI and map the paper onto the code")
subprocess.run([sys.executable, "-m", "pip", "install", "-q", "git+https://github.com/google-research/rrsi.git@be50316e1db05914068a973f322770ef08ed7ba1"], check=True)
from importlib.metadata import version
from rrsi.config import RRSIConfig
from rrsi.evaluate import TaskResult, EvalResult, aggregate, evaluate
from rrsi.calibrate import calibrate
from rrsi.selection import Candidate, cost_rule, judge, select_round
from rrsi.schedule import edit_budget, budget_table
from rrsi.history import History, stall_flag, exploration
from rrsi.components import K, K_STR, normalize, novelty
from rrsi.critic import precheck, review
from rrsi.domain import Domain
print(f" rrsi {version('rrsi')} | anthropic {version('anthropic')} | Python {sys.version.split()[0]}")
print("\n RRSI evolves an agent's HARNESS (prompts, tools, memory, control flow, sub-agents) around a")
print(" frozen model. The full loop drafts edits with Claude Opus on Vertex AI and scores them in Docker")
print(" benchmarks. The part that decides which edits to KEEP is plain Python, and that is what we drive:")
for symbol, where in [
("S_hat, C_hat Eq. (estimate)", "rrsi.evaluate.aggregate"),
("delta noise band", "rrsi.calibrate.calibrate"),
("Algorithm 2 floor + cost rule", "rrsi.selection.judge / select_round"),
("b_t Eq. (anneal)", "rrsi.schedule.edit_budget"),
("Critic leakage screen", "rrsi.critic.precheck / review"),
("L_t, g_t, B_t history, yield, prune", "rrsi.history.History"),
("sigma_t, U_t stall + exploration", "rrsi.history.stall_flag / exploration"),
("nu structural novelty", "rrsi.components.novelty"),
]:
print(f" {symbol:40s} -> {where}")
CFG = RRSIConfig()
print(f"\n paper defaults: T={CFG.T} rounds, k={CFG.k} trials/task, m={CFG.m} candidates/round,"
f" b in [{CFG.b_min},{CFG.b_max}]")
print(f" beta0={CFG.beta0} beta1={CFG.beta1} w_s={CFG.w_s} w_c={CFG.w_c} w_n={CFG.w_n}"
f" delta_z={CFG.delta_z}")
print("\n Nothing below needs an API key, a GPU or a dataset download.")
Nous installons RRSI depuis le dépôt google-research, fixé sur le commit sur lequel ce notebook a été écrit, car le package n’est pas sur PyPI. Sa seule dépendance est le client Anthropic, que les rôles de recherche utilisent pour appeler Claude et que nous ne utilisons jamais. Nous imprimons ensuite la correspondance que le répositoire lui-même documente entre les symboles du papier et les fonctions qui les mettent en œuvre : le score empirique et l’estimation de coût dans « evaluate », la bande de bruit dans « calibrate », l’algorithme 2 dans « selection », le budget d’édition « annealed » dans « schedule », le filtre de fuite dans « critic », et l’histoire des éditions avec ses résumés de rendu, de coupe, de retard et d’exploration dans « history ». RRSIConfig contient les hyperparamètres du papier, et chaque fonction qui suit le reçoit exactement comme le fait le boucle réelle.
Copier le code@section("2. Evaluate(H): a score and a cost, and why a crash counts as zero")
def estimator():
base = {
"task_000": TaskResult(rewards=[1, 1], tokens=[11_800, 12_400]),
"task_001": TaskResult(rewards=[1, 0], tokens=[15_100, 14_600]),
"task_002": TaskResult(rewards=[0, 0], tokens=[21_000, 19_500]),
}
ev = aggregate("H0", 2, base)
print(f" three tasks x k=2 trials -> S_hat = {ev.S:.3f} C_hat = {ev.C:,.0f} tokens/trial"
f" ({ev.n_expected} trials expected, {ev.missing} missing)")
crashy = dict(base)
crashy["task_002"] = TaskResult(rewards=[0.0, 0.0], tokens=[None, None], missing=2)
ev_crash = aggregate("crashy", 2, crashy)
dropped = {t: r for t, r in crashy.items() if not r.missing}
naive = sum(sum(r.rewards) for r in dropped.values()) / sum(len(r.rewards) for r in dropped.values())
print("\n A candidate crashes on the hardest task instead of failing it:")
print(f" an estimator that drops missing trials reports {naive:.3f} <- looks like a gain")
print(f" RRSI's aggregate (missing = 0, full denominator) {ev_crash.S:.3f} <- no reward for crashing")
print(f" ...and C_hat uses only recorded token counts: {ev_crash.C:,.0f}")
rubric = {"memo": TaskResult(rewards=[0.5, 1.0], weights=[10, 10]),
"brief": TaskResult(rewards=[0.0, 0.0], weights=[90, 90])}
ev_w = aggregate("rubric", 2, rubric)
print("\n Weighted rewards (Harvey LAB style: weight = number of rubric criteria):")
print(f" mean of per-task means = {st.mean(r.mean for r in rubric.values()):.3f}"
f" vs RRSI's S_hat = {ev_w.S:.3f} (fraction of all criteria passed)")
return f"crash scored {ev_crash.S:.3f} under RRSI vs {naive:.3f} if dropped"
estimator()
RRSI mesure deux chiffres pour chaque harnais : S, la récompense moyenne sur chaque essai de chaque tâche, et C, les tokens de politique moyens par essai. TaskResult enregistre les essais d’une tâche et les agrège. La détail qui vaut la peine d’être copié dans toute évaluation d’un agent est la manière dont les essais manquants sont gérés. Lorsqu’un candidat fait une erreur sur la tâche la plus difficile, un estimateur qui omet les essais manquants indique 0,750 et fait en sorte que l’erreur ressemble à une amélioration. En même temps, le RRSI compte chaque essai manquant comme une récompense nulle avec un dénominateur complet, et indique toujours 0,500, comme avant. Ainsi, un candidat ne peut pas se montrer mieux en supprimant les essais qu’il trouve difficiles. Les récompenses pondérées couvrent des séries classées selon des critères comme le Harvey LAB, où S représente la fraction de tous les critères remplis, plutôt que la moyenne des moyennes par tâche.
Copier le codeFAMILIES = ["parse", "search", "edit", "test"]
H0 = {"skill": {f: 0.0 for f in FAMILIES}, "memo": [], "cost": 1.0}
def make_world(seed, n_evolve, n_heldout=80):
"""Tasks with a family and a difficulty. Evolve ids are task_NNN, held-out ids held_NNN."""
r = random.Random(seed)
world = {f"task_{i:03d}": (FAMILIES[i % 4], r.gauss(0, 1)) for i in range(n_evolve)}
world.update({f"held_{i:03d}": (FAMILIES[i % 4], r.gauss(0, 1)) for i in range(n_heldout)})
return world
def p_success(h, world, task):
"""The frozen policy: a logistic in harness skill minus task difficulty, or 0.97 if memorised."""
if task in h["memo"]:
return 0.97
family, difficulty = world[task]
return 1 / (1 + math.exp(-(0.3 + h["skill"][family] - difficulty)))
def run_trials(h, world, ids, k, rng):
return {t: TaskResult(rewards=[float(rng.random() < p_success(h, world, t)) for _ in range(k)],
tokens=[round(12_000 * h["cost"] * math.exp(rng.gauss(0, 0.08))) for _ in range(k)])
for t in ids}
def true_score(h, world, ids):
return sum(p_success(h, world, t) for t in ids) / len(ids)
def calibrated_delta(world, k, seed, repeats=3):
"""delta from `repeats` independent evaluations of the SAME harness H0."""
ids = [t for t in world if t.startswith("task_")]
rng = random.Random(10_000 + seed)
evals = [aggregate(f"base{j}", k, run_trials(H0, world, ids, k, rng)) for j in range(repeats)]
return calibrate(evals, z=CFG.delta_z, reps=200), evals
@section("3. The noise band: how far one harness's score moves between two evaluations")
def noise_band():
world = make_world(0, 40)
ids = [t for t in world if t.startswith("task_")]
rng = random.Random(1)
scores = [aggregate(f"H0#{j}", 2, run_trials(H0, world, ids, 2, rng)).S for j in range(6)]
print(f" H0 evaluated 6 times on 40 tasks x k=2: {[round(s, 3) for s in scores]}")
print(f" true expected score {true_score(H0, world, ids):.3f}; spread of the estimates"
f" {max(scores) - min(scores):.3f}. Nothing about the harness changed.")
print(f"\n calibrate() turns repeated evaluations into delta = z * sd(null dS), z = {CFG.delta_z}:")
print(f" {'evaluator':>12s} {'trials':>7s} {'delta':>8s} method")
deltas = {}
for n, k in [(40, 2), (100, 4), (400, 8)]:
cal, _ = calibrated_delta(make_world(0, n), k, seed=0)
deltas[f"{n}x{k}"] = cal["delta"]
print(f" {f'{n} x k={k}':>12s} {n * k:>7d} {cal['delta']:>8.4f} {cal['method']}")
print("\n The paper's calibrated bands are 0.017 (coding), 0.004 (workspace) and 0.020 (engineering).")
print(" A gain smaller than delta is indistinguishable from re-running the same harness, and")
print(" Algorithm 2 treats it that way. Remember the first row: it becomes the lesson of step 11.")
return "delta " + ", ".join(f"{k}={v:.3f}" for k, v in deltas.items())
noise_band()
Avant qu’une règle ne puisse distinguer un gain réel de la chance, il est nécessaire de savoir jusqu’où le score d’un harnais peut se déplacer seul. Nous construisons un petit agent simulé, dont le succès pour chaque tâche est une fonction logistique de l’habileté du harnais moins la difficulté de la tâche. Nous évaluons le harnais de départ inchangé six fois sur quarante tâches, avec deux essais par tâche : les scores varient de 0,113, bien que rien ne soit modifié. Le calibrage transforme les évaluations répétées du même harnais en delta, soit le double de la variance standard de la différence entre deux essais. Avec 80 essais, le delta est d’environ 0,108 ; avec 3 200 essais, il tombe à environ 0,013, dans la plage rapportée par l’étude pour ses cas (0,004 à 0,020). Aux fins de sélection, tout gain inférieur au delta ne peut être distingué de la répétition du même harnais.
Copier le codedef ev_at(S, C, job, n=200):
"""An EvalResult with exactly score S (in steps of 1/n) and C tokens per trial."""
hits = round(S * n)
return aggregate(job, 1, {f"task_{i:03d}": TaskResult(rewards=[1.0 if i < hits else 0.0], tokens=[C])
for i in range(n)})
INCUMBENT = ev_at(0.630, 10_000, "incumbent")
S_STAR, DELTA = 0.640, 0.020
@section("4. Algorithm 2, branch by branch: the floor, the cost rule and the noise band")
def algorithm_2():
print(f" incumbent S = {INCUMBENT.S:.3f}, C = {INCUMBENT.C:,.0f} best ever S* = {S_STAR} delta = {DELTA}")
print(f" floor = S* - delta = {S_STAR - DELTA:.3f}\n")
cases = [
("A slipped below the floor", 0.600, 10_000, ["prompt"], None),
("B real gain, pays for its tokens", 0.690, 12_000, ["prompt"], None),
("C real gain, far too expensive", 0.660, 25_000, ["subagent"], None),
("D slightly WORSE but cheaper", 0.625, 8_000, ["context_mgmt"], None),
("E in the band but costlier", 0.640, 11_500, ["prompt"], None),
("F in the band, new sub-agent", 0.630, 10_000, ["subagent"], None),
("F' in the band, prompt tweak", 0.630, 10_000, ["prompt"], None),
("G like B, breaks a domain guard", 0.690, 12_000, ["prompt"], ["valid-output rate fell"]),
]
rows = []
for label, S, C, comps, guards in cases:
cand = Candidate(label[:2].strip(), [{"id": "C1", "component": c} for c in comps], ev=ev_at(S, C, label))
dec = judge(cand, INCUMBENT, S_STAR, DELTA, CFG, incumbent_counts={}, guards=guards)
rows.append((label, dec))
print(f" {label:36s} S={S:.3f} C={C:>6,} {'ADMIT ' if dec.admissible else 'reject'}")
print(textwrap.indent(textwrap.fill(dec.reason, 88), " " * 6))
print("\n Three rules, in the order RRSI applies them:")
print(" 1. never fall below the best score ever seen, minus the noise band (A)")
print(" 2. a gain bigger than delta must pay for any extra tokens: dC <= 0.10 + 40*dS (B, C)")
print(" 3. inside the band scores are a tie, so prefer the cheaper harness, and let a")
print(" never-tried structural component break the tie (D, E, F, F')")
print(" D is the surprising one: RRSI admits a harness that scored LOWER than the incumbent.")
return f"{sum(d.admissible for _, d in rows)}/{len(rows)} admissible; D admitted at dS={rows[3][1].delta_S:+.3f}"
algorithm_2()
L’algorithme 2 est implémenté en tant que fonctions pures ; nous pouvons donc lui transmettre des candidats et lire ses raisons mot pour mot. Nous fixons un score de 0,630 et 10 000 tokens pour S 0, et un score optimal de 0,640 avec un delta de 0,020. Huit candidats sont ensuite envoyés au juge. Un candidat inférieur au seuil, avec un score optimal jamais atteint moins le delta, est rejeté immédiatement. Une gain supérieure à delta doit être payée pour ses tokens supplémentaires, selon la règle selon laquelle le changement de coût relatif reste inférieur à 0,10 plus 40 fois le gain ; un gain de +6 points avec +20 % de tokens est accepté, tandis qu’un gain de +3 points avec +150 % de tokens ne l’est pas. À l’intérieur de la bande, les scores sont considérés comme un match nul ; un score de forme de 100 fois la gain plus moins 15 fois le changement de coût, plus un petit bonus pour une composante structurelle jamais acceptée, détermine la décision. C’est ainsi que RRSI accepte la candidate D, qui a obtenu un score inférieur à celui du candidat actuel mais coûte 20 % moins de tokens. De même, un nouveau sous-agent résout un match nul que ne fait pas une modification similaire du prompt. Un gardien de domaine vétuste indépendamment du score.
Copier le code@section("5. One round of selection: the highest score does not always win")
def one_round():
c_hi = Candidate("C", [{"id": "C1", "component": "subagent"}], ev=ev_at(0.660, 25_000, "C"))
d_lo = Candidate("D", [{"id": "C1", "component": "context_mgmt"}], ev=ev_at(0.625, 8_000, "D"))
leak = Candidate("X", [{"id": "C1", "component": "memory"}], gate_failure="critic_reject")
winner, decisions = select_round([c_hi, d_lo, leak], INCUMBENT, S_STAR, DELTA, CFG, incumbent_counts={})
for d in decisions:
print(f" {d.variant}: {'admissible' if d.admissible else 'rejected '} {d.reason[:92]}")
new_star = max(S_STAR, winner.ev.S)
print(f"\n winner: {winner.variant} (S {winner.ev.S:.3f}, C {winner.ev.C:,.0f})")
print(f" the new incumbent scores {winner.ev.S - INCUMBENT.S:+.3f} vs the old one and costs"
f" {(winner.ev.C - INCUMBENT.C) / INCUMBENT.C:+.0%} tokens; S* stays {new_star:.3f}")
print("\n C scored highest and still lost: its +3pp does not pay for +150% tokens. D moved the")
print(" incumbent DOWN inside the noise band because it is 20% cheaper. Because S* only ever")
print(" rises, the floor never follows the incumbent down, so a chain of 'cheaper but slightly")
print(" worse' swaps cannot walk the score away. X never reached evaluation at all.")
return f"winner {winner.variant} at dS={winner.ev.S - INCUMBENT.S:+.3f}, dC={(winner.ev.C - INCUMBENT.C) / INCUMBENT.C:+.0%}"
one_round()
select_round applique un juge à chaque candidat dans une round et garde celui qui obtient le score le plus élevé autorisé. Nous lui attribuons le candidat coûteux, le candidat moins cher mais légèrement moins bon, et un candidat que le critique a déjà rejeté. Le meilleur scoreur perd parce que sa gain de trois points ne couvre pas 150 % de plus de tokens, le candidat rejeté par le critique n’atteint jamais l’évaluation, et le gagnant fait baisser le concurrent en demi point tout en réduisant son coût en tokens de un cinquième. Le détail qui garantit cette sécurité est S*, qui ne monte jamais : le sol est ancré au meilleur score jamais mesuré, plutôt que dans le score actuel, de sorte qu’une série de échanges moins coûteux mais légèrement moins bons ne peut pas éliminer ce score après de nombreuses rondes.
Copier le code@section("6. The proposal side: an edit budget that anneals from 4 toward 1")
def edit_budget_schedule():
table = budget_table(CFG.T, CFG.b_min, CFG.b_max)
print(" b_t = ceil(b_min + (b_max - b_min) * (1 + cos(pi t / T)) / 2)")
print(f" T={CFG.T}, b in [{CFG.b_min}, {CFG.b_max}]:")
print(" t : " + " ".join(f"{t:>2d}" for t in range(CFG.T)))
print(" b_t : " + " ".join(f"{b:>2d}" for b in table))
print(f" round {CFG.T} (after the run) -> {edit_budget(CFG.T, CFG.T, CFG.b_min, CFG.b_max)}")
print("\n Early candidates may bundle up to 4 coordinated edits. Note the ceil(): the cosine term is")
print(f" only exactly zero at t = T, so inside a {CFG.T}-round run the budget bottoms out at"
f" {min(table)}, not {CFG.b_min}.")
print(" Every edit in a bundle inherits the bundle's single measurement, so the shrinking budget is")
print(" what makes late history attributable to fewer components. It caps how MANY edits ride")
print(" together, never WHICH mechanisms the harness may eventually contain.")
return "budget " + "".join(str(b) for b in table)
edit_budget_schedule()
La partie de proposition régule la manière dont les modifications sont rédigées, mais ne spécifie pas quelles sont conservées. edit_budget met en œuvre le budget L0 ajusté rapporté dans le papier : un programme en cosinus allant de b_max à b_min, et par défaut il permet jusqu’à quatre modifications coordonnées par candidat pour les huit premières tours, trois pour les cinq suivants, et deux pour les sept derniers. Le plafond dans la formule signifie que le budget n’atteint son minimum de un qu’à t = T, un pas après la fin du tour, ce qui est facilement négligé lors de la lecture de l’équation. Comme chaque modification dans un bundle hérite de la mesure du bundle lui-même, le budget réduit est ce qui fait que l’histoire de la mise en œuvre tardive peut être attribuée à moins de composants. Le budget limite le nombre d’éditions qui peuvent se produire simultanément, et ne limite pas non plus les mécanismes que le harnais pourrait contenir par la suite.
Copier le codeCRITIC_PATTERNS = [
(r"\btask_\d{3}\b", "hard-codes an evolve-set task id"),
(r"expected_output|grader|rubric\[", "reads the grader or the expected answer"),
]
COMPONENT_SIGNALS = [
("control_flow", [r"\bretry\(", r"max_attempts"]),
("context_mgmt", [r"compress_context", r"keep_last"]),
("config", [r"CONFIG\["]),
]
class ScreenOnly(Domain):
name = "screen"
critic_patterns = CRITIC_PATTERNS
briefs = {"critic": "A coding agent harness."}
@section("7. The critic's deterministic layer, and how edits are tagged")
def critic_and_tags():
dom = ScreenOnly()
diffs = {
"memorise answers": "+ memory = Memory('answers')\n+ memory.remember('task_007', cached_patch)",
"peek at the grader": "+ if os.path.exists('/grader/expected_output.txt'): return read_it()",
"leaked credential": "+ api_key = 'sk-live-0123456789abcdefghijkl'",
"empty diff": " ",
"general retry rule": "+ for attempt in range(max_attempts): result = retry(step)",
}
for label, diff in diffs.items():
try:
verdict = review(dom, diff, summary=label, targets_mode="evolve")
print(f" {label:20s} -> {verdict['verdict']:6s} {verdict['reasons']}")
except ZeroDivisionError as e:
print(f" {label:20s} -> passed the deterministic layer; the LLM layer raised"
f" ZeroDivisionError: {e}")
print("\n Gotcha: with RRSI_VERTEX_PROJECTS unset, rrsi.llm.generate computes `x % len(projects)`")
print(" outside its try block, so the helpful 'set RRSI_VERTEX_PROJECTS' error is never reached.")
print(" A clean diff is supposed to go to Claude for an intent review; here there is no Claude.")
print("\n Every edit is tagged with the component it touches, and a tag needs evidence in the diff:")
tag_cases = [
("skill", "+ \"Remember to run the tests before finishing.\""),
("control_flow", "+ for attempt in range(max_attempts): result = retry(step)"),
(None, "+ review = subcall('reviewer', transcript)"),
("memory", "+ context = compress_context(context, keep_last=8)"),
]
for declared, diff in tag_cases:
tag = normalize(declared, diff, COMPONENT_SIGNALS)
print(f" declared {str(declared):13s} -> tagged {tag:13s} {diff[2:60]!r}")
print(" A proposer cannot label a prompt tweak as a new 'skill' to look novel: without evidence")
print(" the tag falls back to what the diff actually is.")
counts = {"prompt": 3, "subagent": 1}
print(f"\n novelty(nu) counts STRUCTURAL components {K_STR} the incumbent has never accepted.")
print(f" against an incumbent with accepted edits {counts}:")
for comps in (["memory"], ["subagent"], ["prompt", "client_tool", "memory"]):
print(f" {str(comps):38s} nu = {novelty(comps, counts)}")
return "precheck rejected 4/5 diffs without an LLM call"
critic_and_tags()
Le critique vérifie chaque candidature diff avant qu’une évaluation ne soit effectuée, en deux étapes. La première est une vérification déterministe basée sur un modèle de crediteur générique ainsi que sur la liste de blocage propre au domaine ; avec des modèles pour les identifiants de tâches en mode évolutif et les parcours de juge, il rejette une diff qui mémorise la réponse à la tâche_007, une diff qui lit la sortie attendue, une diff qui divulgue une clé API, et une diff vide, sans appeler un modèle. Un diff propre passe au deuxième niveau, une revue d’intention effectuée par Claude, et ici le notebook révèle un vrai problème : lorsque RRSI_VERTEX_PROJECTS n’est pas défini, rrsi.llm.generate calcule un index modulo le nombre de projets en dehors de son bloc try, ce qui entraîne une ZeroDivisionError ; donc l’erreur de configuration utile dans le code ne se produit jamais. Nous examinons également la manière dont les modifications sont étiquetées : normaliser ne conserve un composant déclaré que lorsque le diff fournit des preuves à cet effet ; ainsi, une modification mineure ne peut pas se présenter comme une nouvelle compétence permettant d’obtenir le bonus de nouveauté, et une modification dans la gestion du contexte mal étiquetée est étiquetée telle quelle.
Copier le code@section("8. The edit history: what was tried, what paid off, what to prune")
def edit_history():
with tempfile.TemporaryDirectory() as tmp:
h = History(Path(tmp) / "history.jsonl")
log = [
(0, "A", [("prompt", "tell the agent to read the failing test first")], "ACCEPTED", 0.030, 0.02, True, 0.66),
(1, "A", [("prompt", "ask for a plan before editing")], "REJECTED", -0.010, 0.05, False, 0.65),
(1, "B", [("subagent", "add a reviewer sub-agent"), ("memory", "persist lint rules")],
"REJECTED", -0.020, 0.40, False, 0.64),
(2, "A", [("config", "raise the step limit")], "ACCEPTED", 0.005, -0.03, True, 0.665),
(3, "A", [("prompt", "shorter system prompt")], "LOST", -0.001, -0.10, False, 0.664),
(4, "B", [("memory", "cache task_014 solution")], "critic_reject", None, None, False, None),
(5, "A", [("prompt", "stricter output format")], "REJECTED", -0.004, 0.00, False, 0.661),
]
for t, v, edits, outcome, dS, dC, acc, S in log:
h.append_candidate(t, v, [{"id": f"C{i + 1}", "component": c, "hypothesis": hyp}
for i, (c, hyp) in enumerate(edits)],
outcome, dS, dC, acc, S, 12_000 if S else None, diff=None)
t_now = 6
print(f" {len(h.records())} per-edit records from {len(log)} candidates"
f" (the two-edit bundle in round 1 wrote two records with ONE measurement)")
print(f" T_t, tried components : {sorted(h.tried())} (the critic-rejected edit is not 'tried')")
g_t = h.yield_g(t_now, CFG.n_prune)
print(" g_t, best gain in the last n_prune rounds: "
+ ", ".join(f"{c} {'none measured' if g == -math.inf else f'{g:+.3f}'}" for c, g in sorted(g_t.items())))
prune = h.prune_set(t_now, CFG.n_prune)
print(f" B_t, prune set : {[p['component'] for p in prune]}")
for p in prune:
if p["accepted_edits_in_incumbent"]:
print(f" {p['component']}: still in the incumbent but no recent gain ->"
f" {[e['hypothesis'] for e in p['accepted_edits_in_incumbent']]}")
trajectory = [0.630, 0.660, 0.660, 0.665, 0.665, 0.665, 0.665]
sigma = stall_flag(trajectory, t_now, CFG.w, DELTA)
ex = exploration(t_now, sigma, h.tried(), CFG.m_draft)
print(f"\n S over rounds {trajectory}: moved {trajectory[t_now] - trajectory[t_now - CFG.w]:+.3f}"
f" in the last w={CFG.w} rounds -> sigma_t = {sigma}")
print(" what the proposer is told next round:")
print(textwrap.indent(textwrap.fill(ex["text"], 84), " "))
print("\n The proposer sees this history, so a falsified hypothesis ('ask for a plan first', -1pp)")
print(" is not redrawn, and a stalled run is pushed toward components it has never touched.")
return f"prune set {[p['component'] for p in prune]}, stall flag {sigma}"
edit_history()
L’histoire enregistre un enregistrement JSONL par modification, et le cycle extrait quatre résumés à partir de celui-ci. L’ensemble testé exclut les modifications que le critique a supprimées, car elles n’ont jamais été mesurées. Le rendement récent g_t représente la meilleure amélioration mesurée par composant au cours des dernières n_prune tours, et chaque composant dont le rendement récent n’est pas positif est inscrit dans l’ensemble de prunage B_t, ainsi que toute machine appartenant à ce composant qui reste en place ; dans notre histoire, cela indique une modification urgente acceptée au tour zéro qui n’a pas abouti depuis. Le drapeau de stand est activé lorsque le score a diminué de moins que delta au cours des dernières w rondes, et l’exploration écrit alors la directive reçue par le proposant, réservant une place pour les composants qui n’ont jamais été utilisés dans cette séance. Comme le proposant est conditionné par tout cela, une hypothèse falsifiée ne sera plus soumise à nouveau.
Copier le codedef write_harness(root, h):
root = Path(root)
root.mkdir(parents=True, exist_ok=True)
(root / "harness.json").write_text(json.dumps(h))
return root
class SimulatedAgentDomain(Domain):
"""A Domain adapter over the simulated agent: the same contract RRSI's coding,
workspace and engineering instances implement."""
name = "simulated"
critic_patterns = CRITIC_PATTERNS
component_signals = COMPONENT_SIGNALS
briefs = {"critic": "A coding agent harness evaluated on parse/search/edit/test tasks."}
def __init__(self, world, seed):
self.world, self.rng = world, random.Random(seed)
def evolve_ids(self):
return [t for t in self.world if t.startswith("task_")]
def heldout_ids(self):
return [t for t in self.world if t.startswith("held_")]
def smoke_ids(self, incumbent_per_task=None):
return self.evolve_ids()[:3]
def run(self, root, runs_dir, job, ids, k, log_prefix=""):
out = Path(runs_dir) / "jobs" / job
if (out / "trials.json").exists():
return # resume-safe, as the contract requires
h = json.loads((Path(root) / "harness.json").read_text())
per = run_trials(h, self.world, ids, k, self.rng)
out.mkdir(parents=True, exist_ok=True)
(out / "trials.json").write_text(json.dumps(
{t: {"rewards": r.rewards, "tokens": r.tokens} for t, r in per.items()}))
def score(self, runs_dir, job, ids, k):
d = json.loads((Path(runs_dir) / "jobs" / job / "trials.json").read_text())
return {t: TaskResult(rewards=d[t]["rewards"], tokens=d[t]["tokens"]) for t in ids}, {}
@section("9. A Domain adapter: plugging an environment into RRSI's own evaluate()")
def domain_adapter():
dom = SimulatedAgentDomain(make_world(0, 40), seed=3)
with tempfile.TemporaryDirectory() as tmp:
root, runs = write_harness(Path(tmp) / "wt_H0", H0), Path(tmp) / "runs"
ev = evaluate(dom, root, runs, "H0", dom.evolve_ids(), k=2)
files = sorted(str(p.relative_to(tmp)) for p in Path(tmp).rglob("*") if p.is_file())
print(f" evaluate(domain, worktree, runs_dir, 'H0', 40 ids, k=2) -> S={ev.S:.3f} C={ev.C:,.0f}")
print(f" files: {files}")
again = evaluate(dom, root, runs, "H0", dom.evolve_ids(), k=2)
print(f" evaluate() again on the same job -> S={again.S:.3f} (read back, not re-run)")
print("\n The harness lives in files under a worktree root, exactly as RRSI's real instances keep")
print(" one git worktree per candidate. RRSI never runs an agent or grades anything itself; the")
print(" adapter's run() and score() do, and everything in steps 2-8 consumes what they return.")
return f"adapter evaluated H0 at S={ev.S:.3f} through rrsi.evaluate.evaluate"
domain_adapter()
RRSI ne lance jamais un agent ni ne valide les livrables par lui-même ; c’est l’adaptateur de domaine qui le fait, et la même interface soutient la codification du papier, le espace de travail et les instances d’ingénierie. Nous implémentons une pour l’agent simulé : des divisions de tâches évolutives et déployées, une méthode de exécution qui lit les données depuis les fichiers situés sous le répertoire racine du travailier et écrit les résultats d’essai dans un répertoire runs, une méthode de score qui les lit à nouveau en objets TaskResult, ainsi que les schémas critiques du domaine et les signaux des composants. La fonction d’évaluation de RRSI évalue ensuite le harnais de départ via l’adaptateur ; en l’appelant une deuxième fois pour le même travail, on lit à nouveau les essais stockés, plutôt que de les réexécuter, conformément aux exigences de sécurité du contrat.
Copier le codedef propose_edit(r, evolve_ids):
"""The scripted proposer. Each draw is one edit whose TRUE effect we know."""
u = r.random()
if u < 0.35:
fam = r.choice(FAMILIES)
comp = r.choice(["prompt", "control_flow", "context_mgmt"])
diff = {"prompt": f"+ \"On {fam} tasks, check the edge cases before finishing.\"",
"control_flow": f"+ for attempt in range(max_attempts): result = retry({fam}_step)",
"context_mgmt": f"+ context = compress_context(context, keep_last=12) # {fam}"}[comp]
return {"kind": "general", "component": comp, "family": fam, "effect": r.gauss(0.20, 0.30),
"dcost": 0.02, "diff": diff}
if u < 0.55:
ids = r.sample(evolve_ids[:40], 3)
return {"kind": "leaky", "component": "memory", "ids": ids, "dcost": 0.03,
"diff": "+ memory = Memory('solutions')\n" + "\n".join(
f"+ memory.remember('{i}', cached_patch)" for i in ids)}
if u < 0.70:
return {"kind": "inert", "component": "config", "dcost": r.uniform(-0.02, 0.05),
"diff": "+ CONFIG['log_level'] = 'debug'"}
if u < 0.85:
return {"kind": "expensive", "component": "subagent", "effect": 0.15, "mult": 1.5,
"diff": "+ review = subcall('reviewer', transcript)"}
return {"kind": "compress", "component": "context_mgmt", "effect": -0.03, "mult": 0.85,
"diff": "+ context = compress_context(context, keep_last=6)"}
def apply_edit(h, e):
h = copy.deepcopy(h)
if e["kind"] == "general":
h["skill"][e["family"]] += e["effect"]
h["cost"] *= 1 + e["dcost"]
elif e["kind"] == "leaky":
h["memo"] = sorted(set(h["memo"]) | set(e["ids"]))
h["cost"] *= 1 + e["dcost"]
elif e["kind"] == "inert":
h["cost"] *= 1 + e["dcost"]
else: # expensive / compress: every family
for f in FAMILIES:
h["skill"][f] += e["effect"]
h["cost"] *= e["mult"]
return h
def search(world, mode, *, seed, k, delta, T=CFG.T, m=CFG.m):
"""One run of harness search. mode: 'greedy' | 'critic only' | 'rrsi'. Same proposer stream in all."""
dom = SimulatedAgentDomain(world, seed=seed * 7 + 1)
proposer = random.Random(seed * 13 + 5)
ids = dom.evolve_ids()
with tempfile.TemporaryDirectory() as tmp:
tmp, h = Path(tmp), copy.deepcopy(H0)
runs, hist = tmp / "runs", History(tmp / "history.jsonl")
inc = evaluate(dom, write_harness(tmp / "H0", h), runs, "H0", ids, k)
S_star, audit, n_evals = inc.S, [], 1
for t in range(T):
b_t = edit_budget(t, T, CFG.b_min, CFG.b_max)
cands, drafts = [], []
for v in "AB"[:m]:
edits = [propose_edit(proposer, ids) for _ in range(proposer.randint(1, b_t))]
hc = h
for e in edits:
hc = apply_edit(hc, e)
diff = "\n".join(e["diff"] for e in edits)
tags = [{"id": f"C{i + 1}", "component": normalize(e["component"], e["diff"], COMPONENT_SIGNALS),
"hypothesis": e["kind"]} for i, e in enumerate(edits)]
c = Candidate(f"r{t}{v}", tags)
if mode != "greedy" and precheck(diff, dom.critic_patterns):
c.gate_failure = "critic_reject" # screened BEFORE any evaluation is spent
else:
c.ev = evaluate(dom, write_harness(tmp / c.variant, hc), runs, c.variant, ids, k)
n_evals += 1
cands.append(c)
drafts.append((hc, edits))
if mode == "rrsi":
winner, decisions = select_round(cands, inc, S_star, delta, CFG,
hist.incumbent_component_counts())
for c, d in zip(cands, decisions): # the records loop.round() writes
outcome = (c.gate_failure if c.ev is None else
"ACCEPTED" if c is winner else "LOST" if d.admissible else "REJECTED")
hist.append_candidate(t, c.variant, c.edits, outcome, d.delta_S, d.delta_C,
c is winner, d.S, d.C, None, d.reason)
else: # unregularized: keep the best score if it rose
live = [c for c in cands if c.ev is not None and c.ev.S > inc.S]
winner = max(live, key=lambda c: c.ev.S) if live else None
for c, (hc, edits) in zip(cands, drafts):
audit += [(e["kind"], c is winner) for e in edits]
if winner is not None:
h, inc = drafts[cands.index(winner)][0], winner.ev
S_star = max(S_star, inc.S)
return {"evolve_measured": inc.S, "evolve_true": true_score(h, world, ids),
"heldout_true": true_score(h, world, dom.heldout_ids()), "tokens": h["cost"],
"memorised": len(h["memo"]), "evals": n_evals, "audit": audit}
MODES = ["greedy", "critic only", "rrsi"]
def compare(n_evolve, k, seeds):
out = {mode: [] for mode in MODES}
deltas = []
for s in seeds:
world = make_world(s, n_evolve)
cal, _ = calibrated_delta(world, k, seed=s)
deltas.append(cal["delta"])
for mode in MODES:
out[mode].append(search(world, mode, seed=s, k=k, delta=cal["delta"]))
return out, st.mean(deltas)
def fmt(xs, d=3):
return f"{st.mean(xs):.{d}f}±{st.pstdev(xs):.{d}f}"
@section("10. Greedy vs RRSI in a world where we know the truth")
def miniature():
seeds = range(8)
out, delta = compare(40, 2, seeds)
globals()["MINIATURE"] = (out, delta) # step 11 reuses this row
h0_held = st.mean(true_score(H0, make_world(s, 40), [f"held_{i:03d}" for i in range(80)]) for s in seeds)
print(f" 40 evolve tasks x k=2, 80 held-out tasks, T={CFG.T}, m={CFG.m}, {len(seeds)} seeds,"
f" mean calibrated delta {delta:.3f}")
print(f" H0 held-out (true) = {h0_held:.3f}\n")
print(f" {'mode':12s} {'evolve meas':>12s} {'evolve TRUE':>12s} {'held-out TRUE':>14s}"
f" {'tokens':>11s} {'memorised':>10s} {'evals':>6s}")
for mode in MODES:
rs = out[mode]
print(f" {mode:12s} {fmt([r['evolve_measured'] for r in rs]):>12s} {fmt([r['evolve_true'] for r in rs]):>12s}"
f" {fmt([r['heldout_true'] for r in rs]):>14s} {fmt([r['tokens'] for r in rs], 2) + 'x':>11s}"
f" {st.mean(r['memorised'] for r in rs):>10.1f} {st.mean(r['evals'] for r in rs):>6.1f}")
print("\n Ground-truth audit: share of proposed edits of each kind that ended up in the incumbent")
kinds = ["general", "expensive", "compress", "inert", "leaky"]
print(f" {'mode':12s}" + "".join(f"{k:>11s}" for k in kinds))
for mode in MODES:
tally = {k: [0, 0] for k in kinds}
for r in out[mode]:
for kind, accepted in r["audit"]:
tally[kind][0] += 1
tally[kind][1] += accepted
print(f" {mode:12s}" + "".join(f"{tally[k][1]:>5d}/{tally[k][0]:<5d}" for k in kinds))
print("\n Where the evolve-set score stops transferring, split in two:")
print(f" {'mode':12s} {'winner curse':>14s} {'memorisation':>14s} (evolve meas - evolve true | evolve true - held-out true)")
for mode in MODES:
curse = st.mean(r["evolve_measured"] - r["evolve_true"] for r in out[mode])
memo = st.mean(r["evolve_true"] - r["heldout_true"] for r in out[mode])
print(f" {mode:12s} {curse:>+14.3f} {memo:>+14.3f}")
print(" Selecting the best of noisy scores inflates every mode about equally; no rule here removes")
print(" that - only re-measuring on tasks the search never saw does. The critic removes almost all")
print(" of the memorisation, and it does so before evaluation, which is why its runs cost fewer evals.")
g, c, r_ = (st.mean(x["heldout_true"] for x in out[m_]) for m_ in MODES)
tg, tc, tr = (st.mean(x["tokens"] for x in out[m_]) for m_ in MODES)
return f"held-out {g:.3f} / {c:.3f} / {r_:.3f}, tokens x{tg:.2f} / x{tc:.2f} / x{tr:.2f} (greedy / critic / rrsi)"
miniature()
Maintenant, nous exécutons toute la sélection côté recherche : vingt tours, deux candidats par tour, les ensembles étant déterminés par le budget annelé, les éditions étiquetées par normalisation, les candidats filtrés par pré-check et évalués via l’adaptateur, les gagnants choisis par select_round, et chaque résultat est enregistré dans Histoire selon l’ordre utilisé par le cycle de RRSI. Le proposant scripté propose cinq types de modifications dont les effets réels sont connus : des modifications générales généralement utiles, des modifications « fuites » qui mémorisent les réponses des tâches évolutives, des changements de configuration inactifs, des sous-agents coûteux qui aident un peu partout avec 1,5 fois le nombre de tokens, et une compression de contexte moins coûteuse. Nous comparons trois règles d’acceptation sur la même chaîne de candidats à travers huit séries de tests : le gourmand conserve le meilleur score si celui-ci augmente, le critique ne ajoute qu’une écran de fuite, et RRSI applique l’algorithme 2 avec un delta calibré. Le gourmand atteint le score le plus élevé retenu, 0,681, contre 0,616 pour RRSI, tout en mémorisant environ dix réponses et en triplant presque son coût de token ; RRSI ne mémorise rien, se termine à environ la moitié du coût de token (1,53 fois le harnais de départ contre 2,97 fois), et utilise un tiers moins d’évaluations. Diviser l’écart entre « evolve-to-held-out » est le résultat le plus utile ici : sélectionner les meilleurs scores bruyants augmente chaque mode d’environ neuf points, et aucune règle dans ce cycle ne l’élimine. En même temps, le critère élimine presque tout le composant de mémorisation.
Copier le code@section("11. Turn the evaluator up: RRSI's caution is calibrated, not configured")
def noise_sweep():
rows = [("40x2", MINIATURE[1], MINIATURE[0], 8)] # from step 10
for n, k in [(100, 4), (400, 8)]:
out, delta = compare(n, k, range(5))
rows.append((f"{n}x{k}", delta, out, 5))
print(f" {'evaluator':>10s} {'seeds':>5s} {'delta':>7s} "
+ "".join(f"{m_ + ' held / tokens':>26s}" for m_ in MODES))
for label, delta, out, n_seeds in rows:
cells = "".join(f"{st.mean(r['heldout_true'] for r in out[m_]):>13.3f} /"
f" x{st.mean(r['tokens'] for r in out[m_]):<9.2f}" for m_ in MODES)
print(f" {label:>10s} {n_seeds:>5d} {delta:>7.3f} {cells}")
first, last = rows[0], rows[-1]
rr_first = st.mean(r["heldout_true"] for r in first[2]["rrsi"])
rr_last = st.mean(r["heldout_true"] for r in last[2]["rrsi"])
tok_ratio = (st.mean(r["tokens"] for r in last[2]["greedy"]) /
st.mean(r["tokens"] for r in last[2]["rrsi"]))
print(f"\n As the evaluator sharpens, delta falls from {first[1]:.3f} to {last[1]:.3f} (the paper: 0.004-0.020),"
f" and RRSI's held-out score rises from {rr_first:.3f} to {rr_last:.3f}.")
print(f" At the sharpest setting the unregularized search is spending {tok_ratio:.1f}x RRSI's tokens.")
print("\n Read the table honestly: this world has no diminishing returns, so every sub-agent the")
print(" greedy search stacks keeps buying accuracy. That is the most favourable world possible for")
print(" spending, and the greedy search does score higher. RRSI trades some of that score for a")
print(" bounded token bill, zero memorised answers and fewer wasted evaluations - and the size of")
print(" the trade is set by delta, which it measures from your evaluator rather than taking from you.")
return (f"delta {first[1]:.3f} -> {last[1]:.3f}; RRSI held-out {rr_first:.3f} -> {rr_last:.3f};"
f" greedy uses {tok_ratio:.1f}x the tokens")
noise_sweep()
Notre premier évaluateur était bruyant, et le delta a été calibré à partir de lui ; ainsi, la prudence de RRSI lors de l’étape 10 reflétait l’évaluateur plutôt que la méthode elle-même. Nous répétons la comparaison avec des ensembles plus grands et plus de tentatives. À mesure que l’évaluateur s’améliore, le delta passe de 0,096 à 0,025, et la note retenue par RRSI passe de 0,616 à 0,759, tandis que la recherche non régularisée à l’état le plus aigu utilise 6,5 fois plus de tokens que RRSI. Greedy reste plus performant, et le notebook explique pourquoi : ce monde simulé ne présente aucune réduction des retours, donc chaque sous-agent de la recherche gourmande continue à acquérir de la précision, ce qui est le monde le plus favorable possible pour les dépenses. Les RRSI échangent une partie de cette performance contre un billet de token limité, sans réponses mémorisées et avec moins d’évaluations inutiles, et la taille de l’échange est définie par delta, qui est mesurée plutôt que demandée.
Copier le codebanner("SUMMARY")
for name, res in RESULTS.items():
print(f" {name:<76s} {res}")
print("""
What the miniature does not model
- Edits that help the evolve suite but hurt a different suite. The paper's LLM critic and its
held-out and out-of-distribution splits exist for those; a regex screen cannot catch them.
- A proposer that reads the history. Ours is scripted, so it redraws falsified ideas freely.
Where to go next
- Run a real instance: `python3 rrsi.py --domain coding smoke` after configuring Claude on Vertex AI
(RRSI_VERTEX_PROJECTS) and harbor; see domains/coding/README.md.
- Add a domain: implement rrsi.domain.Domain in domains/<name>/adapter.py, as step 9 did in the notebook.
- Re-adjudicate a stored round under a different delta or cost rule without re-running anything:
`python3 rrsi.py --domain <name> readjudicate --t <t>`.
- Paper: arxiv.org/abs/2609.24972 Code: github.com/google-research/rrsi
""")
Le résumé affichait le résultat de ligne unique fourni par chaque section, indiquant clairement ce que la miniature ne modélise pas : les modifications qui aident l’ensemble des outils tandis qu’elles nuisent à un autre ensemble d’outils – c’est pour cela que les critiques du LLM et les splits proposées existent. De plus, il y avait une proposition qui lit l’histoire, puis indique de mettre en œuvre une instance réelle, ajouter un domaine, et rejouir une évaluation stockée avec un différentiel, sans avoir à relancer quoi que ce soit.
En conclusion, nous avons décomposé RRSI à l’endroit où réside l’idée du document : les règles qui déterminent quelles modifications un agent auto-améliorant conserve, et nous les avons exécutées hors ligne sur une CPU sans modèle ni clé API. L’estimateur refuse de récompenser un crash ; la bande de bruit est mesurée plutôt que choisie, et l’algorithme 2 s’avère plus subtil que « rejeter les gains bruyants » : il ne permet jamais que le score tombe en dessous de la meilleure valeur observée moins la bande de bruit, il fait en sorte que les gains réels paient pour leurs tokens supplémentaires, et à l’intérieur de la bande, il préfère délibérément le harnais moins cher, même si celui-ci obtient une valeur légèrement inférieure. Vérifier ces règles dans un monde dont nous contrôlions la vérité fondamentale était la partie la plus informative : le critique déterministe a éliminé presque toute la mémorisation avant que l’évaluation ne soit terminée, l’algorithme 2 a maintenu le chiffre de billet en dessous d’une recherche non régulière plusieurs fois, et ni l’un ni l’autre n’a éliminé l’inflation résultant de choisir le meilleur des scores bruyants, ce que seule une nouvelle mesure sur des tâches inconnues peut faire. La remarque honnête est également claire : dans un monde où les tokens assurent toujours une précision, les scores de recherche non réguliers sont plus élevés. Ainsi, la valeur de la régularisation du RRSI dépend de l’ampleur des coûts liés aux tokens et du prix qu’une réponse volée vous coûterait. Le notebook vous fournit les outils nécessaires pour mesurer ces deux éléments avec votre propre évaluateur.
Veuillez consulter les codes complet ici. Tout le crédit revient au chercheur de ce projet. De plus, n’hésitez pas à nous suivre sur Twitter, et n’oubliez pas de rejoindre notre subreddit de 150 000+ utilisateurs sur ML ainsi que notre bulletin newsletter. Attendez ! êtes-vous sur Telegram ? Maintenant, vous pouvez également nous rejoindre sur Telegram.
La publication Guide de recherche RRSI Google : Maîtriser les agents d’IA auto-améliorant a été publiée en premier sur MarkTechPost.