Explorar el Código

增加测试报告工具

wangyang hace 5 días
padre
commit
68f0915655
Se han modificado 1 ficheros con 354 adiciones y 0 borrados
  1. 354 0
      llmops/folder_scanner.py

+ 354 - 0
llmops/folder_scanner.py

@@ -0,0 +1,354 @@
+import os
+import logging
+from typing import Dict, List, Optional
+from dataclasses import dataclass
+
+# 配置日志
+logging.basicConfig(
+    level=logging.INFO,
+    format='%(asctime)s - %(levelname)s - %(message)s'
+)
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class ScanConfig:
+    """扫描配置"""
+    # 决策agent的期望数量,-1表示不校验
+    expected_decision_count: int = -1
+    # 大纲生成agent的期望数量,-1表示不校验
+    expected_outline_count: int = -1
+    # 指标计算agent的期望数量,-1表示不校验
+    expected_metric_count: int = -1
+
+
+@dataclass
+class ScanResult:
+    """单个文件夹扫描结果"""
+    folder_name: str
+    decision_files: List[str]
+    outline_files: List[str]
+    metric_files: List[str]
+    success_metric_files: List[str]
+
+    # 实际数量
+    actual_decision_count: int
+    actual_outline_count: int
+    actual_metric_count: int
+    actual_success_metric_count: int
+
+    # 准确率(如果配置了期望值)
+    decision_accuracy: Optional[float] = None
+    outline_accuracy: Optional[float] = None
+    metric_accuracy: Optional[float] = None
+
+
+class FolderScanner:
+    """文件夹扫描工具类"""
+
+    def __init__(self, base_path: str = "."):
+        """
+        初始化扫描器
+
+        Args:
+            base_path: 基础路径,默认为当前目录
+        """
+        self.base_path = base_path
+        self.results: List[ScanResult] = []
+
+    def scan_folders(self, folder_count: int, config: Optional[ScanConfig] = None) -> Dict:
+        """
+        扫描指定数量的文件夹
+
+        Args:
+            folder_count: 要扫描的文件夹数量
+            config: 扫描配置,包含各agent的期望数量
+
+        Returns:
+            扫描统计结果
+        """
+        if config is None:
+            config = ScanConfig()
+
+        logger.info(f"开始扫描 {folder_count} 个文件夹...")
+        logger.info(f"期望配置: 决策agent={config.expected_decision_count}, "
+                    f"大纲agent={config.expected_outline_count}, "
+                    f"指标agent={config.expected_metric_count}")
+
+        self.results = []
+
+        for i in range(1, folder_count + 1):
+            folder_name = f"api_results_{i}"
+            folder_path = os.path.join(self.base_path, folder_name)
+
+            if not os.path.exists(folder_path):
+                logger.warning(f"文件夹不存在: {folder_path}")
+                continue
+
+            result = self._scan_single_folder(folder_name, folder_path, config)
+            self.results.append(result)
+
+        return self._generate_report(config)
+
+    def _calculate_accuracy(self, actual: int, expected: int) -> float:
+        """计算准确率"""
+        if actual == expected:
+            return 100.0
+        elif actual < expected:
+            # 实际少于期望
+            return (actual / expected) * 100
+        else:
+            # 实际多于期望,也应该计算偏差
+            # 比如期望3个,实际4个,准确率 = 3/4 * 100 = 75%
+            return (expected / actual) * 100
+
+    def _scan_single_folder(self, folder_name: str, folder_path: str, config: ScanConfig) -> ScanResult:
+        """扫描单个文件夹"""
+        all_files = os.listdir(folder_path)
+
+        # 分类文件
+        decision_files = []
+        outline_files = []
+        metric_files = []
+        success_metric_files = []
+
+        for file in all_files:
+            if file.endswith('.md'):
+                if '规划决策' in file:
+                    decision_files.append(file)
+                elif '大纲生成' in file:
+                    outline_files.append(file)
+            elif file.endswith('.json'):
+                metric_files.append(file)
+                if '_success.json' in file:
+                    success_metric_files.append(file)
+
+        # 计算实际数量
+        actual_decision_count = len(decision_files)
+        actual_outline_count = len(outline_files)
+        actual_metric_count = len(metric_files)
+        actual_success_metric_count = len(success_metric_files)
+
+        # 计算准确率(如果配置了期望值)
+        decision_accuracy = None
+        outline_accuracy = None
+        metric_accuracy = None
+
+        if config.expected_decision_count != -1 and config.expected_decision_count > 0:
+            decision_accuracy = self._calculate_accuracy(actual_decision_count, config.expected_decision_count)
+
+        if config.expected_outline_count != -1 and config.expected_outline_count > 0:
+            outline_accuracy = self._calculate_accuracy(actual_outline_count, config.expected_outline_count)
+
+        if config.expected_metric_count != -1 and config.expected_metric_count > 0:
+            metric_accuracy = self._calculate_accuracy(actual_metric_count, config.expected_metric_count)
+
+        return ScanResult(
+            folder_name=folder_name,
+            decision_files=decision_files,
+            outline_files=outline_files,
+            metric_files=metric_files,
+            success_metric_files=success_metric_files,
+            actual_decision_count=actual_decision_count,
+            actual_outline_count=actual_outline_count,
+            actual_metric_count=actual_metric_count,
+            actual_success_metric_count=actual_success_metric_count,
+            decision_accuracy=decision_accuracy,
+            outline_accuracy=outline_accuracy,
+            metric_accuracy=metric_accuracy
+        )
+
+    def _generate_report(self, config: ScanConfig) -> Dict:
+        """生成报告"""
+        total_folders = len(self.results)
+
+        # 计算平均准确率
+        avg_decision_accuracy = 0
+        avg_outline_accuracy = 0
+        avg_metric_accuracy = 0
+
+        if total_folders > 0:
+            # 决策agent平均准确率
+            decision_accuracies = [r.decision_accuracy for r in self.results if r.decision_accuracy is not None]
+            if decision_accuracies:
+                avg_decision_accuracy = sum(decision_accuracies) / len(decision_accuracies)
+
+            # 大纲agent平均准确率
+            outline_accuracies = [r.outline_accuracy for r in self.results if r.outline_accuracy is not None]
+            if outline_accuracies:
+                avg_outline_accuracy = sum(outline_accuracies) / len(outline_accuracies)
+
+            # 指标agent平均准确率
+            metric_accuracies = [r.metric_accuracy for r in self.results if r.metric_accuracy is not None]
+            if metric_accuracies:
+                avg_metric_accuracy = sum(metric_accuracies) / len(metric_accuracies)
+
+        # 计算综合成功率(基于所有配置的校验项)
+        comprehensive_success_rate = 0
+        success_items = []
+
+        # 添加所有配置的准确率
+        if config.expected_decision_count != -1:
+            success_items.append(avg_decision_accuracy)
+        if config.expected_outline_count != -1:
+            success_items.append(avg_outline_accuracy)
+        if config.expected_metric_count != -1:
+            success_items.append(avg_metric_accuracy)
+
+        if success_items:
+            comprehensive_success_rate = sum(success_items) / len(success_items)
+
+        # 统计成功指标文件总数
+        total_success_metric = sum(r.actual_success_metric_count for r in self.results)
+        total_metric_files = sum(r.actual_metric_count for r in self.results)
+
+        report = {
+            "total_folders_scanned": total_folders,
+            "accuracy_statistics": {
+                "comprehensive_success_rate": f"{comprehensive_success_rate:.2f}%",  # 综合成功率
+                "decision_accuracy": f"{avg_decision_accuracy:.2f}%" if config.expected_decision_count != -1 else "未配置",
+                "outline_accuracy": f"{avg_outline_accuracy:.2f}%" if config.expected_outline_count != -1 else "未配置",
+                "metric_accuracy": f"{avg_metric_accuracy:.2f}%" if config.expected_metric_count != -1 else "未配置",
+            },
+            "metric_statistics": {
+                "total_metric_files": total_metric_files,
+                "total_success_metric_files": total_success_metric,
+                "success_ratio": f"{(total_success_metric / total_metric_files * 100):.2f}%" if total_metric_files > 0 else "0.00%"
+            },
+            "expected_counts": {
+                "decision": config.expected_decision_count,
+                "outline": config.expected_outline_count,
+                "metric": config.expected_metric_count
+            },
+            "details": []
+        }
+
+        # 添加详细信息
+        for result in self.results:
+            detail = {
+                "folder": result.folder_name,
+                "actual_counts": {
+                    "decision": result.actual_decision_count,
+                    "outline": result.actual_outline_count,
+                    "metric": result.actual_metric_count,
+                    "success_metric": result.actual_success_metric_count
+                },
+                "accuracies": {}
+            }
+
+            # 只添加有准确率的项
+            if result.decision_accuracy is not None:
+                detail["accuracies"]["decision"] = f"{result.decision_accuracy:.2f}%"
+            if result.outline_accuracy is not None:
+                detail["accuracies"]["outline"] = f"{result.outline_accuracy:.2f}%"
+            if result.metric_accuracy is not None:
+                detail["accuracies"]["metric"] = f"{result.metric_accuracy:.2f}%"
+
+            report["details"].append(detail)
+
+        return report
+
+    def print_report(self, report: Dict):
+        """打印报告"""
+        logger.info("=" * 50)
+        logger.info("文件夹扫描统计报告")
+        logger.info("=" * 50)
+        logger.info(f"扫描文件夹总数: {report['total_folders_scanned']}")
+
+        # 准确率统计
+        logger.info("\n📊 准确率统计:")
+        acc_stats = report['accuracy_statistics']
+
+        # 显示综合成功率
+        logger.info(f"  综合成功率: {acc_stats['comprehensive_success_rate']}")
+
+        # 只显示配置了的准确率
+        if acc_stats['decision_accuracy'] != "未配置":
+            logger.info(f"  决策agent准确率: {acc_stats['decision_accuracy']}")
+        if acc_stats['outline_accuracy'] != "未配置":
+            logger.info(f"  大纲agent准确率: {acc_stats['outline_accuracy']}")
+        if acc_stats['metric_accuracy'] != "未配置":
+            logger.info(f"  指标agent准确率: {acc_stats['metric_accuracy']}")
+
+        # 指标文件统计
+        logger.info("\n📈 指标文件统计:")
+        metric_stats = report['metric_statistics']
+        logger.info(f"  指标文件总数: {metric_stats['total_metric_files']}")
+        logger.info(f"  成功指标文件数: {metric_stats['total_success_metric_files']}")
+        logger.info(f"  指标成功率: {metric_stats['success_ratio']}")
+
+        # 期望值显示
+        logger.info("\n⚙️ 配置期望值:")
+        expected = report['expected_counts']
+        if expected['decision'] != -1:
+            logger.info(f"  决策agent: {expected['decision']}个")
+        if expected['outline'] != -1:
+            logger.info(f"  大纲agent: {expected['outline']}个")
+        if expected['metric'] != -1:
+            logger.info(f"  指标agent: {expected['metric']}个")
+
+        logger.info("=" * 50)
+
+        # 打印前3个文件夹的详情(避免日志太长)
+        if report['details']:
+            logger.info("前3个文件夹详情:")
+            for i, detail in enumerate(report['details'][:3]):
+                logger.info(f"\n  📂 {detail['folder']}:")
+                actual = detail['actual_counts']
+
+                # 构建实际数量字符串
+                count_parts = []
+                if expected['decision'] != -1:
+                    count_parts.append(f"决策:{actual['decision']}")
+                if expected['outline'] != -1:
+                    count_parts.append(f"大纲:{actual['outline']}")
+                if expected['metric'] != -1:
+                    count_parts.append(f"指标:{actual['metric']}")
+                count_parts.append(f"成功指标:{actual['success_metric']}")
+
+                logger.info(f"    实际数量 - {', '.join(count_parts)}")
+
+                # 构建准确率字符串
+                if detail['accuracies']:
+                    accuracy_parts = []
+                    for key, value in detail['accuracies'].items():
+                        if key == 'decision':
+                            accuracy_parts.append(f"决策:{value}")
+                        elif key == 'outline':
+                            accuracy_parts.append(f"大纲:{value}")
+                        elif key == 'metric':
+                            accuracy_parts.append(f"指标:{value}")
+
+                    logger.info(f"    准确率 - {', '.join(accuracy_parts)}")
+
+            if len(report['details']) > 3:
+                logger.info(f"  ... 还有{len(report['details']) - 3}个文件夹未显示")
+
+        logger.info("=" * 50)
+
+
+# 使用示例
+def main():
+    """使用示例"""
+    # 创建扫描器
+    scanner = FolderScanner(base_path=".")
+
+    # 配置期望值
+    config = ScanConfig(
+        expected_decision_count=3,  # 期望每个文件夹有3个决策agent文件
+        expected_outline_count=1,  # 期望每个文件夹有1个大纲生成agent文件
+        expected_metric_count=20  # 期望每个文件夹有20个指标计算agent文件
+    )
+
+    # 扫描100个文件夹
+    report = scanner.scan_folders(folder_count=2, config=config)
+
+    # 打印报告
+    scanner.print_report(report)
+
+    # 也可以返回报告数据供进一步处理
+    return report
+
+
+if __name__ == "__main__":
+    main()