في هذا البرنامج التعليمي، ننفّذ RRSI (Regularized Recursive Self-Improvement)، وهي طريقة تتيح لوكيل LLM إعادة كتابة هيكله الخاص، بما في ذلك المطالبات والأدوات والذاكرة وتدفق التحكم والوكلاء الفرعيين، حول نموذج مجمّد، دون أن يتجاوز الهيكل في ملاءمته للمهام التي يتطور عليها. حلقة RRSI الكاملة تعدّ المسودات باستخدام Claude Opus على Vertex AI وتقيّمها داخل معايير أداء تعمل في Docker، وهذا ليس شيئاً يمكن لدفتر ملاحظات مجاني تشغيله. أما الجزء من RRSI الذي يحمل فعلياً فكرة الورقة البحثية، أي القواعد التي تقرر أي التعديلات المقترحة يُحتفظ بها، فهو بلغة بايثون بسيطة، وهذا ما نستخدمه مباشرة. نثبّت الحزمة من المستودع الرسمي، ونستعرض مقدّرها، ونطاق الضوضاء المعاير لديها، وكلا فرعي خوارزمية الاختيار لديها، وميزانية التعديلات المتدرجة، وفحص التسريب الحتمي، وسجل تعديلاتها، ثم نوصّل وكيلاً محاكى في واجهة Domain الخاصة بـ RRSI نفسها. ولأننا بنينا البيئة المحاكاة بأنفسنا، نعرف الأثر الحقيقي لكل تعديل، وهو ما يتيح لنا تدقيق قرارات 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.")
نثبّت RRSI من مستودع google-research، مثبّتاً على الالتزام (commit) الذي كُتب هذا الدفتر ضده، إذ إن الحزمة غير متوفرة على PyPI. اعتمادها الوحيد هو عميل Anthropic، الذي تستخدمه أدوار البحث لاستدعاء Claude ونحن لا نستخدمه أبداً. ثم نطبع التطابق الذي يوثّقه المستودع نفسه بين رموز الورقة البحثية والدوال التي تنفّذها: الدرجة التجريبية وتقدير التكلفة في evaluate، ونطاق الضوضاء في calibrate، والخوارزمية 2 في selection، وميزانية التعديلات المتدرجة في schedule، وفحص التسريب في critic، وسجل التعديلات مع ملخصاته للإنتاجية والتقليم والجمود والاستكشاف في history. يحمل RRSIConfig معاملات الورقة الفائقة (hyperparameters)، وتستقبل كل دالة أدناه هذه الإعدادات تماماً كما تفعل الحلقة الحقيقية.
نسخ الكود@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 رقمين لكل هيكل: S، المكافأة الموسّطة عبر كل محاولة من كل مهمة، وC، متوسط رموز السياسة لكل محاولة. تسجّل 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()
قبل أن يتمكن أي قاعدة من التمييز بين مكسب حقيقي والحظ، تحتاج إلى معرفة مدى تحرك درجة هيكل واحد من تلقاء نفسها. نبني وكيلاً محاكى صغيراً، يكون نجاحه في كل مهمة دالة لوجستية لمهارة الهيكل مطروحاً منها صعوبة المهمة، ونقيّم الهيكل الابتدائي غير المغيّر ست مرات على أربعين مهمة بمحاولتين لكل منها: تتباعد الدرجات بمقدار 0.113 رغم عدم تغيير أي شيء. يحوّل calibrate التقييمات المتكررة لنفس الهيكل إلى delta، أي ضعف الانحراف المعياري للفرق بين تشغيلين. مع 80 محاولة تكون delta نحو 0.108؛ ومع 3,200 محاولة تنخفض إلى نحو 0.013، وهو ضمن النطاق الذي تبلّغ عنه الورقة لحالاتها (من 0.004 إلى 0.020). ولأغراض الاختيار، أي مكسب أصغر من delta لا يمكن تمييزه عن إعادة تشغيل نفس الهيكل.
نسخ الكود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 رمز، وأفضل درجة تم تحقيقها على الإطلاق عند 0.640 وdelta عند 0.020، ونمرّر ثمانية مرشحين عبر judge. يُرفض المرشّح دون الحد الأدنى، وهو أفضل درجة شوهدت على الإطلاق ناقص delta، رفضاً قاطعاً. ويدفع المكسب الأكبر من delta ثمن رموزه الإضافية بموجب القاعدة التي توجب بقاء التغير النسبي في التكلفة دون 0.10 زائد 40 ضعف المكسب؛ فمكسب +6 نقاط مقابل +20% من الرموز يُقبل، ومكسب +3 نقاط مقابل +150% من الرموز لا يُقبل. داخل النطاق، تُعامل الدرجات كتعادل، وتقرر درجة مشكّلة تساوي 100 ضعف المكسب ناقص 15 ضعف تغير التكلفة، بالإضافة إلى مكافأة صغيرة لمكوّن بنيوي لم يُقبل من قبل؛ وبهذا تقبل RRSI المرشّح D، الذي حقق درجة أقل من الشاغل لكنه يكلف 20% رموزاً أقل، وبهذا أيضاً يكسر وكيل فرعي جديد تعادلاً لا يكسره تعديل مطالبة بنفس الدرجة. وحارس النطاق (domain guard) يمنع بغض النظر عن الدرجة.
نسخ الكود@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()
apply الدالة select_round الحكم على كل مرشح في جولة وتُبقي الأعلى score من بين المقبولة. نمنحها المرشح المكلف، والمرشح الأرخص الأسوأ قليلاً، ومرشحاً رفضه الناقد مسبقاً. يخسر صاحب الscore الأعلى لأن مكسبه من ثلاث نقاط لا يعوّض 150% إضافية من الرموز، ولا يصل المرشح المرفوض من الناقد إلى التقييم أصلاً، ويفوز المرشح الأخير بخفض المرجع الحالي بنصف نقطة مع تقليص تكلفة رموزه إلى الخمس. التفصيلة التي تُبقي هذا آمناً هي S*، التي لا ترتفع إلا: فالحد الأدنى مرتكز على أفضل score قُيس على الإطلاق وليس على المرجع الحالي، وبالتالي لا يمكن لسلسلة من الاستبدالات الأرخص الأسوأ قليلاً أن تُسرّق الscore على مدى جولات كثيرة.
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، وبشكل افتراضي تسمح بما يصل إلى أربعة تعديلات منسقة لكل مرشح في الجولات الثماني الأولى، وثلاثة في الجولات الخمس التالية، واثنين في الجولات السبع الأخيرة. تعني سقفية الحد الأقصى في الصيغة أن الميزانية لا تصل إلى حدها الأدنى البالغ واحداً إلا عند t = T، خطوة واحدة بعد انتهاء التشغيل، وهو أمر سهل الإغفال عند قراءة المعادلة. ولأن كل تعديل في الحزمة يرث قياس الحزمة الواحد، فإن الميزانية المتقلصة هي ما يجعل تاريخ المرحلة المتأخرة قابلاً للإسناد إلى مكوّنات أقل. تحدّ الميزانية عدد التعديلات التي تسافر معاً ولا تقيّد أبداً الآليات التي قد يحتويها 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()
يفحص الناقد كل diff مرشح قبل إنفاق أي تقييم، على طبقتين. الأولى فحص مسبق حتمي ضد نمط عام للاعتماديات مع قائمة رفض خاصة بالنطاق نفسه؛ وبأنماط لمعرّفات مهام evolve-set ومسارات graders يرفض diff يحفظ إجابة task_007، وdiff يقرأ المخرج المتوقع، وdiff يسرّب مفتاح API، وdiff فارغ، كل ذلك دون استدعاء نموذج. يمرّ diff نظيف إلى الطبقة الثانية، وهي مراجعة نية من Claude، وهنا تُبرز دفتر الملاحظات مشكلة حقيقية: مع ترك RRSI_VERTEX_PROJECTS بلا ضبط، تحسب rrsi.llm.generate فهرساً بترقيم دوري على عدد المشاريع خارج كتلة try وتُطلق ZeroDivisionError، فلا يُبلغ خطأ الإعداد المفيد الموجود في الكود أبداً. ننظر أيضاً إلى كيفية وسم التعديلات: يُبقي normalize المكوّن المعلن عنه فقط عندما يحمل diff دليلاً عليه، فلا يمكن لتعديل prompt أن يتظاهر بأنه مهارة جديدة ليكسب مكافأة الجِدّة، ولا أن يوسم change لإدارة السياق بطريقة خاطئة إلا بوصفه ما هو عليه فعلاً.
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()
يكتب السجل سجل JSONL واحداً لكل تعديل، وتشتق الحلقة منه أربعة ملخصات. تستبعد مجموعة tried التعديلات التي أسقطها الناقد، إذ لم تُقَس أبداً. العائد الأخير g_t هو أفضل عائد مُقاس لكل مكوّن ضمن آخر n_prune جولة، وكل مكوّن لا يكون عاده الأخير موجباً يدخل مجموعة prune B_t، مع أي آليات من ذلك المكوّن ما تزال في المرجع الحالي؛ وفي سجلّنا يوسم ذلك تعديل prompt قُبل في الجولة صفر ولم يثمر منذ ذلك الحين. تُطلق علامة stall عندما يتحرك الscore بمقدار أقل من delta خلال آخر w جولات، فيكتب الاستكشاف حينها التوجيه الذي يتلقاه المُقترح، محجوزاً خانة مرشح للمكوّنات التي لم يمارسها التشغيل قط. ولأن المُقترح مُشروط بهذا كله، فالفرضية المفندة لا تُسحب مرة أخرى.
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، والواجهة نفسها تدعم حالات البرمجة وworkspace والهندسة في الورقة. ننفّذ واحداً للagent المحاكى: تقسيمات مهام evolve ومهام محتجزة، وmethod باسم run يقرأ الharness من ملفات تحت جذر worktree ويكتب نتائج التجارب تحت دليل runs، وmethod باسم score يقرأها عائدة ككائنات TaskResult، وأنماط الناقد وإشارات المكوّنات الخاصة بالنطاق. ثم تقيّم دالة evaluate الخاصة بـ RRSI نفسها ال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()
الآن نشغّل جانب الاختيار بأكمله كبحث: عشرون جولة، مرشحان في كل جولة، حزم يحدد حجمها الميزانية المتدرجة، تعديلات موسومة بـ normalize، مرشحون يُفرزون عبر precheck ويُسجَّلون من خلال المُحوِّل، الفائزون يُختارون عبر select_round، وكل نتيجة تُكتب إلى History بالترتيب الذي تستخدمه حلقة RRSI. يرسم المُقترِح المُبرمَج خمسة أنواع من التعديل نعرف آثارها الحقيقية: تعديلات عامة مفيدة عادةً، تعديلات مُسربة تحفظ إجابات مهام evolve، تغييرات إعدادات خاملة، وكلاء فرعيون مكلفون يفيدون قليلاً في كل مكان مقابل 1.5 ضعف الرموز، وضغط سياق أرخص. نقارن ثلاث قواعد قبول على نفس تدفق المرشحين عبر ثماني بذور: الجشع (greedy) يحتفظ بأفضل نتيجة إذا ارتفعت، والناقد (critic) يضيف فقط فرز التسريب، وRRSI يضيف Algorithm 2 مع دلتا مُعايَرة. الجشع يصل إلى أعلى نتيجة محجوزة، 0.681 مقابل 0.616 لـ RRSI، مع حفظ نحو عشر إجابات وتقريب تكلفة رموزه من ثلاثة أضعاف؛ بينما RRSI لا يحفظ شيئاً، وينتهي بنحو نصف تكلفة الرموز (1.53 ضعف الهيكل الابتدائي مقابل 2.97 ضعف)، ويستخدم تقييمات أقل بالثلث. تفكيك الفجوة بين evolve والمجموعة المحجوزة هو النتيجة الأكثر فائدة هنا: اختيار أفضل النتائج الضوضائية يضخّم كل نمط بنحو تسع نقاط، وهو ما لا تزيله أي قاعدة في هذه الحلقة. وفي الوقت نفسه، يزيل الناقد تقريباً كل مكوّن الحفظ.
Copy Code@section("11. Turn the evaluator up: RRSI's caution is calibrated, not configured")
def noise_sweep():
rows = [("40x2", MINIATURE[1], MINIATURE[0], 8)] # from step 10
for n, k in [(100, 4), (400, 8)]:
out, delta = compare(n, k, range(5))
rows.append((f"{n}x{k}", delta, out, 5))
print(f" {'evaluator':>10s} {'seeds':>5s} {'delta':>7s} "
+ "".join(f"{m_ + ' held / tokens':>26s}" for m_ in MODES))
for label, delta, out, n_seeds in rows:
cells = "".join(f"{st.mean(r['heldout_true'] for r in out[m_]):>13.3f} /"
f" x{st.mean(r['tokens'] for r in out[m_]):<9.2f}" for m_ in MODES)
print(f" {label:>10s} {n_seeds:>5d} {delta:>7.3f} {cells}")
first, last = rows[0], rows[-1]
rr_first = st.mean(r["heldout_true"] for r in first[2]["rrsi"])
rr_last = st.mean(r["heldout_true"] for r in last[2]["rrsi"])
tok_ratio = (st.mean(r["tokens"] for r in last[2]["greedy"]) /
st.mean(r["tokens"] for r in last[2]["rrsi"]))
print(f"\n As the evaluator sharpens, delta falls from {first[1]:.3f} to {last[1]:.3f} (the paper: 0.004-0.020),"
f" and RRSI's held-out score rises from {rr_first:.3f} to {rr_last:.3f}.")
print(f" At the sharpest setting the unregularized search is spending {tok_ratio:.1f}x RRSI's tokens.")
print("\n Read the table honestly: this world has no diminishing returns, so every sub-agent the")
print(" greedy search stacks keeps buying accuracy. That is the most favourable world possible for")
print(" spending, and the greedy search does score higher. RRSI trades some of that score for a")
print(" bounded token bill, zero memorised answers and fewer wasted evaluations - and the size of")
print(" the trade is set by delta, which it measures from your evaluator rather than taking from you.")
return (f"delta {first[1]:.3f} -> {last[1]:.3f}; RRSI held-out {rr_first:.3f} -> {rr_last:.3f};"
f" greedy uses {tok_ratio:.1f}x the tokens")
noise_sweep()
كان مُقيّمنا الأول ضوضائياً، ودلتا عُايرت منه، لذا فإن حذر RRSI في الخطوة 10 كان يعكس المُقيّم لا الطريقة. نكرر المقارنة بمجموعات evolve أكبر ومحاولات أكثر. ومع تحسّن دقة المُقيّم، تنخفض دلتا من 0.096 إلى 0.025، وترتفع نتيجة RRSI المحجوزة من 0.616 إلى 0.759، بينما يستخدم البحث غير المنتظم في أدق إعداد 6.5 ضعف رموز RRSI. لا يزال الجشع يسجّل أعلى، والدفتر يشرح السبب: هذا العالم المحاكى لا يعرف التناقص في العوائد، فكل وكيل فرعي يكدّسه البحث الجشع يستمر في شراء الدقة، وهو العالم الأكثر ملاءمة ممكنة للإنفاق. أما RRSI فيقايض جزءاً من تلك النتيجة مقابل فاتورة رموز محدودة، دون إجابات محفوظة، وتقييمات مهدورة أقل، وحجم هذه المقايضة تحدده دلتا التي يقيسها بنفسه بدلاً من طلبها.
Copy Codebanner("SUMMARY")
for name, res in RESULTS.items():
print(f" {name:<76s} {res}")
print("""
What the miniature does not model
- Edits that help the evolve suite but hurt a different suite. The paper's LLM critic and its
held-out and out-of-distribution splits exist for those; a regex screen cannot catch them.
- A proposer that reads the history. Ours is scripted, so it redraws falsified ideas freely.
Where to go next
- Run a real instance: `python3 rrsi.py --domain coding smoke` after configuring Claude on Vertex AI
(RRSI_VERTEX_PROJECTS) and harbor; see domains/coding/README.md.
- Add a domain: implement rrsi.domain.Domain in domains/<name>/adapter.py, as step 9 did in the notebook.
- Re-adjudicate a stored round under a different delta or cost rule without re-running anything:
`python3 rrsi.py --domain <name> readjudicate --t <t>`.
- Paper: arxiv.org/abs/2609.24972 Code: github.com/google-research/rrsi
""")
يطبع الملخص النتيجة ذات السطر الواحد التي أرجعها كل قسم، ويذكر صراحةً ما لا تحاكيه هذه النسخة المصغّرة، أي التعديلات التي تفيد مجموعة evolve بينما تضر بمجموعة أخرى، وهو الغرض من وجود الناقد اللغوي الضخم والمقسّمات المحجوزة في الورقة، ومُقترِح يقرأ السجل، ثم يشير إلى تشغيل نسخة حقيقية، وإضافة مجال، وإعادة التحكيم في جولة محفوظة بدلتا مختلفة دون إعادة تشغيل أي شيء.
وفي الختام، فكّكنا RRSI عند الخط الذي تعيش فيه فكرة الورقة فعلاً، وهي القواعد التي تقرر أي التعديلات يحتفظ بها وكيل يُحسّن نفسه بنفسه، وشغّلناها دون اتصال على وحدة معالجة مركزية دون نموذج أو مفتاح API. المُقدِّر يرفض مكافأة الانهيار، ونطاق الضوضاء يُقاس بدلاً من أن يُختار، ويتضح أن Algorithm 2 أكثر دقة من مجرد «رفض المكاسب الضوضائية»: فهو لا يسمح أبداً للنتيجة بالهبوط تحت أفضل نتيجة شوهدت مطروحاً منها نطاق الضوضاء، ويجعل الكسب الحقيقي يدفع ثمن رموزه الإضافية، وداخل النطاق يفضّل عمداً الهيكل الأرخص، حتى لو سجّل نتيجة أدنى قليلاً. مراجعة تلك القواعد مقابل عالَم نتحكم نحن في حقيقته كانت الجزء الأكثر إفادة: الناقد الحتمي أزال تقريباً كل الحفظ قبل إنفاق أي تقييم، وأبقت Algorithm 2 فاتورة الرموز أقل من بحث غير منظم عدة أضعاف، ولم يُزل أيٌّ منهما التضخم الناتج عن اختيار أفضل النتائج الضوضائية، وهو ما لا يمكن إلا إعادة القياس على مهام غير مرئية. والتحذير الصادق واضح بالمثل: في عالَم حيث تشتري الرموز الدقة دائماً، يسجّل البحث غير المنظم نتيجة أعلى، فتعتمد قيمة تنظيم RRSI على مدى غلاء الرموز وعلى تكلفة الإجابة المسرّبة عليك، والدفتر يمنحك الأدوات لقياس الاثنين على مُقيّمك أنت.
اطّلع على FULL CODES here. كل الفضل يعود إلى باحث هذا المشروع. كما لا تتردد في متابعتنا على Twitter ولا تنسَ الانضمام إلى 150k+ML SubReddit والاشتراك في our Newsletter. مهلاً! هل أنت على تيليجرام؟ يمكنك الآن الانضمام إلينا على تيليجرام أيضاً.
ظهر المقال Google Research RRSI Guide: Mastering Self-Improving AI Agents أولاً على MarkTechPost.