RAG 系统测试——以电商 APP 客服问答为例

"""测试集构建器：从知识库 + 生产日志中构建分层测试集"""

from dataclasses import dataclass, field
from typing import Optional
import json
import random
from pathlib import Path


@dataclass
class TestCase:
    """一条 RAG 测试用例"""
    query_id: str                          # 唯一 ID
    user_query: str                        # 原始用户问题
    rewritten_query: str = ""              # 改写后的问题（可选）
    category_l1: str = ""                  # 一级分类（40 大类之一）
    category_l2: str = ""                  # 二级分类（1500+ 小类之一）

    # Ground truth：标注的相关文档 ID 列表
    relevant_doc_ids: list[str] = field(default_factory=list)

    # 期望的答案关键点（用于评估生成质量）
    expected_key_points: list[str] = field(default_factory=list)

    # 用例元数据
    difficulty: str = "medium"            # easy / medium / hard
    source: str = ""                       # 来源：production_log / manual / llm_generated


class TestSetBuilder:
    """
    分层抽样测试集构建器。

    核心原则：
    - 按生产流量的类别分布确定每类的样本数
    - 每类内按难度分层（简单/中等/困难）
    - 最终的测试集分布 ≈ 生产流量分布
    """

    # 类别权重（基于生产日志统计，每月更新一次）
    CATEGORY_WEIGHTS = {
        "售后": 0.25,
        "物流": 0.20,
        "支付": 0.15,
        "优惠券": 0.15,
        "会员": 0.05,
        "活动": 0.10,
        "其他": 0.10,
    }

    # 难度分布（经验值）
    DIFFICULTY_DISTRIBUTION = {
        "easy": 0.30,    # 关键词精确匹配："7天无理由退货的条件"
        "medium": 0.50,  # 需要语义理解："拆了包装还能退吗"
        "hard": 0.20,    # 需要多文档推理："用了一个月坏了，用券买的，能退差价吗"
    }

    def __init__(self, total_size: int = 500):
        self.total_size = total_size
        self.test_cases: list[TestCase] = []

    def build(self) -> list[TestCase]:
        """构建完整测试集"""
        for category, weight in self.CATEGORY_WEIGHTS.items():
            n_samples = int(self.total_size * weight)

            for difficulty, diff_weight in self.DIFFICULTY_DISTRIBUTION.items():
                n_diff = int(n_samples * diff_weight)

                # 最低保证：每类每难度至少 3 条
                n_diff = max(n_diff, 3)

                # 从各来源汇聚
                cases = (
                    self._sample_from_production_logs(category, difficulty, n_diff // 2)
                    + self._sample_from_manual_annotation(category, difficulty, n_diff // 4)
                    + self._generate_with_llm(category, difficulty, n_diff // 4)
                )

                self.test_cases.extend(cases)

        # 去重 + 补充到目标总数
        self._deduplicate()
        self._fill_to_target()

        return self.test_cases

    def _sample_from_production_logs(
        self, category: str, difficulty: str, n: int
    ) -> list[TestCase]:
        """
        从生产日志中采样真实用户 query。

        这是最宝贵的测试数据——它是用户真实说话方式。
        LLM 生成的 query 和真人的表达方式有差异，
        过度依赖 LLM 生成 → 测试集与生产分布偏离。
        """
        # 实际项目中从日志数据库读取
        # SELECT query FROM chat_logs
        # WHERE category = %s AND difficulty = %s
        # ORDER BY RANDOM() LIMIT %s
        return []  # 占位——实际从 DB 取

    def _sample_from_manual_annotation(
        self, category: str, difficulty: str, n: int
    ) -> list[TestCase]:
        """
        人工标注的测试用例。

        人工标注用例覆盖 LLM 难以生成的场景：
        - 模糊表达："那个东西""之前那个"
        - 口语化表达："咋整啊""还能不能弄了"
        - 长尾场景：某个具体活动的特殊规则
        """
        return []

    def _generate_with_llm(
        self, category: str, difficulty: str, n: int
    ) -> list[TestCase]:
        """
        用 LLM 从知识库文档中生成 query。

        ⚠️ LLM 生成的 query 更规整、更接近书面语，
        不能替代真实用户 query，只作为补充。
        """
        return []

"""验证测试集的独立性——确保没有数据泄漏"""

import hashlib
from pathlib import Path


class TestSetValidator:
    """
    测试集隔离性检查。

    检查项：
    1. 测试集的 query 是否与训练/调优用的 query 重叠
    2. 测试集的 query 对应的知识库文档，是否已作为"示例"喂给过 LLM
    3. 测试集的 query 的 embedding 与训练集是否过于相似（> 0.95）
    """

    def __init__(self, test_set_path: str, train_set_path: str):
        self.test_set = self._load(test_set_path)
        self.train_set = self._load(train_set_path)

    def check_overlap(self) -> dict:
        """检查严格文本重叠"""
        train_queries = {self._normalize(q) for q in self.train_set}
        overlaps = []

        for q in self.test_set:
            normalized = self._normalize(q)
            if normalized in train_queries:
                overlaps.append(q)

        return {
            "total_test": len(self.test_set),
            "overlaps": len(overlaps),
            "pass": len(overlaps) == 0,
            "details": overlaps[:10],  # 前 10 条重叠
        }

    def check_semantic_similarity(self, threshold: float = 0.95) -> dict:
        """
        检查语义过度相似。

        即使文本不完全相同，如果 embedding 相似度 > 0.95，
        也应该标记为可疑——可能是换了一种说法问同一件事。
        """
        # 实际用 embedding 模型计算余弦相似度
        # 这里给出接口定义
        suspicious_pairs = []
        return {
            "total_pairs_checked": len(self.test_set) * len(self.train_set),
            "suspicious_pairs": len(suspicious_pairs),
            "pass": len(suspicious_pairs) == 0,
        }

    @staticmethod
    def _normalize(text: str) -> str:
        """标准化文本用于比对"""
        import re
        text = text.lower().strip()
        text = re.sub(r"\s+", " ", text)
        return text

    @staticmethod
    def _load(path: str) -> list[str]:
        with open(path, "r", encoding="utf-8") as f:
            return [line.strip() for line in f if line.strip()]

"""
意图识别准确率测试。

测试目标：
- Top-1 准确率 ≥ 92%（40 大类正确率）
- Top-3 准确率 ≥ 98%（前 3 个候选中包含正确类别）
- 置信度过低的比例 ≤ 10%（需要人工介入的比例）
"""

import json
import time
import pytest
import numpy as np
from pathlib import Path
from sklearn.metrics import (
    classification_report,
    confusion_matrix,
    top_k_accuracy_score,
)
from core.rag.intent_classifier import IntentClassifier


# ============================================================
# 加载意图识别测试集
# ============================================================

def load_intent_test_set() -> list[dict]:
    """
    格式：
    [
      {
        "query": "拆了包装还能退吗",
        "expected_category_l1": "售后-退货",
        "expected_category_l2": "已拆封退货",
        "difficulty": "medium",
      },
      ...
    ]
    """
    with open("tests/rag/intent_test_set.json", "r", encoding="utf-8") as f:
        return json.load(f)


# ============================================================
# 指标计算
# ============================================================

def compute_intent_metrics(
    y_true: list[str],
    y_pred: list[str],
    y_proba: list[list[float]] | None = None,
    category_names: list[str] | None = None,
) -> dict:
    """
    计算意图识别的全部指标。

    返回的指标：
    - top1_accuracy: Top-1 准确率
    - top3_accuracy: Top-3 准确率（如果提供了概率）
    - per_class_accuracy: 每类的准确率
    - confusion_pairs: 最常见的混淆对（类A 被误判为 类B）
    - low_confidence_ratio: 置信度 < 阈值的比例
    - macro_f1: 宏平均 F1（考虑类不平衡）
    """
    n = len(y_true)
    correct = sum(1 for t, p in zip(y_true, y_pred) if t == p)

    metrics = {
        "total_samples": n,
        "top1_correct": correct,
        "top1_accuracy": correct / n if n > 0 else 0,
    }

    # Top-3 准确率
    if y_proba is not None and category_names is not None:
        # 将 y_true 和 y_proba 转为 sklearn 格式
        y_true_indices = [category_names.index(t) for t in y_true]
        y_proba_array = np.array(y_proba)
        top3_correct = sum(
            1 for i, true_idx in enumerate(y_true_indices)
            if true_idx in np.argsort(y_proba_array[i])[-3:]
        )
        metrics["top3_accuracy"] = top3_correct / n if n > 0 else 0

    # 每类准确率
    per_class = {}
    for true_label in set(y_true):
        class_samples = [(t, p) for t, p in zip(y_true, y_pred) if t == true_label]
        if class_samples:
            correct_in_class = sum(1 for t, p in class_samples if t == p)
            per_class[true_label] = correct_in_class / len(class_samples)
    metrics["per_class_accuracy"] = per_class

    # 找出准确率最低的 5 个类
    worst_classes = sorted(per_class.items(), key=lambda x: x[1])[:5]
    metrics["worst_5_classes"] = worst_classes

    # 找出最常见的 10 个混淆对
    confusion_pairs = {}
    for t, p in zip(y_true, y_pred):
        if t != p:
            pair = f"{t} → {p}"
            confusion_pairs[pair] = confusion_pairs.get(pair, 0) + 1
    metrics["top_confusion_pairs"] = sorted(
        confusion_pairs.items(), key=lambda x: x[1], reverse=True
    )[:10]

    # 置信度过低的比例（如果分类器返回置信度）
    if y_proba is not None:
        max_probs = [max(probs) for probs in y_proba]
        metrics["mean_confidence"] = np.mean(max_probs)
        metrics["low_confidence_ratio"] = sum(
            1 for p in max_probs if p < 0.5
        ) / n
        metrics["confidence_distribution"] = {
            "p10": float(np.percentile(max_probs, 10)),
            "p25": float(np.percentile(max_probs, 25)),
            "p50": float(np.percentile(max_probs, 50)),
            "p75": float(np.percentile(max_probs, 75)),
            "p90": float(np.percentile(max_probs, 90)),
        }

    return metrics


# ============================================================
# 测试用例
# ============================================================

class TestIntentAccuracy:
    """意图识别准确率测试"""

    @pytest.fixture(scope="class")
    def classifier(self):
        return IntentClassifier()

    @pytest.fixture(scope="class")
    def test_data(self):
        return load_intent_test_set()

    def test_overall_top1_accuracy(self, classifier, test_data):
        """整体 Top-1 准确率 ≥ 92%"""
        y_true = [c["expected_category_l1"] for c in test_data]
        y_pred = []

        for case in test_data:
            result = classifier.classify(case["query"])
            y_pred.append(result.top_category)

        metrics = compute_intent_metrics(y_true, y_pred)

        print(f"\n意图识别 Top-1 准确率：{metrics['top1_accuracy']:.1%}")
        print(f"总样本数：{metrics['total_samples']}")

        # 记录指标（供 CI 面板展示）
        _record_metric("intent.top1_accuracy", metrics["top1_accuracy"])

        assert metrics["top1_accuracy"] >= 0.92, (
            f"意图识别 Top-1 准确率 {metrics['top1_accuracy']:.1%} 低于目标 92%\n"
            f"最差 5 类：{metrics['worst_5_classes']}\n"
            f"最常混淆：{metrics['top_confusion_pairs'][:5]}"
        )

    def test_per_class_minimum_accuracy(self, classifier, test_data):
        """每类准确率不低于 75%（不能有灾难性失败的类）"""
        y_true = [c["expected_category_l1"] for c in test_data]
        y_pred = [classifier.classify(c["query"]).top_category for c in test_data]

        metrics = compute_intent_metrics(y_true, y_pred)
        per_class = metrics["per_class_accuracy"]

        failures = [
            (cls, acc) for cls, acc in per_class.items()
            if acc < 0.75
        ]

        if failures:
            failure_report = "\n".join(
                f"  {cls}: {acc:.1%}" for cls, acc in failures
            )
            pytest.fail(
                f"以下类别的意图识别准确率低于 75% 底线：\n{failure_report}"
            )

    def test_top3_accuracy(self, classifier, test_data):
        """Top-3 准确率 ≥ 98%（前 3 个候选中包含正确答案）"""
        y_true = []
        y_pred_top = []
        y_proba_list = []

        for case in test_data:
            result = classifier.classify_with_scores(case["query"])
            y_true.append(case["expected_category_l1"])
            y_pred_top.append(result.top_category)
            y_proba_list.append(result.all_scores)

        category_names = sorted(set(y_true))
        metrics = compute_intent_metrics(
            y_true, y_pred_top,
            y_proba=y_proba_list,
            category_names=category_names,
        )

        top3 = metrics.get("top3_accuracy", 0)
        print(f"\n意图识别 Top-3 准确率：{top3:.1%}")

        assert top3 >= 0.98, (
            f"Top-3 准确率 {top3:.1%} 低于目标 98%——"
            f"有 {metrics['total_samples'] - int(top3 * metrics['total_samples'])} "
            f"条 query 的正确类别不在前 3 候选中"
        )

    def test_low_confidence_rate(self, classifier, test_data):
        """低置信度（< 0.5）的比例 ≤ 10%"""
        low_conf_count = 0

        for case in test_data:
            result = classifier.classify_with_scores(case["query"])
            if result.top_score < 0.5:
                low_conf_count += 1

        ratio = low_conf_count / len(test_data)

        print(f"\n低置信度比例：{ratio:.1%} ({low_conf_count}/{len(test_data)})")

        assert ratio <= 0.10, (
            f"低置信度比例 {ratio:.1%} 超过 10% 阈值——"
            f"太多 query 意图模糊，需要优化分类器或增加兜底策略"
        )

    def test_confusion_matrix_output(self, classifier, test_data):
        """
        输出混淆矩阵（不通过/不失败，纯信息输出）。

        目的是让开发和产品看到"哪些类之间容易混淆"，
        从而决定是否合并某些意图，还是增加二分类器。
        """
        y_true = [c["expected_category_l1"] for c in test_data]
        y_pred = [classifier.classify(c["query"]).top_category for c in test_data]

        labels = sorted(set(y_true + y_pred))
        cm = confusion_matrix(y_true, y_pred, labels=labels)

        # 输出到文件供后续分析
        output_dir = Path("tests/rag/output")
        output_dir.mkdir(exist_ok=True)

        with open(output_dir / "intent_confusion_matrix.json", "w", encoding="utf-8") as f:
            json.dump({
                "labels": labels,
                "matrix": cm.tolist(),
                "generated_at": time.strftime("%Y-%m-%d %H:%M:%S"),
            }, f, ensure_ascii=False, indent=2)

        # 打印 Top 5 混淆对
        metrics = compute_intent_metrics(y_true, y_pred)
        print("\n=== Top 5 混淆对 ===")
        for pair, count in metrics["top_confusion_pairs"][:5]:
            print(f"  {pair}: {count} 次")


def _record_metric(name: str, value: float) -> None:
    """记录指标到 CI 系统（替换为你们实际用的）"""
    # 例如：GitHub Actions 的 $GITHUB_STEP_SUMMARY
    # 或者写入 InfluxDB / Prometheus pushgateway
    pass

"""
检索质量测试——RAG 测试中最核心、最需要反复跑的部分。

跑一次检索评估只需要 embedding + 向量搜索，
不涉及 LLM 生成，因此成本低、速度快（500 条 ≈ 2 分钟）。
"""

import json
import time
import numpy as np
import pytest
from pathlib import Path
from core.rag.retriever import HybridRetriever
from core.rag.embedder import Embedder


# ============================================================
# 检索指标计算
# ============================================================

class RetrievalMetrics:
    """检索质量指标计算器"""

    @staticmethod
    def precision_at_k(
        retrieved_doc_ids: list[str],
        relevant_doc_ids: set[str],
        k: int = 5,
    ) -> float:
        """Precision@K：检索到的前 K 篇中有多少是相关的"""
        retrieved_k = retrieved_doc_ids[:k]
        if not retrieved_k:
            return 0.0
        relevant_in_k = sum(1 for doc_id in retrieved_k if doc_id in relevant_doc_ids)
        return relevant_in_k / len(retrieved_k)

    @staticmethod
    def recall_at_k(
        retrieved_doc_ids: list[str],
        relevant_doc_ids: set[str],
        k: int = 5,
    ) -> float:
        """Recall@K：所有相关文档中，有多少在前 K 篇检索结果中"""
        if not relevant_doc_ids:
            return 1.0  # 没有标注相关文档 → 不扣分
        retrieved_k = set(retrieved_doc_ids[:k])
        return len(retrieved_k & relevant_doc_ids) / len(relevant_doc_ids)

    @staticmethod
    def mrr(retrieved_doc_ids: list[str], relevant_doc_ids: set[str]) -> float:
        """MRR (Mean Reciprocal Rank)：第一个相关文档排在第几（倒数平均）"""
        for i, doc_id in enumerate(retrieved_doc_ids, start=1):
            if doc_id in relevant_doc_ids:
                return 1.0 / i
        return 0.0

    @staticmethod
    def ndcg_at_k(
        retrieved_doc_ids: list[str],
        relevant_doc_ids: set[str],
        relevance_scores: dict[str, int] | None = None,
        k: int = 5,
    ) -> float:
        """
        NDCG@K：考虑排序位置和相关性程度的排序质量指标。

        如果提供了 relevance_scores（如 {doc_id: 3, doc_id2: 1}），
        使用分级相关性计算；否则使用二值（相关=1，不相关=0）。
        """
        retrieved_k = retrieved_doc_ids[:k]
        if not retrieved_k:
            return 0.0

        # 计算 DCG
        dcg = 0.0
        for i, doc_id in enumerate(retrieved_k, start=1):
            if relevance_scores:
                rel = relevance_scores.get(doc_id, 0)
            else:
                rel = 1 if doc_id in relevant_doc_ids else 0
            dcg += rel / np.log2(i + 1)

        # 计算 IDCG（理想排序）
        if relevance_scores:
            ideal_rels = sorted(relevance_scores.values(), reverse=True)[:k]
        else:
            n_relevant = len(relevant_doc_ids & set(retrieved_k))
            ideal_rels = [1] * min(n_relevant, k) + [0] * max(0, k - n_relevant)

        idcg = sum(
            rel / np.log2(i + 1) for i, rel in enumerate(ideal_rels, start=1)
        )

        return dcg / idcg if idcg > 0 else 0.0

    @staticmethod
    def hit_rate_at_k(
        retrieved_doc_ids: list[str],
        relevant_doc_ids: set[str],
        k: int = 5,
    ) -> bool:
        """前 K 篇中是否至少有一篇相关文档"""
        return any(doc_id in relevant_doc_ids for doc_id in retrieved_doc_ids[:k])


# ============================================================
# 检索质量评估器
# ============================================================

class RetrievalEvaluator:
    """
    检索质量评估器。

    遍历测试集中的每条 query，执行检索，计算指标。
    """

    def __init__(self, retriever: HybridRetriever):
        self.retriever = retriever
        self.metrics = RetrievalMetrics()

    def evaluate(
        self,
        test_cases: list[dict],
        k_values: list[int] = [1, 3, 5, 10],
    ) -> dict:
        """执行完整评估，返回所有指标"""
        results = {
            "n_queries": len(test_cases),
            "per_k": {},
            "per_category": {},
            "per_difficulty": {},
        }

        all_records = []  # 每条的详细记录

        for case in test_cases:
            # 执行检索
            start = time.perf_counter()
            retrieved = self.retriever.search(
                query=case["user_query"],
                top_k=max(k_values),
                filters={"category": case.get("category_l1")},  # 如果意图识别先路由
            )
            latency_ms = (time.perf_counter() - start) * 1000

            relevant_ids = set(case.get("relevant_doc_ids", []))
            retrieved_ids = [doc["id"] for doc in retrieved]

            record = {
                "query_id": case["query_id"],
                "query": case["user_query"],
                "category": case.get("category_l1", ""),
                "difficulty": case.get("difficulty", "medium"),
                "n_relevant": len(relevant_ids),
                "latency_ms": latency_ms,
            }

            for k in k_values:
                record[f"precision@{k}"] = self.metrics.precision_at_k(
                    retrieved_ids, relevant_ids, k
                )
                record[f"recall@{k}"] = self.metrics.recall_at_k(
                    retrieved_ids, relevant_ids, k
                )
                record[f"hit@{k}"] = self.metrics.hit_rate_at_k(
                    retrieved_ids, relevant_ids, k
                )

            record["mrr"] = self.metrics.mrr(retrieved_ids, relevant_ids)
            record["ndcg@5"] = self.metrics.ndcg_at_k(retrieved_ids, relevant_ids, k=5)

            all_records.append(record)

        # 汇总指标
        for k in k_values:
            results["per_k"][k] = {
                "precision": float(np.mean([r[f"precision@{k}"] for r in all_records])),
                "recall": float(np.mean([r[f"recall@{k}"] for r in all_records])),
                "hit_rate": float(np.mean([r[f"hit@{k}"] for r in all_records])),
            }

        results["mrr"] = float(np.mean([r["mrr"] for r in all_records]))
        results["ndcg@5"] = float(np.mean([r["ndcg@5"] for r in all_records]))
        results["avg_latency_ms"] = float(np.mean([r["latency_ms"] for r in all_records]))
        results["p95_latency_ms"] = float(np.percentile(
            [r["latency_ms"] for r in all_records], 95
        ))

        # 按类别汇总
        categories = set(r["category"] for r in all_records)
        for cat in categories:
            cat_records = [r for r in all_records if r["category"] == cat]
            if cat_records:
                results["per_category"][cat] = {
                    "n": len(cat_records),
                    "recall@5": float(np.mean([r["recall@5"] for r in cat_records])),
                    "precision@3": float(np.mean([r["precision@3"] for r in cat_records])),
                    "mrr": float(np.mean([r["mrr"] for r in cat_records])),
                }

        # 按难度汇总
        for difficulty in ["easy", "medium", "hard"]:
            diff_records = [r for r in all_records if r["difficulty"] == difficulty]
            if diff_records:
                results["per_difficulty"][difficulty] = {
                    "n": len(diff_records),
                    "recall@5": float(np.mean([r["recall@5"] for r in diff_records])),
                    "precision@3": float(np.mean([r["precision@3"] for r in diff_records])),
                }

        results["detail_records"] = all_records
        return results


# ============================================================
# 测试用例
# ============================================================

class TestRetrievalQuality:
    """检索质量测试"""

    @pytest.fixture(scope="class")
    def retriever(self):
        return HybridRetriever(
            dense_model="text-embedding-3-large",
            sparse_model="bm25",
            milvus_uri="http://localhost:19530",
        )

    @pytest.fixture(scope="class")
    def test_cases(self):
        with open("tests/rag/retrieval_test_set.json", "r", encoding="utf-8") as f:
            return json.load(f)

    def test_recall_at_5(self, retriever, test_cases):
        """Recall@5 ≥ 90%"""
        evaluator = RetrievalEvaluator(retriever)
        results = evaluator.evaluate(test_cases, k_values=[5])

        recall5 = results["per_k"][5]["recall"]

        print(f"\nRecall@5: {recall5:.1%}")
        print(f"MRR: {results['mrr']:.3f}")
        print(f"NDCG@5: {results['ndcg@5']:.3f}")
        print(f"平均延迟: {results['avg_latency_ms']:.1f}ms")
        print(f"P95 延迟: {results['p95_latency_ms']:.1f}ms")

        assert recall5 >= 0.90, (
            f"Recall@5 = {recall5:.1%} 低于目标 90%\n"
            f"检查 per_category 指标定位低召回类别"
        )

    def test_precision_at_3(self, retriever, test_cases):
        """Precision@3 ≥ 75%"""
        evaluator = RetrievalEvaluator(retriever)
        results = evaluator.evaluate(test_cases, k_values=[3])

        precision3 = results["per_k"][3]["precision"]

        print(f"\nPrecision@3: {precision3:.1%}")

        assert precision3 >= 0.75, (
            f"Precision@3 = {precision3:.1%} 低于目标 75%\n"
            f"说明检索结果中有太多不相关文档混入——检查 reranker 效果"
        )

    def test_mrr(self, retriever, test_cases):
        """MRR ≥ 0.80（第一个相关文档平均排在 1.25 位以内）"""
        evaluator = RetrievalEvaluator(retriever)
        results = evaluator.evaluate(test_cases)

        assert results["mrr"] >= 0.80, (
            f"MRR = {results['mrr']:.3f} 低于目标 0.80\n"
            f"第一个相关文档排得太靠后——检查 embedding 模型和排序策略"
        )

    def test_hit_rate_at_5(self, retriever, test_cases):
        """Hit Rate@5 ≥ 95%（95% 的 query 至少找到 1 篇相关文档）"""
        evaluator = RetrievalEvaluator(retriever)
        results = evaluator.evaluate(test_cases, k_values=[5])

        hit_rate = results["per_k"][5]["hit_rate"]

        print(f"\nHit Rate@5: {hit_rate:.1%}")

        assert hit_rate >= 0.95, (
            f"Hit Rate@5 = {hit_rate:.1%} 低于目标 95%\n"
            f"有 {int((1-hit_rate) * results['n_queries'])} 条 query 完全找不到相关文档——"
            f"检查这些 query 对应的知识库是否有对应内容"
        )

    def test_per_category_no_disaster(self, retriever, test_cases):
        """没有任何大类的 Recall@5 低于 70%（不能有灾难性失败）"""
        evaluator = RetrievalEvaluator(retriever)
        results = evaluator.evaluate(test_cases)

        disasters = [
            (cat, data["recall@5"])
            for cat, data in results["per_category"].items()
            if data["recall@5"] < 0.70
        ]

        if disasters:
            report = "\n".join(
                f"  {cat}: Recall@5 = {recall:.1%} ({data['n']} 条)"
                for (cat, recall), (_, data) in zip(disasters, [
                    (c, results["per_category"][c]) for c, _ in disasters
                ])
            )
            pytest.fail(f"以下类别检索严重不足：\n{report}")

    def test_retrieval_latency(self, retriever, test_cases):
        """检索 P95 延迟 ≤ 500ms"""
        evaluator = RetrievalEvaluator(retriever)
        results = evaluator.evaluate(test_cases)

        p95 = results["p95_latency_ms"]

        print(f"\n检索延迟 P95: {p95:.0f}ms")

        assert p95 <= 500, (
            f"检索 P95 延迟 {p95:.0f}ms 超过 500ms 阈值——"
            f"检查 Milvus 索引类型和 embedding API 延迟"
        )

    def test_output_per_difficulty_breakdown(self, retriever, test_cases):
        """按难度拆分指标（不通过/不失败，诊断用）"""
        evaluator = RetrievalEvaluator(retriever)
        results = evaluator.evaluate(test_cases)

        print("\n=== 按难度拆分 ===")
        for difficulty in ["easy", "medium", "hard"]:
            if difficulty in results["per_difficulty"]:
                d = results["per_difficulty"][difficulty]
                print(
                    f"  {difficulty:6s} (n={d['n']:3d}): "
                    f"Recall@5={d['recall@5']:.1%}, "
                    f"Precision@3={d['precision@3']:.1%}"
                )