1
2
3
4
5
6

{
  "question": "如何配置 Redis 缓存？",
  "ground_truth": "需要在配置文件中设置...",
  "contexts": ["doc_id_1", "doc_id_2"],  // 期望检索到的文档ID
  "answer": "配置 Redis 缓存的步骤如下..." // 期望的标准答案
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176

from typing import List, Tuple, Set
import numpy as np

class RelevanceScorer:
    """相关性评分工具"""

    @staticmethod
    def mean_reciprocal_rank(
        retrieved_items: List[bool],  # True表示相关
        threshold_position: int = 10
    ) -> float:
        """
        平均倒数排名 (Mean Reciprocal Rank, MRR)

        衡量第一个相关项目出现的位置。
        越靠前，分数越高。

        公式: MRR = (1/position of first relevant item)

        示例:
        - [True, False, False] -> 1/1 = 1.0 (最优)
        - [False, True, False] -> 1/2 = 0.5
        - [False, False, False] -> 0.0 (无相关项)
        """

        for i, is_relevant in enumerate(retrieved_items[:threshold_position]):
            if is_relevant:
                return 1.0 / (i + 1)

        return 0.0

    @staticmethod
    def recall_at_k(
        retrieved_items: List[bool],
        relevant_item_count: int,
        k: int = 10
    ) -> float:
        """
        召回率 (Recall@k)

        在前k个结果中，找到了多少个相关项。

        公式: Recall@k = (前k个结果中的相关数) / 总相关数

        示例:
        - 总共10个相关项，前20个结果中找到8个 -> 8/10 = 0.8
        - 总共5个相关项，前10个结果中找到3个 -> 3/5 = 0.6
        """

        if relevant_item_count == 0:
            return 0.0

        relevant_in_top_k = sum(retrieved_items[:k])
        return relevant_in_top_k / relevant_item_count

    @staticmethod
    def precision_at_k(
        retrieved_items: List[bool],
        k: int = 10
    ) -> float:
        """
        精确度 (Precision@k)

        前k个结果中有多少比例是相关的。

        公式: Precision@k = (前k个中的相关数) / k

        示例:
        - 前10个结果中有8个相关 -> 8/10 = 0.8
        - 前5个结果中有1个相关 -> 1/5 = 0.2
        """

        if k == 0:
            return 0.0

        relevant_in_top_k = sum(retrieved_items[:k])
        return relevant_in_top_k / k

    @staticmethod
    def f1_score(
        retrieved_items: List[bool],
        relevant_item_count: int,
        k: int = 10
    ) -> float:
        """
        F1分数 (F1 Score)

        Precision和Recall的调和平均。
        在精确度和召回率之间找到平衡。

        公式: F1 = 2 * (Precision * Recall) / (Precision + Recall)
        """

        precision = RelevanceScorer.precision_at_k(retrieved_items, k)
        recall = RelevanceScorer.recall_at_k(retrieved_items, relevant_item_count, k)

        if precision + recall == 0:
            return 0.0

        return 2 * (precision * recall) / (precision + recall)

    @staticmethod
    def ndcg(
        relevance_scores: List[float],
        k: int = 10,
        ideal_dcg: float = None
    ) -> float:
        """
        归一化折损累计增益 (Normalized Discounted Cumulative Gain, NDCG)

        考虑相关性的"程度"（不只是相关/不相关）。
        排名靠前的相关项贡献更大。

        公式:
        DCG@k = sum(relevance[i] / log2(i+2)) for i in 0..k-1
        NDCG@k = DCG@k / IDCG@k

        其中IDCG是完美排序下的DCG。

        示例:
        relevance_scores = [0.9, 0.7, 0.0, 0.5, 0.8]
        DCG@5 = 0.9/1 + 0.7/log2(3) + 0/log2(4) + 0.5/log2(5) + 0.8/log2(6)
        """

        if not relevance_scores or k == 0:
            return 0.0

        # 计算实际的DCG
        dcg = 0.0
        for i, score in enumerate(relevance_scores[:k]):
            # 注意：log2(i+2)表示从位置1开始计数
            dcg += score / np.log2(i + 2)

        # 计算理想的DCG（如果没有给出，则用排序后的最优值）
        if ideal_dcg is None:
            sorted_scores = sorted(relevance_scores, reverse=True)
            ideal_dcg = 0.0
            for i, score in enumerate(sorted_scores[:k]):
                ideal_dcg += score / np.log2(i + 2)

        if ideal_dcg == 0:
            return 0.0

        return dcg / ideal_dcg


# 使用示例
if __name__ == "__main__":
    # 场景：搜索"Python教程"，返回了10个结果
    # 其中标注的相关性：
    retrieved = [
        True,   # 位置1: 很相关
        False,  # 位置2: 不相关
        True,   # 位置3: 相关
        False,  # 位置4: 不相关
        False,  # 位置5: 不相关
        True,   # 位置6: 相关
        False,  # 位置7: 不相关
        False,  # 位置8: 不相关
        True,   # 位置9: 相关
        False,  # 位置10: 不相关
    ]

    total_relevant_count = 6  # 来自黄金标注：语料库中共有6个相关项

    scorer = RelevanceScorer()

    mrr = scorer.mean_reciprocal_rank(retrieved)
    recall = scorer.recall_at_k(retrieved, total_relevant_count, k=10)
    precision = scorer.precision_at_k(retrieved, k=10)
    f1 = scorer.f1_score(retrieved, total_relevant_count, k=10)

    print(f"MRR: {mrr:.2f}")        # 应该是1.0（第1个位置）
    print(f"Recall@10: {recall:.2f}")  # 应该是0.67（找到6个相关项中的4个）
    print(f"Precision@10: {precision:.2f}")  # 应该是0.4（4/10）
    print(f"F1: {f1:.2f}")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68

class GradedRelevanceScorer:
    """处理分级相关性的评估工具"""

    # 常见的相关性等级
    RELEVANCE_GRADES = {
        0: "完全不相关",
        1: "部分相关",
        2: "相关",
        3: "高度相关",
    }

    @staticmethod
    def dcg_with_grades(
        relevance_grades: List[int],  # [3, 2, 0, 1, 3, ...]
        k: int = 10
    ) -> float:
        """
        使用分级相关性计算DCG。
        """

        dcg = 0.0
        for i, grade in enumerate(relevance_grades[:k]):
            # 标准NDCG使用 (2^grade - 1) 作为相关性权重
            weight = (2 ** grade) - 1
            dcg += weight / np.log2(i + 2)

        return dcg

    @staticmethod
    def ndcg_with_grades(
        relevance_grades: List[int],
        k: int = 10
    ) -> float:
        """
        使用分级相关性计算NDCG。
        """

        dcg = GradedRelevanceScorer.dcg_with_grades(relevance_grades, k)

        # 理想的DCG：把同一批候选的人工相关性等级按从高到低排序
        ideal_grades = sorted(relevance_grades, reverse=True)[:k]
        ideal_dcg = GradedRelevanceScorer.dcg_with_grades(ideal_grades, k)

        if ideal_dcg == 0:
            return 0.0

        return dcg / ideal_dcg

    @staticmethod
    def metric_at_each_level(
        relevance_grades: List[int],
        k: int = 10
    ) -> dict:
        """
        计算每个相关性等级的指标分布。
        """

        grade_counts = {grade: 0 for grade in range(4)}

        for grade in relevance_grades[:k]:
            grade_counts[grade] += 1

        return {
            'not_relevant': grade_counts[0],
            'partially_relevant': grade_counts[1],
            'relevant': grade_counts[2],
            'highly_relevant': grade_counts[3],
        }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94

class RetrievalQualityEvaluator:
    """综合的检索质量评估"""

    def __init__(self):
        self.scorer = RelevanceScorer()

    def evaluate_retrieval(
        self,
        query: str,
        retrieved_results: List[dict],  # [{'id': '...', 'is_relevant': True/False, 'relevance_score': 0.8}]
        total_relevant_count: int,
        top_k_values: List[int] = None
    ) -> dict:
        """
        对单个查询的检索结果进行多维度评估。

        total_relevant_count 必须来自黄金标注或完整候选集，不能从 retrieved_results 推断。
        """

        if top_k_values is None:
            top_k_values = [1, 5, 10]

        # 提取相关性标签
        is_relevant = [r.get('is_relevant', False) for r in retrieved_results]
        relevance_scores = [r.get('relevance_score', 0.0) for r in retrieved_results]

        evaluation = {
            'query': query,
            'total_retrieved': len(retrieved_results),
            'total_relevant': total_relevant_count,
        }

        # 计算多个k值下的指标
        for k in top_k_values:
            k_metrics = {
                'mrr': self.scorer.mean_reciprocal_rank(is_relevant, k),
                'recall': self.scorer.recall_at_k(is_relevant, total_relevant_count, k),
                'precision': self.scorer.precision_at_k(is_relevant, k),
                'f1': self.scorer.f1_score(is_relevant, total_relevant_count, k),
            }

            evaluation[f'metrics@{k}'] = k_metrics

        return evaluation

    def evaluate_retrieval_batch(
        self,
        queries: List[str],
        retrieved_results_batch: List[List[dict]],
        total_relevant_counts: List[int],
        top_k_values: List[int] = None
    ) -> dict:
        """
        对多个查询的检索结果进行批量评估，并计算平均指标。
        """

        if top_k_values is None:
            top_k_values = [5, 10]

        individual_results = []
        aggregated_metrics = {f'@{k}': {} for k in top_k_values}

        for query, results, total_relevant_count in zip(queries, retrieved_results_batch, total_relevant_counts):
            indiv = self.evaluate_retrieval(query, results, total_relevant_count, top_k_values)
            individual_results.append(indiv)

            # 累积指标
            for k in top_k_values:
                k_str = f'@{k}'
                k_metrics = indiv[f'metrics@{k}']

                for metric_name, metric_value in k_metrics.items():
                    if metric_name not in aggregated_metrics[k_str]:
                        aggregated_metrics[k_str][metric_name] = []

                    aggregated_metrics[k_str][metric_name].append(metric_value)

        # 计算平均值
        for k_str in aggregated_metrics:
            for metric_name in aggregated_metrics[k_str]:
                values = aggregated_metrics[k_str][metric_name]
                aggregated_metrics[k_str][metric_name] = {
                    'mean': np.mean(values),
                    'std': np.std(values),
                    'min': np.min(values),
                    'max': np.max(values),
                }

        return {
            'individual_results': individual_results,
            'aggregated_metrics': aggregated_metrics,
            'total_queries': len(queries),
            'average_retrieval_count': np.mean([r['total_retrieved'] for r in individual_results]),
        }

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153

class ContextEfficiencyAnalyzer:
    """分析上下文的使用效率"""

    @staticmethod
    def information_density(
        context: str,
        relevant_spans: List[str]
    ) -> float:
        """
        信息密度 = 相关信息量 / 总信息量

        衡量上下文中"有用信息"的比例。

        relevant_spans 是上下文中被判定为“与当前任务相关”的片段原文，
        来自人工标注或评估模型。密度必须由两侧的实际规模算出：只传入一个
        现成的 relevance_score 再乘回总量，分子分母会约掉，得到的永远是
        传进来的那个分数本身，与 context 无关。
        """

        total_tokens = len(context) // 4  # 粗略估计
        if total_tokens <= 0:
            return 0.0

        useful_tokens = sum(len(span) // 4 for span in relevant_spans)

        return min(1.0, useful_tokens / total_tokens)

    @staticmethod
    def redundancy_ratio(
        context: str
    ) -> float:
        """
        冗余度 = 重复内容的比例

        通过n-gram重复频率估计冗余度。
        """

        # 简化：提取所有5-gram并统计重复
        words = context.split()

        if len(words) < 5:
            return 0.0

        ngrams = []
        for i in range(len(words) - 4):
            ngram = ' '.join(words[i:i+5])
            ngrams.append(ngram)

        unique_ngrams = len(set(ngrams))
        total_ngrams = len(ngrams)

        # 冗余度 = 1 - (unique / total)
        redundancy = 1 - (unique_ngrams / total_ngrams) if total_ngrams > 0 else 0

        return redundancy

    @staticmethod
    def context_coverage_ratio(
        context: str,
        query: str
    ) -> float:
        """
        覆盖率 = 查询中被上下文覆盖的关键词比例

        衡量上下文对查询的覆盖程度。
        """

        query_keywords = set(query.lower().split())
        context_lower = context.lower()

        covered = sum(1 for kw in query_keywords if kw in context_lower)
        total = len(query_keywords)

        return covered / total if total > 0 else 0

    @staticmethod
    def efficiency_score(
        context: str,
        query: str,
        relevant_spans: List[str],
        max_tokens: int = 2000
    ) -> float:
        """
        综合效率分数 (0-100)

        综合考虑：
        - 信息密度（越高越好）
        - 冗余度（越低越好）
        - 覆盖率（越高越好）
        - Token利用率（在预算内）

        四项里只有信息密度需要外部标注（relevant_spans），其余三项都能
        从 context 和 query 直接算出。权重是可调的工程约定，不是通用常数。
        """

        density = ContextEfficiencyAnalyzer.information_density(context, relevant_spans)
        redundancy = ContextEfficiencyAnalyzer.redundancy_ratio(context)
        coverage = ContextEfficiencyAnalyzer.context_coverage_ratio(context, query)

        # Token利用率
        context_tokens = len(context) // 4
        token_utilization = min(1.0, context_tokens / max_tokens)

        # 加权综合
        efficiency = (
            density * 0.35 +           # 信息密度权重35%
            (1 - redundancy) * 0.25 +  # 非冗余度权重25%
            coverage * 0.25 +           # 覆盖率权重25%
            token_utilization * 0.15   # Token利用率权重15%
        )

        # 规范化到0-100
        return efficiency * 100


# 使用示例
if __name__ == "__main__":
    evaluator = RetrievalQualityEvaluator()

    # 评估单个查询
    query = "Python教程"
    results = [
        {'id': '1', 'is_relevant': True, 'relevance_score': 0.95},
        {'id': '2', 'is_relevant': False, 'relevance_score': 0.2},
        {'id': '3', 'is_relevant': True, 'relevance_score': 0.85},
        {'id': '4', 'is_relevant': False, 'relevance_score': 0.15},
        {'id': '5', 'is_relevant': True, 'relevance_score': 0.8},
    ]

    total_relevant_count = 4  # 来自黄金标注：语料库中共有4个相关文档
    evaluation = evaluator.evaluate_retrieval(query, results, total_relevant_count)

    print("单个查询评估:")
    print(f"  总相关: {evaluation['total_relevant']}")
    print(f"  Recall@5: {evaluation['metrics@5']['recall']:.2f}")
    print(f"  Precision@5: {evaluation['metrics@5']['precision']:.2f}")

    # 评估效率
    context = """
    Python是一种高级编程语言，以其简洁易读的语法而闻名。
    Python支持多种编程范式，包括过程式、面向对象和函数式编程。
    Python广泛用于数据科学、机器学习和Web开发。
    Python有一个庞大的标准库和第三方库生态系统。
    """

    # 标注出上下文里真正回答了"Python教程"的片段
    relevant_spans = [
        "Python是一种高级编程语言，以其简洁易读的语法而闻名。",
        "Python支持多种编程范式，包括过程式、面向对象和函数式编程。",
    ]

    efficiency = ContextEfficiencyAnalyzer.efficiency_score(context, query, relevant_spans)
    print(f"\n上下文效率分数: {efficiency:.1f}/100")

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153

from scipy import stats

class ABTestFramework:
    """
    A/B测试框架
    用于统计显著性验证
    """

    @staticmethod
    def paired_ttest(
        control_group: List[float],
        treatment_group: List[float],
        alpha: float = 0.05
    ) -> dict:
        """
        配对t检验

        同一批查询分别跑 control 和 treatment 时，应该比较逐查询差值。
        """

        t_stat, p_value = stats.ttest_rel(control_group, treatment_group)

        is_significant = p_value < alpha

        return {
            't_statistic': t_stat,
            'p_value': p_value,
            'is_significant': is_significant,
            'control_mean': np.mean(control_group),
            'treatment_mean': np.mean(treatment_group),
            'improvement': (
                (np.mean(treatment_group) - np.mean(control_group)) /
                np.mean(control_group) * 100
                if np.mean(control_group) != 0 else 0
            ),
        }

    @staticmethod
    def cohen_d(
        control_group: List[float],
        treatment_group: List[float]
    ) -> float:
        """
        计算效果量 (Effect Size)

        衡量两组差异的实际大小（不只是统计显著性）。
        """

        deltas = np.array(treatment_group) - np.array(control_group)
        mean_diff = np.mean(deltas)

        pooled_std = np.std(deltas)

        if pooled_std == 0:
            return 0

        return mean_diff / pooled_std

    @staticmethod
    def calculate_sample_size(
        effect_size: float = 0.5,  # Cohen's d
        alpha: float = 0.05,
        power: float = 0.8
    ) -> int:
        """
        计算所需样本量

        为了达到统计显著性，需要多少样本？
        """

        # 简化公式
        z_alpha = stats.norm.ppf(1 - alpha / 2)
        z_beta = stats.norm.ppf(power)

        sample_size = int(
            2 * ((z_alpha + z_beta) / effect_size) ** 2
        )

        return sample_size


class ContextOptimizationABTest:
    """上下文优化的A/B测试设计"""

    def __init__(self, control_strategy: str, treatment_strategy: str):
        self.control_strategy = control_strategy
        self.treatment_strategy = treatment_strategy
        self.control_results = []
        self.treatment_results = []

    def run_test(
        self,
        queries: List[str],
        ground_truth_answers: List[str],
        llm_generate_func
    ) -> dict:
        """
        执行A/B测试。

        control_strategy: 原始的上下文工程方案
        treatment_strategy: 新的上下文工程方案
        """

        print(f"Running A/B Test: {self.control_strategy} vs {self.treatment_strategy}")

        # 测试Control组
        for query, ground_truth in zip(queries, ground_truth_answers):
            context = self._prepare_context(query, strategy=self.control_strategy)
            answer = llm_generate_func(context, query)
            similarity = self._similarity(answer, ground_truth)
            self.control_results.append(similarity)

        # 测试Treatment组
        for query, ground_truth in zip(queries, ground_truth_answers):
            context = self._prepare_context(query, strategy=self.treatment_strategy)
            answer = llm_generate_func(context, query)
            similarity = self._similarity(answer, ground_truth)
            self.treatment_results.append(similarity)

        # 统计分析
        test_results = ABTestFramework.paired_ttest(
            self.control_results,
            self.treatment_results
        )

        effect_size = ABTestFramework.cohen_d(
            self.control_results,
            self.treatment_results
        )

        test_results['effect_size'] = effect_size

        return test_results

    def _prepare_context(self, query: str, strategy: str) -> str:
        """准备上下文（这里是简化实现）"""
        # 实际应该调用不同的上下文处理流程
        return f"Using {strategy}: {query}"

    def _similarity(self, text1: str, text2: str) -> float:
        """计算两个文本的相似度"""
        # 简化：基于词重叠。生产评估应换成经过校准的语义指标、
        # LLM-as-a-judge 量表或人工黄金标注，避免把复述当成正确性。
        words1 = set(text1.lower().split())
        words2 = set(text2.lower().split())

        if not words1 or not words2:
            return 0.0

        overlap = len(words1 & words2)
        union = len(words1 | words2)

        return overlap / union