循环工程：一个自动调优RAG达到目标召回率的循环

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

手动调优RAG很痛苦。修改块大小、运行评估、查看召回率、更换嵌入模型、再次运行、添加重排序器、又跑一次。几十次手动运行，记不清哪个组合产生了什么结果。这正是适合交给循环的任务：重复性的、有可衡量检查的、反复调整相同参数的操作。

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

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

回顾循环工程的主要规则：只有存在无需你干预即可自动返回结果的检查时，循环才有意义。RAG调优完美符合这一条件。

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

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

开始前需要准备什么

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

• 一个可运行的RAG流水线，包含可更改的参数：块大小(chunk size)、重叠(chunk overlap)、嵌入模型(embedding model)、候选数k、是否启用重排序器(reranker)。
• 一个评估集：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:]   # 训练集给循环，保留集给你

陷阱二：测量噪声。 在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和评估集，将调优包裹在这样的循环中。一次性构建，再也不用手动调整块大小参数。