feat: summary新增label_stats属性

dingo/exec/local.py

@@ -134,6 +134,7 @@ def execute(self) -> SummaryModel:
                         for label in label_set:
                             self.summary.type_count[field_key].setdefault(label, 0)
                             self.summary.type_count[field_key][label] += 1
+                        self.summary.add_label_sample(field_key, label_set)
 
                     for field_key, eval_detail_list in result_info.token_usage_details.items():
                         for eval_detail in eval_detail_list:
@@ -291,10 +292,12 @@ def summarize(
     ) -> SummaryModel:
         new_summary = copy.deepcopy(summary)
         if new_summary.total == 0:
+            new_summary.calculate_label_stats_ratios()
             return new_summary
         new_summary.score = round(new_summary.num_good / new_summary.total * 100, 2)
         if refresh_type_ratio:
             new_summary = self.refresh_type_ratio(new_summary)
+        new_summary.calculate_label_stats_ratios()
 
         # 计算指标分数的平均值、最小值、最大值、标准差等
         new_summary.calculate_metrics_score_averages()
@@ -436,6 +439,7 @@ def write_summary(self, path: str, input_args: InputArgs, summary: SummaryModel)
 
     def get_summary(self):
         self.summary = self.refresh_type_ratio(self.summary)
+        self.summary.calculate_label_stats_ratios()
         return self.summary
 
     def get_info_list(self, high_quality: bool) -> list:

dingo/exec/spark.py

@@ -9,7 +9,7 @@
 
 from dingo.config import InputArgs
 from dingo.exec.base import ExecProto, Executor
-from dingo.io import Data, ResultInfo, SummaryModel
+from dingo.io import Data, LabelStat, ResultInfo, SummaryModel
 from dingo.io.output.eval_detail import EvalDetail
 from dingo.model import Model
 
@@ -81,6 +81,7 @@ def _aggregate_eval_details(acc, item):
         """聚合单个 item 的 eval_details 到累加器中，同时收集 scores"""
         eval_details_dict = item.get('eval_details', {})
         statistics_details = item.get('_statistics_details', {})
+        label_path_counts = acc.setdefault('label_path_counts', {})
 
         # 遍历第一层：字段名，第二层是 List[EvalDetail] (序列化为 list of dicts)
         for field_key, eval_detail_list in eval_details_dict.items():
@@ -89,6 +90,8 @@ def _aggregate_eval_details(acc, item):
                 acc['label_counts'][field_key] = {}
             if field_key not in acc['metric_scores']:
                 acc['metric_scores'][field_key] = {}
+            if field_key not in label_path_counts:
+                label_path_counts[field_key] = {}
 
             # 遍历 List[EvalDetail]，同时收集指标分数和标签
             label_set = set()
@@ -115,6 +118,16 @@ def _aggregate_eval_details(acc, item):
                 else:
                     acc['label_counts'][field_key][label] += 1
 
+            label_path_set = {
+                path
+                for label in label_set
+                for path in SummaryModel.expand_label_paths(label)
+            }
+            for label_path in label_path_set:
+                label_path_counts[field_key][label_path] = (
+                    label_path_counts[field_key].get(label_path, 0) + 1
+                )
+
         statistics_counts = acc.setdefault('statistics', {})
         for field_key, eval_detail_list in statistics_details.items():
             field_counts = statistics_counts.setdefault(field_key, {})
@@ -147,6 +160,12 @@ def _merge_eval_details(acc1, acc2):
                     else:
                         acc1['label_counts'][field_key][label] += count
 
+        target_label_path_counts = acc1.setdefault('label_path_counts', {})
+        for field_key, label_dict in acc2.get('label_path_counts', {}).items():
+            target = target_label_path_counts.setdefault(field_key, {})
+            for label, count in label_dict.items():
+                target[label] = target.get(label, 0) + count
+
         # 合并 metric scores（按 field_key 分组）
         for field_key, metrics_dict in acc2['metric_scores'].items():
             if field_key not in acc1['metric_scores']:
@@ -341,17 +360,20 @@ def summarize(self, summary: SummaryModel) -> SummaryModel:
             aggregated_results = self.data_info_list.aggregate(
                 {
                     'label_counts': {},
+                    'label_path_counts': {},
                     'metric_scores': {},
                     'statistics': {},
                 },  # 初始累加器
                 SparkExecutor._aggregate_eval_details,  # 聚合单个元素
                 SparkExecutor._merge_eval_details  # 合并累加器
             )
             type_ratio_counts = aggregated_results['label_counts']
+            label_path_counts = aggregated_results['label_path_counts']
             metric_scores = aggregated_results['metric_scores']
             statistics = aggregated_results['statistics']
         else:
             type_ratio_counts = {}
+            label_path_counts = {}
             metric_scores = {}
             statistics = {}
 
@@ -364,6 +386,18 @@ def summarize(self, summary: SummaryModel) -> SummaryModel:
                     type_ratio_counts[field_name][eval_details] / new_summary.total, 6
                 )
 
+        new_summary.label_stats = {
+            field_name: [
+                LabelStat(
+                    label=label,
+                    sample_count=count,
+                    sample_ratio=round(count / new_summary.total, 6),
+                )
+                for label, count in sorted(label_counts.items())
+            ]
+            for field_name, label_counts in sorted(label_path_counts.items())
+        }
+
         # 添加收集到的 metric scores 到 summary（按 field_key 分组）
         for field_key, metrics in metric_scores.items():
             for metric_name, scores in metrics.items():

dingo/io/__init__.py

@@ -1,3 +1,3 @@
 from dingo.io.input.data import Data  # noqa E402.
 from dingo.io.output.result_info import ResultInfo  # noqa E402.
-from dingo.io.output.summary_model import SummaryModel  # noqa E402.
+from dingo.io.output.summary_model import LabelStat, SummaryModel  # noqa E402.

dingo/io/output/__init__.py

@@ -1,3 +1,3 @@
 # from dingo.io.output.benchmark_report import BenchmarkReport  # noqa E402.
 from dingo.io.output.result_info import ResultInfo  # noqa E402.
-from dingo.io.output.summary_model import SummaryModel  # noqa E402.
+from dingo.io.output.summary_model import LabelStat, SummaryModel  # noqa E402.

dingo/io/output/summary_model.py

@@ -1,11 +1,17 @@
 import statistics
-from typing import Any, Dict
+from typing import Any, Dict, Iterable, List
 
 from pydantic import BaseModel, Field
 
 from dingo.io.output.eval_detail import TokenUsage
 
 
+class LabelStat(BaseModel):
+    label: str
+    sample_count: int
+    sample_ratio: float
+
+
 class SummaryModel(BaseModel):
     task_id: str = ''
     task_name: str = ''
@@ -20,6 +26,7 @@ class SummaryModel(BaseModel):
     total: int = 0
     type_count: Dict[str, Dict[str, int]] = Field(default_factory=dict)
     type_ratio: Dict[str, Dict[str, float]] = Field(default_factory=dict)
+    label_stats: Dict[str, List[LabelStat]] = Field(default_factory=dict)
 
     # 新增：指标分数统计（用于RAG等评估场景）
     # 结构：{field_key: {metric_name: {scores, score_average, ...}}}
@@ -28,6 +35,51 @@ class SummaryModel(BaseModel):
     statistics: Dict[str, Dict[str, Dict[str, int]]] = Field(default_factory=dict)
     input_args: Dict[str, Any] = Field(default_factory=dict)
 
+    @staticmethod
+    def expand_label_paths(label: str) -> List[str]:
+        """Expand one label into its dot-delimited parent paths."""
+        parts = label.split('.')
+        return ['.'.join(parts[:index]) for index in range(1, len(parts) + 1)]
+
+    def add_label_sample(self, field_key: str, labels: Iterable[str]):
+        """Count each label path once for one sample and field."""
+        label_paths = {
+            path
+            for label in labels
+            for path in self.expand_label_paths(label)
+        }
+        if not label_paths:
+            return
+        stats_by_label = {
+            stat.label: stat
+            for stat in self.label_stats.setdefault(field_key, [])
+        }
+        for label_path in label_paths:
+            stat = stats_by_label.get(label_path)
+            if stat is None:
+                stat = LabelStat(
+                    label=label_path,
+                    sample_count=0,
+                    sample_ratio=0.0,
+                )
+                self.label_stats[field_key].append(stat)
+                stats_by_label[label_path] = stat
+            stat.sample_count += 1
+
+    def calculate_label_stats_ratios(self):
+        """Refresh label ratios after the summary total is known."""
+        for field_key, stats in self.label_stats.items():
+            for stat in stats:
+                stat.sample_ratio = (
+                    round(stat.sample_count / self.total, 6)
+                    if self.total > 0
+                    else 0.0
+                )
+            self.label_stats[field_key] = sorted(
+                stats,
+                key=lambda stat: stat.label,
+            )
+
     def add_metric_score(self, field_key: str, metric_name: str, score: float):
         """
         添加指标分数到统计中
@@ -181,6 +233,13 @@ def to_dict(self):
             field_key: dict(sorted(type_ratios.items()))
             for field_key, type_ratios in sorted(self.type_ratio.items())
         }
+        sorted_label_stats = {
+            field_key: [
+                stat.model_dump()
+                for stat in sorted(stats, key=lambda item: item.label)
+            ]
+            for field_key, stats in sorted(self.label_stats.items())
+        }
         result = {
             'task_id': self.task_id,
             'task_name': self.task_name,
@@ -195,6 +254,7 @@ def to_dict(self):
             'total': self.total,
             'type_count': sorted_type_count,
             'type_ratio': sorted_type_ratio,
+            'label_stats': sorted_label_stats,
         }
 
         # 如果有指标分数统计，以层级结构添加到输出中（与 type_ratio 结构一致）

test/data/test_local_jsonl.jsonl

@@ -1,2 +1,4 @@
 {"id": 0, "content": "�I am 8 years old. ^I love apple because:"}
 {"id": 1, "content": "[I like blue best. Because blue is the color of the sky. I like blue best. Because blue is the color of the sky. I like blue best. Because blue is the color of the sky.I like blue best. Because blue is the color of the sky.I like blue best. Because blue is the color of the sky.I like blue best. Because blue is the color of the sky.I like blue best. Because blue is the color of the sky.I like blue best. Because blue is the color of the sky.I like blue best. Because blue is the color of the sky.I like blue best. Because blue is the color of the sky. "}
+{"id": 0, "content": "�I am 8 years old. ^I love apple because:"}
+{"id": 1, "content": "[I like blue best. Because blue is the color of the sky. I like blue best. Because blue is the color of the sky. I like blue best. Because blue is the color of the sky.I like blue best. Because blue is the color of the sky.I like blue best. Because blue is the color of the sky.I like blue best. Because blue is the color of the sky.I like blue best. Because blue is the color of the sky.I like blue best. Because blue is the color of the sky.I like blue best. Because blue is the color of the sky.I like blue best. Because blue is the color of the sky. "}

test/scripts/exec/test_local.py

@@ -420,6 +420,14 @@ def test_all_labels_config(self):
         assert all([item in result.type_ratio.get('content') for item in ["QUALITY_BAD_EFFECTIVENESS.RuleColonEnd",
                                                       "QUALITY_BAD_EFFECTIVENESS.RuleSpecialCharacter",
                                                       "QUALITY_GOOD"]])
+        label_stats = {
+            item.label: item
+            for item in result.label_stats["content"]
+        }
+        assert label_stats["QUALITY_BAD_EFFECTIVENESS"].sample_count == 1
+        assert label_stats["QUALITY_BAD_EFFECTIVENESS"].sample_ratio == 1.0
+        assert label_stats["QUALITY_GOOD"].sample_count == 1
+        assert "Effectiveness" not in label_stats
 
         input_data["executor"]["result_save"]["all_labels"] = False
         input_args = InputArgs(**input_data)

test/scripts/exec/test_spark.py

@@ -77,6 +77,35 @@ def test_aggregate_eval_details_with_scores(self):
         assert acc['metric_scores']['field1']['LLMRAGFaithfulness'] == [9.5, 8.3]
         assert 'field1' in acc['label_counts']
         assert acc['label_counts']['field1']['good'] == 2
+        assert acc['label_path_counts']['field1']['good'] == 2
+
+    def test_aggregate_label_paths_deduplicates_per_item(self):
+        item = {
+            "eval_details": {
+                "content": [
+                    {
+                        "label": [
+                            "Completeness.Table_Header_Corruption",
+                            "Completeness.Table_Cell_Corruption",
+                            "QUALITY_GOOD",
+                            "QUALITY_BAD_EFFECTIVENESS.RuleColonEnd",
+                        ]
+                    }
+                ]
+            }
+        }
+        acc = {'label_counts': {}, 'metric_scores': {}}
+
+        result = SparkExecutor._aggregate_eval_details(acc, item)
+
+        counts = result['label_path_counts']['content']
+        assert counts['Completeness'] == 1
+        assert counts['Completeness.Table_Header_Corruption'] == 1
+        assert counts['Completeness.Table_Cell_Corruption'] == 1
+        assert counts['QUALITY_GOOD'] == 1
+        assert counts['QUALITY_BAD_EFFECTIVENESS'] == 1
+        assert counts['QUALITY_BAD_EFFECTIVENESS.RuleColonEnd'] == 1
+        assert 'Effectiveness' not in counts
 
     def test_merge_eval_details_with_scores(self):
         """测试合并函数正确合并多个累加器的分数"""
@@ -263,6 +292,15 @@ def mock_aggregate(init_acc, seq_func, comb_func):
         assert stats['score_average'] == 8.9  # (9.5 + 8.3) / 2
         assert stats['score_min'] == 8.3
         assert stats['score_max'] == 9.5
+        label_stats = {
+            item["label"]: item
+            for item in result_dict["label_stats"]["field1"]
+        }
+        assert label_stats["good"] == {
+            "label": "good",
+            "sample_count": 2,
+            "sample_ratio": 1.0,
+        }
 
     def test_spark_executor_summarize_multiple_metrics(self):
         """测试 SparkExecutor.summarize 处理多个指标"""

test/scripts/io/test_summary_model.py

@@ -19,6 +19,45 @@
 class TestSummaryModel:
     """测试 SummaryModel 的指标分数统计功能"""
 
+    def test_label_stats_expand_paths_and_deduplicate_per_sample(self):
+        summary = SummaryModel(total=2)
+
+        summary.add_label_sample(
+            "content",
+            [
+                "Completeness.Table_Header_Corruption",
+                "Completeness.Table_Cell_Corruption",
+                "QUALITY_GOOD",
+            ],
+        )
+        summary.add_label_sample(
+            "content",
+            [
+                "Completeness.Table_Header_Corruption",
+                "QUALITY_BAD_EFFECTIVENESS.RuleColonEnd",
+            ],
+        )
+        summary.calculate_label_stats_ratios()
+
+        stats = {
+            item.label: item
+            for item in summary.label_stats["content"]
+        }
+        assert stats["Completeness"].sample_count == 2
+        assert stats["Completeness"].sample_ratio == 1.0
+        assert stats["Completeness.Table_Header_Corruption"].sample_count == 2
+        assert stats["Completeness.Table_Cell_Corruption"].sample_count == 1
+        assert stats["QUALITY_GOOD"].sample_count == 1
+        assert stats["QUALITY_BAD_EFFECTIVENESS"].sample_count == 1
+        assert (
+            stats["QUALITY_BAD_EFFECTIVENESS.RuleColonEnd"].sample_count
+            == 1
+        )
+        assert "Effectiveness" not in stats
+
+        serialized = summary.to_dict()["label_stats"]["content"]
+        assert [item["label"] for item in serialized] == sorted(stats)
+
     def test_input_args_are_serialized(self):
         input_args = InputArgs(**{
             "input_path": "data.jsonl",