前面几篇已经为 RAG 系统搭好了底座：

• 三类专用 Agent：检索 / 解释 / 评估；
• 完整的可观测与质量闭环：对每次问答都有结构化的评估结果（score + labels）；
• 低分样本会被打包成 QualityEvent，进入统一队列。

如果这些质量数据只用来看报表、画看板，系统的能力顶多是“会自我监控”；
要让系统“越用越懂自己的业务”，还差最后一环：

把质量事件自动转成监督微调（SFT）数据集，
驱动模型参数本身随时间持续变好。

这篇文章就专门讲这件事：如何从质量事件出发，构建一套 RAG 专用的 SFT 策略，并落到代码。

---

一、为什么 RAG 需要“专用”的 SFT？

1.1 RAG 的错误分布，和普通对话不一样

通用 LLM 的 SFT 通常优化：

• 语言流畅度；
• 指令遵从；
• 安全性与风格。

但在 RAG 里，你更关心的是：

• 事实性（Factuality）​：答案是否“完全基于文档，而不是瞎编”；
• 完整性（Completeness）​：是否把文档中的关键信息都覆盖到了；
• 相关性（Relevance）​：回答是否越过了问题的边界，开始乱扩展；
• 语气控制（Overconfidence）​：在“文档没说清”的时候，能不能少点绝对化表述。

简单说：

RAG 的关键，不是“说得好听”，而是“说得对、说得全、说得谨慎”。

所以，RAG 的 SFT 不应该只是“再训一遍通用对话数据”，而是要围绕这几类错误做定向监督。

1.2 质量事件 = 现成的标注样本

在之前的质量闭环中，每次回答都会产出一个 evaluation 结构，大致包含：

• score: 本次问答的整体打分；
• labels: 错误标签列表，如：
  • factuality_error
  • incomplete_answer
  • low_relevance
  • overconfidence
• checks: 各维度子评分（事实性、完整性、相关性、过度自信等）；
• suggestion: 给系统的修正建议。

再加上你已经保存的：

• question：用户问题；
• docs：本次使用的文档片段；
• answer：当前给用户的回答。

这套信息天然就是监督样本的“原料”：​

• 输入侧：问题 + 文档；
• 输出侧：应该怎样回答 / 怎样修正；
• 标签：哪里错了、为什么错。

不需要额外人工标注，就可以把“错误案例”转成“训练样本”。

---

二、从质量事件到训练样本：完整流水线

整体来看，这条链路可以分为四步：

1. 事件落盘：把 QualityEvent 稳定写入数据库；
2. 清洗标注：根据 evaluation.labels 识别错误类型，生成纠正规则；
3. 样本构造：把 question + docs + instruction 拼成 prompt，生成目标 completion；
4. 导出数据：落成标准 JSONL，交给训练脚本。

2.1 事件落盘：一张原始事件表就够

推荐的表结构示意：

CREATE TABLE raw_quality_events (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    request_id VARCHAR(64) UNIQUE,
    question  TEXT NOT NULL,
    docs      JSON NOT NULL,  -- [{"id": "...", "content": "...", "score": 0.87}, ...]
    answer    TEXT NOT NULL,
    evaluation JSON NOT NULL, -- {"score": 0.53, "labels": [...], "checks": {...}}
    timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    processed BOOLEAN DEFAULT FALSE
);

在 RAG Orchestrator 末尾加一段逻辑（伪代码）：

if evaluation["score"] < 0.7:
    event = {
        "request_id": request_id,
        "question": question,
        "docs": docs,         # 检索结果
        "answer": answer,
        "evaluation": evaluation
    }
    # 写入 DB 或消息队列

关键点只有一个：低质样本不能丢，否则 SFT 没“养料”。

2.2 清洗标注：把事件变成“样本骨架”

我们先定义一个简单的清洗器，用来：

1. 识别主错误类型（label）；
2. 选出适合作为上下文的 top-k 文档；
3. 生成一段“指令”，告诉模型要如何修正。

def detect_label(evaluation: dict) -> str:
    """从 labels 中选出一个主错误类型"""
    labels = evaluation.get("labels", [])
    if not labels:
        return "unspecified"
    priority = ["factuality_error", "incomplete_answer", "low_relevance", "overconfidence"]
    for p in priority:
        if p in labels:
            return p
    return labels[0]


def select_context(docs: list[dict], top_k: int = 3) -> list[dict]:
    """按照 score 取前 top_k 个文档片段"""
    return sorted(docs, key=lambda d: d.get("score", 0), reverse=True)[:top_k]


def build_correction_instruction(label: str) -> str:
    """针对不同错误类型，生成不同的纠正要求"""
    if label == "factuality_error":
        return (
            "请严格依据文档内容回答，避免任何事实性错误。"
            "如果文档未提及相关信息，请明确回答“文档中未提及”。"
        )
    if label == "incomplete_answer":
        return (
            "请根据文档内容重新回答问题，尽量覆盖所有关键信息，"
            "避免遗漏重要细节。"
        )
    if label == "low_relevance":
        return (
            "请重新理解问题，使回答紧扣问题本身和文档内容，"
            "不要引入与文档无关的信息。"
        )
    if label == "overconfidence":
        return (
            "请用更谨慎、中性的语气回答，"
            "避免使用“肯定”“必须”“一定”等过度自信的表达。"
        )
    return "请根据文档内容重新回答问题。"


def annotate_event(raw: dict) -> dict:
    """把原始事件转成一个结构化的样本骨架"""
    label = detect_label(raw["evaluation"])
    context_docs = select_context(raw["docs"], top_k=3)
    instruction = build_correction_instruction(label)
    confidence = raw["evaluation"].get("score", 0.0) * 1.5  # 简单放大，用于后续样本权重

    return {
        "sample_id": f"{raw['request_id']}_{label}",
        "question": raw["question"],
        "context_docs": context_docs,
        "original_answer": raw["answer"],
        "label": label,
        "instruction": instruction,
        "score": raw["evaluation"].get("score", 0.0),
        "confidence": confidence,
    }

2.3 组装 prompt / completion：两种策略

Prompt 模板建议固定下来，方便后续分析和复用：

def build_prompt(annotated: dict) -> str:
    parts = []
    for i, d in enumerate(annotated["context_docs"], start=1):
        parts.append(f"[文档{i}] {d['content']}")
    ctx_block = "\n\n".join(parts)

    prompt = (
        f"{ctx_block}\n\n"
        f"{annotated['instruction']}\n\n"
        f"# 问题\n{annotated['question']}"
    )
    return prompt

Completion 有两种做法：​

1. 快速版（拼接式）​：
   不依赖额外 LLM，直接用规则“纠正”或“补充”原回答；
2. 高质量版（LLM 生成）​：
   用更强的、可信度高的模型生成理想答案，作为 SFT 的目标。

例如快速版（适合打通流程阶段）：

def build_naive_completion(annotated: dict) -> str:
    if annotated["label"] == "factuality_error":
        prefix = "【修正事实】"
    elif annotated["label"] == "incomplete_answer":
        prefix = "【补充说明】"
    elif annotated["label"] == "overconfidence":
        prefix = "【语气调整】"
    else:
        prefix = "【重新回答】"
    return f"{prefix}{annotated['original_answer']}"

高质量版（适合以后上线前的正式数据集）：

def build_llm_completion(high_quality_llm, annotated: dict) -> str:
    prompt = build_prompt(annotated)
    return high_quality_llm.chat(
        messages=[{"role": "user", "content": prompt}],
        temperature=0.2,
        max_tokens=512
    ).strip()

2.4 导出 JSONL：标准训练数据格式

多数训练框架都支持如下 JSONL 结构：

{"prompt": "...", "completion": "...", "label": "factuality_error", "confidence": 0.93}

一个批处理示例：

class DatasetBuilder:
    def __init__(self, db_conn, output_path="rag_tuning_data.jsonl"):
        self.conn = db_conn
        self.output_path = output_path

    def fetch_unprocessed(self, limit=200):
        cur = self.conn.cursor()
        cur.execute("""
            SELECT request_id, question, docs, answer, evaluation
            FROM raw_quality_events
            WHERE processed = FALSE
            ORDER BY timestamp ASC
            LIMIT ?
        """, (limit,))
        rows = cur.fetchall()
        events = []
        for r in rows:
            events.append({
                "request_id": r[0],
                "question": r[1],
                "docs": json.loads(r[2]),
                "answer": r[3],
                "evaluation": json.loads(r[4]),
            })
        return events

    def mark_processed(self, request_ids):
        if not request_ids:
            return
        cur = self.conn.cursor()
        cur.executemany(
            "UPDATE raw_quality_events SET processed=TRUE WHERE request_id=?",
            [(rid,) for rid in request_ids]
        )
        self.conn.commit()

    def process_batch(self, llm=None, min_confidence=0.6):
        events = self.fetch_unprocessed(limit=200)
        if not events:
            return 0

        processed_ids = []
        with open(self.output_path, "a", encoding="utf-8") as f:
            for ev in events:
                annotated = annotate_event(ev)
                if annotated["confidence"] < min_confidence:
                    continue

                prompt = build_prompt(annotated)
                if llm is not None:
                    completion = build_llm_completion(llm, annotated)
                else:
                    completion = build_naive_completion(annotated)

                record = {
                    "prompt": prompt,
                    "completion": completion,
                    "label": annotated["label"],
                    "confidence": annotated["confidence"],
                    "score": annotated["score"],
                }
                f.write(json.dumps(record, ensure_ascii=False) + "\n")
                processed_ids.append(ev["request_id"])

        self.mark_processed(processed_ids)
        return len(processed_ids)

---

四、RAG 专用 SFT 策略：不只是“喂数据”

有了数据集，还需要一套面向 RAG 的训练策略，核心包含三块：

1. 样本级权重（按置信度）；
2. 标签级权重（按类别不平衡）；
3. 损失函数级别的加权。

4.1 样本权重：confidence 越高，越值得学

简单策略：

def calculate_sample_weight(confidence: float, baseline=0.7, max_weight=5.0) -> float:
    """基于置信度计算样本权重，置信度越高权重越大"""
    if confidence < baseline:
        return 0.5
    return min(max_weight, 1.0 + (confidence - baseline) * 4)

使用时，可以在 Dataset 中预计算每条样本的 weight，然后用 WeightedRandomSampler 来做采样。

4.2 标签权重：让“严重错误”多练几遍

例如：你希望模型优先改正事实性错误和低相关性问题：

def get_label_weights(label_counts: dict) -> dict:
    """基于类别数量动态算权重，样本少/更致命的类别权重大"""
    total = sum(label_counts.values())
    weights = {}
    for label, count in label_counts.items():
        rare_factor = total / max(count, 1)
        # 可以手动给某些label再乘一个系数
        if label == "factuality_error":
            base = 1.5
        elif label == "low_relevance":
            base = 1.3
        else:
            base = 1.0
        weights[label] = base + 0.6 * rare_factor
    return weights

后续可以在损失里使用这个 label_weights。

4.3 损失函数加权：在 loss 层强化重点错误

示例实现：

class WeightedLoss(torch.nn.Module):
    def __init__(self, label_weights: dict):
        super().__init__()
        self.label_weights = label_weights
        self.ce = torch.nn.CrossEntropyLoss(reduction="none")

    def forward(self, logits, labels, batch_labels):
        """
        logits: [batch, seq, vocab]
        labels: [batch, seq]
        batch_labels: [batch]，每条样本对应的错误类型
        """
        # 简化起见，这里先把 seq 维度合并
        loss_per_token = self.ce(
            logits.view(-1, logits.size(-1)),
            labels.view(-1)
        ).view(logits.size(0), -1).mean(dim=1)  # [batch]

        for i, lbl in enumerate(batch_labels):
            loss_per_token[i] *= self.label_weights.get(lbl, 1.0)

        return loss_per_token.mean()

这样，同一批数据中，带有 factuality_error 的样本，会在反向传播中“被放大”，从而引导模型更快地修正这些错误行为。

---

五、训练后怎么验证“值不值”？

你前面已经有 A/B 测试框架和可观测看板，这里只需要把 RAG 相关的指标纳入对比即可。

关键观察：

• 事实性错误率（factuality_error_rate）​
• 回答不完整率（incomplete_answer_rate）​
• 过度自信率（overconfidence_rate）​
• 平均延迟（avg_latency_ms）​
• 每次调用平均 token & 费用（方便计算 $/请求）

一个典型的对比表可以是：

指标	基线模型	微调模型	变化
factuality_error_rate	12%	4%	-67%
incomplete_answer_rate	22%	11%	-50%
overconfidence_rate	8%	2%	-75%
avg_latency_ms	1450	1480	+30ms

如果看到事实性和过度自信相关指标有明显下降，而延迟和成本可接受，就说明这轮 SFT 是“赚的”。

---

六、如何低成本地先跑起来？

建议按下面节奏渐进式落地，而不是一口气大重构。

1. 先打通数据流水线

   • 只做采集 + 清洗 + 导出 JSONL；
   • completion 先用简单拼接版（不依赖额外大模型）；
   • 不急着训练，把样本抽查几百条，看质量是否 OK。

2. 在一个小场景试训

   • 选一个风险低、业务闭环清晰的场景（比如内部 FAQ）；
   • 用 LoRA/QLoRA 做小规模 SFT；
   • 用你现有的 A/B 测试框架灰度 10–20% 流量。

3. 体验改善明显后，再扩大范围

   • 这时可以把 completion 换成“高质量 LLM 自动修正版”；
   • 加入样本权重、标签权重、加权损失等策略；
   • 把训练好的 Adapter 逐步扩展到更多 RAG 工作流里。

4. 最后才考虑“自动触发训练”

   • 比如：当过去一周累计新样本 >= N 时，自动触发一轮训练；
   • 训练完成后自动注册一个新模型版本，接入灰度；
   • 整条链路就从“人工调参”变成“半自动自适应”。

---

七、总结

这一篇的目标是：把你已有的 RAG 质量闭环，扩展成一条完整的 数据 → 模型 → 线上效果的闭环。

核心步骤可以归纳成 6 点：

1. 采集：把低分 / 高风险回答记录成结构化 QualityEvent，落盘到 DB；
2. 清洗：基于 evaluation.labels 自动识别错误类型，选取合适的文档上下文；
3. 构造：用统一的 prompt 模板 + 策略（拼接 / LLM 修正）生成 (prompt, completion)；
4. 加权：通过样本置信度、错误类型和损失函数加权，引导模型重点学习“最致命”的错误；
5. 评估：用事实性、完整性、过度自信等 RAG 特有指标做 A/B 测试；
6. 发布：通过灰度 + 回滚策略，把新模型稳妥地推到线上。

做到这一步，RAG 系统就已经从：

“有监控、有日志、有人手动调整”

升级成了：

“能看懂问题在哪、能自己攒数据、能自己长本事”的系统。