评测集建设与防过拟合

from dataclasses import dataclass, field, asdict
from typing import Literal
import json, random, hashlib
Layer = Literal["smoke", "regression", "boundary", "adversarial"]
@dataclass
class EvalCase:
    id: str
    layer: Layer
    input: str
    # 期望行为（可判定），而不是期望文本
    expect_tool: str | None = None
    expect_args: dict | None = None
    expect_refuse: bool = False        # 是否应该拒绝
    expect_clarify: bool = False       # 是否应该追问澄清
    # 溯源信息：没有这个字段的样本是无法维护的
    origin: str = ""                   # 来源：工单号 / trace id / 人工设计
    reason: str = ""                   # 为什么加入：xxx 场景下会崩
    added_at: str = ""
    tags: list[str] = field(default_factory=list)
    def is_judgeable(self) -> bool:
        """
        可判定性检查：一条样本如果既没有期望工具也没有明确的拒绝/澄清要求，
        就只能靠人读输出判断 —— 这类样本不能进回归集（无法自动化）。
        """
        return bool(self.expect_tool or self.expect_refuse or self.expect_clarify)
class Dataset:
    def __init__(self):
        self.cases: dict[str, EvalCase] = {}
        # 保留池划分：用稳定的哈希决定，保证多次运行划分一致
        self.holdout_ratio = 0.2
    def add(self, case: EvalCase) -> str:
        if not case.is_judgeable():
            return "rejected: 样本不可判定，请补充期望行为（工具/拒绝/澄清）"
        if not case.reason:
            return "rejected: 缺少加入原因，无法维护"
        self.cases[case.id] = case
        return "added"
    def holdout_id(self, case_id: str) -> bool:
        """
        用内容哈希划分，而不是随机数 ——
        随机数会导致每次运行划分不同，分数无法比较。
        """
        h = int(hashlib.sha256(case_id.encode()).hexdigest()[:8], 16)
        return (h % 100) < self.holdout_ratio * 100
    def split(self) -> tuple[list[EvalCase], list[EvalCase]]:
        dev = [c for c in self.cases.values() if not self.holdout_id(c.id)]
        hold = [c for c in self.cases.values() if self.holdout_id(c.id)]
        return dev, hold
    def by_layer(self, layer: Layer) -> list[EvalCase]:
        return [c for c in self.cases.values() if c.layer == layer]
    def health(self) -> dict:
        """健康度报告：用来发现评测集自身的问题"""
        layers = {l: len(self.by_layer(l)) for l in
                  ("smoke", "regression", "boundary", "adversarial")}
        no_origin = [c.id for c in self.cases.values() if not c.origin]
        return {
            "total": len(self.cases),
            "layers": layers,
            "missing_origin": len(no_origin),        # 无溯源的样本无法维护
            "advice": (
                "冒烟集偏少，建议至少 10 条" if layers["smoke"] < 10 else
                "对抗集缺失，安全类问题不会被评测覆盖" if layers["adversarial"] == 0 else
                "结构基本健康"
            ),
        }
    def to_json(self) -> str:
        return json.dumps([asdict(c) for c in self.cases.values()],
                          ensure_ascii=False, indent=1)
    @classmethod
    def from_json(cls, text: str) -> "Dataset":
        d = cls()
        for item in json.loads(text):
            d.cases[item["id"]] = EvalCase(**item)
        return d
def from_badcase(trace_id: str, expect_tool: str, expect_args: dict,
                 reason: str, layer: Layer = "regression") -> EvalCase:
    """从线上 badcase 构造样本的标准入口"""
    return EvalCase(
        id=f"bc-{trace_id}", layer=layer,
        input="",                    # 由 trace 回填（已脱敏）
        expect_tool=expect_tool, expect_args=expect_args,
        origin=f"trace:{trace_id}", reason=reason,
    )