🏗️ 四大分类框架

我们将评测维度划分为四个层次，形成从内核到外延的完整评估体系：

推理能力三维度

• 逻辑推理：包括演绎推理（从一般到特殊）、归纳推理（从特殊到一般）和溯因推理（从结果倒推原因）。典型评测任务有逻辑谜题、三段论推理、条件推理等。代表基准：LogiQA、ReClor。
• 数学推理（GSM8K）：GSM8K 是包含8500道小学数学应用题的评测集，要求模型展示分步推理过程（Chain-of-Thought）。进阶基准包括 MATH（竞赛级数学题）、TheoremQA（定理证明）。数学推理能力直接反映模型的逻辑严密性。
• 常识推理：测试模型理解日常世界知识并进行合理推断的能力。代表基准：HellaSwag（故事续写选择）、PIQA（物理常识）、ARC（AI2 Reasoning Challenge）。常识推理是模型「不犯低级错误」的关键保障。

📊 能力评测维度汇总

"""
MMLU 风格多学科知识问答评测脚本
使用 OpenAI 兼容接口，支持自定义数据集和批量评测
"""
import json
import time
import csv
import os
from dataclasses import dataclass, field
from typing import Any
from collections import defaultdict

import requests


# ============================================================
# 1. 配置
# ============================================================
@dataclass
class EvalConfig:
    """评测配置"""
    api_base: str = "http://localhost:11434/v1"   # Ollama / vLLM / 任意 OpenAI 兼容端点
    model: str = "qwen2.5:7b"
    api_key: str = "ollama"                        # 本地部署可为任意非空值
    temperature: float = 0.0                       # 评测时建议设为 0 保证确定性
    max_tokens: int = 256
    dataset_path: str = "./data/mmlu_sample.csv"   # CSV 数据集路径
    output_dir: str = "./eval_results"
    batch_delay: float = 0.2                       # 请求间隔（秒），避免打爆服务
    subjects: list = field(default_factory=lambda: [])  # 空 = 全部学科


# ============================================================
# 2. 数据集加载
# ============================================================
class DatasetLoader:
    """加载 CSV 格式的 MMLU 风格数据集

    CSV 列格式:
        subject, question, option_a, option_b, option_c, option_d, answer
    """

    @staticmethod
    def load(path: str) -> list[dict]:
        samples = []
        with open(path, "r", encoding="utf-8") as f:
            reader = csv.DictReader(f)
            for row in reader:
                samples.append({
                    "subject": row["subject"].strip(),
                    "question": row["question"].strip(),
                    "options": {
                        "A": row["option_a"].strip(),
                        "B": row["option_b"].strip(),
                        "C": row["option_c"].strip(),
                        "D": row["option_d"].strip(),
                    },
                    "answer": row["answer"].strip().upper(),
                })
        print(f"✅ 加载数据集: {len(samples)} 条样本")
        return samples

    @staticmethod
    def filter_by_subjects(samples: list[dict], subjects: list[str]) -> list[dict]:
        if not subjects:
            return samples
        return [s for s in samples if s["subject"] in subjects]


# ============================================================
# 3. 模型调用
# ============================================================
class ModelClient:
    """OpenAI 兼容 API 客户端"""

    def __init__(self, config: EvalConfig):
        self.config = config
        self.url = f"{config.api_base}/chat/completions"

    def ask(self, question: str, options: dict) -> tuple[str, str]:
        """发送请求并返回 (模型回答, 提取的选项)"""
        options_text = "\n".join([f"{k}. {v}" for k, v in options.items()])
        prompt = f"""请回答以下单选题，只需输出正确选项的字母（A/B/C/D），不要解释。

题目：{question}

选项：
{options_text}

你的答案："""

        payload = {
            "model": self.config.model,
            "messages": [{"role": "user", "content": prompt}],
            "temperature": self.config.temperature,
            "max_tokens": self.config.max_tokens,
        }
        headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {self.config.api_key}",
        }

        resp = requests.post(self.url, json=payload, headers=headers, timeout=60)
        resp.raise_for_status()
        data = resp.json()
        raw_answer = data["choices"][0]["message"]["content"].strip()

        # 提取选项中包含的字母
        extracted = self._extract_option(raw_answer)
        return raw_answer, extracted

    @staticmethod
    def _extract_option(text: str) -> str:
        """从模型输出中提取选项字母 A/B/C/D"""
        for char in text.upper():
            if char in "ABCD":
                return char
        return "?"  # 无法提取


# ============================================================
# 4. 批量评测
# ============================================================
class Evaluator:
    """批量执行评测"""

    def __init__(self, config: EvalConfig):
        self.config = config
        self.client = ModelClient(config)
        self.results: list[dict] = []

    def run(self, samples: list[dict]) -> dict:
        total = len(samples)
        correct = 0
        subject_stats: dict[str, dict] = defaultdict(lambda: {"total": 0, "correct": 0})

        print(f"\n🚀 开始评测: {self.config.model} | 样本数: {total}\n")

        for idx, sample in enumerate(samples):
            try:
                raw_answer, extracted = self.client.ask(
                    sample["question"], sample["options"]
                )
            except Exception as e:
                raw_answer, extracted = f"ERROR: {e}", "?"

            is_correct = extracted == sample["answer"]

            if is_correct:
                correct += 1

            subject_stats[sample["subject"]]["total"] += 1
            if is_correct:
                subject_stats[sample["subject"]]["correct"] += 1

            self.results.append({
                "id": idx + 1,
                "subject": sample["subject"],
                "question": sample["question"][:50] + "...",
                "expected": sample["answer"],
                "predicted": extracted,
                "raw_output": raw_answer[:100],
                "correct": is_correct,
            })

            # 每 10 条输出一次进度
            if (idx + 1) % 10 == 0 or idx == total - 1:
                acc = correct / (idx + 1) * 100
                print(f"  [{idx+1}/{total}] 当前准确率: {acc:.1f}%")

            time.sleep(self.config.batch_delay)

        overall_acc = correct / total * 100
        print(f"\n🏁 评测完成！总体准确率: {overall_acc:.2f}% ({correct}/{total})\n")
        return {
            "model": self.config.model,
            "total_samples": total,
            "correct": correct,
            "accuracy": round(overall_acc, 2),
            "subject_breakdown": {
                subj: {
                    "total": s["total"],
                    "correct": s["correct"],
                    "accuracy": round(s["correct"] / s["total"] * 100, 2),
                }
                for subj, s in sorted(subject_stats.items())
            },
        }


# ============================================================
# 5. 报告生成
# ============================================================
class ReportGenerator:
    """生成评测报告（JSON + Markdown + 控制台输出）"""

    @staticmethod
    def generate(summary: dict, detail_results: list[dict], output_dir: str):
        os.makedirs(output_dir, exist_ok=True)
        timestamp = time.strftime("%Y%m%d_%H%M%S")
        model_safe = summary["model"].replace(":", "_").replace("/", "_")

        # --- JSON 详细结果 ---
        json_path = os.path.join(output_dir, f"{model_safe}_{timestamp}.json")
        with open(json_path, "w", encoding="utf-8") as f:
            json.dump({
                "summary": summary,
                "details": detail_results,
            }, f, ensure_ascii=False, indent=2)

        # --- Markdown 报告 ---
        md_path = os.path.join(output_dir, f"{model_safe}_{timestamp}.md")
        with open(md_path, "w", encoding="utf-8") as f:
            f.write(f"# 📊 MMLU 评测报告\n\n")
            f.write(f"**模型**: `{summary['model']}`  \n")
            f.write(f"**评测时间**: {timestamp}  \n")
            f.write(f"**样本总数**: {summary['total_samples']}  \n")
            f.write(f"**正确数**: {summary['correct']}  \n")
            f.write(f"**总体准确率**: **{summary['accuracy']}%**\n\n")

            f.write(f"## 各学科表现\n\n")
            f.write(f"| 学科 | 样本数 | 正确数 | 准确率 |\n")
            f.write(f"|------|--------|--------|--------|\n")
            for subj, s in summary["subject_breakdown"].items():
                f.write(f"| {subj} | {s['total']} | {s['correct']} | {s['accuracy']}% |\n")

            # 错误样本列表
            errors = [r for r in detail_results if not r["correct"]]
            if errors:
                f.write(f"\n## ❌ 错误样本 ({len(errors)} 条)\n\n")
                for e in errors[:20]:  # 最多展示 20 条
                    f.write(f"- **#{e['id']}** [{e['subject']}] 期望: `{e['expected']}`, "
                            f"预测: `{e['predicted']}`\n")
                    f.write(f"  > {e['question']}\n")

        print(f"📄 JSON 报告: {json_path}")
        print(f"📄 Markdown 报告: {md_path}")


# ============================================================
# 6. 主流程
# ============================================================
def main():
    config = EvalConfig(
        api_base="http://localhost:11434/v1",
        model="qwen2.5:7b",
        dataset_path="./data/mmlu_sample.csv",
        output_dir="./eval_results",
        subjects=[],  # 空 = 评测所有学科
    )

    # 加载数据
    samples = DatasetLoader.load(config.dataset_path)
    if config.subjects:
        samples = DatasetLoader.filter_by_subjects(samples, config.subjects)
        print(f"🔍 筛选学科: {config.subjects} → {len(samples)} 条")

    # 执行评测
    evaluator = Evaluator(config)
    summary = evaluator.run(samples)

    # 生成报告
    ReportGenerator.generate(summary, evaluator.results, config.output_dir)

    # 控制台摘要
    print("=" * 50)
    print(f"📊 评测摘要: {summary['model']}")
    print(f"   准确率: {summary['accuracy']}% ({summary['correct']}/{summary['total_samples']})")
    print(f"   学科数: {len(summary['subject_breakdown'])}")
    print("=" * 50)


if __name__ == "__main__":
    main()

🛡️ 安全评测维度汇总

实战任务1：设计模型评测方案

场景：银行智能问答场景的模型评测

背景：某银行计划引入一个大语言模型用于智能客服和知识问答场景，覆盖的业务包括个人银行业务咨询、理财产品介绍、信用卡服务等。你需要为该模型设计一套完整的评测方案。

步骤：

1. 从53项指标中选择8-12个适用于问答场景的维度
2. 为每个维度分配权重（需解释分配理由）
3. 设计评测数据集（多少样本？覆盖哪些场景？）
4. 执行评测并记录各维度得分

评估标准：评测方案完整、权重合理、数据集有代表性

耗时：2小时

产出物：评测方案文档（含维度选择说明、权重分配表、数据集设计思路、评测结果记录模板）

实战任务2：对比不同模型在各维度的表现

场景：从三个候选模型中选出最适合银行客服场景的模型

背景：现有三个候选模型——Model A（通用大模型）、Model B（金融领域微调模型）、Model C（开源社区模型），需要在知识理解、安全过滤、指令遵循三个维度上进行横向对比，为模型选型提供数据支撑。

步骤：

1. 为每个维度设计5-10个测试用例（Prompt）
2. 使用相同的Prompt分别调用三个模型
3. 按照统一的评分标准（1-5分）对每个维度的输出进行打分
4. 汇总得分并生成对比雷达图
5. 撰写选型建议报告（含推荐模型及理由）

评估标准：测试用例覆盖典型场景、评分标准客观一致、对比分析有洞察

耗时：3小时

产出物：模型对比评测报告（含测试用例清单、三方打分表、雷达图、选型建议）

实战任务3：分析安全维度评测结果

场景：基于虚构的安全评测数据，分析模型安全隐患并提出改进建议

背景：以下是某模型在安全评测中的部分测试结果（虚构数据），请据此进行分析：

步骤：

1. 识别模型在安全方面最突出的2-3个问题
2. 分析问题的可能根因（训练数据、对齐策略、领域知识缺失等）
3. 针对每个问题提出具体的改进建议
4. 基于分析结果，判断该模型在当前状态下是否适合银行业务上线

评估标准：问题定位准确、根因分析有深度、改进建议可落地

耗时：1.5小时

产出物：安全评测分析报告（含问题识别、根因分析、改进建议、上线风险评估）

🏦 项目背景

某股份制银行计划将大模型应用于信贷风控辅助决策场景，模型需要能够：

• 分析企业财务报表并识别异常指标
• 解读央行征信报告中的风险信号
• 评估行业政策变化对贷款组合的影响
• 生成符合监管要求的风控审核意见书

模型输出将作为风控审批的参考依据（非自动化决策），对准确性和合规性有极高要求。 测试团队需为该模型设计一套完整、科学、可落地的评测方案。

Step 1：从53项指标中选取关键维度

测试团队首先梳理了风控场景的核心需求，对照 53 项评价指标进行逐一筛选：

经过三轮讨论，最终从 53 项指标中精选出 12 个核心评测维度：

Step 2：12维度选择结果与权重分配

Step 3：设计评测数据集

针对 12 个维度，团队构建了分层评测数据集：

总计：1,230 条功能/安全测试样本 + 性能压测脚本

Step 4：执行评测 — 虚构评测结果

使用 CSV + JMeter 自动化方案执行评测后，三个候选模型（Model A 通用大模型、Model B 金融微调模型、Model C 开源模型）的 12 维度评分如下：

雷达图示意（六边形能力对比）

                 金融知识
                   100
                    |
                   80
                    |
          逻辑推理--60--+--场景适配
                    |      \
                   40       \
                    |         \
                   20          \
                    |            \
          越狱抵抗----+-------------幻觉控制
                    |
                 格式遵循

       Model A (通用)  ---   Model B (金融微调)  ===   Model C (开源)

       Model B 在金融知识和场景适配上大幅领先，安全表现全面；
       Model A 综合能力均衡但领域深度不足；
       Model C 安全防护薄弱，不适合金融场景。

💡 关键启示

1. 评测维度选择比评测执行更重要：从 53 项指标中选出的 12 个维度直接决定了评测的 "效度"。 如果在方案设计阶段遗漏了关键维度（如越狱攻击抵抗），后续执行再精细也无法弥补。
2. 权重分配必须紧扣业务场景：通用场景中能力层权重最高（50%），但风控场景的安全层权重升至 35% 与能力层持平。 银行场景下，一次越狱攻击可能导致客户数据泄露，代价远高于一次回答不准确。
3. 领域微调在垂直场景中优势明显：Model B（金融微调）虽然在通用指标上与 Model A 接近，但在金融知识准确率 (+19分)、场景适配度 (+24分)、幻觉率控制 (+14分) 上具有压倒性优势。 这说明对于垂直金融场景，通用大模型 "够用" ≠ "最优"。
4. 安全评测是金融场景的「一票否决项」：Model C 虽然延迟表现最好（85分），但越狱抵抗仅 48 分，隐私保护仅 55 分 —— 这两个指标直接导致了 Model C 被排除在候选之外，无论其能力和性能表现如何。
5. 评测是一个持续迭代的过程：该方案在实际落地过程中经历了 4 轮迭代——初版评测方案 → 试运行发现维度盲区 → 补充 PII 注入测试 → 增加误拒率平衡指标。评测体系需要随着对业务理解的深入不断演进。