在本教程中，我们实现 RRSI(Regularized Recursive Self-Improvement,正则化递归自我改进)，这是一种让 LLM 智能体围绕冻结模型重写自身 harness、提示词、工具、记忆、控制流和子智能体的方法，同时避免 harness 对其所演化的任务过拟合。完整的 RRSI 循环使用 Vertex AI 上的 Claude Opus 起草编辑，并在 Docker 基准测试中进行评分，这是免费笔记本无法运行的。而 RRSI 中真正承载论文思想的部分——决定保留哪些提议编辑的规则——是纯 Python 代码，这正是我们直接驱动的部分。我们从官方仓库安装该包，逐步讲解其估计器、校准噪声带、选择算法的两个分支、退火编辑预算、确定性泄漏审查以及编辑历史，然后将一个模拟智能体接入 RRSI 自身的 Domain 接口。由于我们自己构建了模拟环境，我们知道每次编辑的真实效果，这让我们能够对照真实情况审计 RRSI 的决策，并将其与一种简单保留得分最高结果的未正则化搜索进行比较。
复制代码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.")
我们从 google-research 仓库安装 RRSI，并固定到编写本笔记本时所用的 commit，因为该包不在 PyPI 上。它唯一的依赖是 Anthropic 客户端，搜索角色用它来调用 Claude，而我们从不使用它。然后我们打印仓库自身文档记录的论文符号与实现这些符号的函数之间的映射：evaluate 中的经验得分和成本估计，calibrate 中的噪声带，selection 中的 Algorithm 2，schedule 中的退火编辑预算，critic 中的泄漏审查，以及 history 中带有 yield、prune、stall 和 exploration 摘要的编辑历史。RRSIConfig 保存论文的超参数，下面的每个函数都像真实循环那样接收它。
复制代码@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 为每个 harness 测量两个数字：S，即所有任务的每个试验的平均奖励；C，即每次试验的平均策略 token 数。TaskResult 记录一个任务的试验并对其进行汇总。任何智能体评估都值得借鉴的细节是如何处理缺失的试验。当一个候选在最难的任务上崩溃时，丢弃缺失试验的估计器会报告 0.750，并使崩溃看起来像一种改进。与此同时，RRSI 将每个缺失试验记为零奖励并保留完整的分母，报告与之前相同的 0.500，因此候选无法通过摧毁它觉得困难的试验来显得更好。加权奖励适用于诸如 Harvey LAB 这类按量规评分的套件，此时 S 变成所有通过标准的比例，而不是各任务均值的均值。
复制代码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()
在任何规则能够区分真实增益与运气之前，它需要知道一个 harness 的得分自身会在多大范围内波动。我们构建一个小的模拟智能体，其在每个任务上的成功率是 harness 技能减去任务难度的 logistic 函数，并在 40 个任务、每个任务 2 次试验的设置下对未更改的起始 harness 进行 6 次评估：尽管什么都没变，得分却波动了 0.113。calibrate 将同一 harness 的重复评估转换为 delta，即两次运行之间差异标准差的两倍。在 80 次试验下，delta 约为 0.108；在 3,200 次试验下，它降至约 0.013，处于论文为其实例报告的范围(0.004 到 0.020)之内。就选择而言，任何小于 delta 的增益都与重新运行同一 harness 无法区分。
复制代码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()
Algorithm 2 被实现为纯函数，因此我们可以把候选交给它并逐字读取其理由。我们设定一个现任者为 S 0.630 和 10,000 个 token，历史最佳得分为 0.640,delta 为 0.020，并将 8 个候选送入 judge。低于下限——即历史上见过的最佳得分减去 delta——的候选会被直接拒绝。大于 delta 的增益必须为其额外的 token 付出代价，规则是相对成本变化保持在 0.10 加上 40 倍增益以下；+6 点增益伴随 +20% 的 token 会被接受，而 +3 点增益伴随 +150% 的 token 则不会被接受。在噪声带之内，得分被视为平局，由整形得分——100 倍增益减去 15 倍成本变化，再加上对从未被接受过的结构组件的小额奖励——来决出胜负；这就是 RRSI 接受候选 D 的方式，该候选得分低于现任者但 token 成本少 20%，也是新的子智能体能够打破平局而得分相同的提示词微调却不能的原因。无论得分如何，领域守卫都可以否决。
复制代码@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 对一轮中的每个候选应用 judge 并保留得分最高的可接受者。我们给它一个昂贵候选、一个更便宜但稍差的候选，以及一个已被 critic 拒绝的候选。最高分者落选，因为它 3 分的提升抵不上 150% 的 token 增量，被 critic 拒绝的候选从未进入评估，而胜出者将现任版本拉低半分，同时把 token 成本削减了五分之一。保证这一切安全的关键细节是 S*，它只会上升：下限锚定在曾有过的最佳得分上，而不是现任版本上，因此一连串“更便宜但稍差”的替换无法在多轮中把得分慢慢挪走。
Copy 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()
提案侧规范的是编辑如何被起草，而不是哪些编辑被保留。edit_budget 实现了论文中的退火 L0 预算：一个从 b_max 到 b_min 的余弦调度，默认情况下前 8 轮每个候选允许最多 4 个协同编辑，接下来 5 轮为 3 个，最后 7 轮为 2 个。公式中的上限意味着预算只在 t = T 时达到其最小值 1，也就是运行结束后的一步，这在阅读公式时很容易被忽略。由于 bundle 中的每个编辑都继承该 bundle 的唯一一次测量，收缩的预算正是让后期运行历史能归因到更少组件的原因。预算限制的是多少编辑一起出行，从不限制 harness 最终可能包含哪些机制。
Copy 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()
critic 在任何评估开销发生之前筛查每个候选 diff，分为两层。第一层是针对通用凭据模式加上领域自身 denylist 的确定性预检；借助针对 evolve 集任务 id 和 grader 路径的模式，它拒绝了一个记住 task_007 答案的 diff、一个读取期望输出的 diff、一个泄露 API 密钥的 diff，以及一个空 diff，全程无需调用模型。干净的 diff 则进入第二层，即由 Claude 进行的意图审查，这里 notebook 揭示了一个真实的坑：在 RRSI_VERTEX_PROJECTS 未设置时，rrsi.llm.generate 会在其 try 块之外对项目数量取模计算索引并抛出 ZeroDivisionError，因此代码中那条有用的配置错误提示永远不会被触及。我们还考察了编辑如何被打标签：normalize 只有在 diff 带有证据时才保留声明的组件，因此 prompt 上的小调整不能伪装成新技能来赢取新颖性奖励，而被错误标记的上下文管理改动会被打上它本来的标签。
Copy 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()
History 为每个编辑写入一条 JSONL 记录，循环从其中导出四个摘要。tried 集合排除 critic 丢弃的编辑，因为它们从未被测量。近期产出 g_t 是最近 n_prune 轮内每个组件的最佳测量增益，近期产出不为正的每个组件都会进入 prune 集合 B_t，连同现任版本中仍属于该组件的任何机制；在我们的历史中，这标记了第 0 轮被接受、此后再无回报的一次 prompt 编辑。stall 标志在得分于最近 w 轮内变动小于 delta 时触发，随后探索会写入 proposer 收到的指令，为运行从未使用过的组件保留一个候选槽位。由于 proposer 以所有这些信息为条件，一个被证伪的假设不会被再次抽取。
Copy 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 从不亲自运行 agent 或为交付物评分；由 Domain 适配器来做，而论文的 coding、workspace 和 engineering 实例也由同一接口支撑。我们为模拟 agent 实现了一个：evolve 与 held-out 任务划分、一个从 worktree 根目录下文件读取 harness 并把试验结果写入 runs 目录的 run 方法、一个把它们读回为 TaskResult 对象的 score 方法，以及该领域的 critic 模式和组件信号。RRSI 自身的 evaluate 函数随后通过该适配器对起始 harness 评分，而对同一任务再次调用它会读回已存储的试验而不是重新运行，这正是契约所要求的恢复安全性。
Copy 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()
现在我们将整个选择侧作为搜索来运行：二十轮，每轮两个候选，bundle 大小按退火预算确定，编辑由 normalize 打标签，候选由 precheck 筛选并通过 adapter 打分，胜者由 select_round 选出，每个结果按 RRSI 循环使用的顺序写入 History。脚本化的提出器生成五种我们已知真实效果的编辑：通常有益的一般性改动、会记忆 evolve 任务答案的泄漏性编辑、无效的配置改动、在 1.5 倍 token 下处处小有帮助的昂贵子代理，以及更便宜的上下文压缩。我们在相同的候选流上用 8 个种子比较三种接受规则：greedy 在分数上升时保留最佳分数，critic 只增加泄漏筛查，RRSI 则加入带校准 delta 的 Algorithm 2。Greedy 达到了最高的留出集分数，0.681 对 RRSI 的 0.616，同时记住了大约十个答案，token 成本几乎翻了三倍；RRSI 没有记住任何答案，最终 token 成本约为其一半（1.53 倍起始 harness 对 2.97 倍），并且少用了三分之一的评估次数。拆分 evolve 到留出集之间的差距是这里最有用的结果：从含噪分数中挑选最佳会使每种模式膨胀约九分，而此循环中的任何规则都无法消除这一点。与此同时，critic 几乎消除了全部的记忆化成分。
复制代码@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()
我们的第一个评估器是有噪声的，delta 是根据它校准的，因此第 10 步中 RRSI 的谨慎反映的是评估器而非方法本身。我们用更大的 evolve 集和更多试验重复该比较。随着评估器变得更精确，delta 从 0.096 降至 0.025，RRSI 的留出集分数从 0.616 升至 0.759，而在最精确设置下的未正则化搜索使用的 token 是 RRSI 的 6.5 倍。Greedy 仍然得分更高，notebook 说明了原因：这个模拟世界没有收益递减，因此 greedy 搜索堆叠的每个子代理都会继续换来精度，这是对花钱最有利的世界。RRSI 用部分分数换取有界的 token 账单、零记忆答案和更少的浪费评估，而交换的规模由 delta 决定——它自己去测量，而不是索取。
复制代码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
""")
摘要部分打印每个部分返回的单行结果，明确说明这个迷你模型未涵盖的内容，即在帮助 evolve 套件的同时损害另一套件的编辑（这正是论文的 LLM critic 和留出集划分存在的原因），以及一个读取历史的提出器，然后指向运行一个真实实例、添加一个领域，以及在不重新运行任何内容的情况下用不同的 delta 对存储的一轮进行重新裁决。
总而言之，我们沿着论文思想真正所在的那条线拆解了 RRSI，即决定自改进代理保留哪些编辑的规则，并在没有模型或 API 密钥的情况下在 CPU 上离线运行它们。估计器拒绝奖励崩溃，噪声带是测量出来的而非人为选择的，而 Algorithm 2 比“拒绝有噪声的收益”更微妙：它绝不允许分数降到已见最佳值减去噪声带之下，它让真实收益为其额外的 token 付出代价，并且在带内它有意偏好更便宜的 harness，即使是得分略低的那个。在一个我们控制了真值的世界中审计这些规则是最有启发性的部分：确定性 critic 在任何评估被消耗之前就消除了几乎全部记忆化，Algorithm 2 将 token 账单保持在未正则化搜索的几倍以下，而两者都无法消除从含噪分数中挑选最佳带来的膨胀，这只有在未见任务上重新测量才能做到。诚实的警告同样清晰：在一个 token 总能换来精度的世界里，未正则化的搜索得分更高，因此 RRSI 正则化的价值取决于 token 有多贵以及泄漏一个答案会让你付出多大代价，而 notebook 为你提供了在自己的评估器上测量这两者的机制。
查看 完整代码请点击此处。所有功劳归于该项目的研究者。另外，欢迎在 Twitter 上关注我们，别忘了加入我们的 150k+ML SubReddit 并订阅 我们的新闻通讯。等等！你在用 telegram 吗？ 现在你也可以在 telegram 上加入我们了。
这篇文章 Google Research RRSI Guide: Mastering Self-Improving AI Agents 首发于 MarkTechPost.