Руководство Google Research по RRSI: освоение самоулучшающихся ИИ-агентов

В этом руководстве мы реализуем RRSI (Regularized Recursive Self-Improvement — регуляризованное рекурсивное самоулучшение) — метод, который позволяет агенту на базе LLM переписывать собственную оболочку, промпты, инструменты, память, логику управления и субагентов вокруг замороженной модели, не допуская переобучения оболочки на задачах, на которых она развивается. Полный цикл RRSI создает варианты изменений с помощью Claude Opus в Vertex AI и оценивает их в Docker-бенчмарках, а это невозможно запустить в бесплатном ноутбуке. Та часть RRSI, которая действительно воплощает идею статьи, — правила, определяющие, какие предложенные изменения оставить, — написана на обычном Python, и именно ее мы запускаем напрямую. Мы устанавливаем пакет из официального репозитория, рассматриваем его оценщик, калиброванный диапазон шума, обе ветви алгоритма отбора, постепенно уменьшаемый бюджет изменений, детерминированную проверку утечек и историю изменений, а затем подключаем имитированного агента к собственному интерфейсу 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, зафиксировав коммит, на основе которого написан этот ноутбук, поскольку пакета нет в PyPI. Единственная его зависимость — клиент Anthropic, который роли поиска используют для вызова Claude, но мы его не задействуем. Затем мы выводим соответствие, задокументированное в самом репозитории, между обозначениями статьи и функциями, которые их реализуют: эмпирической оценкой и оценкой стоимости в evaluate, диапазоном шума в calibrate, алгоритмом 2 в selection, постепенно уменьшаемым бюджетом изменений в schedule, проверкой утечек в critic и историей изменений с ее сводками по эффективности, удалению, простою и исследованию в history. 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 измеряет два показателя для каждой оболочки: 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 неотличимо от повторного запуска той же оболочки.

Алгоритм 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% меньше токенов, а новый субагент разрешает ничью, которую не может разрешить изменение промпта с такой же оценкой. Ограничение домена отклоняет вариант независимо от его оценки.

select_round применяет judge ко всем кандидатам раунда и оставляет самого высоко оцененного допустимого кандидата. Мы передаем ему дорогого кандидата, немного худший, но более дешевый вариант и кандидата, заранее отклоненного критиком. Кандидат с наивысшей оценкой проигрывает, потому что его прирост на три процентных пункта не оправдывает увеличение стоимости на 150%; отклоненный критиком кандидат вообще не доходит до оценки, а победитель снижает оценку текущего варианта на полпроцента, одновременно сокращая стоимость токенов на пятую часть. Безопасность здесь обеспечивает S*, которое может только расти: порог привязан к лучшей когда-либо измеренной оценке, а не к текущему варианту, поэтому цепочка замен «дешевле, но немного хуже» не сможет постепенно ухудшать результат в течение многих раундов.

На стороне предложений регуляризуется способ создания изменений, а не то, какие изменения сохраняются. edit_budget реализует L0-бюджет изменений из статьи с косинусным расписанием от b_max до b_min; по умолчанию он разрешает до четырех согласованных изменений на кандидата в первых восьми раундах, трех — в следующих пяти и двух — в последних семи. Потолок в формуле означает, что бюджет достигает минимального значения, равного единице, только при t = T, то есть через один шаг после окончания запуска, — это легко упустить при чтении уравнения. Поскольку каждое изменение в наборе наследует единственный результат измерения этого набора, именно сокращение бюджета позволяет связывать историю поздних раундов с меньшим числом компонентов. Бюджет ограничивает количество изменений, перемещающихся вместе, но никогда не ограничивает набор механизмов, которые оболочка в итоге может содержать.

Критик проверяет каждый diff кандидата до того, как на оценку будут потрачены ресурсы, используя два слоя. Первый — детерминированная предварительная проверка по общему шаблону учетных данных и собственному черному списку домена; с шаблонами для идентификаторов задач из набора развития и путей к файлам проверяющего он отклоняет diff, который запоминает ответ на task_007, diff, читающий ожидаемый результат, diff с утечкой API-ключа и пустой diff — и все это без вызова модели. Чистый diff переходит ко второму слою: проверке намерения Claude. Здесь ноутбук показывает реальную проблему: если RRSI_VERTEX_PROJECTS не задана, rrsi.llm.generate вычисляет индекс по модулю количества проектов вне блока try и вызывает ZeroDivisionError, поэтому полезное сообщение об ошибке конфигурации не появляется. Мы также рассматриваем маркировку изменений: normalize сохраняет объявленный компонент только тогда, когда diff содержит подтверждение его использования. Поэтому изменение промпта не может выдать себя за новый навык ради бонуса за новизну, а неправильно обозначенное изменение управления контекстом получает фактическую метку.

History записывает одну JSONL-запись для каждого изменения, а цикл выводит из нее четыре сводки. Множество tried исключает изменения, отброшенные критиком, поскольку они никогда не измерялись. Недавняя эффективность g_t — это наилучший измеренный прирост для каждого компонента за последние n_prune раундов; каждый компонент, чья недавняя эффективность не положительна, попадает в набор удаления B_t вместе с механизмами этого компонента, которые все еще присутствуют в текущем варианте. В нашей истории это отмечает изменение промпта, принятое в нулевом раунде, но с тех пор не принесшее результата. Флаг простоя срабатывает, когда оценка изменилась меньше чем на delta за последние w раундов; после этого exploration формирует указание, которое получает предлагающая сторона, резервируя место кандидата для компонентов, еще не опробованных в запуске. Поскольку предложения создаются с учетом всей этой информации, опровергнутая гипотеза не генерируется снова.

RRSI сам не запускает агента и не оценивает результат; это делает адаптер Domain, причем тот же интерфейс лежит в основе примеров статьи для программирования, рабочих пространств и инженерных задач. Мы реализуем такой адаптер для смоделированного агента: разделение задач на развиваемые и отложенные, метод run, который читает оболочку из файлов в корне рабочего дерева и записывает результаты попыток в каталог runs, метод score, считывающий их обратно в виде объектов TaskResult, а также шаблоны критика и сигналы компонентов домена. Затем собственная функция evaluate в RRSI оценивает начальную оболочку через адаптер, а повторный вызов для той же задачи считывает сохраненные результаты, а не запускает оценку заново, обеспечивая требуемую контрактом безопасность возобновления.

Теперь мы запускаем всю сторону отбора как поиск: двадцать раундов, по два кандидата в раунде, наборы изменений размером согласно постепенно уменьшаемому бюджету, маркировка изменений с помощью normalize, предварительная проверка кандидатов через precheck и их оценка через адаптер, выбор победителей посредством select_round и запись каждого результата в History в том порядке, который использует цикл RRSI. Скриптовый генератор предложений создает пять типов изменений, истинные эффекты которых нам известны: обычно полезные общие изменения, утекающие изменения, запоминающие ответы задач набора развития, инертные изменения конфигурации, дорогие субагенты, немного помогающие везде при стоимости в 1,5 раза выше, и более дешевое сжатие контекста. Мы сравниваем три правила принятия на одном и том же потоке кандидатов для восьми начальных значений: жадный алгоритм сохраняет лучший результат, если он вырос; режим «только критик» добавляет проверку утечек; RRSI добавляет алгоритм 2 с калиброванной delta. Жадный поиск достигает наивысшей оценки на отложенных задачах — 0.681 против 0.616 у RRSI, — но запоминает около десяти ответов и почти втрое увеличивает стоимость токенов. RRSI ничего не запоминает, заканчивает с примерно половиной исходной стоимости токенов (1.53 от начальной оболочки против 2.97) и использует на треть меньше оценок. Наиболее полезный результат дает разделение разрыва между развиваемыми и отложенными задачами: выбор лучшей из шумных оценок увеличивает результат каждого режима примерно на девять пунктов, и ни одно правило этого цикла не устраняет эффект. Одновременно критик устраняет почти весь компонент запоминания.

Наш первый оценщик был шумным, и delta калибровалась на его основе, поэтому осторожность RRSI на шаге 10 отражала свойства оценщика, а не самого метода. Мы повторяем сравнение с большими наборами развиваемых задач и большим числом попыток. По мере повышения точности оценщика delta снижается с 0.096 до 0.025, а оценка RRSI на отложенных задачах растет с 0.616 до 0.759; при самых точных настройках нерегуляризованный поиск использует в 6,5 раза больше токенов, чем RRSI. Жадный алгоритм по-прежнему набирает больше, и ноутбук объясняет почему: в этой смоделированной среде нет убывающей отдачи, поэтому каждый добавленный жадным поиском субагент продолжает покупать дополнительную точность — это наиболее благоприятный для больших расходов сценарий. RRSI отказывается от части оценки ради ограниченного бюджета токенов, отсутствия запомненных ответов и меньшего числа бесполезных оценок; размер этой уступки определяется delta, которую он измеряет, а не запрашивает у пользователя.

В сводке выводится однострочный результат каждого раздела, прямо указывается, чего не моделирует миниатюрный пример, а именно изменений, которые помогают набору развиваемых задач, но вредят другому набору, — для таких случаев и существуют LLM-критик статьи и отложенные разбиения, — а также генератора предложений, который читает историю. После этого сводка указывает, как запустить реальный пример, добавить домен и повторно вынести решение по сохраненному раунду с другой delta без повторного запуска.

В заключение мы разобрали RRSI по линии, где действительно находится идея статьи, — по правилам, определяющим, какие изменения сохраняет самоулучшающийся агент, — и запустили их офлайн на CPU без модели и API-ключа. Оценщик не вознаграждает сбой, диапазон шума измеряется, а не выбирается вручную, и алгоритм 2 оказывается тоньше, чем простое «отклонять шумные улучшения»: он не позволяет оценке опуститься ниже лучшего достигнутого значения за вычетом диапазона шума, требует, чтобы реальное улучшение оправдывало дополнительные токены, а внутри диапазона намеренно предпочитает более дешевую оболочку, даже если она набрала немного меньше. Проверка этих правил в среде, где мы контролировали эталонную истину, оказалась наиболее информативной частью: детерминированный критик устранил почти все запоминание еще до начала оценки, алгоритм 2 удержал расходы на токены в несколько раз ниже, чем нерегуляризованный поиск, но ни один из них не устранил завышение, возникающее при выборе лучшей из шумных оценок; это может сделать только повторная оценка на невиденных задачах. Не менее очевидна и честная оговорка: в мире, где токены всегда покупают точность, нерегуляризованный поиск набирает больше. Поэтому ценность регуляризации RRSI зависит от стоимости токенов и от того, насколько дорого вам обойдется утечка ответа, а ноутбук дает инструменты, позволяющие измерить оба фактора на собственном оценщике.

Ознакомьтесь с ПОЛНЫМ КОДОМ здесь. Все заслуги принадлежат исследователю этого проекта. Также подписывайтесь на нас в Twitter и не забудьте присоединиться к нашему ML-сабреддиту с более чем 150 тысячами участников и подписаться на нашу рассылку. Постойте! Вы есть в Telegram? теперь к нам можно присоединиться и в Telegram.