在本教程中，我们实现 RRSI（Regularized Recursive Self-Improvement，正则化递归自我改进），这是一种让 LLM 智能体围绕一个冻结的模型重写自身的 harness、提示词、工具、记忆、控制流和子智能体的方法，同时避免 harness 对其演化所基于的任务过拟合。完整的 RRSI 循环使用 Vertex AI 上的 Claude Opus 起草编辑，并在 Docker 基准测试中为其打分，这不是一个免费 notebook 能运行的。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，固定到本 notebook 编写时所对应的 commit，因为该包不在 PyPI 上。它唯一的依赖是 Anthropic 客户端，搜索角色用它来调用 Claude，而我们从不使用它。然后我们打印仓库自身记录的论文符号与实现这些符号的函数之间的映射：evaluate 中的经验分数和成本估计，calibrate 中的噪声带，selection 中的 Algorithm 2，schedule 中的退火编辑预算，critic 中的泄漏筛查，以及 history 中带有 yield、prune、stall 和 exploration 摘要的编辑历史。RRSIConfig 保存论文的超参数，下面的每个函数都像真实循环那样接收它。
复制代码已复制使用其他浏览器
@section("2. Evaluate(H)：一个分数和一个成本，以及为什么崩溃算作零")
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" 三个任务 x k=2 次试验 -> S_hat = {ev.S:.3f} C_hat = {ev.C:,.0f} tokens/试验"
f" （预期 {ev.n_expected} 次试验，{ev.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 一个候选在最难的任务上崩溃，而不是失败：")
print(f" 一个丢弃缺失试验的估计器报告 {naive:.3f} <- 看起来像是增益")
print(f" RRSI 的 aggregate（缺失 = 0，完整分母） {ev_crash.S:.3f} <- 崩溃没有奖励")
print(f" ...而 C_hat 只使用记录的 token 计数：{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 加权奖励（Harvey LAB 风格：权重 = 评分标准条目数）：")
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 变为所有通过标准的比例，而非各任务均值的均值。
Copy CodeCopiedUse a different Browser
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):
来自对同一 harness H0 的 `repeats` 次独立评估的 delta。
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. 噪声带：同一个 harness 的分数在两次评估之间会波动多大")
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 在 40 个任务 x k=2 上评估 6 次：{[round(s, 3) for s in scores]}")
print(f" 真实期望分数 {true_score(H0, world, ids):.3f}；估计值的离散程度"
f" {max(scores) - min(scores):.3f}。harness 本身没有任何变化。")
print(f"\n calibrate() 将重复评估转化为 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 论文中校准后的区间为 0.017（coding）、0.004（workspace）和 0.020（engineering）。")
print(" 小于 delta 的增益与重新运行同一个 harness 无法区分，而")
print(" Algorithm 2 正是这样处理的。记住第一行：它将成为第 11 步的教训。")
return "delta " + ", ".join(f"{k}={v:.3f}" for k, v in deltas.items())
noise_band()
在任何规则能够区分真正的增益和运气之前，它需要知道一个 harness 的分数自身会波动多大。我们构建一个小型模拟智能体，它在每个任务上的成功是 harness 技能减去任务难度的 logistic 函数，并在四十个任务上对未改动的起始 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):
"""一个 EvalResult，其分数恰好为 S（以 1/n 为步长），每次试验消耗 C 个 token。"""
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. 算法 2，逐分支：下限、成本规则和噪声带")
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 跌破下限", 0.600, 10_000, ["prompt"], None),
("B 真正的增益，为其 token 付费", 0.690, 12_000, ["prompt"], None),
("C 真正的增益，但过于昂贵", 0.660, 25_000, ["subagent"], None),
("D 略差但更便宜", 0.625, 8_000, ["context_mgmt"], None),
("E 在带内但更贵", 0.640, 11_500, ["prompt"], None),
("F 在带内，新 sub-agent", 0.630, 10_000, ["subagent"], None),
("F' 在带内，prompt 微调", 0.630, 10_000, ["prompt"], None),
("G 类似 B，但破坏了领域防护", 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 RRSI 按顺序应用的三条规则：")
print(" 1. 永远不低于历史最佳分数减去噪声带 (A)")
print(" 2. 大于 delta 的增益必须为其额外 token 买单：dC <= 0.10 + 40*dS (B, C)")
print(" 3. 在噪声带内分数视为平局，因此优先选择更便宜的 harness，并让一个")
print(" 从未尝试过的结构性组件来打破平局 (D, E, F, F')")
print(" D 是最令人意外的一个：RRSI 接纳了一个分数低于现任的 harness。")
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 tokens，历史最佳分数为 0.640，delta 为 0.020，然后让八个候选通过 judge。低于下限（历史最佳分数减去 delta）的候选会被直接拒绝。大于 delta 的增益必须为其额外 token 买单，规则是相对成本变化保持在 0.10 加上 40 倍增益以下；+6 个点的增益伴随 +20% tokens 会被接纳，而 +3 个点的增益伴随 +150% tokens 则不会。在噪声带内，分数被视为平局，由 100 倍增益减去 15 倍成本变化构成的塑形分数，再加上一个从未被接受的结构性组件的小额加分来决定；RRSI 正是这样接纳了候选 D——它的分数低于现任，但 token 成本低 20%——也正是这样让一个新的子智能体打破平局，而一个分数相同的提示词微调则做不到。领域守卫无论分数如何都拥有一票否决权。
Copy CodeCopiedUse a different Browser
@section("5. 一轮选择：最高分并不总是胜出")
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 会对一轮中的每个候选应用评判器，并保留得分最高的可接受候选。我们给它一个昂贵的候选、一个更便宜但稍差的候选，以及一个已被评判器拒绝的候选。得分最高的那个输了，因为它的三个百分点增益不足以支付多出的 150% token；被评判器拒绝的候选从未进入评估；而获胜者把现任者拉低了半分，同时将其 token 成本削减了五分之一。让这一切保持安全的细节是 S*，它只会上升：下限锚定在曾测得的最佳分数上，而不是锚定在现任者上，因此一连串更便宜但稍差的替换无法在多轮中把分数带偏。
Copy CodeCopiedUse a different Browser
@section("6. 提案侧：一个从 4 退火到 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 早期的候选可以捆绑最多 4 个协同编辑。注意 ceil()：余弦项
print(f" 仅在 t = T 时恰好为零，因此在 {CFG.T} 轮的运行中，预算最低只到
f" {min(table)}，而不是 {CFG.b_min}。")
print(" 捆绑中的每个编辑都继承该捆绑的单一测量值，因此不断缩减的预算
print(" 正是让后期历史可归因于更少组件的原因。它限制的是有多少个编辑一起
print(" 同行，而从不限制 harness 最终可能包含哪些机制。")
return "budget " + "".join(str(b) for b in table)
edit_budget_schedule()
提案侧规范的是编辑如何起草，而不是保留哪些编辑。edit_budget 实现了论文中的退火 L0 预算：从 b_max 到 b_min 的余弦调度，默认情况下它允许每个候选在前八轮最多进行 4 个协同编辑，接下来五轮为 3 个，最后七轮为 2 个。公式中的上取整意味着预算仅在 t = T 时达到最小值 1，也就是运行结束后的下一步，这一点在阅读方程时很容易被忽略。因为捆绑中的每个编辑都继承该捆绑的单一测量值，所以不断缩减的预算正是让后期运行历史可归因于更少组件的原因。预算限制了有多少个编辑一起同行，而从不限制 harness 最终可能包含哪些机制。
复制代码已复制使用其他浏览器
CRITIC_PATTERNS = [
(r"\btask_\d{3}\b", "硬编码了一个 evolve-set 任务 id"),
(r"expected_output|grader|rubric\[", "读取了评分器或预期答案"),
]
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": "一个编码智能体 harness。"}
@section("7. 评论者的确定性层，以及编辑如何被打标签")
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-set 任务 id 和 grader 路径的模式，它拒绝了一个记住了 task_007 答案的 diff、一个读取预期输出的 diff、一个泄露 API key 的 diff，以及一个空 diff，全程无需调用模型。干净的 diff 会落到第二层，即由 Claude 进行的意图审查，而这里 notebook 揭示了一个真实的坑：在 RRSI_VERTEX_PROJECTS 未设置时，rrsi.llm.generate 在其 try 块之外计算一个对项目数取模的索引并抛出 ZeroDivisionError，因此代码中那个有用的配置错误永远不会被触达。我们还审视了编辑如何被打标签：normalize 仅在 diff 携带证据时才保留声明的组件，因此一个 prompt 微调无法伪装成新技能来赢得新颖性奖励，而一个被错误标记的上下文管理变更会被按其本来面目打标签。
复制代码已复制使用其他浏览器
@section("8. 编辑历史：尝试过什么、什么有回报、什么该剪掉")
def edit_history():
with tempfile.TemporaryDirectory() as tmp:
h = History(Path(tmp) / "history.jsonl")
log = [
(0, "A", [("prompt", "告诉智能体先读取失败的测试")], "ACCEPTED", 0.030, 0.02, True, 0.66),
(1, "A", [("prompt", "在编辑前先要求一个计划")], "REJECTED", -0.010, 0.05, False, 0.65),
(1, "B", [("subagent", "添加一个审查者子智能体"), ("memory", "持久化 lint 规则")],
"REJECTED", -0.020, 0.40, False, 0.64),
(2, "A", [("config", "提高步数上限")], "ACCEPTED", 0.005, -0.03, True, 0.665),
(3, "A", [("prompt", "更短的 system prompt")], "LOST", -0.001, -0.10, False, 0.664),
(4, "B", [("memory", "缓存 task_014 的解答")], "critic_reject", None, None, False, None),
(5, "A", [("prompt", "更严格的输出格式")], "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())} 条每次编辑的记录，来自 {len(log)} 个候选"
f"（第 1 轮中的双编辑捆绑用一次测量写入了两条记录）")
print(f" T_t，已尝试的组件 : {sorted(h.tried())} （被 critic 拒绝的编辑不算‘已尝试’）")
g_t = h.yield_g(t_now, CFG.n_prune)
print(" g_t，最近 n_prune 轮中的最佳增益: "
+ ", ".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，剪枝集合 : {[p['component'] for p in prune]}")
for p in prune:
if p["accepted_edits_in_incumbent"]:
print(f" {p['component']}: 仍在 incumbent 中但近期没有增益 ->"
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 在各轮次 {trajectory}: 变动 {trajectory[t_now] - trajectory[t_now - CFG.w]:+.3f}"
f" 在最近 w={CFG.w} 轮中 -> sigma_t = {sigma}")
print(" 下一轮告知 proposer 的内容:")
print(textwrap.indent(textwrap.fill(ex["text"], 84), " "))
print("\n proposer 看到这段历史，因此被证伪的假设（‘先要一个计划’，-1pp）"
print(" 不会被重新抽取，而停滞的运行会被推向它从未接触过的组件。")
return f"prune set {[p['component'] for p in prune]}, stall flag {sigma}"
edit_history()
History 每次编辑写一条 JSONL 记录，循环从中推导出四个摘要。已尝试集合排除了 critic 丢弃的编辑，因为它们从未被测量。近期收益 g_t 是最近 n_prune 轮内每个组件测得的最佳增益，任何近期收益不为正的组件都会进入剪枝集合 B_t，同时该组件中仍在 incumbent 里的任何机制也会一并进入；在我们的历史中，这标记了第零轮接受的一次 prompt 编辑，此后它再未带来回报。当分数在最近 w 轮内变化小于 delta 时，停滞标志触发，探索随后写入 proposer 收到的指令，为本次运行从未尝试过的组件保留一个候选槽位。由于 proposer 以这一切为条件，被证伪的假设不会被再次抽取。
Copy CodeCopiedUse a different Browser
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):
"""一个针对模拟智能体的 Domain 适配器：RRSI 的 coding、
workspace 和 engineering 实例所实现的同一契约。"""
name = "simulated"
critic_patterns = CRITIC_PATTERNS
component_signals = COMPONENT_SIGNALS
briefs = {"critic": "一个在 parse/search/edit/test 任务上评估的 coding agent harness。"}
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 # 可安全恢复，正如契约所要求
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. 领域适配器：将环境接入 RRSI 自己的 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 与留出任务划分、一个从 worktree 根目录下的文件读取 harness 并将试验结果写入 runs 目录的 run 方法、一个将这些结果读回为 TaskResult 对象的 score 方法，以及该领域的 critic 模式和组件信号。随后 RRSI 自己的 evaluate 函数通过该适配器为起始 harness 打分，而在同一任务上第二次调用它时会读回已存储的试验结果，而不是重新运行它们，这正是契约所要求的恢复安全性。
Copy CodeCopiedUse a different Browser
def propose_edit(r, evolve_ids):
"""脚本化的提议者。每次抽取都是一次编辑，其真实效果我们已知。"""
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" # 在任何评估开销之前就筛掉
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): # loop.round() 写入的记录
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: # 未正则化：如果分数上升就保留最佳分数
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 与 RRSI 的对比")
def miniature():
seeds = range(8)
out, delta = compare(40, 2, seeds)
globals()["MINIATURE"] = (out, delta) # 第 11 步复用这一行
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 在所有地方都略有帮助的昂贵子智能体，以及更便宜的上下文压缩。我们在八个种子上对同一候选流比较三种接受规则：greedy 在分数上升时保留最佳分数，critic 只增加泄漏筛查，RRSI 增加带校准 delta 的 Algorithm 2。Greedy 达到最高的 held-out 分数，0.681，而 RRSI 为 0.616，同时记住了约十个答案，token 成本几乎增至三倍；RRSI 没有记住任何答案，最终 token 成本约为一半（相对起始 harness 为 1.53 倍，而后者为 2.97 倍），并且评估次数少了三分之一。拆分 evolve 到 held-out 的差距是这里最有用的结果：从带噪分数中挑选最佳会使每种模式膨胀约九个百分点，而这个循环中没有任何规则能消除这一点。与此同时，critic 消除了几乎所有的记忆成分。
Copy CodeCopiedUse a different Browser
@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 倍。贪心搜索的分数仍然更高，notebook 解释了原因：这个模拟世界没有收益递减，所以贪心搜索堆叠的每个子智能体都在持续换来准确率，这是对花费最有利的世界。RRSI 用其中一部分分数换来了有界的 token 账单、零记忆答案和更少的浪费评估，而这场交换的幅度由 delta 决定，delta 是它测量出来的，而不是向你索取的。
Copy CodeCopiedUse a different Browser
banner("SUMMARY")
for name, res in RESULTS.items():
print(f" {name:<76s} {res}")
print("""
这个微型实现没有建模的内容
- 那些有助于 evolve 套件但会损害另一个套件的编辑。论文的 LLM 评审及其
留出集和分布外划分正是为此而存在；正则筛查无法捕捉它们。
- 一个会读取历史的提议者。我们的是脚本化的，所以它会随意重新提出已被证伪的想法。
下一步该往哪里走
- 运行一个真实实例：在 Vertex AI 上配置好 Claude 后，执行 `python3 rrsi.py --domain coding smoke`
（RRSI_VERTEX_PROJECTS）以及 harbor；参见 domains/coding/README.md。
- 添加一个领域：在 domains/<name>/adapter.py 中实现 rrsi.domain.Domain，就像 notebook 中第 9 步所做的那样。
- 在不同的 delta 或成本规则下重新裁定一个已存储的轮次，而无需重新运行任何东西：
`python3 rrsi.py --domain <name> readjudicate --t <t>`。
- 论文：arxiv.org/abs/2609.24972 代码：github.com/google-research/rrsi
""")
摘要会打印每个部分返回的单行结果，直白地说明这个微型实现没有建模的内容，即那些有助于 evolve 套件但会损害另一个套件的编辑，而这正是论文的 LLM 评审和留出集划分存在的原因，以及一个会读取历史的提议者，然后指向运行一个真实实例、添加一个领域，以及在不同的 delta 下重新裁定一个已存储的轮次而无需重新运行任何东西。
In conclusion, we took RRSI apart along the line where the paper’s idea actually lives, the rules that decide which edits a self-improving agent keeps, and ran them offline on a CPU without a model or an API key. The estimator refuses to reward a crash, the noise band is measured rather than chosen, and Algorithm 2 turns out to be subtler than ‘reject noisy gains’: it never lets the score fall below the best seen minus the noise band, it makes a real gain pay for its extra tokens, and inside the band it deliberately prefers the cheaper harness, even one that scored slightly lower. Auditing those rules against a world whose ground truth we controlled was the most informative part: the deterministic critic removed almost all of the memorization before any evaluation was spent, Algorithm 2 held the token bill several times below an unregularized search, and neither removed the inflation that comes from picking the best of noisy scores, which only re-measurement on unseen tasks can do. The honest caveat is equally clear: in a world where tokens always buy accuracy, the unregularized search scores higher, so the value of RRSI’s regularization depends on how expensive tokens are and how much a leaked answer would cost you, and the notebook gives you the machinery to measure both on your own evaluator.
Check out the FULL CODES here. All credit goes to the researcher of this project. Also, feel free to follow us on Twitter and don’t forget to join our 150k+ML SubReddit and Subscribe to our Newsletter. Wait! are you on telegram? now you can join us on telegram as well.
The post Google Research RRSI Guide: Mastering Self-Improving AI Agents appeared first on MarkTechPost.