| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415 |
- 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
- # 数据转换agent的期望数量,-1表示不校验
- expected_transform_count: int = -1
- # 期望扫描的文件夹总数(用于计算文件成功率)
- expected_folder_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]
- transform_files: List[str] # 新增:数据转换文件
- # 实际数量
- actual_decision_count: int
- actual_outline_count: int
- actual_metric_count: int
- actual_success_metric_count: int
- actual_transform_count: int # 新增:数据转换数量
- # 准确率(如果配置了期望值)
- decision_accuracy: Optional[float] = None
- outline_accuracy: Optional[float] = None
- metric_accuracy: Optional[float] = None
- transform_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}, "
- f"数据转换agent={config.expected_transform_count}")
- self.results = []
- missing_folders = [] # 记录缺失的文件夹
- 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}")
- missing_folders.append(folder_name)
- continue
- result = self._scan_single_folder(folder_name, folder_path, config)
- self.results.append(result)
- return self._generate_report(config, folder_count, missing_folders)
- 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 = []
- transform_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 '数据转换' in file: # 新增:识别数据转换文件
- transform_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)
- actual_transform_count = len(transform_files) # 新增:数据转换数量
- # 计算准确率(如果配置了期望值)
- decision_accuracy = None
- outline_accuracy = None
- metric_accuracy = None
- transform_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)
- if config.expected_transform_count != -1 and config.expected_transform_count > 0:
- transform_accuracy = self._calculate_accuracy(actual_transform_count, config.expected_transform_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,
- transform_files=transform_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,
- actual_transform_count=actual_transform_count, # 新增
- decision_accuracy=decision_accuracy,
- outline_accuracy=outline_accuracy,
- metric_accuracy=metric_accuracy,
- transform_accuracy=transform_accuracy # 新增
- )
- def _generate_report(self, config: ScanConfig, expected_folder_count: int, missing_folders: List[str]) -> Dict:
- """生成报告"""
- total_folders = len(self.results)
- # 计算文件成功率
- file_success_rate = 0
- if expected_folder_count > 0:
- file_success_rate = (total_folders / expected_folder_count) * 100
- # 计算平均准确率
- avg_decision_accuracy = 0
- avg_outline_accuracy = 0
- avg_metric_accuracy = 0
- avg_transform_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)
- # 数据转换agent平均准确率
- transform_accuracies = [r.transform_accuracy for r in self.results if r.transform_accuracy is not None]
- if transform_accuracies:
- avg_transform_accuracy = sum(transform_accuracies) / len(transform_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 config.expected_transform_count != -1:
- success_items.append(avg_transform_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,
- "expected_folder_count": expected_folder_count,
- "missing_folders": missing_folders,
- "file_success_rate": f"{file_success_rate:.2f}%",
- "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 "未配置",
- "transform_accuracy": f"{avg_transform_accuracy:.2f}%" if config.expected_transform_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,
- "transform": config.expected_transform_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,
- "transform": result.actual_transform_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}%"
- if result.transform_accuracy is not None:
- detail["accuracies"]["transform"] = f"{result.transform_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['expected_folder_count']}")
- logger.info(f"实际扫描文件夹数: {report['total_folders_scanned']}")
- logger.info(f"文件成功率: {report['file_success_rate']}")
- if report['missing_folders']:
- logger.info(f"缺失文件夹: {', '.join(report['missing_folders'][:5])}")
- if len(report['missing_folders']) > 5:
- logger.info(f" ... 还有{len(report['missing_folders']) - 5}个缺失文件夹")
- # 准确率统计
- 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']}")
- if acc_stats['transform_accuracy'] != "未配置":
- logger.info(f" 数据转换agent准确率: {acc_stats['transform_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']}个")
- if expected['transform'] != -1:
- logger.info(f" 数据转换agent: {expected['transform']}个")
- logger.info("=" * 50)
- # 打印前3个文件夹的详情(避免日志太长)
- if report['details']:
- logger.info("前3个文件夹详情:")
- for i, detail in enumerate(report['details'][:50]):
- 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']}")
- if expected['transform'] != -1:
- count_parts.append(f"数据转换:{actual['transform']}")
- 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}")
- elif key == 'transform':
- accuracy_parts.append(f"数据转换:{value}")
- logger.info(f" 准确率 - {', '.join(accuracy_parts)}")
- if len(report['details']) > 50:
- logger.info(f" ... 还有{len(report['details']) - 3}个文件夹未显示")
- logger.info("=" * 50)
- # 使用示例
- def main():
- """使用示例"""
- # 创建扫描器
- scanner = FolderScanner(base_path=".")
- # 配置期望值
- config = ScanConfig(
- expected_decision_count=-1, # 期望每个文件夹有3个决策agent文件
- expected_outline_count=1, # 期望每个文件夹有1个大纲生成agent文件
- expected_metric_count=19, # 期望每个文件夹有20个指标计算agent文件
- expected_transform_count=2, # 期望每个文件夹有2个数据转换agent文件
- expected_folder_count=10 # 期望扫描100个文件夹
- )
- # 扫描100个文件夹
- report = scanner.scan_folders(folder_count=10, config=config)
- # 打印报告
- scanner.print_report(report)
- # 也可以返回报告数据供进一步处理
- return report
- if __name__ == "__main__":
- main()
|