RadarAI· MarkTechPost（RSS）·· 9 天前AI 评分39
MarkTechPost 教程拆解 Google Research 的 RRSI：在离线模拟环境中掌握自改进 AI 智能体的编辑保留规则
AI 导读
MarkTechPost 发布教程，从 google-research 仓库安装 RRSI（Regularized Recursive Self-Improvement），只驱动其决定保留哪些编辑的纯 Python 规则部分，并接入自建模拟 agent 环境做 ground truth 审计。
正文 · AI 翻译
MarkTechPost· Sana Hassan·2026-10-09 13:06· 2 小时前AI 评分55
Google Research RRSI 指南：掌握自我改进的 AI 智能体
AI 导读
MarkTechPost 发布教程，从 google-research 仓库安装 RRSI（Regularized Recursive Self-Improvement），只驱动其决定保留哪些编辑的纯 Python 规则部分，并接入自建模拟 agent 环境做 ground truth 审计。
正文
在本教程中，我们实现 RRSI（Regularized Recursive Self-Improvement），这是一种让 LLM 智能体围绕冻结模型重写自身 harness、提示词、工具、记忆、控制流和子智能体的方法，同时避免 harness 对其演化所基于的任务过拟合。完整的 RRSI 循环使用 Vertex AI 上的 Claude Opus 起草编辑，并在 Docker 基准测试中为其评分，这不是免费 notebook 能运行的。RRSI 中真正承载论文思想的部分——决定保留哪些提议编辑的规则——是纯 Python 的，这正是我们直接驱动的部分。我们从官方仓库安装该包，逐一讲解其估计器、校准噪声带、选择算法的两个分支、退火编辑预算、确定性泄漏筛查及其编辑历史，然后将一个模拟智能体接入 RRSI 自有的 Domain 接口。由于模拟环境是我们自己构建的，我们知道每次编辑的真实效果，这让我们能够对照 ground truth 审计 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，固定到本 notebook 编写时所对应的提交，因为该包不在 PyPI 上。它唯一的依赖是 Anthropic 客户端，搜索角色用它调用 Claude，而我们从未使用它。然后我们打印仓库自身记录的论文符号与实现这些符号的函数之间的映射：evaluate 中的经验分数和成本估计，calibrate 中的噪声带，selection 中的算法 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 技能减去任务难度的逻辑函数，并在四十个任务上以每次两个试验对未改动的起始 harness 评估六次：尽管什么都没变，分数却散布了 0.113。calibrate 将同一 harness 的重复评估转化为 delta，即两次运行之间差值的标准差的两倍。在 80 次试验下 delta 约为 0.108；在 3,200 次试验下它降至约 0.013，处于论文为其实例报告的范围（0.004 到 0.020）内。出于选择目的，任何小于 delta 的增益都与重新运行同一 harness 无法区分。
Copy CodeCopiedUse a different Browser
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()
算法 2 实现为纯函数，因此我们可以把候选交给它并逐字读取其理由。我们将现任固定为 S 0.630 和 10,000 个 token，历史最佳分数为 0.640，delta 为 0.020，并将八个候选送入 judge。低于下限（历史最佳分数减去 delta）的候选会被直接拒绝。大于 delta 的增益必须为其额外 token 付出代价，规则是相对成本变化保持在 0.10 加上 40 倍增益以下；在 +20% token 下 +6 点的增益被接受，而在 +150% token 下 +3 点的增益不被接受。在区间内，分数被视为平局，由 100 倍增益减去 15 倍成本变化，再加上对从未被接受的结构组件的小额奖励所构成的整形分数来决定；这就是 RRSI 如何接受候选 D——它的得分低于现任但 token 成本少 20%——以及一个新的子智能体如何打破一个同等得分的提示微调无法打破的平局。领域守卫无论分数如何都拥有一票否决权。
Copy CodeCopiedUse a different Browser
@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 拒绝的候选。得分最高者落选，因为其三点增益不足以支付多出 150% 的 token，被 critic 拒绝的候选从未进入评估，而获胜者将现任降低半分，同时将其 token 成本削减五分之一。保持这一安全性的细节是 S*，它只会上升：下限锚定于历史测得的最佳分数而非现任，因此一连串更便宜但稍差的交换无法在多个轮次中让分数漂移走。
Copy CodeCopiedUse a different Browser
@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 的余弦调度，默认情况下它在前八轮允许每个候选最多四个协调编辑，接下来五轮为三个，最后七轮为两个。公式中的上限意味着预算只在 t = T 时达到其最小值一，即运行结束后的下一步，这在阅读方程时很容易被忽略。由于一个捆绑中的每个编辑都继承该捆绑的一次测量，不断缩减的预算正是使运行后期历史可归因于更少组件的原因。预算限制了有多少编辑一起移动，而从不限制 harness 最终可能包含哪些机制。
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()
评审器在任何评估开销产生之前，分两层筛选每个候选 diff。第一层是针对通用凭据模式加上该域自身拒绝列表的确定性预检；借助针对 evolve-set 任务 id 和评分器路径的模式，它无需调用模型即可拒绝一个记住了 task_007 答案的 diff、一个读取预期输出的 diff、一个泄露 API 密钥的 diff，以及一个空 diff。干净的 diff 会进入第二层，即由 Claude 进行的意图审查，而这里 notebook 暴露了一个真实的坑：当 RRSI_VERTEX_PROJECTS 未设置时，rrsi.llm.generate 在其 try 块之外计算项目数取模的索引并抛出 ZeroDivisionError，因此代码中那个有用的配置错误永远不会被触发。我们还考察了编辑如何被标记：normalize 仅在 diff 携带证据时才保留声明的组件，因此提示词微调无法伪装成新技能来赢得新颖性奖励，而被错误标记的上下文管理变更会按其真实性质被标记。
@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 记录，循环从中推导出四种摘要。已尝试集合排除评审器丢弃的编辑，因为它们从未被测量。近期收益 g_t 是最近 n_prune 轮内每个组件的最佳实测增益，任何近期收益不为正的组件都会进入剪枝集合 B_t，同时该组件中仍存在于当前方案中的任何机制也会一并进入；在我们的历史中，这标记了第零轮接受的一个提示词编辑，此后一直未产生收益。当分数在最近 w 轮内变化小于 delta 时，停滞标志会触发，随后探索会写入提议者收到的指令，为运行从未尝试过的组件保留一个候选槽位。由于提议者以所有这些为条件，被证伪的假设不会被再次抽取。
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 从不运行智能体或自行对交付物评分；由 Domain 适配器来完成，同一接口支撑了论文中的编码、工作区和工程实例。我们为模拟智能体实现了一个：evolve 和 held-out 任务划分，一个 run 方法从 worktree 根目录下的文件读取 harness 并将试验结果写入 runs 目录，一个 score 方法将它们读回为 TaskResult 对象，以及该域的评审器模式和组件信号。随后 RRSI 自身的 evaluate 函数通过适配器对起始 harness 评分，而在同一作业上第二次调用它会读回已存储的试验，而不是重新运行它们，这正是契约所要求的恢复安全性。
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()
现在我们以搜索的方式运行整个选择侧：二十轮，每轮两个候选，束大小由退火预算决定，编辑由 normalize 标记，候选由 precheck 筛选并通过 adapter 评分，胜者由 select_round 选出，每个结果都按 RRSI 循环使用的顺序写入 History。脚本化的提议者抽取五种编辑，其真实效果我们已知：通常有帮助的通用更改、会记住 evolve 任务答案的泄漏性编辑、惰性配置更改、以 1.5 倍 token 在所有地方都略有帮助的昂贵子代理，以及更便宜的上下文压缩。我们在八个种子上对同一候选流比较三种接受规则：greedy 在分数上升时保留最佳分数，critic 只添加泄漏筛查，RRSI 添加带有校准 delta 的 Algorithm 2。Greedy 达到最高的留出分数，0.681，而 RRSI 为 0.616，同时记住了约十个答案，token 成本几乎增至三倍；RRSI 没有记住任何答案，最终 token 成本约为一半（起始 harness 的 1.53 倍对 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 是根据它校准的，因此 RRSI 在第 10 步的谨慎反映的是评估器而非方法本身。我们用更大的 evolve 集和更多试验重复比较。随着评估器变得更锐利，delta 从 0.096 降至 0.025，RRSI 的留出分数从 0.616 升至 0.759，而在最锐利设置下未正则化的搜索使用的 token 是 RRSI 的 6.5 倍。Greedy 仍然得分更高，笔记本说明了原因：这个模拟世界没有收益递减，因此 greedy 搜索堆叠的每个子代理都在持续购买准确率，这是对花费最有利的世界。RRSI 用部分分数换取有界的 token 账单、没有记住的答案和更少的浪费评估，而交易的大小由 delta 设定，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 上离线运行了它们。估计器拒绝奖励崩溃，噪声带是测量出来的而非选定的，而算法 2 比“拒绝有噪声的增益”要微妙得多：它绝不让分数低于已见最佳值减去噪声带，它让真正的增益为其额外 token 买单，而在噪声带内，它刻意偏好更便宜的测试框架，即使那个框架得分略低。将这些规则对照一个我们掌控其真实基准的世界进行审计，是最有信息量的部分：确定性评判器在任何评估花费之前就移除了几乎所有的记忆化，算法 2 将 token 账单保持在无正则化搜索的数倍以下，而两者都没有消除因从有噪声的分数中挑选最佳值而带来的虚高，这只有在对未见任务重新测量时才能做到。诚实的告诫同样清楚：在一个 token 总能买到准确率的世界里，无正则化搜索得分更高，因此 RRSI 正则化的价值取决于 token 有多贵，以及泄露的答案会让你付出多大代价，而该 notebook 为你提供了在你自己的评估器上测量这两者的机制。
在此查看 完整代码。所有功劳归于该项目的研究者。另外，欢迎在 Twitter 上关注我们，别忘了加入我们的 15 万+ 机器学习 SubReddit 并订阅 我们的新闻通讯。等等！你在 Telegram 上吗？现在你也可以在 Telegram 上加入我们。
文章 Google Research RRSI 指南：掌握自我改进 AI 智能体 首次发表于 MarkTechPost。
来源：MarkTechPost · marktechpost.com
来源：RadarAI · aihot.news