folder_scanner.py 13 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354
  1. import os
  2. import logging
  3. from typing import Dict, List, Optional
  4. from dataclasses import dataclass
  5. # 配置日志
  6. logging.basicConfig(
  7. level=logging.INFO,
  8. format='%(asctime)s - %(levelname)s - %(message)s'
  9. )
  10. logger = logging.getLogger(__name__)
  11. @dataclass
  12. class ScanConfig:
  13. """扫描配置"""
  14. # 决策agent的期望数量,-1表示不校验
  15. expected_decision_count: int = -1
  16. # 大纲生成agent的期望数量,-1表示不校验
  17. expected_outline_count: int = -1
  18. # 指标计算agent的期望数量,-1表示不校验
  19. expected_metric_count: int = -1
  20. @dataclass
  21. class ScanResult:
  22. """单个文件夹扫描结果"""
  23. folder_name: str
  24. decision_files: List[str]
  25. outline_files: List[str]
  26. metric_files: List[str]
  27. success_metric_files: List[str]
  28. # 实际数量
  29. actual_decision_count: int
  30. actual_outline_count: int
  31. actual_metric_count: int
  32. actual_success_metric_count: int
  33. # 准确率(如果配置了期望值)
  34. decision_accuracy: Optional[float] = None
  35. outline_accuracy: Optional[float] = None
  36. metric_accuracy: Optional[float] = None
  37. class FolderScanner:
  38. """文件夹扫描工具类"""
  39. def __init__(self, base_path: str = "."):
  40. """
  41. 初始化扫描器
  42. Args:
  43. base_path: 基础路径,默认为当前目录
  44. """
  45. self.base_path = base_path
  46. self.results: List[ScanResult] = []
  47. def scan_folders(self, folder_count: int, config: Optional[ScanConfig] = None) -> Dict:
  48. """
  49. 扫描指定数量的文件夹
  50. Args:
  51. folder_count: 要扫描的文件夹数量
  52. config: 扫描配置,包含各agent的期望数量
  53. Returns:
  54. 扫描统计结果
  55. """
  56. if config is None:
  57. config = ScanConfig()
  58. logger.info(f"开始扫描 {folder_count} 个文件夹...")
  59. logger.info(f"期望配置: 决策agent={config.expected_decision_count}, "
  60. f"大纲agent={config.expected_outline_count}, "
  61. f"指标agent={config.expected_metric_count}")
  62. self.results = []
  63. for i in range(1, folder_count + 1):
  64. folder_name = f"api_results_{i}"
  65. folder_path = os.path.join(self.base_path, folder_name)
  66. if not os.path.exists(folder_path):
  67. logger.warning(f"文件夹不存在: {folder_path}")
  68. continue
  69. result = self._scan_single_folder(folder_name, folder_path, config)
  70. self.results.append(result)
  71. return self._generate_report(config)
  72. def _calculate_accuracy(self, actual: int, expected: int) -> float:
  73. """计算准确率"""
  74. if actual == expected:
  75. return 100.0
  76. elif actual < expected:
  77. # 实际少于期望
  78. return (actual / expected) * 100
  79. else:
  80. # 实际多于期望,也应该计算偏差
  81. # 比如期望3个,实际4个,准确率 = 3/4 * 100 = 75%
  82. return (expected / actual) * 100
  83. def _scan_single_folder(self, folder_name: str, folder_path: str, config: ScanConfig) -> ScanResult:
  84. """扫描单个文件夹"""
  85. all_files = os.listdir(folder_path)
  86. # 分类文件
  87. decision_files = []
  88. outline_files = []
  89. metric_files = []
  90. success_metric_files = []
  91. for file in all_files:
  92. if file.endswith('.md'):
  93. if '规划决策' in file:
  94. decision_files.append(file)
  95. elif '大纲生成' in file:
  96. outline_files.append(file)
  97. elif file.endswith('.json'):
  98. metric_files.append(file)
  99. if '_success.json' in file:
  100. success_metric_files.append(file)
  101. # 计算实际数量
  102. actual_decision_count = len(decision_files)
  103. actual_outline_count = len(outline_files)
  104. actual_metric_count = len(metric_files)
  105. actual_success_metric_count = len(success_metric_files)
  106. # 计算准确率(如果配置了期望值)
  107. decision_accuracy = None
  108. outline_accuracy = None
  109. metric_accuracy = None
  110. if config.expected_decision_count != -1 and config.expected_decision_count > 0:
  111. decision_accuracy = self._calculate_accuracy(actual_decision_count, config.expected_decision_count)
  112. if config.expected_outline_count != -1 and config.expected_outline_count > 0:
  113. outline_accuracy = self._calculate_accuracy(actual_outline_count, config.expected_outline_count)
  114. if config.expected_metric_count != -1 and config.expected_metric_count > 0:
  115. metric_accuracy = self._calculate_accuracy(actual_metric_count, config.expected_metric_count)
  116. return ScanResult(
  117. folder_name=folder_name,
  118. decision_files=decision_files,
  119. outline_files=outline_files,
  120. metric_files=metric_files,
  121. success_metric_files=success_metric_files,
  122. actual_decision_count=actual_decision_count,
  123. actual_outline_count=actual_outline_count,
  124. actual_metric_count=actual_metric_count,
  125. actual_success_metric_count=actual_success_metric_count,
  126. decision_accuracy=decision_accuracy,
  127. outline_accuracy=outline_accuracy,
  128. metric_accuracy=metric_accuracy
  129. )
  130. def _generate_report(self, config: ScanConfig) -> Dict:
  131. """生成报告"""
  132. total_folders = len(self.results)
  133. # 计算平均准确率
  134. avg_decision_accuracy = 0
  135. avg_outline_accuracy = 0
  136. avg_metric_accuracy = 0
  137. if total_folders > 0:
  138. # 决策agent平均准确率
  139. decision_accuracies = [r.decision_accuracy for r in self.results if r.decision_accuracy is not None]
  140. if decision_accuracies:
  141. avg_decision_accuracy = sum(decision_accuracies) / len(decision_accuracies)
  142. # 大纲agent平均准确率
  143. outline_accuracies = [r.outline_accuracy for r in self.results if r.outline_accuracy is not None]
  144. if outline_accuracies:
  145. avg_outline_accuracy = sum(outline_accuracies) / len(outline_accuracies)
  146. # 指标agent平均准确率
  147. metric_accuracies = [r.metric_accuracy for r in self.results if r.metric_accuracy is not None]
  148. if metric_accuracies:
  149. avg_metric_accuracy = sum(metric_accuracies) / len(metric_accuracies)
  150. # 计算综合成功率(基于所有配置的校验项)
  151. comprehensive_success_rate = 0
  152. success_items = []
  153. # 添加所有配置的准确率
  154. if config.expected_decision_count != -1:
  155. success_items.append(avg_decision_accuracy)
  156. if config.expected_outline_count != -1:
  157. success_items.append(avg_outline_accuracy)
  158. if config.expected_metric_count != -1:
  159. success_items.append(avg_metric_accuracy)
  160. if success_items:
  161. comprehensive_success_rate = sum(success_items) / len(success_items)
  162. # 统计成功指标文件总数
  163. total_success_metric = sum(r.actual_success_metric_count for r in self.results)
  164. total_metric_files = sum(r.actual_metric_count for r in self.results)
  165. report = {
  166. "total_folders_scanned": total_folders,
  167. "accuracy_statistics": {
  168. "comprehensive_success_rate": f"{comprehensive_success_rate:.2f}%", # 综合成功率
  169. "decision_accuracy": f"{avg_decision_accuracy:.2f}%" if config.expected_decision_count != -1 else "未配置",
  170. "outline_accuracy": f"{avg_outline_accuracy:.2f}%" if config.expected_outline_count != -1 else "未配置",
  171. "metric_accuracy": f"{avg_metric_accuracy:.2f}%" if config.expected_metric_count != -1 else "未配置",
  172. },
  173. "metric_statistics": {
  174. "total_metric_files": total_metric_files,
  175. "total_success_metric_files": total_success_metric,
  176. "success_ratio": f"{(total_success_metric / total_metric_files * 100):.2f}%" if total_metric_files > 0 else "0.00%"
  177. },
  178. "expected_counts": {
  179. "decision": config.expected_decision_count,
  180. "outline": config.expected_outline_count,
  181. "metric": config.expected_metric_count
  182. },
  183. "details": []
  184. }
  185. # 添加详细信息
  186. for result in self.results:
  187. detail = {
  188. "folder": result.folder_name,
  189. "actual_counts": {
  190. "decision": result.actual_decision_count,
  191. "outline": result.actual_outline_count,
  192. "metric": result.actual_metric_count,
  193. "success_metric": result.actual_success_metric_count
  194. },
  195. "accuracies": {}
  196. }
  197. # 只添加有准确率的项
  198. if result.decision_accuracy is not None:
  199. detail["accuracies"]["decision"] = f"{result.decision_accuracy:.2f}%"
  200. if result.outline_accuracy is not None:
  201. detail["accuracies"]["outline"] = f"{result.outline_accuracy:.2f}%"
  202. if result.metric_accuracy is not None:
  203. detail["accuracies"]["metric"] = f"{result.metric_accuracy:.2f}%"
  204. report["details"].append(detail)
  205. return report
  206. def print_report(self, report: Dict):
  207. """打印报告"""
  208. logger.info("=" * 50)
  209. logger.info("文件夹扫描统计报告")
  210. logger.info("=" * 50)
  211. logger.info(f"扫描文件夹总数: {report['total_folders_scanned']}")
  212. # 准确率统计
  213. logger.info("\n📊 准确率统计:")
  214. acc_stats = report['accuracy_statistics']
  215. # 显示综合成功率
  216. logger.info(f" 综合成功率: {acc_stats['comprehensive_success_rate']}")
  217. # 只显示配置了的准确率
  218. if acc_stats['decision_accuracy'] != "未配置":
  219. logger.info(f" 决策agent准确率: {acc_stats['decision_accuracy']}")
  220. if acc_stats['outline_accuracy'] != "未配置":
  221. logger.info(f" 大纲agent准确率: {acc_stats['outline_accuracy']}")
  222. if acc_stats['metric_accuracy'] != "未配置":
  223. logger.info(f" 指标agent准确率: {acc_stats['metric_accuracy']}")
  224. # 指标文件统计
  225. logger.info("\n📈 指标文件统计:")
  226. metric_stats = report['metric_statistics']
  227. logger.info(f" 指标文件总数: {metric_stats['total_metric_files']}")
  228. logger.info(f" 成功指标文件数: {metric_stats['total_success_metric_files']}")
  229. logger.info(f" 指标成功率: {metric_stats['success_ratio']}")
  230. # 期望值显示
  231. logger.info("\n⚙️ 配置期望值:")
  232. expected = report['expected_counts']
  233. if expected['decision'] != -1:
  234. logger.info(f" 决策agent: {expected['decision']}个")
  235. if expected['outline'] != -1:
  236. logger.info(f" 大纲agent: {expected['outline']}个")
  237. if expected['metric'] != -1:
  238. logger.info(f" 指标agent: {expected['metric']}个")
  239. logger.info("=" * 50)
  240. # 打印前3个文件夹的详情(避免日志太长)
  241. if report['details']:
  242. logger.info("前3个文件夹详情:")
  243. for i, detail in enumerate(report['details'][:3]):
  244. logger.info(f"\n 📂 {detail['folder']}:")
  245. actual = detail['actual_counts']
  246. # 构建实际数量字符串
  247. count_parts = []
  248. if expected['decision'] != -1:
  249. count_parts.append(f"决策:{actual['decision']}")
  250. if expected['outline'] != -1:
  251. count_parts.append(f"大纲:{actual['outline']}")
  252. if expected['metric'] != -1:
  253. count_parts.append(f"指标:{actual['metric']}")
  254. count_parts.append(f"成功指标:{actual['success_metric']}")
  255. logger.info(f" 实际数量 - {', '.join(count_parts)}")
  256. # 构建准确率字符串
  257. if detail['accuracies']:
  258. accuracy_parts = []
  259. for key, value in detail['accuracies'].items():
  260. if key == 'decision':
  261. accuracy_parts.append(f"决策:{value}")
  262. elif key == 'outline':
  263. accuracy_parts.append(f"大纲:{value}")
  264. elif key == 'metric':
  265. accuracy_parts.append(f"指标:{value}")
  266. logger.info(f" 准确率 - {', '.join(accuracy_parts)}")
  267. if len(report['details']) > 3:
  268. logger.info(f" ... 还有{len(report['details']) - 3}个文件夹未显示")
  269. logger.info("=" * 50)
  270. # 使用示例
  271. def main():
  272. """使用示例"""
  273. # 创建扫描器
  274. scanner = FolderScanner(base_path=".")
  275. # 配置期望值
  276. config = ScanConfig(
  277. expected_decision_count=-1, # 期望每个文件夹有3个决策agent文件
  278. expected_outline_count=1, # 期望每个文件夹有1个大纲生成agent文件
  279. expected_metric_count=20 # 期望每个文件夹有20个指标计算agent文件
  280. )
  281. # 扫描100个文件夹
  282. report = scanner.scan_folders(folder_count=2, config=config)
  283. # 打印报告
  284. scanner.print_report(report)
  285. # 也可以返回报告数据供进一步处理
  286. return report
  287. if __name__ == "__main__":
  288. main()