Datawhale学习笔记：LLM后训练之目标诊断与阶段准入

Jqz2005523

257人浏览 · 2026-10-11 23:54:53

Jqz2005523 · 2026-10-11 23:54:53 发布

从失败证据到训练分支：LLM 后训练「目标诊断与阶段准入」学习笔记

学习来源

：DataWhale 社区・llm-algo-leetcode 仓库（202610 后训练优化营・Task 0）
教程仓库：

https://github.com/datawhalechina/llm-algo-leetcode
Task 0 Issue：

https://github.com/datawhalechina/llm-algo-leetcode/issues/190
配套课程：DeepLearning.AI — Post-Training LLMs：

https://www.deeplearning.ai/courses/fine-tuning-and-reinforcement-learning-for-llms-intro-to-post-training
本文按 Task 0 推荐学习顺序组织：

稳定失败证据 → 根因定位 → 最小干预 → 数据、反馈与评测准入 → 训练分支

，并附可直接运行的决策引擎代码

0. 写在前面：Task 0 到底在回答什么

模型上线后表现不好，第一反应不应该是 “赶紧微调”。Task 0 把 “要不要训练、训练到什么阶段” 变成一条可检查的决策链：

当前失败能否由 Prompt、RAG 或工具解决？

（不修改参数）
若必须更新参数：

应停在 SFT、进入离线偏好，还是需要在线奖励与探索？

核心原则一句话：失败根因决定候选，硬门槛决定候选能否执行，最小干预决定候选顺序。

贯穿全文的例子：你负责一个电商客服机器人，老板说 “用户投诉变多，你赶紧把模型微调一下”。本文所有概念都挂在这个例子（以及一个数学推理助手的例子）上讲。

全流程总览：

1. 先搞懂这几个术语

下文会反复出现一些专业名词，先在这里进行补充讲解。

1.1 失败切片（Failure Slice）

定义：把 “模型在哪些地方失败” 按可枚举的维度切分成的独立单元。每个切片是一个 “场景 / 任务类型 / 样本组” 的最小失败单位。先知道失败长在哪，才能对症下药—— 而不是笼统地说 “模型不行”。

具体例子（电商客服）：把 1000 条客服测试样本按意图切成 4 个切片：

失败切片	样本数	答对	失败率
售后咨询	40	16	60%
物流查询	300	282	6%
价格咨询	350	337	3.7%
活动规则	310	295	4.8%

整体答对率 93%，看起来不错；但 “售后咨询” 切片失败率高达 60%，说明问题高度集中在售后。结论不是 “模型整体不行”，而是 “售后切片是主战场”—— 切片决定了后面所有动作的靶心。

判定标准：

切片要互斥、可枚举、有稳定规模（每个切片样本数不能太少，否则是噪声不是证据）；
粒度太粗定位不到问题，粒度太细统计不可靠 —— 一般按 “业务意图 / 任务类型 / 题型” 来切；
在决策引擎代码里，Evidence.fail_slices（失败切片）和 Evidence.baseline_covers（基线覆盖的切片）就是它的实现形态。

1.2 基线（Baseline）

定义：不经过任何新干预的 “当前版本”（模型权重 + 提示词 + 检索 / 工具链路）在固定评测集上的表现，是衡量一切改进的对照组。没有基线，就无法判断失败到底是模型缺陷还是评测缺口。

具体例子：线上客服模型 v3 + 当前提示模板 = 你的基线。它在 “售后咨询” 切片上正确率只有 40%。现在你加了一段提示词，在同一个 1000 条测试集上复测：

如果售后切片还是 40% → 提示词没有覆盖这个切片（残留失败），才允许继续往下走；
如果变成 90% → 到此为止，不需要训练。

为什么要求 “基线覆盖全部失败切片”：如果基线评测集里根本没有 “售后咨询” 这个切片，那 “售后失败” 可能只是你没测到，而不是模型真不行 —— 这时正确的动作是先补评测、修基线，而不是训模型。

两种常见的基线：

行为基线：Prompt / RAG / 工具方案复测后的表现，用于判断非参数方案是否覆盖全部失败切片；
SFT 基线：微调后的模型版本，是后续 RL 的起点、对照和参考策略来源。

1.3 稳定复现（Reproducible Failure）

定义：在固定输入、固定采样参数（如 temperature=0）、固定评测程序下，失败反复出现，而不是偶发。

具体例子：同一 100 条售后样本连跑 5 次，5 次都错同样 24 条 → 稳定，可以进入干预链；只有 1 次出错 → 可能是采样抖动或评测程序有 bug，先别动模型，去查评测。

代码里对应 reproduce_count >= min_repeat 且 fail_rate >= 0.5。

1.4 独立评测集（Independent Eval Set）

定义：训练数据之外、从未参与训练的 “未见数据” 样本集，用来检验泛化，而不是检验背诵。

具体例子：从 10000 条客服语料中切出 8000 条训练 + 2000 条评测；评测集任何人不得改动、不得混进训练池。如果训练集指标一直涨、但留出的评测集不涨 → 过拟合，先别升级训练，先查数据。

1.5 最小干预（Minimal Intervention）

定义：在能解决问题的方案里，选择侵入性最小、成本最低、最容易回滚的那一个。固定优先序：Prompt → RAG → 工具 → 持续预训练 → SFT → 离线偏好 → 在线 RL。

具体例子：售后答错是因为 “退款政策上个月刚改过” → 更新知识库（RAG）就能解决，完全不需要训练；如果是因为 “模型总爱在回答开头加无意义寒暄，改了 prompt 也压不住” → 这是行为问题，才轮到 SFT。

两个判据（缺一不可）：

机制覆盖：这个方案能否覆盖当前的失败信号？
执行条件：数据、评测、预算、链路是否齐备？

1.6 硬门槛（Hard Gate）

定义：某个候选方案 “能执行” 的最低资产要求；不满足就视为 “未准入 /blocked”，而不是勉强执行。

具体例子：SFT 的硬门槛是 “版本化监督数据 + 冻结评测清单 + GPU 小时预算” 三件套。哪怕你判断 “行为缺口该走 SFT”，只要监督数据还没整理好，决策结果就是 blocked（先补齐硬门槛），而不是硬着头皮开训。

1.7 SFT 给后续阶段留下的三样东西

策略初始化：RL 从 SFT 模型的权重开始训练（π_θ 的初始值），而不是从预训练模型开始；
稳定基线：SFT 版本作为对照，衡量 RL 是否真的变好、是否跑偏；
参考策略 π_ref：DPO 损失里做比值分母、PPO 的 KL 正则参照，防止新策略偏离太远。

1.8 偏好对（Preference Pair）

定义：(prompt, chosen, rejected) 三元组：同一个问题下，人类 / 评审认为 “更好” 的回答（chosen）和 “更差” 的回答（rejected）。

具体例子：售后问题，回答 A 直接给出退款路径和时限（chosen），回答 B 绕圈子不解决问题（rejected）。DPO 学习的目标就是 “提高生成 A 的概率、降低生成 B 的概率”。

注意：偏好对 ≠ 示范数据。示范数据教模型 “怎么做”，偏好对教模型 “哪个更好”—— 这就是 SFT 与 DPO 的信号差异。

1.9 可验证奖励与奖励退化

可验证奖励（Verifiable Reward）：不用人类打分、能由程序 / 规则自动判定对错的奖励：数学最终答案是否等于参考答案、代码是否通过单元测试、Agent 是否成功完成任务、JSON 是否符合 schema。
奖励退化（Reward Degeneration）：奖励信号 “失去区分度”—— 所有回答都得同样的分（全 0 或全 1）、或奖励分数与独立评测成功率背离。此时先修奖励 / 验证器 / 采样分布，再谈训练，否则 RL 会把模型带偏。

1.10 质量平台期（Plateau）

定义：继续增加监督数据或训练轮数，独立评测指标不再明显提升。

具体例子：SFT 数据从 5k → 10k → 20k，独立评测 acc 从 0.85 → 0.88 → 0.881，后面基本不动 —— 说明 “示范学习” 的收益吃完了。此时若仍有稳定残余错误，且错误属于 “需要判断哪个更好” 或 “需要探索新行为”，才考虑升级 RL。

1.11 版本化监督数据 / 去重 / 去污染 / 冻结评测清单

版本化：数据按版本管理（data_v1、data_v2…），记录来源、清洗规则、筛选标准，保证 “这次训练用的数据” 随时可复查；
去重：删除重复样本，避免同一题反复出现导致过拟合；
去污染：检查训练数据是否混入了评测集样本 —— 否则 “评测提升” 是背答案，不是泛化；
冻结评测清单：训练期间，评测集、指标、判分程序一律不许改动，否则前后结果无法对比。

1.12 GPU 小时预算与回滚检查点

GPU 小时预算：训练前锁定 “这次允许花多少算力 / 多少钱”（如 40 卡时 A100），既是准入证据，也防止训练失控；
回滚检查点（Rollback Checkpoint）：训练前的模型权重备份。RL 在线探索可能让模型变差，随时能切回上一个可用版本。

2. 背景：预训练 vs 后训练

两者复用同一套底层机制（前向 / 反向传播、优化器、训练状态），差别集中在数据对象、学习信号、采用证据三件事上：

对比维度	预训练（Pre-training）	后训练（Post-training）
核心目标	大规模通用语料上学 “下一词元预测”，获得语言规律与世界知识	把基础模型接到任务接口、人类偏好和环境反馈上
数据	万亿级 token，无标注	几千～几十万条：指令 - 回答、偏好对、带奖励的轨迹
学习信号	自监督 next-token 交叉熵	SFT 监督交叉熵 / DPO 偏好边际 / PPO 优势 / RLVR 可验证奖励
标注	无	需要人工筛选与质量把关
算力	几千卡跑数月	单机几小时～几天
典型产物	Base Model（只会续写）	Instruct / Chat / Reasoning Model
验收证据	通用能力、训练稳定性、规模效率	任务质量、安全、鲁棒性、奖励一致性

一句话：

预训练打地基，后训练做装修、改 bug、对齐行为。

后训练同样可以注入领域知识（持续预训练）、强化推理与工具使用，所以它不是 “知识 vs 行为” 的绝对二分，而是 “目标与信号” 的切换。

3. 第一步：稳定失败证据（Evidence Gate）

没有稳定证据之前，一切干预都是拍脑袋。进入干预链的前提是全部满足：

可测目标 + 目标阈值：指标明确（acc /pass@1 /format_ok），验收门槛明确。例：售后切片目标正确率 ≥ 90%，当前只有 40%；
稳定复现：同一固定工作负载下，失败重复次数达到下限、失败率达到门槛（见 1.3）；
失败切片：明确哪些场景 / 任务 / 样本组持续失败（见 1.1）；
独立评测：能检验未见数据上的泛化（见 1.4）；
基线覆盖全部失败切片：当前基线必须覆盖所有切片，否则修基线而不是训模型（见 1.2）。

代码实现（Python 标准库，可直接运行）：

from dataclasses import dataclass
from typing import Dict, List


@dataclass
class Evidence:
    task: str                       # 任务名
    metric: str                     # 可测指标，如 acc / pass@1 / format_ok
    target_threshold: float         # 目标阈值，如 0.90
    current: float                  # 当前指标
    fail_slices: List[str]          # 失败切片（场景 / 任务 / 样本组）
    baseline_covers: List[str]      # 基线覆盖的失败切片
    min_repeat: int = 3             # 稳定复现的最小重复次数
    fail_rate: float = 0.0          # 失败率（同一固定工作负载下）
    reproduce_count: int = 0        # 已复现次数
    independent_eval: bool = False  # 是否有独立评测集（未见数据泛化）


def evidence_gate(ev: Evidence) -> Dict[str, object]:
    """4 项全过，失败证据才算"稳定"，才允许进入干预链。"""
    checks = {
        "可测目标+阈值": ev.metric != "" and ev.target_threshold > 0,
        "当前未达标": ev.current < ev.target_threshold,
        "稳定复现": ev.reproduce_count >= ev.min_repeat and ev.fail_rate >= 0.5,
        "失败切片非空": len(ev.fail_slices) > 0,
        "独立评测可用": ev.independent_eval,
        "基线覆盖全部切片": all(s in ev.baseline_covers for s in ev.fail_slices),
    }
    return {"ready": all(checks.values()), "checks": checks,
            "gap": round(ev.target_threshold - ev.current, 4)}

典型反例

：只在 2 次运行里见过一次错误、没有独立评测集、失败率 20% ——

evidence_gate

返回

ready=False

，此时正确的动作是

先稳定证据

（补评测、修复现、加切片），而不是启动任何训练。

4. 第二步：根因定位：知识 / 行为 / 偏好 / 探索

证据稳定后，要回答 “失败到底缺什么”。Task 0 把根因分成四类，每类根因对应不同的候选方案族：

根因	主要失败信号	首选候选	为什么	具体例子
外部 / 时效知识缺失	事实错误、最新信息、私有知识、需要溯源	RAG / 工具	知识在库不在参数，可更新、可溯源	售后答错 “最新退款政策”（政策上个月刚改）
内部知识缺口	领域概念从未见过、表示层缺失	持续预训练	需要把领域表示写进参数	医学助手从不认识 “司美格鲁肽” 这类领域词
行为缺口	格式乱、不遵循指令、风格不符、任务接口不稳	SFT	用高质量示范固化行为，是最小的参数干预	客服总在回答开头加无意义寒暄，改了 prompt 也压不住
偏好权衡缺口	回答 “能用” 但不够好，需要 “哪个更好” 的判断	离线偏好优化（DPO）	偏好对直接优化，无需在线探索	数学助手答案对但过程冗长，用户更喜欢简洁推导
探索缺口	需要新行为、多步推理、奖励最大化、可验证结果	在线 RL（PPO / GRPO / RLVR）	需要策略在环境中试错与探索	竞赛数学要尝试多种解法路径，用最终答案正确率当奖励

判定代码（互斥优先级：行为 > 偏好 > 探索 > 知识；实际项目以残留失败切片中最集中的信号为准）：

ROOT_CAUSES = {
    "knowledge_external": {"name": "外部/时效知识缺失", "signal": "事实错误、最新信息、私有知识",
                           "candidate": "rag"},
    "knowledge_internal": {"name": "内部知识缺口（领域表示）", "signal": "领域概念从未见过",
                           "candidate": "continued_pretraining"},
    "behavior":           {"name": "行为缺口", "signal": "格式乱、不遵循指令、风格不符",
                           "candidate": "sft"},
    "preference":         {"name": "偏好权衡缺口", "signal": "回答能用但不够好",
                           "candidate": "offline_preference"},
    "exploration":        {"name": "探索缺口", "signal": "需要新行为/多步推理/奖励最大化",
                           "candidate": "online_rl"},
}


def classify_root_cause(signals: Dict[str, bool]) -> Dict[str, str]:
    priority = ["behavior", "preference", "exploration",
                "knowledge_internal", "knowledge_external"]
    for kind in priority:
        if signals.get(kind, False):
            rc = ROOT_CAUSES[kind]
            return {"kind": kind, "name": rc["name"], "signal": rc["signal"],
                    "candidate": rc["candidate"]}
    return {"kind": "unknown", "name": "根因不明", "signal": "证据不足，先回到证据稳定",
            "candidate": "no_entry"}

5. 第三步：最小干预：先试不改参数的方案

固定优先序：Prompt → RAG → 工具 → 持续预训练 → SFT → 离线偏好 → 在线 RL。

缺口是表达、格式、指令歧义 → 首选 Prompt：零训练成本、迭代最快，永远先试；
缺口是外部 / 时效 / 私有知识 → 首选 RAG：知识在库不在参数，重点检查检索命中与引用正确性；
缺口是实时查询、精确计算、外部动作 → 首选工具（API / 计算器 / 验证器）：参数学不会实时事实，也做不可靠算术；
缺口是模型内部行为（风格、策略、格式习惯） → 才轮到 SFT：只有改参数才能固化行为。

关键纪律：走 SFT 必须意味着 Prompt 和 RAG 都没覆盖全部失败信号。非参数方案要在 “同一工作负载、同一验收门槛” 下复测，而不是 “好像试过”。

维度	Prompt	RAG	SFT
本质	推理时给模型的指令和上下文	检索外部知识拼进 Prompt	用标注数据继续训练模型权重
改什么	不改变模型，只改变输入	不改变模型，改变输入中的知识	改变模型本身的行为模式
成本	最低，改文字即可	中等，需建知识库 + 检索链路	较高，需标注数据 + 训练资源
擅长解决	输出格式、角色设定、简单约束	私有知识问答、需溯源的场景	领域话术、固定格式、行为习惯
知识更新	修改上下文即可	更新知识库即可	需要再训练或参数编辑

6. 第四步：必须更新参数时，如何选训练分支

如果非参数方案已确认无法覆盖全部失败切片，才进入 “改参数” 决策。失败根因决定候选；硬门槛决定候选能否执行。

候选	针对根因	进入执行前的硬门槛	方法边界
持续预训练	内部知识缺口	领域语料 + GPU 预算 + 独立评测	补表示层知识；成本高，需防遗忘
SFT	行为缺口	版本化监督数据 + 冻结评测清单 + 非零预算	最小参数干预；是后续 RL 的基线
离线偏好（DPO）	偏好权衡	SFT 基线 + 偏好对数据 + 评测	离线、轻量；不探索新行为
在线 RL（PPO / GRPO / RLVR）	探索缺口	SFT 基线 + 奖励信号 + 预算 + 回滚检查点	在线探索与奖励最大化；成本最高

三条铁律：

SFT 是 “改参数” 里最轻的一步：没有 SFT 基线时，行为缺口直接进 SFT；格式 / 指令遵循还不稳定、或独立质量仍随高质量示范提升时，不要跨到 RL；
升级 RL 的条件是 “反馈信号换类型”，不是 loss 数值：

有可复查的偏好对（chosen /rejected）且无需探索 → 离线偏好优化（DPO）；
数学答案 / 代码测试 / 环境终态可验证，候选间奖励可区分 → RLVR（GRPO / PPO）；
有学习型奖励模型且需要在线探索多步轨迹 → PPO；
没有任何新反馈信号 → 停在 SFT（stop_at_sft）。

奖励退化要前置检查：奖励大量同分、全正 / 全负，或与独立成功率背离 → 先修奖励 / 验证器 / 采样分布，再谈训练。

SFT 给后续阶段稳定留下三样东西：策略初始化、稳定基线和冻结参考策略来源（DPO 的 π_ref、PPO 的 KL 正则都依赖它，见 1.7）。

7. 第五步：数据、反馈与评测准入

进入任何训练分支前，资产必须齐备（硬门槛），否则视为 “未准入”：

训练分支	数据准入	反馈准入	评测准入
SFT	版本化监督数据（去重 / 去污染 / 切分隔离，见 1.11）	无需反馈	独立评测集 + 冻结指标清单 + GPU 小时预算
离线偏好 DPO	偏好对 (prompt, chosen, rejected)，可复查	偏好数据质量稳定	SFT 基线可复查 + 评测
在线 RL	采样 / 轨迹预算	奖励模型 / 验证器 / 规则可靠	评测 + 回滚检查点 + 安全约束

一个常见的误区

：

幻觉不是升级 RL 的充分理由

。模型不知道某个事实，优先 RAG / 工具 / 知识更新；只有当 “示范学习出现瓶颈（平台期）+ 优化目标改变 + 有可靠奖励信号 + 收益覆盖成本” 四个条件同时成立时，才值得从 SFT 升级到 RL。

8. 第六步：训练配套代码（可直接运行）

8.1 完整决策引擎 `posttraining_stage_selector.py`

把第 3～7 章的逻辑实现为一个可运行脚本，覆盖 Task 0 的完整问题链（纯 Python 标准库，无第三方依赖）：

# -*- coding: utf-8 -*-
"""
后训练问题链 · 阶段选择 —— 微调准入与最小干预（Task 0 决策引擎）

对应 DataWhale llm-algo-leetcode 202610 Task 0：
    稳定失败证据 -> 知识/行为/偏好/探索根因 -> 最小干预 -> 数据、反馈与评测准入 -> 训练分支

纯 Python 标准库实现，无第三方依赖，Python 3.8+ 可直接运行。
运行：python posttraining_stage_selector.py
"""

from __future__ import annotations

from dataclasses import dataclass, field
from typing import Dict, List


# ===========================================================================
# 第一步：稳定失败证据（Evidence Gate）
#    可测目标 + 目标阈值 + 稳定复现 + 失败切片 + 独立评测 + 基线覆盖全部切片
# ===========================================================================

@dataclass
class Evidence:
    task: str                       # 任务名
    metric: str                     # 可测指标，如 acc / pass@1 / format_ok
    target_threshold: float         # 目标阈值，如 0.90
    current: float                  # 当前指标
    fail_slices: List[str]          # 失败切片（场景 / 任务 / 样本组）
    baseline_covers: List[str]      # 基线覆盖的失败切片
    min_repeat: int = 3             # 稳定复现的最小重复次数
    fail_rate: float = 0.0          # 失败率（同一固定工作负载下）
    reproduce_count: int = 0        # 已复现次数
    independent_eval: bool = False  # 是否有独立评测集（未见数据泛化）


def evidence_gate(ev: Evidence) -> Dict[str, object]:
    """4 项全过，失败证据才算"稳定"，才允许进入干预链。"""
    checks = {
        "可测目标+阈值": ev.metric != "" and ev.target_threshold > 0,
        "当前未达标": ev.current < ev.target_threshold,
        "稳定复现": ev.reproduce_count >= ev.min_repeat and ev.fail_rate >= 0.5,
        "失败切片非空": len(ev.fail_slices) > 0,
        "独立评测可用": ev.independent_eval,
        "基线覆盖全部切片": all(s in ev.baseline_covers for s in ev.fail_slices),
    }
    ready = all(checks.values())
    return {
        "ready": ready,
        "checks": checks,
        "gap": round(ev.target_threshold - ev.current, 4),
    }


# ===========================================================================
# 第二步：根因定位（知识 / 行为 / 偏好 / 探索）
#    根因决定候选：同一类失败信号，只能由对应候选族覆盖
# ===========================================================================

ROOT_CAUSES = {
    "knowledge_external": {
        "name": "外部/时效知识缺失",
        "signal": "事实错误、最新信息、私有知识、需要溯源",
        "candidate": "rag",
        "cheaper_first": ["prompt", "rag", "tool"],
    },
    "knowledge_internal": {
        "name": "内部知识缺口（领域表示）",
        "signal": "领域概念从未见过、表示层缺失",
        "candidate": "continued_pretraining",
        "cheaper_first": ["prompt", "rag"],
    },
    "behavior": {
        "name": "行为缺口",
        "signal": "格式乱、不遵循指令、风格不符、任务接口不稳",
        "candidate": "sft",
        "cheaper_first": ["prompt", "rag", "tool"],
    },
    "preference": {
        "name": "偏好权衡缺口",
        "signal": "回答「能用」但不够好：需要选择更优答案而非示范答案",
        "candidate": "offline_preference",
        "cheaper_first": ["prompt", "rag", "tool"],
    },
    "exploration": {
        "name": "探索缺口",
        "signal": "需要新行为 / 多步推理 / 奖励最大化 / 可验证结果",
        "candidate": "online_rl",
        "cheaper_first": ["prompt", "rag", "tool"],
    },
}


def classify_root_cause(signals: Dict[str, bool]) -> Dict[str, str]:
    """
    输入四个根因信号，输出主导根因（互斥优先级：行为 > 偏好 > 探索 > 知识）。
    实际项目中优先看"残留失败切片"里最集中的信号类型。
    """
    priority = ["behavior", "preference", "exploration", "knowledge_internal", "knowledge_external"]
    for kind in priority:
        if signals.get(kind, False):
            rc = ROOT_CAUSES[kind]
            return {"kind": kind, "name": rc["name"], "signal": rc["signal"], "candidate": rc["candidate"]}
    return {"kind": "unknown", "name": "根因不明", "signal": "证据不足，先回到证据稳定", "candidate": "no_entry"}


# ===========================================================================
# 第三步：候选方案表 —— 失败根因决定候选；硬门槛决定候选能否执行
# ===========================================================================

@dataclass
class Assets:
    """执行候选所需的资产与资源（硬门槛）。"""
    prompt_template_ok: bool = False      # 可编辑提示模板 + 复测用例
    rag_stack_ok: bool = False            # 知识库 + 检索器 + 引用链路
    tool_stack_ok: bool = False           # API / 计算器 / 验证器可调用
    corpus_ok: bool = False               # 领域语料（持续预训练）
    sft_data_ready: bool = False          # 版本化监督数据（去重/去污染/切分）
    eval_suite_ready: bool = False        # 独立评测集 + 指标 + 冻结清单
    gpu_budget_ready: bool = False        # 非零训练预算
    sft_baseline_ready: bool = False      # 已有可复查的 SFT 基线
    preference_data_ready: bool = False   # 偏好对 (prompt, chosen, rejected)
    reward_signal_ok: bool = False        # 奖励模型 / 验证器 / 规则可计算
    rollback_ok: bool = False             # 回滚检查点 + 安全约束


CANDIDATES: List[Dict[str, object]] = [
    {
        "name": "prompt", "cost": 1, "family": "non_parameter", "root_cause": ["behavior"],
        "gates": ["prompt_template_ok"],
        "boundary": "只改上下文与示例；解决表达/格式/指令表达不充分，不补知识、不固化行为",
    },
    {
        "name": "rag", "cost": 2, "family": "non_parameter", "root_cause": ["knowledge_external"],
        "gates": ["rag_stack_ok"],
        "boundary": "知识在库不在参数；动态更新、可溯源，但依赖检索质量，增加延迟",
    },
    {
        "name": "tool", "cost": 3, "family": "non_parameter", "root_cause": ["knowledge_external"],
        "gates": ["tool_stack_ok"],
        "boundary": "实时查询 / 精确计算 / 外部动作交给 API 与验证器；参数学不会实时事实",
    },
    {
        "name": "continued_pretraining", "cost": 4, "family": "parameter", "root_cause": ["knowledge_internal"],
        "gates": ["corpus_ok", "gpu_budget_ready", "eval_suite_ready"],
        "boundary": "用领域语料补表示层知识；成本高，需防遗忘与灾难性退化",
    },
    {
        "name": "sft", "cost": 5, "family": "parameter", "root_cause": ["behavior"],
        "gates": ["sft_data_ready", "eval_suite_ready", "gpu_budget_ready"],
        "boundary": "用高质量示范固化行为/格式/风格；最小参数干预，是后续 RL 的基线与参考策略来源",
    },
    {
        "name": "offline_preference", "cost": 6, "family": "parameter", "root_cause": ["preference"],
        "gates": ["sft_baseline_ready", "preference_data_ready", "eval_suite_ready"],
        "boundary": "DPO 用偏好对直接优化，离线、轻量；不探索新行为，偏好数据质量敏感",
    },
    {
        "name": "online_rl", "cost": 7, "family": "parameter", "root_cause": ["exploration"],
        "gates": ["sft_baseline_ready", "reward_signal_ok", "gpu_budget_ready", "rollback_ok"],
        "boundary": "PPO/GRPO/RLVR 在线探索与奖励最大化；需奖励信号可靠、预算充足、可回滚",
    },
]


def viable_candidates(root_kind: str, assets: Assets) -> List[Dict[str, object]]:
    """根因决定候选 + 硬门槛决定能否执行：返回按成本升序的可用候选。"""
    out = []
    for c in CANDIDATES:
        if root_kind not in c["root_cause"]:
            continue
        if all(getattr(assets, g) for g in c["gates"]):
            out.append(c)
    return out


# ===========================================================================
# 第四步：训练分支选择（最小干预 + 数据/反馈/评测准入）
# ===========================================================================

@dataclass
class SFTStatus:
    """SFT 完成度：格式/指令遵循稳定 + 独立质量进入平台期。"""
    format_stable: bool = False       # 格式与指令遵循已稳定
    quality_plateau: bool = False     # 继续加示范数据不再明显提升
    residual_is_preference: bool = False  # 残余错误属于"哪个更好"的偏好
    residual_is_exploration: bool = False # 残余错误需要探索新行为/多步推理
    verifiable_reward: bool = False   # 数学答案/代码测试/环境终态可验证
    learned_reward: bool = False      # 有学习型奖励模型
    reward_degenerate: bool = False   # 奖励大量同分/全正全负/与独立成功率背离


def decide_next_step(ev: Evidence, signals: Dict[str, bool], assets: Assets,
                     sft: SFTStatus | None = None,
                     non_param_covered_slices: List[str] | None = None) -> Dict[str, object]:
    """
    主决策函数，按 Task 0 问题链顺序执行：
      稳定证据 -> 根因 -> 先试不改参数的方案 -> 必须改参数时选训练分支。
    返回 (stage, reason, gates, candidates)。

    non_param_covered_slices：提示词/RAG/工具在"同一工作负载、同一验收门槛"
    下复测后，能覆盖的失败切片列表。只有覆盖全部失败切片，才选中非参数方案。
    """
    # ① 证据门
    eg = evidence_gate(ev)
    if not eg["ready"]:
        return {"stage": "no_entry", "reason": "失败证据未稳定，先补齐：%s" % str(eg["checks"]),
                "candidates": [], "evidence": eg}

    # ② 根因
    rc = classify_root_cause(signals)
    if rc["kind"] == "unknown":
        return {"stage": "no_entry", "reason": rc["signal"], "candidates": [], "evidence": eg}

    # ③ 最小干预：先试不修改参数的候选（同一工作负载、同一验收门槛复测）
    non_param = viable_candidates(rc["kind"], assets)
    covered = set(non_param_covered_slices or [])
    if non_param and all(s in covered for s in ev.fail_slices):
        chosen = non_param[0]
        return {"stage": chosen["name"], "reason": "最小干预优先（不改参数）且已覆盖全部失败切片：%s" % chosen["boundary"],
                "candidates": [c["name"] for c in non_param], "evidence": eg}

    # ④ 必须更新参数：停在 SFT / 离线偏好 / 在线奖励与探索
    if rc["kind"] == "knowledge_internal":
        cands = viable_candidates("knowledge_internal", assets)
        if cands:
            return {"stage": cands[0]["name"], "reason": "知识缺口 -> 领域表示需参数更新：%s" % cands[0]["boundary"],
                    "candidates": [c["name"] for c in cands], "evidence": eg}
        return {"stage": "blocked", "reason": "知识缺口但语料/预算/评测未齐，先补齐硬门槛",
                "candidates": [], "evidence": eg}

    if rc["kind"] == "behavior":
        # SFT 是"改参数"里最轻的一步；没有 SFT 基线就先进 SFT
        if not assets.sft_baseline_ready:
            cands = viable_candidates("behavior", assets)
            if not cands:
                return {"stage": "blocked", "reason": "行为缺口但 SFT 硬门槛未齐：需要 监督数据+评测+预算",
                        "candidates": [], "evidence": eg}
            return {"stage": "sft", "reason": "行为缺口 + 无 SFT 基线 -> 先上 SFT（最小参数干预）",
                    "candidates": ["sft"], "evidence": eg}
        # 已有 SFT 基线：按 SFT 完成度 + 反馈信号类型升级
        s = sft or SFTStatus()
        if not s.format_stable or not s.quality_plateau:
            return {"stage": "continue_sft", "reason": "SFT 未完成：格式/指令遵循不稳或质量仍在提升 -> 补数据继续 SFT",
                    "candidates": ["sft"], "evidence": eg}
        if s.reward_degenerate:
            return {"stage": "fix_reward", "reason": "奖励退化（同分/全正全负/与独立成功率背离）-> 先修奖励与验证器",
                    "candidates": [], "evidence": eg}
        if s.verifiable_reward and s.residual_is_exploration:
            cands = viable_candidates("exploration", assets)
            if cands:
                return {"stage": "rlvr", "reason": "可验证奖励 + 需探索 -> RLVR(GRPO/PPO) 在线优化",
                        "candidates": ["online_rl"], "evidence": eg}
        if s.learned_reward and s.residual_is_exploration:
            cands = viable_candidates("exploration", assets)
            if cands:
                return {"stage": "ppo", "reason": "学习型奖励模型 + 需在线探索 -> PPO",
                        "candidates": ["online_rl"], "evidence": eg}
        if s.residual_is_preference:
            cands = viable_candidates("preference", assets)
            if cands:
                return {"stage": "dpo", "reason": "偏好对可复查 + 无需探索 -> 离线偏好优化 DPO",
                        "candidates": ["offline_preference"], "evidence": eg}
        return {"stage": "stop_at_sft", "reason": "无新反馈信号（偏好对/奖励模型/验证器）-> 停在 SFT",
                "candidates": [], "evidence": eg}

    if rc["kind"] == "preference":
        cands = viable_candidates("preference", assets)
        if not cands:
            return {"stage": "blocked", "reason": "偏好缺口但硬门槛未齐：需要 SFT基线+偏好对+评测",
                    "candidates": [], "evidence": eg}
        return {"stage": "dpo", "reason": "偏好权衡 -> 离线偏好优化 DPO（无需在线探索）",
                "candidates": ["offline_preference"], "evidence": eg}

    if rc["kind"] == "exploration":
        cands = viable_candidates("exploration", assets)
        if not cands:
            return {"stage": "blocked", "reason": "探索缺口但硬门槛未齐：需要 SFT基线+奖励信号+预算+回滚",
                    "candidates": [], "evidence": eg}
        if sft and sft.verifiable_reward:
            return {"stage": "rlvr", "reason": "可验证奖励 -> RLVR(GRPO) 组内相对优势在线优化",
                    "candidates": ["online_rl"], "evidence": eg}
        return {"stage": "ppo", "reason": "学习型奖励 -> PPO 在线策略更新",
                "candidates": ["online_rl"], "evidence": eg}

    return {"stage": "unknown", "reason": "无法判定", "candidates": [], "evidence": eg}


# ===========================================================================
# 配套训练代码示例：SFT 数据契约（标签掩码）与 GRPO 组内优势
# ===========================================================================

def build_sft_sample(instruction: str, response: str) -> Dict[str, List[int]]:
    """
    把 (指令, 输入, 参考回答) 变成 input_ids + labels + loss_mask。
    仅用整数 token 下标示意；真实训练中由 tokenizer + 聊天模板完成。
    原则：只对"有效监督 token（回答部分）"算交叉熵，指令部分掩码掉。
    """
    # 示意 tokenizer：按字符编号，实际使用 HuggingFace tokenizer
    vocab = {ch: i for i, ch in enumerate(set(instruction + response))}
    input_ids = [vocab[ch] for ch in instruction] + [vocab[ch] for ch in response]
    labels = [-100] * len(instruction) + [vocab[ch] for ch in response]  # -100 不参与损失
    loss_mask = [0] * len(instruction) + [1] * len(response)
    return {"input_ids": input_ids, "labels": labels, "loss_mask": loss_mask,
            "n_supervised_tokens": len(response)}


def grpo_group_advantage(rewards: List[float]) -> List[float]:
    """
    GRPO 组内相对优势：A_i = (r_i - mean) / std（同一 prompt 的一组采样）。
    高于组内均值 -> 提高概率；低于 -> 降低概率。无需 Critic。
    """
    mean = sum(rewards) / len(rewards)
    std = (sum((r - mean) ** 2 for r in rewards) / len(rewards)) ** 0.5 or 1.0
    return [round((r - mean) / std, 2) for r in rewards]


# ===========================================================================
# 场景测试
# ===========================================================================

def _mk_evidence(**kw) -> Evidence:
    base = dict(task="数学推理", metric="pass@1", target_threshold=0.90, current=0.72,
                fail_slices=["代数", "几何"], baseline_covers=["代数", "几何"],
                min_repeat=3, fail_rate=0.6, reproduce_count=5, independent_eval=True)
    base.update(kw)
    return Evidence(**base)


def main() -> None:
    # 场景 A：行为缺口（格式乱），Prompt/RAG/工具试过仍残留 -> 进 SFT
    ev_a = _mk_evidence(task="格式化输出", metric="format_ok", target_threshold=0.95, current=0.60)
    assets_a = Assets(prompt_template_ok=True, rag_stack_ok=True, tool_stack_ok=True,
                      sft_data_ready=True, eval_suite_ready=True, gpu_budget_ready=True)
    r_a = decide_next_step(ev_a, {"behavior": True}, assets_a, non_param_covered_slices=[])
    assert r_a["stage"] == "sft", r_a
    print("[A] 行为缺口、外部方案试过仍残留 ->", r_a["stage"], "|", r_a["reason"])

    # 场景 B：外部时效知识缺失，RAG 复测覆盖全部切片 -> 最小干预选中 RAG（不改参数）
    ev_b = _mk_evidence(task="实时新闻问答", metric="fact_accuracy", target_threshold=0.95, current=0.55)
    assets_b = Assets(prompt_template_ok=True, rag_stack_ok=True)
    r_b = decide_next_step(ev_b, {"knowledge_external": True}, assets_b,
                           non_param_covered_slices=ev_b.fail_slices)
    assert r_b["stage"] == "rag", r_b
    print("[B] 外部知识缺口 + RAG 就绪且覆盖全部切片 ->", r_b["stage"], "|", r_b["reason"])

    # 场景 B2：提示模板已覆盖全部失败切片 -> 直接选 Prompt，不进入训练
    ev_b2 = _mk_evidence(task="角色扮演格式", metric="format_ok", target_threshold=0.95, current=0.60)
    assets_b2 = Assets(prompt_template_ok=True)
    r_b2 = decide_next_step(ev_b2, {"behavior": True}, assets_b2,
                            non_param_covered_slices=ev_b2.fail_slices)
    assert r_b2["stage"] == "prompt", r_b2
    print("[B2] 提示模板覆盖全部切片 ->", r_b2["stage"], "|", r_b2["reason"])

    # 场景 C：SFT 完成、格式稳定、质量平台期，出现偏好权衡 -> DPO
    ev_c = _mk_evidence(task="客服话术", metric="prefer_win_rate", target_threshold=0.80, current=0.62)
    assets_c = Assets(sft_baseline_ready=True, preference_data_ready=True, eval_suite_ready=True)
    sft_c = SFTStatus(format_stable=True, quality_plateau=True, residual_is_preference=True)
    r_c = decide_next_step(ev_c, {"preference": True}, assets_c, sft=sft_c, non_param_covered_slices=[])
    assert r_c["stage"] == "dpo", r_c
    print("[C] 偏好权衡 + 偏好对就绪 ->", r_c["stage"], "|", r_c["reason"])

    # 场景 D：数学推理，答案可验证 + 需要探索 -> RLVR（GRPO 组内优势）
    ev_d = _mk_evidence(task="竞赛数学", metric="pass@1", target_threshold=0.90, current=0.72)
    assets_d = Assets(sft_baseline_ready=True, reward_signal_ok=True,
                      gpu_budget_ready=True, rollback_ok=True, eval_suite_ready=True)
    sft_d = SFTStatus(format_stable=True, quality_plateau=True,
                      residual_is_exploration=True, verifiable_reward=True)
    r_d = decide_next_step(ev_d, {"exploration": True}, assets_d, sft=sft_d, non_param_covered_slices=[])
    assert r_d["stage"] == "rlvr", r_d
    print("[D] 可验证奖励 + 需探索 ->", r_d["stage"], "|", r_d["reason"])

    # 场景 E：证据不稳定（复现次数不足）-> 禁止进入干预链
    ev_e = _mk_evidence(reproduce_count=1, fail_rate=0.2)
    r_e = decide_next_step(ev_e, {"behavior": True}, Assets())
    assert r_e["stage"] == "no_entry", r_e
    print("[E] 失败证据未稳定 ->", r_e["stage"], "|", r_e["reason"])

    # 场景 F：SFT 已完成但无新反馈信号 -> stop_at_sft
    ev_f = _mk_evidence(task="摘要生成", metric="rouge", target_threshold=0.60, current=0.52)
    assets_f = Assets(sft_baseline_ready=True, eval_suite_ready=True)
    sft_f = SFTStatus(format_stable=True, quality_plateau=True)
    r_f = decide_next_step(ev_f, {"behavior": True}, assets_f, sft=sft_f, non_param_covered_slices=[])
    assert r_f["stage"] == "stop_at_sft", r_f
    print("[F] 无新反馈信号 ->", r_f["stage"], "|", r_f["reason"])

    # 配套示例：SFT 标签掩码 + GRPO 组内优势
    sample = build_sft_sample("把结果输出为 JSON：", '{"ok": true}')
    print("[G] SFT 样本：监督 token 数 =", sample["n_supervised_tokens"], "| labels 头部掩码 =", sample["labels"][:5])
    rewards = [8.5, 6.8, 4.2, 2.5]
    print("[H] GRPO 组内优势：", grpo_group_advantage(rewards))

    print("\n全部断言通过：决策引擎行为符合 Task 0 问题链。")


if __name__ == "__main__":
    main()

8.2 运行输出（本地 Python 3.13・Windows・全部断言通过）

[A] 行为缺口、外部方案试过仍残留 -> sft | 行为缺口 + 无 SFT 基线 -> 先上 SFT（最小参数干预）
[B] 外部知识缺口 + RAG 就绪且覆盖全部切片 -> rag | 最小干预优先（不改参数）且已覆盖全部失败切片：知识在库不在参数；动态更新、可溯源，但依赖检索质量，增加延迟
[B2] 提示模板覆盖全部切片 -> prompt | 最小干预优先（不改参数）且已覆盖全部失败切片：只改上下文与示例；解决表达/格式/指令表达不充分，不补知识、不固化行为
[C] 偏好权衡 + 偏好对就绪 -> dpo | 偏好权衡 -> 离线偏好优化 DPO（无需在线探索）
[D] 可验证奖励 + 需探索 -> rlvr | 可验证奖励 -> RLVR(GRPO) 组内相对优势在线优化
[E] 失败证据未稳定 -> no_entry | 失败证据未稳定，先补齐：{'可测目标+阈值': True, '当前未达标': True, '稳定复现': False, '失败切片非空': True, '独立评测可用': True, '基线覆盖全部切片': True}
[F] 无新反馈信号 -> stop_at_sft | 无新反馈信号（偏好对/奖励模型/验证器）-> 停在 SFT
[G] SFT 样本：监督 token 数 = 12 | labels 头部掩码 = [-100, -100, -100, -100, -100]
[H] GRPO 组内优势： [1.3, 0.56, -0.56, -1.3]

全部断言通过：决策引擎行为符合 Task 0 问题链。

复现方式

：把 8.1 的完整脚本保存为

posttraining_stage_selector.py

，执行

python posttraining_stage_selector.py

即可得到以上输出。场景 A～F 覆盖六种典型决策路径：进 SFT / 选 RAG / 选 Prompt / 进 DPO / 进 RLVR / 证据不足被拦截 / 停在 SFT。

8.3 配套训练代码示例

以下两个函数已包含在 8.1 的完整脚本中，这里给出使用方式与运行结果。

SFT 数据契约：只对 “有效监督 token” 算交叉熵（标签掩码）

sample = build_sft_sample("把结果输出为 JSON：", '{"ok": true}')
print(sample["n_supervised_tokens"], sample["labels"][:5])
# 输出：12 [-100, -100, -100, -100, -100]   （指令部分全部掩码，只有回答参与损失）

GRPO 组内相对优势（不需要 Critic）

print(grpo_group_advantage([8.5, 6.8, 4.2, 2.5]))
# 输出：[1.3, 0.56, -0.56, -1.3]   （高于组内均值的回答提高概率，低于的降低）

DPO 目标函数（跳过奖励模型，直接用偏好对优化）：

$L^{DPO}(\pi_\theta; \pi_{ref}) = - \mathbb{E}_{(x, y_w, y_l) \sim D} \left[ \log \sigma \left( \beta \log \frac{\pi_\theta(y_w|x)}{\pi_{ref}(y_w|x)} - \beta \log \frac{\pi_\theta(y_l|x)}{\pi_{ref}(y_l|x)} \right) \right]$

PPO 的 clip 目标（限制每次策略更新幅度 + KL 约束）：

$L^{PPO}(\theta) = \mathbb{E} \left[ \min\left( \rho(\theta) A,\ \operatorname{clip}(\rho(\theta), 1-\epsilon, 1+\epsilon) A \right) \right] - \beta\, KL(\pi_\theta \| \pi_{ref})$

9. 概念说明：RLHF / PPO / DPO / GRPO / RLVR 与 RL 四模型

9.1 强化学习中的四个模型

在经典 RLHF-PPO 架构中，通常有四个模型角色：

模型角色	主要功能	是否更新
Actor / Policy（策略模型）	生成回答，实际被训练	更新
Reference（参考模型）	冻结的 SFT 基线，用于 KL 约束 / 偏好参照	冻结
Reward（奖励模型）	对回答质量评分，把人类偏好转为奖励	通常冻结
Critic / Value（价值模型）	估计状态价值，为 PPO 计算优势提供基线	通常更新

注意：GRPO

不需要 Critic

，用组内奖励的均值和标准差估计相对优势，省显存、省训练链路。还要区分

π_old

（生成当前训练数据的策略，算概率比用）与

π_ref

（参考策略，KL 正则用）不是同一个概念。

9.2 RLHF / PPO / DPO / GRPO / RLVR 概念

概念	全称	一句话理解
RLHF	RL from Human Feedback	一种对齐范式：用人类偏好反馈对齐模型，经典实现是 PPO；不是单一算法
PPO	Proximal Policy Optimization	在线策略优化算法：clip 限制更新幅度 + KL 约束，稳定但链路重（Actor+Ref+Reward+Critic）
DPO	Direct Preference Optimization	离线偏好优化：跳过奖励模型，直接用 (prompt, chosen, rejected) 偏好对优化，轻量
GRPO	Group Relative Policy Optimization	去掉 Critic 的策略优化：同一 prompt 采一组答案，用组内相对奖励估计优势
RLVR	RL with Verifiable Rewards	奖励信号来源范式：用可自动验证的奖励（数学答案、单测、终态）替代主观打分，常与 GRPO/PPO 组合

一句话关系图：

RLHF（范式，人类反馈）
 ├── 在线：PPO（Actor+Critic，clip+KL）
 └── 离线：DPO（偏好对直接优化，无需奖励模型）
策略优化去 Critic：GRPO（组内相对优势）
奖励更客观：RLVR（可验证奖励，配 PPO 或 GRPO 用）

RLVR 的关键不是 “必须有 ground truth”，而是奖励能通过可靠的验证机制自动计算—— 数学答案是否等于参考答案、代码是否通过单测、Agent 任务是否成功。

参考资料

DataWhale・llm-algo-leetcode 仓库：https://github.com/datawhalechina/llm-algo-leetcode
Task 0 Issue（目标诊断与阶段准入）：https://github.com/datawhalechina/llm-algo-leetcode/issues/190
DeepLearning.AI · Post-Training LLMs：https://www.deeplearning.ai/courses/fine-tuning-and-reinforcement-learning-for-llms-intro-to-post-training
PPO — Proximal Policy Optimization Algorithms：https://arxiv.org/abs/1707.06347
DPO — Direct Preference Optimization：https://arxiv.org/abs/2305.18290
DeepSeekMath（GRPO）：https://arxiv.org/abs/2402.03300

DAMO开发者矩阵

DAMO开发者矩阵，由阿里巴巴达摩院和中国互联网协会联合发起，致力于探讨最前沿的技术趋势与应用成果，搭建高质量的交流与分享平台，推动技术创新与产业应用链接，围绕“人工智能与新型计算”构建开放共享的开发者生态。

更多推荐

纸机干部粉尘最贵的不是设备，是惰化系统没人管

DAMO开发者矩阵

一问一答：复古理发厅数字人为什么最终走上了_端到端具身交互_

DAMO开发者矩阵

几百块的桌面机器人只能用手柄在跟前遥控？给 Microduck 双足机器人加一层远程遥控面板，再用 cpolar 让在外地的家人也能逗它走两步

DAMO开发者矩阵

所有评论(0)

查看更多评论

Jqz2005523

@Jqz2005523

已为社区贡献3条内容

Datawhale学习笔记：LLM后训练之目标诊断与阶段准入

Jqz2005523

从失败证据到训练分支：LLM 后训练 「目标诊断与阶段准入」学习笔记

0. 写在前面：Task 0 到底在回答什么

1. 先搞懂这几个术语

1.1 失败切片（Failure Slice）

1.2 基线（Baseline）

1.3 稳定复现（Reproducible Failure）

1.4 独立评测集（Independent Eval Set）

1.5 最小干预（Minimal Intervention）

1.6 硬门槛（Hard Gate）

1.7 SFT 给后续阶段留下的三样东西

1.8 偏好对（Preference Pair）

1.9 可验证奖励与奖励退化

1.10 质量平台期（Plateau）

1.11 版本化监督数据 / 去重 / 去污染 / 冻结评测清单

1.12 GPU 小时预算与回滚检查点

2. 背景：预训练 vs 后训练

3. 第一步：稳定失败证据（Evidence Gate）

4. 第二步：根因定位：知识 / 行为 / 偏好 / 探索

5. 第三步：最小干预：先试不改参数的方案

6. 第四步：必须更新参数时，如何选训练分支

7. 第五步：数据、反馈与评测准入

8. 第六步：训练配套代码（可直接运行）

8.1 完整决策引擎 posttraining_stage_selector.py

8.2 运行输出（本地 Python 3.13・Windows・全部断言通过）

8.3 配套训练代码示例

9. 概念说明：RLHF / PPO / DPO / GRPO / RLVR 与 RL 四模型

9.1 强化学习中的四个模型

9.2 RLHF / PPO / DPO / GRPO / RLVR 概念

所有评论(0)

温馨提示：您尚未绑定手机号

Jqz2005523

从失败证据到训练分支：LLM 后训练「目标诊断与阶段准入」学习笔记

8.1 完整决策引擎 `posttraining_stage_selector.py`