跳到主要内容

RSI 开发指南

七天实战

对应 7 天学会 RSI Day 3 到 Day 6：反思记忆、策略补丁、多角色互评和防退化。本文不讲子站如何部署，只讲可复用的循环结构。

组件边界​

一个能维护的 RSI 循环不是「一个大提示词」，而是五块代码，彼此只通过结构体说话。

组件输入输出禁止
执行器规格、策略、检索到的教训轨迹给自己打分
评估器轨迹、评分器版本分数、失败证据改策略
反思器失败证据带引用的教训发明证据里没有的编号
改进器策略、教训、白名单补丁改黑名单字段
门禁补丁、回归基线接受或拒绝在拒绝时仍覆盖当前策略

调度器负责预算、停止条件和日志。把停止条件写进提示词里，模型会忽略；写在调度器里，才会执行。

轨迹结构​

from dataclasses import dataclass, field

@dataclass
class Trace:
    task_id: str
    policy_rev: str
    scorer_version: str
    output: str
    tool_calls: list[str] = field(default_factory=list)
    failures: list[str] = field(default_factory=list)
    score: int = 0

policy_rev 用字段哈希而不是自增整数，避免两台机器的「第 3 版」不是同一版。failures 在评估器返回前保持空列表，执行器不许预填。

评估器：规则层和评审层​

def evaluate(trace: Trace, rules, judge=None) -> Trace:
    failed = [name for name, fn in rules.items() if not fn(trace.output)]
    trace.failures = failed
    trace.score = 2 * (len(rules) - len(failed))
    if judge and not failed:
        judged = judge(trace.output)
        if not quotes_are_substrings(judged, trace.output):
            trace.tool_calls.append("judge_rejected_ungrounded")
        else:
            trace.score = min(trace.score, judged.score)
    return trace

规则失败时不要调用评审器来「翻案」。评审器只在规则通过后检查规则覆盖不到的方面，例如是否答非所问。min 保证评审器不能把分数抬过规则分。

评审提示词单独存放，哈希记入 scorer_version 的后缀，例如 rules-r1r5-v1+judge-a3f1。改一个词就升版本。

反思器：证据闭包​

def reflect(trace: Trace, catalog: dict) -> dict:
    cites = [f for f in trace.failures if f in catalog]
    if not cites:
        return {"grounded": False, "cites": [], "text": ""}
    text = "\n".join(f"{c}: {catalog[c]}" for c in cites)
    return {"grounded": True, "cites": cites, "text": text}

若反思改为模型生成，后处理必须删除不在 trace.failures 里的编号。模型可以说「可能是超时」，但除非证据里有超时记录，否则这句话不能进入 cites。无编号的句子可以留下来当注释，不能触发改进器。

错误归因比不反思更糟：把测试写错的责任记到业务代码上，下一轮会改错文件。事实校验可以很简单——引用必须能在轨迹的 output 或测试日志里找到子串。

改进器：补丁而不是整文件重写​

def refine(policy: dict, lesson: dict, whitelist: set[str]) -> dict | None:
    if not lesson.get("grounded"):
        return None
    patch = {k: policy[k] for k in policy}
    key = "constraints"
    if key not in whitelist:
        return None
    patch[key] = policy[key].rstrip() + "\n回应 " + ",".join(lesson["cites"])
    dirty = [k for k in patch if patch[k] != policy[k]]
    if set(dirty) - whitelist:
        return None
    return patch

生产系统里用统一 diff 更好审查。教学阶段全量字典可以，但日志只记录 dirty 键。一次教训只改一个键。若教训同时建议改提示词和改测试，拆成两个候选，分别过门禁，否则无法归因。

代码补丁额外要求：补丁后的测试命令在沙箱里退出码为 0，且 diff 不包含锁文件、密钥样例和评分器源码。评分器源码出现在 diff 里，按黑名单直接拒绝。

门禁​

def gate(current_failed, new_failed, regression_ok: bool) -> bool:
    if not regression_ok:
        return False
    if set(new_failed) - set(current_failed):
        return False  # 当前题出现新失败
    return set(new_failed) < set(current_failed) or new_failed == current_failed

分数相同但失败集合换成了另一个编号，不算改进。<= 用集合真子集表达「少失败」，相等则只在你显式允许「持平但反思更短」时接受。默认不接受持平，否则日志会被无意义改写填满。

回归基线是任务标识到失败集合的映射，实验开始时冻结。新增回归题是人的决定，记在变更说明里，不由改进器提议后自动合并。

教训库的降权​

def on_retrieve_outcome(item: dict, helped: bool) -> None:
    item["uses"] = item.get("uses", 0) + 1
    item["helps"] = item.get("helps", 0) + (1 if helped else 0)

def keep(item: dict) -> bool:
    if item["uses"] < 3:
        return True
    return item["helps"] / item["uses"] >= 0.4

使用不足三次不淘汰，避免偶然。三次里帮助少于四成，归档而不是删除，归档区不参与检索。错误教训往往来自 Day 3 的「故意注入错误反思」实验，归档能让你看见它，又不会继续污染。

检索时同时取出「被接受」和「被拒绝」的教训，但拒绝教训只能作为负例提示：「不要再尝试删除输入字段」，不能作为新的约束正文直接拼接。负例拼接过多会挤占上下文，每轮最多一条负例。

多角色互评时的接口​

Day 5 的作家、编辑、读者是三个执行器，不是三个调度器。共享同一个评估器和同一份回归集。编辑输出的是对作家策略的补丁建议，仍要经过改进器的白名单。读者只提供失败证据，不直接改文件。

若三个角色用同一个模型，用三套系统提示，并在日志里记录角色名。比较「单角色自省」和「三角色」时，预算按模型调用次数对齐，而不是按墙钟时间。三角色调用更多却分数相同，应判为更差的系统，而不是「更丰富」。

种群进化同理：每一代保留分数最高且回归通过的前 k 个策略，变异只作用在白名单字段。交叉两份策略时，安全段永远取父代中未改过的那份原文，不做字符串拼接。

调度器伪代码​

def run(task, policy, memory, max_rounds=4):
    baseline = load_baseline()
    for round_id in range(1, max_rounds + 1):
        lesson_text = memory.retrieve(task)
        trace = evaluate(execute(task, policy, lesson_text))
        log(round_id, trace, policy)
        if not trace.failures:
            return policy
        lesson = reflect(trace)
        patch = refine(policy, lesson, {"constraints"})
        if patch is None:
            memory.add(lesson, accepted=False)
            continue
        new_trace = evaluate(execute(task, patch, lesson_text))
        ok = gate(trace.failures, new_trace.failures, regression_ok(patch, baseline))
        memory.add(lesson, accepted=ok)
        if ok:
            policy = patch
        if plateau(log_history()):
            break
    return policy

continue 的分支必须写日志。没有日志的拒绝等于没发生，下一轮会重复同一补丁。

配置清单​

  • 白名单、黑名单写在仓库 rsi.config.json，代码启动时加载并打印哈希。
  • 最大轮数、温度、评审器模型名写在同一文件的 budget 段。
  • 回归集路径只读挂载或文件权限设为运行用户不可写。
  • 人工检查点：judge 提示词变更、白名单新增键、回归集新增题。

测试改进器本身​

改进器是普通代码，用单元测试锁住：

  1. 教训未 grounded 时返回 None。
  2. 补丁改动 safety 时返回 None。
  3. 当前题失败集扩大时门禁为假。
  4. 回归新增失败时门禁为假，且策略对象仍是旧字典。
  5. 评分器文件哈希变化时调度器拒绝启动。

这些测试不调用大模型，应该在秒级跑完。模型只出现在标记为集成的可选测试里，默认跳过，避免 CI 因网络抖动改写策略文件。

观测​

每轮日志至少包括 round、policy_rev、scorer_version、failures、accepted、patch_keys。再加上 token_in、token_out 后，你才能回答「改进是否划算」。成本上升而失败集不变，按平台期停止处理。

不要把完整提示词打到公共日志。轨迹可以存本地，对外分享时去掉规格里的用户原文，保留失败编号和补丁键名即可。

下一篇对照开源实现：GitHub 相关项目。门禁在生产习惯上的展开见 最佳实践。

失败分类字典​

规则编号之外，代码循环使用异常分类，避免反思写成散文。

代号含义允许的下一步
ASSERT断言与期望不符改业务函数一行
TIMEOUT超过时限改算法复杂度说明或实现，不改时限常量
SCHEMA输出不是约定 JSON只改格式段
UNSAFE触碰黑名单路径停止，不产生补丁
FLAKY重复三次结果不一致先固定种子，禁止改进

FLAKY 期间改进器直接返回空。把抖动修成「看起来稳定的一次」是在记录噪声。UNSAFE 写安全日志，不写教训推荐区。

版本同时前进​

策略、评分器、回归基线三套版本在日志里同时出现。任意一套变化，对比句子就得重写。调度器启动时打印三行哈希。哈希与上次实验笔记不一致时，拒绝续跑旧的 trace 文件，另开新文件。续跑混版本是分数「突然变好」的常见原因。

参考实现：日志校验、数字规则与停止条件​

七天里只维护两个文件：rsi_loop.py 与 test_rsi_loop.py。下面的函数是 Day 2、Day 4、Day 6 的验收对象，不是另一门课。

日志最低字段：round、task_id、policy_rev、scorer_version、failures、score、accepted、regression_new、safety_hash。校验器发现 regression_new 为真且 accepted 为真，或 patch_keys 含 safety，就把该行记为违规。违规未清零时，调度器不得调用模型。

import json, re
from collections import Counter
from pathlib import Path

REQUIRED = ["round","task_id","policy_rev","scorer_version","failures","score","accepted","regression_new","safety_hash"]

def check_trace(path):
    bad = []
    versions = set()
    for i, line in enumerate(Path(path).read_text(encoding="utf-8").splitlines(), 1):
        if not line.strip():
            continue
        row = json.loads(line)
        missing = [k for k in REQUIRED if k not in row]
        if missing:
            bad.append((i, "missing", missing))
        versions.add(row.get("scorer_version"))
        if row.get("regression_new") and row.get("accepted"):
            bad.append((i, "accepted_on_regression"))
        if "safety" in (row.get("patch_keys") or []):
            bad.append((i, "safety_patched"))
    return versions, bad

数字规则按多重集比较，而不是搜索「有没有数字」。规格里的「15 分钟」「5 次」「99.9%」必须在输出中以同样的数字和单位出现。「约 100%」不能代替「99.9%」。单位词表由人维护，改进器不许往词表里加词来放宽 R6。

NUM = re.compile(r"\d+(?:\.\d+)?%?")
UNITS = ("分钟", "次", "位", "天")

def obligations(spec: str) -> Counter:
    nums = Counter(NUM.findall(spec))
    for unit in UNITS:
        if unit in spec:
            nums[unit] += 1
    return nums

def rule_r6(spec: str, output: str) -> bool:
    need = obligations(spec)
    return all(obligations(output)[k] >= v for k, v in need.items())

停止函数比较失败列表是否相同，不比较分数是否接近。分数同为 8 但编号从 R6 换成 R2，是漂移，不是平台期。成功停止只在失败列表为空时触发。轮数或 token 先到，返回预算原因，并保留最后一次 accepted 为真的策略，而不是再请求一轮。

def should_stop(history, max_rounds, max_tokens):
    if len(history) >= max_rounds:
        return "budget_rounds"
    tokens = sum(h.get("token_in", 0) + h.get("token_out", 0) for h in history)
    if tokens >= max_tokens:
        return "budget_tokens"
    if history and history[-1]["failures"] == []:
        return "success"
    if len(history) >= 2 and history[-1]["failures"] == history[-2]["failures"]:
        if history[-1]["scorer_version"] == history[-2]["scorer_version"]:
            return "plateau"
    return None

单元测试覆盖四件事：差草稿上 R6 为假；「99.9%」被写成「100%」时为假；安全段变化时补丁函数返回 None；失败编号发生替换时 should_stop 不返回 plateau。测试退出码非 0，就不要启动带模型的循环。毕业说明里写明你运行的是 python3 -m unittest test_rsi_loop.py，以及退出码。

这些函数放在开发指南，是因为它们是组件契约的可执行版本。快速开始 的脚本跑通之后，把契约迁到这里，避免两套字段名。