洞察型评分与 LLM-as-Judge

import json, re
from dataclasses import dataclass
RUBRIC = """
你是评审员。请对下面这份分析输出按四个维度打分（1-5 的整数），
并给出理由。评分标准：
【方向相关性】5=直击核心问题且发现了用户未明说但更重要的问题；
　3=回答了明确提出的问题；1=答非所问或只复述数据。
【证据充分性】5=每个结论都有具体数据/来源，并给出量级或对比基线；
　3=主要结论有依据，个别为推测但已标注；1=结论无依据或使用无量化词。
【可操作性】5=给出具体动作、执行主体、判断标准，可直接排期；
　3=给出方向性建议但未细化；1=只有现象描述没有下一步。
【表达清晰度】5=结构清晰、有摘要、关键数字突出；3=结构基本清楚；
　1=大段文字堆砌找不到重点。
注意：
· 不要因为篇幅长而给高分，信息密度高才给高分。
· 不要因为排版漂亮而给高分，格式不计入表达清晰度之外的维度。
· 如果某维度确实无法判断，给 0 并在理由里说明。
只输出 JSON：
{"direction":n,"evidence":n,"actionable":n,"clarity":n,
 "reason":"一句话理由","blocking":"最严重的短板维度名或 null"}
"""
@dataclass
class Verdict:
    direction: int
    evidence: int
    actionable: int
    clarity: int
    reason: str
    blocking: str | None = None
    @property
    def weighted(self) -> float:
        """加权：方向相关性权重最高——方向错了，其余努力全部作废"""
        w = {"direction": 0.35, "evidence": 0.22, "actionable": 0.25, "clarity": 0.18}
        return (self.direction * w["direction"] + self.evidence * w["evidence"]
                + self.actionable * w["actionable"] + self.clarity * w["clarity"])
    def passed(self) -> bool:
        """一票否决：可操作性过低 → 直接判定不合格，不用平均分"""
        if self.actionable <= 2:
            return False
        return self.weighted >= 3.5
    def explain(self) -> str:
        return (f"方向 {self.direction} 证据 {self.evidence} "
                f"动作 {self.actionable} 表达 {self.clarity}　"
                f"加权 {self.weighted:.2f}　"
                f"{'通过' if self.passed() else '不通过'}"
                f"{f'（卡在 {self.blocking}）' if self.blocking else ''}")
def judge(output: str, call_model) -> Verdict:
    raw = call_model(RUBRIC, output, temperature=0)
    data = parse_lenient(raw)
    return Verdict(**{k: data[k] for k in
                      ("direction", "evidence", "actionable", "clarity", "reason")},
                   blocking=data.get("blocking"))
def parse_lenient(raw: str) -> dict:
    """宽容解析：模型常把 JSON 包在代码块里，或带尾随逗号"""
    text = raw.strip()
    text = re.sub(r"^```(?:json)?|```$", "", text, flags=re.M).strip()
    text = re.sub(r",(\s*[}\]])", r"\1", text)     # 去掉尾随逗号
    if "{" in text:
        text = text[text.index("{"): text.rindex("}") + 1]
    return json.loads(text)
def judge_stable(output: str, call_model, repeats: int = 2) -> tuple[Verdict, dict]:
    """
    稳定性检查：同一份输出多次打分。
    方差大的样本不能进入自动决策 —— 它们必须转人工。
    """
    verdicts = [judge(output, call_model) for _ in range(repeats)]
    scores = [v.weighted for v in verdicts]
    spread = max(scores) - min(scores)
    meta = {
        "spread": spread,
        "reliable": spread <= 0.8,     # 分差超过 0.8 视为不可靠
        "verdicts": [v.explain() for v in verdicts],
    }
    # 取中位数（比均值更抗单次异常）
    scores_sorted = sorted(scores)
    median = scores_sorted[len(scores_sorted) // 2]
    best = min(verdicts, key=lambda v: abs(v.weighted - median))
    return best, meta
def pairwise_stable(a: str, b: str, call_model) -> dict:
    """
    两两对比必须交换顺序跑两次。
    两次结论不一致 → 判平局，而不是选一个。
    """
    first = call_model(RUBRIC + "\n请判断 A 与 B 哪个更好。", f"【A】{a}\n【B】{b}", temperature=0)
    second = call_model(RUBRIC + "\n请判断 A 与 B 哪个更好。", f"【A】{b}\n【B】{a}", temperature=0)
    # 解析出「选了谁」，交换后应指向同一个原始样本
    pick1 = "A" if "A" in first[:20] else "B"
    pick2 = "B" if "A" in second[:20] else "A"      # 第二次的位置已交换
    return {
        "consistent": pick1 == pick2,
        "winner": pick1 if pick1 == pick2 else "tie",
        "note": "两次顺序一致的结论才可采信；否则记为平局",
    }