MarkTechPost

Google Research RRSI-Leitfaden: Die Beherrschung selbstverbessernder AI-Agenten

Erkunden Sie einen umfassenden Kodierleitfaden zu Google Research's RRSI (Regularized Recursive Self-Improvement), der erklärt, wie Noise-Bands, Kostenerlasses und Leakage-Screens sichere, effiziente und…

In diesem Tutorial implementieren wir RRSI (Regularized Recursive Self-Improvement), eine Methode, die einem LLM-Agenten ermöglicht, sein eigenes Framework, Anweisungen, Tools, Memo, Kontrollfluss und Sub-Agenten um einen gefrorenen Modell herum zu überarbeiten, ohne dass das Framework überfällig auf die Aufgaben wird, an denen es sich entwickelt. Der vollständige RRSI-Prozess erstellt Änderungen mit Claude Opus auf Vertex AI und bewertet sie in Docker-Benchmarks, was etwas ist, das ein kostenloses Notebook nicht ausführen kann. Der Teil von RRSI, der tatsächlich die Idee des Papers enthält, die Regeln, die entscheiden, welche vorgeschlagenen Änderungen beibehalten werden, ist einfach Python – und das ist es, was wir direkt steuern. Wir installieren das Paket aus dem offiziellen Repository, betrachten seinen Schätzungsmechanismus, seine kalibrierte Störbandbreite, beide Zweige seines Selektionsalgoritmus, sein annealed Änderungsbudget, seine deterministische Leakagescanner und seine Änderungshistorie, und dann fügen wir einen simulierten Agenten in die eigene Domain-Schnittstelle von RRSI ein. Da wir die simulierte Umgebung selbst erstellt haben, kennen wir den tatsächlichen Effekt jeder Änderung, was es uns ermöglicht, RRSIs Entscheidungen gegen die echten Daten zu überprüfen und sie mit einer unregulierten Suche zu vergleichen, die einfach die höchsten Punkte beibehält.

Code kopieren
import 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.")

Wir installieren RRSI aus dem Google-Research-Repository, das auf den Committement dieser Notizbücher gesetzt ist, da das Paket nicht auf PyPI verfügbar ist. Seine einzige Abhängigkeit ist der Anthropic-Client, den die Suchrollen verwenden, um Claude zu kontaktieren, und den wir nie nutzen. Wir drucken anschließend die Zuordnung aus, die das Repository selbst dokumentiert, zwischen den Symbolen des Papers und den Funktionen, die sie implementieren: den empirischen Score und die Kostenabschätzung in „evaluate“, die Rauschband in „calibrate“, das Algorithmus 2 in „selection“, das annealed-Edit-Budget in „schedule“, der Leakage-Screen in „critic“ sowie die Edit-Historie mit ihren Yield-, Prune-, Stall- und Exploration-Zusammenfassungen in „history“. RRSIConfig enthält die Hyperparameter des Buches, und jede Funktion unten erhält sie genau so wie der echte Loop es tut.

Code kopieren
@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()

Der RRSI misst zwei Zahlen pro Harness: S, die durchschnittliche Belohnung für jede Versuchseinheit jedes Aufgabens, und C, die durchschnittlichen Policy-Token pro Versuchseinheit. TaskResult speichert die Versuchseinheiten einer Aufgabe und sammelt sie zusammen. Die Details, die für die Bewertung eines Agenten wichtig sind, betreffen wie die fehlenden Versuchseinheiten behandelt werden. Wenn ein Kandidat bei der schwierigsten Aufgabe abstürzt, berichtet ein Schätzungsmodell, das die fehlenden Versuchseinheiten weglässt, 0,750 und macht den Absturz wie eine Verbesserung aus. Gleichzeitig zählt der RRSI jede fehlende Prüfung als Nullbelohnung mit dem vollen Nenner und meldet wiederum 0,500, wie zuvor, sodass ein Kandidat nicht besser sein kann, indem er die Prüfungen zerstört, an denen es schwierig ist. Gewichtete Belohnungen umfassen rubric-klassifizierte Sets wie das Harvey LAB, bei denen S der Anteil aller erfüllten Kriterien ist, anstatt der Durchschnitt der Einzelkriterien.

Code kopieren
FAMILIES = ["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()

Bevor eine Regel einen echten Vorteil von Glück unterscheiden kann, muss sie wissen, wie weit die Punktzahl eines „Harnesses“ allein voranschreitet. Wir bauen ein kleines simulierte Agentenmodell, dessen Erfolg bei jedem Aufgabentyp eine logistische Funktion der „Harness-Kunst“ minus der Aufgabenkomplexität ist. Die unveränderten Startwerte des „Harnesses“ werden sechsmal bei vierzig Aufgaben mit zwei Versuchen pro Aufgabe bewertet: Die Punktzahlen variieren um 0,113 – obwohl sich nichts änderte. Calibrieren macht die wiederholten Bewertungen desselben Harnesses zu einem Delta, das dem Doppelten des Standardabstands der Differenz zwischen zwei Durchläufen entspricht. Bei 80 Versuchen beträgt das Delta etwa 0,108; bei 3.200 Versuchen sinkt es auf etwa 0,013 – in dem Bereich, den das Paper für seine Beispiele angeben (0,004 bis 0,020). Für die Auswahlzwecke ist jede Gewinnung, die kleiner als das Delta ist, nicht von einer erneuten Durchführung des gleichen Harnesses zu unterscheiden.

Code kopieren
def 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()

Algorithmus 2 wird als reine Funktion implementiert, sodass wir Kandidaten übergeben und seine Gründe wörtlich lesen können. Wir festlegen einen Inhaber bei S 0,630 und 10.000 Tokens, einen besten Wert von 0,640 und einen Delta von 0,020, und senden acht Kandidaten an den Richter. Ein Kandidat unter dem Schwellenwert, der beste bisherige Wert minus der Delta, wird einfach abgelehnt. Ein Gewinn größer als Delta muss für seine zusätzlichen Tokens bezahlt werden, nachdem die relative Kostenänderung unter 0,10 plus 40 Mal der Gewinne bleibt; ein +6-Punkte-Gewinn bei +20% Tokens ist zulässig, und ein +3-Punkte-Gewinn bei +150% Tokens nicht. Innerhalb der Band wird ein Unentschieden als gleichgesetzt betrachtet; eine geformte Punktzahl von 100 Mal dem Gewinn minus 15 Mal der Kostenänderung, plus ein kleiner Bonus für einen nie akzeptierten strukturellen Komponenten, entscheidet über die Wahl des Kandidaten D – dieser erzielte eine niedrigere Punktzahl als der Amtsinhaber, kostete jedoch 20% weniger Tokens. Ebenso entscheidet ein neuer Sub-Agent über einen Unentschieden, den kein gleichwertiges Punktzahl-Anpassungsprompt erreicht. Ein Domänenguard verweigert die Zulassung unabhängig von der Punktzahl.

Code kopieren
@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 wendet den Richter auf jeden Kandidaten in einer Runde an und behält den mit der höchsten möglichen Punktzahl. Wir geben dem teuren Kandidaten, dem etwas schlechteren kostengünstigeren Kandidaten und dem von dem Kritiker bereits abgelehnten Kandidaten die Wahl. Der mit der höchsten Punktzahl verliert, weil sein drei-Punkte-Vorteil nicht den 150% höheren Tokenbedarf deckt; der von dem Kritiker abgelehnte Kandidat erreicht nie die Bewertung; und der Sieger senkt den Amtsinhaber um einen halben Punkt und reduziert gleichzeitig seinen Tokenkosten um ein Fünftel. Die Besonderheit, die das sicherehalten macht, ist S*, die nur immer steigt: der Boden ist an den besten gemessenen Wert angekoppelt, anstatt an den Amtsinhaber, sodass eine Kette kostengünstigerer – aber etwas schlechterer – Austausche das Ergebnis in vielen Runden nicht verlassen kann.

Code kopieren
@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()

Die Vorschlagsseite regelt die Art und Weise, wie Änderungen formuliert werden, nicht, welche Änderungen beibehalten werden. edit_budget implementiert das überarbeitete L0-Budget aus dem Paper: eine Kosinus-Schaltung von b_max bis b_min. Standardmäßig erlaubt es bis zu vier koordinierte Änderungen pro Kandidat in den ersten acht Runden, drei in den nächsten fünf Runden und zwei in den letzten sieben Runden. Der Obergrenzewert in der Formel bedeutet, dass das Budget nur bei t = T, einem Schritt nach Ende des Laufes, sein Minimum von eins erreicht – was beim Lesen der Gleichung leicht übersehen werden kann. Da jede Bearbeitung in einem Bundle die Messung des Bundes übernimmt, ist es die verringerte Budgetrestriktion, die dazu führt, dass die späte Ausführungshistorie auf weniger Komponenten zurückzuführen ist. Das Budget beschränkt die Anzahl der zusammengeführten Bearbeitungen und beschränkt sich nie darauf, welche Mechanismen das System möglicherweise enthält.

Code kopieren
CRITIC_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()

Der Kritiker prüft jeden Kandidaten-Diff vor einer vollständigen Bewertung in zwei Schichten. Die erste Schicht ist eine deterministische Vorprüfung gegen ein generisches Credential-Muster sowie die eigene Ablehnungsliste des Domains; mit Mustern für Task-IDs und Grader-Pfade lehnt sie einen Diff ab, der die Antwort auf Task_007 speichert, einen, der den erwarteten Ausgabewert liest, einen, der eine API-Kennung preisgibt, sowie einen leeren Diff – ohne dass ein Modell aufgerufen wird. Ein sauberer Diff geht in die zweite Schicht über, eine Überprüfung durch Claude, und hier zeigt das Notizbuch einen echten Problempunkt: Wenn RRSI_VERTEX_PROJECTS nicht gesetzt ist, berechnet rrsi.llm.generate einen Index modulo der Anzahl der Projekte außerhalb seines Try-Blocks und löst eine ZeroDivisionError aus, sodass das hilfreiche Konfigurationsfehler im Code nie erreicht wird. Wir betrachten auch, wie die Änderungen markiert werden: Normalisierung behält einen deklarierten Komponenten nur dann bei, wenn der Diff die entsprechenden Beweise liefert; daher kann eine einfache Anpassung nicht als neue Fähigkeit dienen, um den Neuheitenbonus zu erhalten, und eine falsch etikettierte Veränderung im Kontextmanagement wird einfach so markiert, wie sie ist.

Code kopieren
@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()

Die Geschichte schreibt ein JSONL-Record pro Bearbeitung, und der Loop erzeugt vier Zusammenfassungen daraus. Die ausgewählte Sammlung entfernt die Bearbeitungen, die der Kritiker weggelassen hat, da sie nie gemessen wurden. Der jüngste Yield g_t ist der beste gemessene Gewinn pro Komponente in den letzten n_prune Runden. Jede Komponente, deren jüngster Yield nicht positiv ist, tritt in die Prune-Set B_t ein, zusammen mit allen Maschinen dieser Komponente, die noch im Amt sind. In unserer Geschichte zeigt dies auf eine prompte Bearbeitung in Runde null hin, die seitdem nicht erfolgreich war. Das Stall-Flag wird aktiviert, wenn der Score in den letzten w Runden weniger als Delta verändert wurde, und die Exploration schreibt dann die Direktive auf, die der Proposer erhält, indem ein Platz für Komponenten reserviert wird, die im Lauf noch nie genutzt wurden. Da der Proposer von all dem abhängig ist, wird keine falsche Hypothese erstellt.

Code kopieren
def 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 führt nie einen Agenten aus oder bewertet selbst die Ergebnisse; ein Domainadapter tut dies, und dieselbe Schnittstelle unterstützt die Kodierung, den Arbeitsbereich und die Instances der Ingenieuraufgaben. Wir implementieren einen für den simulierten Agenten: Aufgabensplittungen und Durchlaufverfahren, ein Verfahren, das die Daten aus Dateien unter dem Wurzelpunkt des Worktrees liest und Ergebnisse in einem Runs-Verzeichnis schreibt, ein Scoreverfahren, das diese wieder als TaskResult-Objekte liest, sowie die Kritikmuster und Komponentensignale des Domains. Die eigene Bewertungsfunktion von RRSI bewertet dann das Startgehäuse über den Adapter, und wenn sie zum zweiten Mal auf der gleichen Aufgabe aufgerufen wird, werden die gespeicherten Tests wieder abgelesen, anstatt sie erneut auszuführen – das ist die Sicherheit, die der Vertrag vorsieht.

Code kopieren
def 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()

Jetzt führen wir die gesamte Auswahlseite als Suchverfahren durch: zwanzig Runden, zwei Kandidaten pro Runde, Pakete nach dem angegebenen Budget bestimmt, Änderungen mit „Normalisieren“ versehen, Kandidaten durch eine Vorprüfung ausgewählt und bewertet über den Adapter, die Gewinner durch „select_round“ ermittelt, und jede Ergebnis ist in der Reihenfolge des RRSI-Loops in die Geschichte aufgenommen. Der skriptierte Vorschlagende führt fünf Arten von Änderungen durch, deren wirkende Effekte wir kennen: üblicherweise nützliche allgemeine Änderungen, leckende Änderungen, die Antworten auf Evolution-Tasks speichern, inaktive Konfigurationsänderungen, teure Sub-Agenten, die überall ein wenig helfen und 1,5-mal so viele Tokens verbrauchen, sowie günstigere Kontextkomprimierung. Wir vergleichen drei Akzeptierungsregeln auf demselben Kandidatenfluss über acht Seeds: Greedy behält den besten Score, wenn er steigt, der Kritiker fügt nur die Leakage-Screen hinzu, und RRSI fügt Algorithmus 2 mit einem kalibrierten Delta hinzu. Greedy erreicht den höchsten aufbewahrten Score von 0,681 gegenüber 0,616 bei RRSI, während es etwa zehn Antworten merkt und sein Token-Kosten nahezu verdreifacht; RRSI merkt nichts, endet bei etwa der Hälfte der Token-Kosten (1,53 Mal das Start-Kostenniveau gegenüber 2,97 Mal) und verwendet einen Drittel weniger Evaluierungen. Die Aufteilung der Lücke zwischen „evolve-to-held-out“ ist hier das nützlichste Ergebnis: Die Auswahl der besten, chaotischen Punktzahlen erhöht jeden Modus um etwa neun Punkte – und diese Erhöhung wird durch keine Regel in diesem Loop entfernt. Gleichzeitig eliminiert der Kritiker fast den gesamten Teil des Lernens.

Code kopieren
@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()

Unser erster Evaluator war ungenau, und der Delta wurde auf dieser Grundlage kalibriert. Daher spiegelte die Vorsicht von RRSI in Schritt 10 den Evaluator wider, nicht die Methode. Wir wiederholen den Vergleich mit größeren Evolve-Sets und mehr Versuchen. Wenn der Evaluator besser wird, sinkt der Delta von 0,096 auf 0,025, und der durchgehaltene Score von RRSI steigt von 0,616 auf 0,759. Dabei verwendet die unregulierte Suche bei der optimalen Einstellung 6,5 Mal so viele Tokens wie RRSI. Die Greedy-Version erzielt immer noch höhere Punktzahlen, und das Notizbuch erklärt warum: Dieser simulierte Welt hat keine abnehmenden Ergebnisse, daher kauft jeder Sub-Agent der gierigen Suche ständig mehr Genauigkeit – was die günstigste Welt für Ausgaben ist. RRSI tauscht einen Teil dieser Punktzahl gegen ein begrenztes Token-Bill aus, ohne abgespeicherte Antworten und weniger verlorene Bewertungen; die Größe des Tauschs wird durch Delta festgelegt, das Delta wird eher gemessen als verlangt.

Code kopieren
banner("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
""")

Die Zusammenfassung gibt das einzeilige Ergebnis jeder Sektion an, und macht klar, was das Miniaturmodell nicht modelliert – nämlich Änderungen, die dem Evolve-Suite helfen, aber einem anderen Suite schaden. Das sind genau die Zwecke, für die der LLM-Kritiker des Artikels und die verbleibenden Split-Aufgaben existieren. Außerdem gibt es einen Proposer, der die Geschichte liest, dann eine echte Instanz ausführt, ein Domänen hinzufügt und eine gespeicherte Runde unter einem anderen Delta neu bewertet, ohne etwas neu auszuführen.

Zusammenfassend haben wir den RRSI auf der Strecke zerlegt, an der die Idee des Artikels tatsächlich liegt – die Regeln, die entscheiden, welche Änderungen ein selbstverbesserndes Agent beibehält – und diese offline auf einer CPU ohne Modell oder API-Schlüssel ausgeführt. Der Schätzungstyp weigert sich, einen Abfall zu belohnen; die Geräuschbänder werden gemessen, nicht ausgewählt, und Algorithmus 2 ist subtiler als „Ablehnung von störenden Vorteilen“: Er lässt das Score nie unter das beste Ergebnis minus das Geräuschband fallen, er zahlt für echte Vorteile zusätzliche Tokens, und innerhalb des Bandes bevorzugt er absichtlich den günstigeren Hänger, selbst wenn dieser ein etwas niedrigeres Score hat. Die Überprüfung dieser Regeln gegenüber einer Welt, deren Grundwahrheit wir kontrollierten, war der informativste Teil: der deterministische Kritiker entfernte fast alle Erinnerungen, bevor jede Bewertung abgeschlossen war. Das Algorithmus 2 hielt das Token-Bill mehrmals unter eine unregelmäßige Suche, und es entfernte weder die Inflation, die durch die Auswahl der besten, störansten Punkte entsteht – was nur durch erneute Messung auf unbekannten Aufgaben möglich ist. Die ehrliche Bemerkung ist ebenso klar: In einer Welt, in der Tokens immer für Genauigkeit sorgen, sind die unregulierten Suchergebnisse höher. Daher hängt der Wert von RRSI’s Regulierung davon ab, wie teuer die Tokens sind und wie viel eine verlorene Antwort kosten würde. Das Notizbuch gibt Ihnen die Möglichkeit, beides mit Ihrem eigenen Evaluierungsverfahren zu messen.


Schauen Sie sich die VOLLSTEN KODEN hier an. Der ganze Anteil des Lobes geht an den Forscher dieses Projekts. Folgen Sie uns auch gerne auf Twitter und vergessen Sie nicht, sich bei unserem 150k+ML SubReddit anzumelden sowie unseren Newsletter zu abonnieren. Warte! Sind Sie auf Telegram? Jetzt können Sie sich auch auf Telegram bei uns anmelden.

Die Beitrag Google Research RRSI Guide: Mastering Self-Improving AI Agents erschien zuerst auf MarkTechPost.

Originalquelle

MarkTechPost

Hinweise zum Inhalt

Originalveröffentlichung und Rechte liegen bei der Quelle.

Maschinelle Übersetzung · Original beachten