""" 测试评估V2模块 从现有run_context.json读取帖子,使用V2评估模块重新评估,生成统计报告 """ import asyncio import json import sys from pathlib import Path from datetime import datetime from collections import defaultdict # 导入必要的模块 from knowledge_search_traverse import Post from post_evaluator_v2 import evaluate_post_v2, apply_evaluation_v2_to_post async def test_evaluation_v2(run_context_path: str, max_posts: int = 10): """ 测试V2评估模块 Args: run_context_path: run_context.json路径 max_posts: 最多评估的帖子数量(用于快速测试) """ print(f"\n{'='*80}") print(f"📊 评估V2测试 - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") print(f"{'='*80}\n") # 读取run_context.json print(f"📂 读取: {run_context_path}") with open(run_context_path, 'r', encoding='utf-8') as f: run_context = json.load(f) # 提取原始query original_query = run_context.get('o', '') print(f"🔍 原始Query: {original_query}\n") # 提取所有帖子 (从rounds -> search_results -> post_list) post_data_list = [] rounds = run_context.get('rounds', []) for round_idx, round_data in enumerate(rounds): search_results = round_data.get('search_results', []) for search_idx, search in enumerate(search_results): post_list = search.get('post_list', []) for post_idx, post_data in enumerate(post_list): # 生成唯一ID post_id = f"r{round_idx}_s{search_idx}_p{post_idx}" post_data_list.append((round_idx, search_idx, post_id, post_data)) total_posts = len(post_data_list) print(f"📝 找到 {total_posts} 个帖子 (来自 {len(rounds)} 轮)") # 限制评估数量(快速测试) if max_posts and max_posts < total_posts: post_data_list = post_data_list[:max_posts] print(f"⚡ 快速测试模式: 仅评估前 {max_posts} 个帖子\n") else: print() # 将post_data转换为Post对象 posts = [] for round_idx, search_idx, post_id, post_data in post_data_list: post = Post( note_id=post_data.get('note_id', post_id), title=post_data.get('title', ''), body_text=post_data.get('body_text', ''), images=post_data.get('images', []), type=post_data.get('type', 'normal') ) posts.append((round_idx, search_idx, post_id, post)) # 批量评估 print(f"🚀 开始批量评估 (并发数: 5)...\n") semaphore = asyncio.Semaphore(5) tasks = [] for round_idx, search_idx, post_id, post in posts: task = evaluate_post_v2(post, original_query, semaphore) tasks.append((round_idx, search_idx, post_id, post, task)) results = [] for i, (round_idx, search_idx, post_id, post, task) in enumerate(tasks, 1): print(f" [{i}/{len(tasks)}] 评估: {post.note_id}") knowledge_eval, relevance_eval = await task if knowledge_eval: # 应用评估结果（可能只有知识评估，没有相关性评估） apply_evaluation_v2_to_post(post, knowledge_eval, relevance_eval) results.append((round_idx, search_idx, post_id, post, knowledge_eval, relevance_eval)) # 输出已经在 evaluate_post_v2 中打印过了，这里不重复打印 else: print(f" ❌ 评估失败") print(f"\n✅ 评估完成: {len(results)}/{len(posts)} 成功\n") # 更新run_context.json中的帖子数据 print("💾 更新 run_context.json...") for round_idx, search_idx, post_id, post, knowledge_eval, relevance_eval in results: # 定位到对应的post_list if round_idx < len(rounds): search_results = rounds[round_idx].get('search_results', []) if search_idx < len(search_results): post_list = search_results[search_idx].get('post_list', []) # 找到对应的帖子并更新 for p in post_list: if p.get('note_id') == post.note_id: # 更新顶层字段 p['is_knowledge'] = post.is_knowledge p['knowledge_reason'] = post.knowledge_reason p['knowledge_score'] = post.knowledge_score p['knowledge_level'] = post.knowledge_level p['relevance_score'] = post.relevance_score p['relevance_level'] = post.relevance_level p['relevance_reason'] = post.relevance_reason p['relevance_conclusion'] = post.relevance_conclusion p['evaluation_time'] = post.evaluation_time p['evaluator_version'] = post.evaluator_version # 更新嵌套字段 p['knowledge_evaluation'] = post.knowledge_evaluation p['relevance_evaluation'] = post.relevance_evaluation break # 保存更新后的run_context.json output_path = run_context_path.replace('.json', '_v2.json') with open(output_path, 'w', encoding='utf-8') as f: json.dump(run_context, f, ensure_ascii=False, indent=2) print(f"✅ 已保存: {output_path}\n") # 生成统计报告 print(f"\n{'='*80}") print("📊 统计报告") print(f"{'='*80}\n") # 知识评估统计 knowledge_counts = defaultdict(int) knowledge_level_counts = defaultdict(int) knowledge_scores = [] for _, _, _, post, _, _ in results: if post.is_knowledge: knowledge_counts['知识内容'] += 1 else: knowledge_counts['非知识内容'] += 1 if post.knowledge_level: knowledge_level_counts[post.knowledge_level] += 1 if post.knowledge_score is not None: knowledge_scores.append(post.knowledge_score) total = len(results) print("📚 知识评估:") print(f" 知识内容: {knowledge_counts['知识内容']:3d} / {total} ({knowledge_counts['知识内容']/total*100:.1f}%)") print(f" 非知识内容: {knowledge_counts['非知识内容']:3d} / {total} ({knowledge_counts['非知识内容']/total*100:.1f}%)") print() if knowledge_scores: avg_score = sum(knowledge_scores) / len(knowledge_scores) print(f" 平均得分: {avg_score:.1f}分") print(f" 最高得分: {max(knowledge_scores):.0f}分") print(f" 最低得分: {min(knowledge_scores):.0f}分") print() print(" 星级分布:") for level in range(1, 6): count = knowledge_level_counts.get(level, 0) bar = '★' * count print(f" {level}星: {count:3d} {bar}") print() # 相关性评估统计 relevance_conclusion_counts = defaultdict(int) relevance_scores = [] purpose_scores = [] category_scores = [] for _, _, _, post, _, _ in results: if post.relevance_conclusion: relevance_conclusion_counts[post.relevance_conclusion] += 1 if post.relevance_score is not None: relevance_scores.append(post.relevance_score) if post.relevance_evaluation: if 'purpose_score' in post.relevance_evaluation: purpose_scores.append(post.relevance_evaluation['purpose_score']) if 'category_score' in post.relevance_evaluation: category_scores.append(post.relevance_evaluation['category_score']) print("🎯 相关性评估:") for conclusion in ['高度匹配', '中度匹配', '低度匹配', '不匹配']: count = relevance_conclusion_counts.get(conclusion, 0) if count > 0: print(f" {conclusion}: {count:3d} / {total} ({count/total*100:.1f}%)") print() if relevance_scores: avg_score = sum(relevance_scores) / len(relevance_scores) high_relevance = sum(1 for s in relevance_scores if s >= 70) print(f" 平均得分: {avg_score:.1f}分") print(f" 高相关性: {high_relevance} / {total} ({high_relevance/total*100:.1f}%) [≥70分]") print(f" 最高得分: {max(relevance_scores):.0f}分") print(f" 最低得分: {min(relevance_scores):.0f}分") print() if purpose_scores and category_scores: avg_purpose = sum(purpose_scores) / len(purpose_scores) avg_category = sum(category_scores) / len(category_scores) print(f" 目的性平均: {avg_purpose:.1f}分 (权重70%)") print(f" 品类平均: {avg_category:.1f}分 (权重30%)") print() # 综合分析 print("🔥 高质量内容 (知识内容 + 高相关性):") high_quality = sum( 1 for _, _, _, post, _, _ in results if post.is_knowledge and post.relevance_score and post.relevance_score >= 70 ) print(f" {high_quality} / {total} ({high_quality/total*100:.1f}%)") print() print(f"{'='*80}\n") return results if __name__ == "__main__": if len(sys.argv) < 2: print("用法: python3 test_evaluation_v2.py [最大评估数量]") print() print("示例:") print(" python3 test_evaluation_v2.py input/test_case/output/knowledge_search_traverse/20251112/173512_dc/run_context.json") print(" python3 test_evaluation_v2.py input/test_case/output/knowledge_search_traverse/20251112/173512_dc/run_context.json 20") sys.exit(1) run_context_path = sys.argv[1] max_posts = int(sys.argv[2]) if len(sys.argv) > 2 else None asyncio.run(test_evaluation_v2(run_context_path, max_posts))