已经把这条线打通到了：

• RAG 专用 Agent（三件套：检索 / 解释 / 评估）；
• 质量事件（QualityEvent）收集与分桶；
• 从质量事件自动生成 SFT 数据集，并做监督微调。

做到这里，你已经能让模型：

少犯错、别乱编、说话更谨慎。

但在实际业务里，“没错”不等于“好”，你还会遇到很多“偏好层面”的诉求：

• 同样都对，希望回答更像「专业顾问」而不是「搜索引擎摘要」；
• 有两种都说得通的答案，但你更偏爱其中一种表达方式/结构；
• 希望在有歧义时倾向某种解释，而不是平均分配。

这些就属于 偏好对齐（preference alignment）​ 的范畴，需要用到 RLHF / DPO 一类方法。

这篇就结合你已经有的质量事件系统，讲清楚：

如何自动构建偏好样本（chosen/rejected 对）​，
再用 DPO 这类“轻量 RLHF”方法，
让你的 RAG 模型从“不会犯大错”进化到“更懂业务喜好”。

---

一、整体思路：在现有质量闭环外再加一圈「偏好环」

先把整条链路按时间顺序排一下，你现在大致是这样：

用户提问
  └─► 检索 Agent（Retrieve）
        └─► 解释 Agent（Answer）
              └─► 评估 Agent（Evaluate）→ 生成 QualityEvent（低分样本）→ 存DB

上一篇你又加了一步：

QualityEvent → 清洗/标注 → 生成 SFT 样本(JSONL) → 监督微调模型

这篇要做的是再加一条支线，把 “好 vs 不好” 的对比关系也挖出来：

QualityEvent
   └─► 生成「原回答（坏一些）」 vs 「改进回答（好一些）」的样本对
          └─► 得到 (prompt, chosen, rejected)
                 └─► 用 DPO / RLHF 做偏好对齐微调

可以理解为：

• SFT：教模型“什么是对的、不要犯硬错误”；
• DPO/RLHF：教模型“在多个看似都对的选项里，哪一种更符合业务偏好”。

---

二、从 QualityEvent 自动构建偏好样本对

2.1 已有信息回顾

一条质量事件（QualityEvent）里，至少有这些字段：

• question：用户问题；
• docs：检索到的文档片段（含 content 和 score）；
• answer：当时返回给用户的回答（当前模型的输出）；
• evaluation：
  • score：整体质量评分（0~1）；
  • labels：错误或缺陷标签（如 incomplete_answer / low_relevance / overconfidence 等）；
  • checks：子维度打分（事实性、完整性、相关性、语气等）。

这已经非常接近 RLHF / DPO 需要的 偏好数据原料 了——差的一步就是：

在「原回答」的基础上，再造一个「明显更优的改进版回答」，
然后把两者组成 (chosen, rejected) 对。

2.2 核心接口：从事件构造偏好对

可以设计一个「偏好样本生成器」：

from typing import Dict, Optional

class PreferenceGenerator:
    def __init__(self, retriever_agent, explainer_agent, llm_for_improve):
        self.retriever = retriever_agent        # 你现有的检索 Agent
        self.explainer = explainer_agent        # 你现有的解释 Agent
        self.llm = llm_for_improve              # 一个质量更高的 LLM（或同模型+更强约束）

    def generate_pair(self, event: Dict, min_score: float = 0.5) -> Optional[Dict]:
        """
        从一条 QualityEvent 生成 DPO 需要的 (prompt, chosen, rejected) 样本。
        :return: {prompt, chosen, rejected, label, confidence} 或 None
        """
        evaluation = event["evaluation"]
        score = evaluation.get("score", 0.0)
        if score < min_score:
            # 质量过低可能上下文也出大问题，先丢掉
            return None

        label = (evaluation.get("labels") or ["unspecified"])[0]
        original_answer = event["answer"]

        # 基于标签决定“如何生成改进版”
        improved_answer = self._improve_answer(event, label)

        # 构造标准 prompt（问题 + 文档上下文）
        prompt = self._build_prompt(event["question"], event["docs"])

        return {
            "prompt": prompt,
            "chosen": improved_answer,
            "rejected": original_answer,
            "label": label,
            "confidence": score
        }

其中 _improve_answer 和 _build_prompt 是两个关键点。

2.3 用不同策略生成「改进版回答」

根据不同 label，用不同“改进策略”来生成 chosen：

    def _improve_answer(self, event: Dict, label: str) -> str:
        q = event["question"]
        docs = event["docs"]
        ctx = "\n\n".join(d["content"] for d in docs)

        # 按错误类型选择不同提示语
        if label == "incomplete_answer":
            instruction = (
                "你之前的回答不够完整。"
                "现在请根据下方文档内容，补充遗漏的信息，生成一个更全面的答案。"
            )
        elif label == "low_relevance":
            instruction = (
                "你之前的回答和问题、文档的相关性不高。"
                "请阅读文档，只回答与问题高度相关的内容。"
            )
        elif label == "overconfidence":
            instruction = (
                "你之前的回答语气过于肯定。"
                "请在这次回答中，用谨慎、中性的语气描述，只在有充分文档依据时使用肯定语气。"
            )
        elif label == "factuality_error":
            instruction = (
                "你之前的回答中存在事实性错误。"
                "请严格对照文档信息，重新回答，避免任何文档未提及的“脑补”内容。"
            )
        else:
            instruction = "请根据文档内容，重新给出更优回答。"

        prompt = (
            f"【文档】\n{ctx}\n\n"
            f"【问题】\n{q}\n\n"
            f"【要求】\n{instruction}"
        )
        # 用更强 LLM 生成改进答案；或者用同一个模型 + 严格 prompt
        improved = self.llm(prompt)
        return improved.strip()

2.4 标准化 prompt：问题 + 文档

    def _build_prompt(self, question: str, docs: list[Dict]) -> str:
        parts = []
        for i, d in enumerate(docs[:3], start=1):
            parts.append(f"[文档{i}] {d['content']}")
        ctx_block = "\n\n".join(parts)
        prompt = (
            f"{ctx_block}\n\n"
            f"请根据这些文档回答下面的问题，做到准确、完整、谨慎：\n\n"
            f"问题：{question}\n"
        )
        return prompt

这样，你就从一条质量事件里得到了一个标准的偏好样本：

{
  "prompt": "...(文档+问题)...",
  "chosen": "改进后的答案",
  "rejected": "原回答",
  "label": "incomplete_answer",
  "confidence": 0.65
}

这个格式基本可以直接喂给 DPO 等偏好对齐算法使用。

---

三、批量生成偏好数据：从队列到 JSONL

在你原有的质量队列处理逻辑里，加一条分支，把“可用的质量事件”转成偏好样本写到一个 JSONL 文件里：

import json

class PreferenceDatasetBuilder:
    def __init__(self, generator: PreferenceGenerator, output_path: str):
        self.gen = generator
        self.output_path = output_path

    def process_events(self, events: list[Dict], min_confidence: float = 0.6) -> int:
        count = 0
        with open(self.output_path, "a", encoding="utf-8") as f:
            for ev in events:
                pair = self.gen.generate_pair(ev, min_score=min_confidence)
                if pair is None:
                    continue
                # 转成 DPO 常用的 JSONL 结构
                record = {
                    "prompt": pair["prompt"],
                    "chosen": pair["chosen"],
                    "rejected": pair["rejected"],
                    "label": pair["label"],
                    "confidence": pair["confidence"]
                }
                f.write(json.dumps(record, ensure_ascii=False) + "\n")
                count += 1
        return count

你可以用一个定时任务（或后台线程）周期性拉 QualityEvent 表/队列的数据，跑这段逻辑，把偏好样本不断追加到如 rag_dpo_pref_data.jsonl 里。

---

四、如何用这些偏好样本做 DPO 对齐（概念 + 实践骨架）

这里不写完整训练脚本，只给你一个「工程上应该怎么接」的框架思路，方便你后续按自己训练栈落地。

4.1 模型准备

一般会有三个版本：

1. base 模型：原始基础模型（未针对你业务 SFT）；
2. sft 模型：你前几篇用监督微调得到的版本（已掌握业务知识 + RAG 行为）；
3. dpo policy 模型：这次要在 sft 模型基础上再做 DPO，对其偏好。

通常做法是：

• 以 sft 模型为起点；
• 复制一份冻结为 reference_model；
• 另一份作为 policy_model，在偏好数据上训练。

4.2 DPO 损失的大致形态

DPO 的核心思想是：对同一个 prompt 的 chosen 和 rejected 两个回答，希望模型在 chosen 上的「相对概率」比在 rejected 上更高，且相对参考模型有改善。

极简伪代码（逻辑示意）：

def dpo_loss(policy, reference, prompt, chosen, rejected, beta=0.1):
    # 计算 policy 对 chosen/rejected 的 log p
    logp_pi_c = log_prob(policy, prompt, chosen)
    logp_pi_r = log_prob(policy, prompt, rejected)

    # 计算 reference 对 chosen/rejected 的 log p（不反传）
    with torch.no_grad():
        logp_ref_c = log_prob(reference, prompt, chosen)
        logp_ref_r = log_prob(reference, prompt, rejected)

    # DPO 核心：让 policy 在 chosen 上的相对优势超过 reference
    diff_pi = logp_pi_c - logp_pi_r
    diff_ref = logp_ref_c - logp_ref_r
    loss = -torch.logsigmoid(beta * (diff_pi - diff_ref)).mean()
    return loss

真正落地可以直接用 HuggingFace TRL 之类的工具库，重点是：

• 我们用刚生成的 (prompt, chosen, rejected) 作为训练数据；
• 把原 SFT 模型作为 reference；
• 让新的 policy 模型在“偏好维度”上比 reference 更好。

---

五、如何插回你的 RAG 系统：A/B + 安全阈值

对你现有业务来说，不建议“一次性替换掉 SFT 模型”，可以这样做：

1. 先把 DPO policy 接成一个新的模型版本，例如 rag_qa_dpo_v1；
2. 在你已有的「工作流 + A/B 测试」框架里：
   • 基线：rag_qa_sft_vX（现在线上版本）；
   • 实验：rag_qa_dpo_v1（新的偏好对齐模型）；
   • 先给 DPO 模型 5～10% 流量；
3. 指标上重点关注：
   • 事实性/完整性/相关性是否没有明显变差；
   • 用户主观满意度（若有反馈机制）是否有提升；
   • 你关心的“偏好维度”（比如结构化程度、专业语气等）是否肉眼更好。

当你在实验中看到：

• 硬错误没上升（甚至略有下降）；
• 偏好维度明显变好（比如人工抽查更顺眼）；

再考虑逐步提升 DPO 模型的流量占比，最终替代旧的 SFT-only 版本。

---

六、如何低成本地先试一把（推荐落地路线）

你完全可以照下面 4 步“最小可行方案”先跑起来：

1. 只对一种典型错误标签做偏好数据

   • 比如先从 incomplete_answer 入手，
     专门让模型学会「说得更全」；
   • 其它标签先不管，简化复杂度。

2. 先用简单规则生成 chosen

   • 不上额外大模型，只用严格一点的 Prompt 让现有模型「补全回答」；
   • 确认 chosen 真的比 rejected 好，再考虑换更强 LLM。

3. 小数据 + 小 DPO

   • 先用几百到一两千条偏好样本，在 dev 环境跑一轮 DPO；
   • 把 DPO 版本接进开发/测试环境，对比同一批问题效果。

4. 接入你的 A/B 框架做线上实验

   • 确认没有明显回归问题后，再考虑跑大规模 DPO。

---

七、小结：你现在多了一条“偏好级”的进化路径

把这篇和你前面的文章串起来，你现在已经具备：

1. 行为级：通过 RAG 工作流 + Agent 家族，让系统先“能干活”；
2. 质量级：通过评估 Agent + 质量闭环 + SFT，让系统“少犯错、别乱编”；
3. 偏好级（本篇）：通过质量事件 → 偏好样本 → DPO，让系统“更懂你想要怎样的回答”。

整体演进路径可以归纳成：

原始模型
  └─► 业务 SFT（掌握知识 + 行为约束）
        └─► RAG 工作流（可观测 + 自愈）
              └─► SFT 增量微调（修错）
                    └─► DPO 偏好对齐（更像“你的人”）

你可以先在一个很小的业务场景试点这套「质量事件 → 偏好数据 → DPO」闭环：

• 先只做一种标签（比如“回答不完整”）；
• 用少量样本做一次小 DPO；
• 把新旧模型在同一批问题上的输出贴到文章里，对比结构、完整度和语气。