循环工程：自动调优RAG召回率

一个具体的循环工程案例。不是“手动调优 RAG”，而是构建一个循环，它自行搜索配置，在评估集上测量召回率，并在达到目标时停止。附带完整代码。

手动调优 RAG 是件痛苦的事。修改块大小，运行评估，查看召回率，更换嵌入模型，再次运行，添加重排序器，又一次。几十次手动运行下来，你会搞不清楚哪个组合产生了什么结果。这正是应该交给循环的任务：重复性的、有可测量检查点的任务，你反复调整相同的旋钮。

本文展示了如何构建这样的循环。任务很具体：你有一个 RAG 系统和一个评估集，目标是将 recall@5 提升到 0.9，循环自行搜索配置，直到达到目标或达到限制。我们将一步步把循环工程应用到一项真实任务中。

为什么 RAG 调优非常适合循环

回顾循环工程的主要规则：只有在存在自动检查、无需你亲自判断的情况下，循环才有意义。RAG 调优完美符合这一条件。

这里的检查是评估集上的 recall@k。它是一个数值。要么高于阈值，要么低于，没什么可争论的，就像一个绿色测试。循环总是确切地知道是否达到目标。这是一个罕见的案例，指标开箱即用，无需额外创造。

此外，调优是在配置空间中进行搜索，而搜索正是循环胜过人类的地方：它不会疲劳，不会迷失，有条不紊地迭代并记住已经尝试过的内容。人类在第十次运行后会感到困惑，循环不会。

开始前需要准备什么

循环不是从头构建 RAG，而是调优已有的 RAG。因此，在开始之前，你必须具备：

一个可工作的 RAG 流水线，参数可调：块大小、重叠、嵌入模型、候选数量 k、重排序器开关（开/关）。

一个评估集：包含 30-50 个问题，每个问题有已知的正确源块。这是你的检查，没有它就没有循环。

一个运行评估并返回 recall@k 的函数。这是循环的预言机。

## eval.py — 循环的预言机，返回配置的 recall@k
def evaluate(config, eval_set):
    pipeline = build_rag(config)        # 使用这些参数构建 RAG
    hits = 0
    for case in eval_set:
        retrieved = pipeline.retrieve(case["question"], k=config["k"])
        retrieved_ids = {c["id"] for c in retrieved}
        if case["gold_chunk_id"] in retrieved_ids:
            hits += 1
    return hits / len(eval_set)

第一步：定义搜索空间

循环会搜索配置，因此你必须精确定义它要调整什么以及在什么范围内。不是无限空间，而是一组真正影响召回率的合理参数。

## search_space.py — 循环搜索的内容
SEARCH_SPACE = {
    "chunk_size":   [400, 600, 800, 1200],
    "chunk_overlap":[0, 100, 200],
    "embedding":    ["text-embedding-3-small", "bge-large", "e5-large"],
    "k":            [5, 10, 20],
    "reranker":     [None, "bge-reranker", "cohere-rerank"],
    "hybrid":       [False, True],   # 仅向量或向量 + BM25
}

这个空间已经很大了：4×3×3×3×3×2 = 648 种组合。全部搜索既耗时又愚蠢，所以循环不会穷举遍历，而是智能地进行，这是第三步要讲的内容。

第二步：一个你无法欺骗的检查

循环的核心。评估集上的召回率是一个好的检查点，但有两个陷阱需要在启动前解决。

陷阱一：过拟合评估集。如果循环不断搜索配置直到评估集上的召回率上升，它可能会找到一个恰好对这 40 个问题表现良好但在生产环境中不佳的组合。这与代码中的奖励黑客行为相同，只是指标变成了数值。防御措施：将评估集分成两部分，循环在一部分（训练集）上优化，你在第二部分（保留集）上检查最终配置，循环从未见过保留集。

import random

def split_eval(eval_set, holdout_frac=0.3, seed=42):
    random.Random(seed).shuffle(eval_set)
    n = int(len(eval_set) * (1 - holdout_frac))
    return eval_set[:n], eval_set[n:]   # 训练集给循环，保留集给你

train_set, holdout_set = split_eval(eval_set)

陷阱二：测量噪声。在 30 个问题上，召回率差 0.87 与 0.90 可能是噪声，不是真正的改进。如果改进阈值太小，循环会追逐噪声。防御措施：只有当改进超过合理阈值（例如 0.02）时才视为显著，否则循环会因随机波动而反复调整。

第三步：带有智能搜索的循环

穷举遍历 648 种组合既昂贵又无必要。循环采用贪心方式，一次调整一个旋钮：固定其他参数，搜索一个参数的值，取最佳，然后进入下一个。这是坐标下降法，它能在几十次运行中找到好的配置，而不是几百次。

import json

def tune_rag(search_space, train_set, target_recall=0.9, max_evals=40):
    # 从合理的默认值开始
    config = {
        "chunk_size": 600, "chunk_overlap": 100,
        "embedding": "text-embedding-3-small", "k": 10,
        "reranker": None, "hybrid": False,
    }
    best_recall = evaluate(config, train_set)
    evals_used = 1
    log = []

    # 坐标下降：一次调整一个旋钮
    for param, values in search_space.items():
        if evals_used >= max_evals:
            break
        best_value = config[param]
        for value in values:
            if value == config[param]:
                continue
            if evals_used >= max_evals:
                break
            trial = dict(config, **{param: value})
            recall = evaluate(trial, train_set)
            evals_used += 1
            log.append({"config": trial, "recall": recall, "eval": evals_used})

            # 仅当改进超过噪声阈值时才视为显著
            if recall > best_recall + 0.02:
                best_recall = recall
                best_value = value

            # 目标检查：达到目标则立即退出，不浪费后续运行
            if best_recall >= target_recall:
                config[param] = best_value
                save_log(log)
                return config, best_recall, evals_used

        config[param] = best_value   # 锁定该旋钮的最佳值

    save_log(log)
    return config, best_recall, evals_used

def save_log(log):
    with open("tune_log.jsonl", "w") as f:
        for row in log:
            f.write(json.dumps(row) + "\n")

看看这里发生了什么。每次运行都检查目标：达到 target_recall 则立即退出，不消耗剩余运行次数。max_evals 是保险丝，没有它循环会在不幸的空间中遍历一切。0.02 的阈值防止追逐噪声。日志写入 jsonl，以便日后查看哪个旋钮产生了什么结果。

第四步：刹车，因为运行需要花钱

每次评估都需要对整个 RAG 运行几十个问题，每个问题都涉及查询嵌入，可能还有重排序器和 LLM 调用。这会消耗金钱和时间。没有刹车的循环可能会悄然消耗预算，尤其是当空间中包含昂贵配置（大的 k 加上每个问题都使用 cohere-rerank）时。

def tune_rag_safe(search_space, train_set, target_recall=0.9,
                  max_evals=40, max_budget_usd=15):
    config = default_config()
    best_recall = evaluate(config, train_set)
    spent = estimate_cost(config, len(train_set))
    evals_used = 1
    log = []

    for param, values in search_space.items():
        best_value = config[param]
        for value in values:
            if value == config[param]:
                continue
            # 预算刹车：下一次运行会超出限制，停止
            trial = dict(config, **{param: value})
            trial_cost = estimate_cost(trial, len(train_set))
            if spent + trial_cost > max_budget_usd:
                print(f"预算 ${max_budget_usd} 已花完。在第 {evals_used} 次运行后停止。")
                save_log(log)
                return config, best_recall, evals_used
            # 运行次数刹车
            if evals_used >= max_evals:
                save_log(log)
                return config, best_recall, evals_used

            recall = evaluate(trial, train_set)
            spent += trial_cost
            evals_used += 1
            log.append({"config": trial, "recall": recall,
                        "spent": round(spent, 2), "eval": evals_used})

            if recall > best_recall + 0.02:
                best_recall = recall
                best_value = value
            if best_recall >= target_recall:
                config[param] = best_value
                save_log(log)
                return config, best_recall, evals_used

        config[param] = best_value

    save_log(log)
    return config, best_recall, evals_used

def estimate_cost(config, n_questions):
    # 粗略估计：每个问题的查询嵌入 + 可选重排序器
    cost = n_questions * 0.00002   # 查询嵌入
    if config["reranker"] == "cohere-rerank":
        cost += n_questions * config["k"] * 0.000001
    return cost

这里两个刹车是 max_evals（运行次数上限）和 max_budget_usd（金钱上限）。预算刹车在运行前检查：如果下一次评估会超出限制，循环提前停止，而不是在钱花完之后。这就是资产和意外账单之间的差别——前者可预测，后者带来意外。

第五步：启动并在保留集上检查

在训练集上运行循环，然后始终在循环从未见过的保留集上检查发现的配置。

if __name__ == "__main__":
    train_set, holdout_set = split_eval(eval_set)

    # 循环在训练集上调优
    config, train_recall, n = tune_rag_safe(
        SEARCH_SPACE, train_set, target_recall=0.9, max_evals=40
    )
    print(f"在 {n} 次运行中找到。训练集召回率：{train_recall:.2f}")
    print(f"配置：{config}")

    # 关键：在保留集上检查，循环未在此优化
    holdout_recall = evaluate(config, holdout_set)
    print(f"保留集召回率：{holdout_recall:.2f}")

    gap = train_recall - holdout_recall
    if gap > 0.1:
        print("警告：训练/保留集差距过大。"
              "循环过拟合了评估集，配置不可信。")
    else:
        print("差距小，配置可泛化。可以部署。")

保留集检查不是形式，它是区分真正改进与过拟合评估集的关键。如果训练集召回率 0.92 而保留集 0.79，则循环找到的不是好的 RAG，而是一个恰好迎合了你的 28 个训练问题的组合。差距就是警报信号。

此循环的失败方式

与任何循环相同的失败方式，应用于调优。

追逐噪声。循环因随机召回波动而抖动，无法收敛。原因：改进阈值太小或评估集太小。解决方案：0.02 阈值，评估集至少 40 个问题。

过拟合评估集。训练集召回率上升，但生产环境中 RAG 并未更好。原因：循环针对特定问题优化。解决方案：保留集检查，差距提供信号。

预算失控。循环搜索昂贵配置并烧钱。原因：没有预算刹车，或空间中包含昂贵组合。解决方案：在运行前检查 max_budget_usd。

无休止的搜索。目标在给定空间中无法达到，循环搜索所有内容。原因：target_recall 高于你的空间允许的上限。解决方案：设置 max_evals，如果卡住，则扩大空间或降低目标。

你得到什么

最终你不仅得到调优后的 RAG，还有一个可重复的过程。jsonl 日志显示哪个旋钮产生什么结果，保留集检查告诉你是否可信，当数据或模型变化时，整个调优可以通过一条命令重复执行。

这就是循环工程应用于真实任务。不是“一个智能体神奇地调优了 RAG”，而是一个工程循环：一个你无法欺骗的可测量检查，防止过拟合，预算刹车，可审计的日志。与一般循环工程相同的纪律，只是这里的检查是召回率，而不是绿色测试。

拿起你的 RAG 和评估集，将调优封装在这样的循环中。构建一次，再也不用手动调整分块参数了。

• 原文链接： x.com/h100envy/status/20...
• 登链社区 AI 助手，为大家转译优秀英文文章，如有翻译不通的地方，还请包涵～