为什么二元对错评估会毁掉你的大模型生产系统

import json
import re

def parse_truncated_json(raw_string: str) -> dict:
    """
    清洗并尝试解析可能被 LLM 截断的 JSON 字符串。
    """
    # 提取 Markdown 代码块中的 JSON 内容
    json_match = re.search(r'```json\s*(.*?)\s*```', raw_string, re.DOTALL)
    if json_match:
        content = json_match.group(1)
    else:
        content = raw_string.strip()

    # 统计括号数量以修复截断问题
    open_braces = content.count('{')
    close_braces = content.count('}')

    # 如果括号不匹配，尝试在末尾补齐缺失的右括号
    if open_braces > close_braces:
        content += '}' * (open_braces - close_braces)

    try:
        return json.loads(content)
    except json.JSONDecodeError as e:
        # 如果解析仍然失败，使用正则提取关键字段作为兜底
        verdict = "unknown"
        if "\"verdict\"" in content:
            verdict_match = re.search(r'"verdict"\s*:\s*"([^"]+)"', content)
            if verdict_match:
                verdict = verdict_match.group(1)
        return {"verdict": verdict, "error": f"Failed to parse: {str(e)}"}

def evaluate_response(candidate_output: str, ground_truth: dict) -> str:
    """
    对比候选输出与标准答案，并评估错误的严重程度级别。
    """
    candidate = parse_truncated_json(candidate_output)

    candidate_verdict = candidate.get("verdict", "").lower().strip()
    expected_verdict = ground_truth.get("verdict", "").lower().strip()

    # 规则 1：在分析具体数据负载前，优先评估 verdict（判定）字段
    if candidate_verdict == "needs_confirmation" and expected_verdict == "needs_confirmation":
        return "HARMLESS"

    if candidate_verdict == "confirm" and expected_verdict == "needs_confirmation":
        # 模型在需要确认的情况下直接进行了确认，属于风险行为
        return "RISKY"

    if candidate_verdict == "confirm" and expected_verdict == "reject":
        # 模型确认了本应被拒绝的订单（例如已取消的订单），属于致命错误
        return "FATAL"

    if candidate_verdict == "reject" and expected_verdict == "confirm":
        # 模型拒绝了本应确认的有效订单，属于遗漏
        return "MISSED"

    if candidate_verdict == expected_verdict:
        # 如果判定一致，进一步检查细节参数
        candidate_details = candidate.get("details", {})
        expected_details = ground_truth.get("details", {})

        if candidate_details != expected_details:
            # 如果细节错误但判定正确，根据操作类型分类
            if candidate_verdict == "confirm":
                return "FATAL"  # 发错货属于致命错误
            return "MISSED"     # 拒绝错商品属于遗漏

        return "CORRECT"

    return "FATAL"  # 兜底安全策略，默认归为致命错误

{
  "verdict": "needs_confirmation",
  "candidate_details": {
    "boxes": 250,
    "units": 5
  }
}

+------------------+     API 调用     +------------------+     保存至本地     +------------------+
|     n1n.ai       | ---------------> |   模型原始输出   | ---------------------> |   raw_cache.jsonl|
+------------------+                  +------------------+                        +------------------+
                                                                                           |
                                                                                           | --rescore
                                                                                           v
                                                                                  +------------------+
                                                                                  |    评分器引擎    |
                                                                                  +------------------+

import json

def log_result_incremental(file_path: str, result_data: dict):
    with open(file_path, "a", encoding="utf-8") as f:
        f.write(json.dumps(result_data) + "\n")