kg_agent/test_evaluation_v3.py

"""
测试评估V3模块
从现有run_context.json读取帖子,使用V3评估模块重新评估,生成统计报告
"""

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_v3 import evaluate_post_v3, apply_evaluation_v3_to_post, two_stage_batch_evaluate


async def test_evaluation_v3(run_context_path: str, max_posts: int = 10):
    """
    测试V3评估模块

    Args:
        run_context_path: run_context.json路径
        max_posts: 最多评估的帖子数量(用于快速测试)
    """
    print(f"\n{'='*80}")
    print(f"📊 评估V3测试 - {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'),
            video=post_data.get('video', ''),
            interact_info=post_data.get('interact_info', {}),
            note_url=post_data.get('note_url', ''),
            author_name=post_data.get('author_name', ''),
            author_id=post_data.get('author_id', ''),
            publish_time=post_data.get('publish_time', 0),
            cdn_images=post_data.get('cdn_images', []),
            detail_fetched=post_data.get('detail_fetched', False)
        )
        posts.append((round_idx, search_idx, post_id, post))

    # 提取纯post列表用于两阶段评估
    post_list = [post for _, _, _, post in posts]

    # 使用两阶段批量评估
    await two_stage_batch_evaluate(post_list, original_query, quick_concurrent=15, detail_concurrent=15)

    # 处理评估结果（两阶段评估已将结果应用到post对象）
    results = []
    detailed_reports = []  # 收集详细评估报告

    print(f"📊 收集评估报告...\n")
    for i, (round_idx, search_idx, post_id, post) in enumerate(posts, 1):
        # 从post对象的嵌套字段中提取评估结果
        knowledge_eval = post.knowledge_evaluation
        content_eval = post.content_knowledge_evaluation
        purpose_eval = post.purpose_evaluation
        category_eval = post.category_evaluation

        # 收集详细报告
        if knowledge_eval:
            detailed_report = {
                'post_index': i,
                'note_id': post.note_id,
                'title': post.title,
                'final_score': post.final_score,
                'match_level': post.match_level,
                'is_knowledge': post.is_knowledge,
                'is_content_knowledge': post.is_content_knowledge,
                'knowledge_score': post.knowledge_score,
                'evaluations': {
                    'knowledge': {
                        'conclusion': knowledge_eval.get('conclusion') if isinstance(knowledge_eval, dict) else getattr(knowledge_eval, 'conclusion', None),
                        'core_evidence': knowledge_eval.get('core_evidence') if isinstance(knowledge_eval, dict) else getattr(knowledge_eval, 'core_evidence', None),
                        'issues': knowledge_eval.get('issues') if isinstance(knowledge_eval, dict) else getattr(knowledge_eval, 'issues', None)
                    },
                    'content_knowledge': {
                        'summary': content_eval.get('summary') if isinstance(content_eval, dict) else getattr(content_eval, 'summary', None),
                        'final_score': content_eval.get('final_score') if isinstance(content_eval, dict) else getattr(content_eval, 'final_score', None),
                        'level': content_eval.get('level') if isinstance(content_eval, dict) else getattr(content_eval, 'level', None)
                    } if content_eval and post.is_content_knowledge else None,
                    'purpose': {
                        'score': purpose_eval.get('purpose_score') if isinstance(purpose_eval, dict) else getattr(purpose_eval, 'purpose_score', None),
                        'core_motivation': purpose_eval.get('core_motivation') if isinstance(purpose_eval, dict) else getattr(purpose_eval, 'core_motivation', None),
                        'core_basis': purpose_eval.get('core_basis') if isinstance(purpose_eval, dict) else getattr(purpose_eval, 'core_basis', None),
                        'match_level': purpose_eval.get('match_level') if isinstance(purpose_eval, dict) else getattr(purpose_eval, 'match_level', None)
                    } if purpose_eval else None,
                    'category': {
                        'score': category_eval.get('category_score') if isinstance(category_eval, dict) else getattr(category_eval, 'category_score', None),
                        'core_basis': category_eval.get('core_basis') if isinstance(category_eval, dict) else getattr(category_eval, 'core_basis', None),
                        'match_level': category_eval.get('match_level') if isinstance(category_eval, dict) else getattr(category_eval, 'match_level', None)
                    } if category_eval else None
                }
            }
            detailed_reports.append(detailed_report)
            results.append((round_idx, search_idx, post_id, post))

    print(f"\n✅ 评估完成: {len(results)}/{len(posts)} 成功\n")

    # 更新run_context.json中的帖子数据
    print("💾 更新 run_context.json...")
    for round_idx, search_idx, post_id, post 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:
                        # 更新V3顶层字段
                        p['is_knowledge'] = post.is_knowledge
                        p['is_content_knowledge'] = post.is_content_knowledge
                        p['knowledge_score'] = post.knowledge_score

                        p['purpose_score'] = post.purpose_score
                        p['category_score'] = post.category_score
                        p['final_score'] = post.final_score
                        p['match_level'] = post.match_level

                        p['evaluation_time'] = post.evaluation_time
                        p['evaluator_version'] = post.evaluator_version

                        # 更新V3嵌套字段
                        p['knowledge_evaluation'] = post.knowledge_evaluation
                        p['content_knowledge_evaluation'] = post.content_knowledge_evaluation
                        p['purpose_evaluation'] = post.purpose_evaluation
                        p['category_evaluation'] = post.category_evaluation
                        break

    # 保存更新后的run_context.json
    output_path = run_context_path.replace('.json', '_v3.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}")

    # 保存详细评估报告
    report_path = run_context_path.replace('.json', '_evaluation_report.json')
    evaluation_report = {
        'metadata': {
            'original_query': original_query,
            'total_posts': len(results),
            'evaluation_time': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
            'evaluator_version': 'v3.0'
        },
        'detailed_reports': detailed_reports
    }
    with open(report_path, 'w', encoding='utf-8') as f:
        json.dump(evaluation_report, f, ensure_ascii=False, indent=2)
    print(f"📄 已保存详细评估报告: {report_path}\n")

    # 生成统计报告
    print(f"\n{'='*80}")
    print("📊 统计报告")
    print(f"{'='*80}\n")

    # Prompt1: 是否是知识
    is_knowledge_counts = defaultdict(int)
    for _, _, _, post in results:
        if post.is_knowledge:
            is_knowledge_counts['是知识'] += 1
        else:
            is_knowledge_counts['非知识'] += 1

    total = len(results)
    print("🔍 Prompt1 - 是否是知识:")
    print(f"  是知识: {is_knowledge_counts['是知识']:3d} / {total} ({is_knowledge_counts['是知识']/total*100:.1f}%)")
    print(f"  非知识: {is_knowledge_counts['非知识']:3d} / {total} ({is_knowledge_counts['非知识']/total*100:.1f}%)")
    print()

    # Prompt2: 是否是内容知识
    is_content_knowledge_counts = defaultdict(int)
    knowledge_scores = []
    for _, _, _, post in results:
        if post.is_content_knowledge is not None:
            if post.is_content_knowledge:
                is_content_knowledge_counts['是内容知识'] += 1
            else:
                is_content_knowledge_counts['非内容知识'] += 1

        if post.knowledge_score is not None:
            knowledge_scores.append(post.knowledge_score)

    if is_content_knowledge_counts:
        content_total = sum(is_content_knowledge_counts.values())
        print("📚 Prompt2 - 是否是内容知识:")
        print(f"  是内容知识: {is_content_knowledge_counts['是内容知识']:3d} / {content_total} ({is_content_knowledge_counts['是内容知识']/content_total*100:.1f}%)")
        if is_content_knowledge_counts['非内容知识'] > 0:
            print(f"  非内容知识: {is_content_knowledge_counts['非内容知识']:3d} / {content_total} ({is_content_knowledge_counts['非内容知识']/content_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()

    # Prompt3 & Prompt4: 目的性和品类匹配
    purpose_scores = []
    category_scores = []
    final_scores = []
    match_level_counts = defaultdict(int)

    for _, _, _, post in results:
        if post.purpose_score is not None:
            purpose_scores.append(post.purpose_score)
        if post.category_score is not None:
            category_scores.append(post.category_score)
        if post.final_score is not None:
            final_scores.append(post.final_score)
        if post.match_level:
            match_level_counts[post.match_level] += 1

    if purpose_scores:
        avg_purpose = sum(purpose_scores) / len(purpose_scores)
        print("🎯 Prompt3 - 目的性匹配:")
        print(f"  平均得分: {avg_purpose:.1f}分")
        print(f"  最高得分: {max(purpose_scores):.0f}分")
        print(f"  最低得分: {min(purpose_scores):.0f}分")
        print()

    if category_scores:
        avg_category = sum(category_scores) / len(category_scores)
        print("🏷️  Prompt4 - 品类匹配:")
        print(f"  平均得分: {avg_category:.1f}分")
        print(f"  最高得分: {max(category_scores):.0f}分")
        print(f"  最低得分: {min(category_scores):.0f}分")
        print()

    if final_scores:
        avg_final = sum(final_scores) / len(final_scores)
        print("🔥 综合得分 (目的性70% + 品类30%):")
        print(f"  平均得分: {avg_final:.2f}分")
        print(f"  最高得分: {max(final_scores):.2f}分")
        print(f"  最低得分: {min(final_scores):.2f}分")
        print()

    if match_level_counts:
        print("📊 匹配等级分布:")
        for level in ['高度匹配', '基本匹配', '部分匹配', '弱匹配', '不匹配']:
            count = match_level_counts.get(level, 0)
            if count > 0:
                bar = '█' * int(count / total * 50)
                print(f"  {level:8s}: {count:3d} / {total} ({count/total*100:.1f}%) {bar}")
        print()

    # 综合分析
    print("🌟 高质量内容统计:")

    # 是知识 + 是内容知识
    is_quality_knowledge = sum(
        1 for _, _, _, post in results
        if post.is_knowledge and post.is_content_knowledge
    )
    print(f"  知识内容: {is_quality_knowledge} / {total} ({is_quality_knowledge/total*100:.1f}%)")

    # 是知识 + 是内容知识 + 高度匹配
    high_match = sum(
        1 for _, _, _, post in results
        if post.is_knowledge and post.is_content_knowledge and post.match_level == '高度匹配'
    )
    print(f"  高度匹配: {high_match} / {total} ({high_match/total*100:.1f}%)")

    # 是知识 + 是内容知识 + 综合得分>=70
    high_score = sum(
        1 for _, _, _, post in results
        if post.is_knowledge and post.is_content_knowledge and post.final_score and post.final_score >= 70
    )
    print(f"  得分≥70:  {high_score} / {total} ({high_score/total*100:.1f}%)")
    print()

    print(f"{'='*80}\n")

    return results


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("用法: python3 test_evaluation_v3.py <run_context.json路径> [最大评估数量]")
        print()
        print("示例:")
        print("  python3 test_evaluation_v3.py input/test_case/output/knowledge_search_traverse/20251112/173512_dc/run_context.json")
        print("  python3 test_evaluation_v3.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_v3(run_context_path, max_posts))