evaluation/script/evaluation_online.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
模型评估脚本
功能：
1. 读取JSON文件中的对话数据
2. 提取human的value作为query调用server:8020
3. 处理流式返回结果
4. 对比和存储结果
"""

import json
import httpx
import asyncio
import time
import re
import os
from typing import Dict, List, Any
from utils.custom_logging import setup_logging
from utils.extraction import extract_json_from_string
from loguru import logger
from collections import Counter
setup_logging()


class ModelEvaluator:
    def __init__(self, server_url: str = "http://localhost:8020/mcp_end2end/stream"):
        self.server_url = server_url
        self.results = []
        self.client = None
        self.start_time = None
        self.error_count = 0
        self.success_count = 0
        
    def load_data(self, file_path: str) -> List[Dict]:
        """加载JSON数据文件"""
        try:
            with open(file_path, 'r', encoding='utf-8') as f:
                data = json.load(f)
            logger.info(f"成功加载数据文件，共{len(data)}条记录")
            return data
        except Exception as e:
            logger.error(f"加载数据文件失败: {e}")
            return []
    
    def extract_human_queries(self, data: List[Dict]) -> List[Dict]:
        """提取所有human的value作为query"""
        queries = []
        for i, item in enumerate(data):
            if 'conversations' in item:
                for conv in item['conversations']:
                    if conv.get('from') == 'human':
                        query_data = {
                            'index': i,
                            'query': conv.get('value', ''),
                            'original_data': item
                        }
                        queries.append(query_data)
                        break  # 只取第一个human的value
        logger.info(f"提取到{len(queries)}个查询")
        return queries
    
    def parse_sse_events(self, sse_content: str, filter_events: List[str] = None) -> List[Dict]:
        """
        解析SSE格式的内容，提取指定类型的事件
        
        Args:
            sse_content: SSE格式的文本内容（可以是多行）
            filter_events: 需要过滤的事件类型列表，如果为None则解析所有事件
        
        Returns:
            解析成功的事件列表
        """
        events = []
        current_event = {}
        parsed_count = 0
        failed_count = 0
        
        # 如果没有指定过滤事件，设置默认过滤
        if filter_events is None:
            filter_events = ['tool_call.created', 'tool_response.completed']
        
        for line in sse_content.split('\n'):
            line = line.strip()
            if not line:
                # 空行表示一个完整的事件结束
                if current_event and 'event' in current_event and 'data' in current_event:
                    event_type = current_event['event']
                    
                    # 只处理我们关心的事件类型
                    if event_type in filter_events:
                        # 使用extract_json_from_string解析data字段中的JSON
                        data_content = extract_json_from_string(current_event['data'])
                        if data_content is not None:
                            event_obj = {
                                'id': current_event.get('id'),
                                'event': current_event['event'],
                                'data': data_content
                            }
                            events.append(event_obj)
                            parsed_count += 1
                            logger.debug(f"✅ 成功解析事件: {current_event['event']} (ID: {current_event.get('id', 'N/A')})")
                        else:
                            failed_count += 1
                            logger.warning(f"❌ 无法解析事件数据: {current_event['event']} - {current_event['data'][:100]}...")
                current_event = {}
                continue
            
            # 解析SSE格式的字段
            if line.startswith('id: '):
                current_event['id'] = line[4:]
            elif line.startswith('event: '):
                current_event['event'] = line[7:]
            elif line.startswith('data: '):
                current_event['data'] = line[6:]
                # 检查是否是结束标记
                if current_event['data'].strip() == '[DONE]':
                    logger.debug("收到结束标记 [DONE]")
                    break
            else:
                logger.debug(f"未知格式的行: {line}")
        
        # 处理最后一个事件（如果没有空行结尾）
        if current_event and 'event' in current_event and 'data' in current_event:
            event_type = current_event['event']
            
            # 只处理我们关心的事件类型
            if event_type in filter_events:
                data_content = extract_json_from_string(current_event['data'])
                if data_content is not None:
                    event_obj = {
                        'id': current_event.get('id'),
                        'event': current_event['event'],
                        'data': data_content
                    }
                    events.append(event_obj)
                    parsed_count += 1
                    logger.debug(f"✅ 成功解析最后一个事件: {current_event['event']} (ID: {current_event.get('id', 'N/A')})")
                else:
                    failed_count += 1
                    logger.warning(f"❌ 无法解析最后一个事件数据: {current_event['event']} - {current_event['data'][:100]}...")
        
        # 统计和日志输出
        logger.info(f"=== SSE解析结果统计 ===")
        logger.info(f"成功解析事件数: {parsed_count}")
        logger.info(f"解析失败事件数: {failed_count}")
        logger.info(f"总事件数: {len(events)}")
        
        if events:
            event_types = [event.get('event', 'unknown') for event in events]
            event_counts = Counter(event_types)
            logger.info(f"事件类型分布: {dict(event_counts)}")
        else:
            logger.warning("⚠️ 未解析到任何目标事件")
        
        return events

    async def call_server(self, query: str, max_retries: int = 3, retry_delay: float = 2.0) -> List[Dict]:
        """异步调用server:8020端口，处理流式返回，支持重试机制"""
        payload = {
            "user_id": "166",
            "role_code": 1,
            "query": query,
            "save_method": 0
        }
        
        for attempt in range(max_retries):
            try:
                async with httpx.AsyncClient(timeout=30.0) as client:
                    async with client.stream(
                        'POST',
                        self.server_url,
                        json=payload,
                        headers={'Accept': 'text/event-stream'}
                    ) as response:
                        response.raise_for_status()
                        
                        # 收集所有SSE文本内容
                        sse_content = ""
                        async for line in response.aiter_text():
                            logger.debug(f"Received data: {line}")
                            sse_content += line
                            
                            # 检查是否收到结束标记
                            if '[DONE]' in line:
                                logger.debug("收到结束标记 [DONE]")
                                break
                        
                        # 使用封装的方法解析SSE内容，只解析我们关心的事件
                        events = self.parse_sse_events(
                            sse_content, 
                            filter_events=['tool_call.created', 'tool_response.completed']
                        )
                        
                        # 验证关键事件类型
                        has_tool_call = any(event.get('event') == 'tool_call.created' for event in events)
                        has_tool_response = any(event.get('event') == 'tool_response.completed' for event in events)
                        logger.info(f"包含工具调用事件: {'✅' if has_tool_call else '❌'}")
                        logger.info(f"包含工具响应事件: {'✅' if has_tool_response else '❌'}")
                        
                        return events
                        
            except httpx.RequestError as e:
                logger.warning(f"Call server failed (attempt {attempt + 1}/{max_retries}): {e}")
                if attempt < max_retries - 1:
                    logger.info(f"Retrying in {retry_delay} seconds...")
                    await asyncio.sleep(retry_delay)
                    retry_delay *= 2  # 指数退避
                else:
                    logger.error(f"All retry attempts failed for query: {query[:50]}...")
                    raise Exception(f"Server connection failed after {max_retries} attempts: {e}")
            except httpx.TimeoutException as e:
                logger.warning(f"Server timeout (attempt {attempt + 1}/{max_retries}): {e}")
                if attempt < max_retries - 1:
                    logger.info(f"Retrying in {retry_delay} seconds...")
                    await asyncio.sleep(retry_delay)
                    retry_delay *= 2
                else:
                    logger.error(f"Timeout after all retry attempts for query: {query[:50]}...")
                    raise Exception(f"Server timeout after {max_retries} attempts: {e}")
            except Exception as e:
                logger.error(f"Unexpected error processing response (attempt {attempt + 1}/{max_retries}): {e}")
                if attempt < max_retries - 1:
                    logger.info(f"Retrying in {retry_delay} seconds...")
                    await asyncio.sleep(retry_delay)
                    retry_delay *= 2
                else:
                    logger.error(f"Unexpected error after all retry attempts for query: {query[:50]}...")
                    raise Exception(f"Unexpected error after {max_retries} attempts: {e}")
        
        raise Exception("All retry attempts exhausted")
    
    def extract_tool_calls_and_observations(self, events: List[Dict]) -> Dict[str, List]:
        """Extract tool_call.created and tool_response.completed content from events"""
        tool_calls = []
        tool_responses = []
        
        logger.debug(f"开始提取工具调用和响应，共 {len(events)} 个事件")
        
        for event in events:
            event_type = event.get('event')
            event_data = event.get('data', {})
            
            if event_type == 'tool_call.created':
                logger.debug(f"Extract tool_call.created content: {event}")
                # Extract tool_call information
                tool_call_info = event_data.get('tool_call', {})
                if tool_call_info:
                    tool_calls.append(tool_call_info)  # 直接存储JSON对象
                    logger.debug(f"✅ 提取工具调用: {tool_call_info.get('name', 'unknown')}")
                else:
                    logger.warning(f"❌ tool_call.created 事件中缺少 tool_call 信息")
                    
            elif event_type == 'tool_response.completed':
                logger.debug(f"Extract tool_response.completed content: {event}")
                # Extract tool_response information

                if 'result_delta' in event_data:
                    tool_response = event_data['result_delta'].get('result', [])
                    tool_responses.append(tool_response)  # 直接存储JSON对象
                    logger.debug(f"✅ 提取工具响应: {len(str(tool_response))} 字符")
                else:
                    tool_response = []
                    tool_responses.append(tool_response)  # 直接存储JSON对象
        
        logger.info(f"Extract {len(tool_calls)} tool calls, {len(tool_responses)} tool responses")
        return {
            'tool_calls': tool_calls,
            'tool_responses': tool_responses
        }
    
    def extract_original_data(self, original_data: Dict) -> Dict[str, List]:
        """Extract function_call and observation content from original data"""
        function_calls = []
        observations = []
        
        if 'conversations' in original_data:
            for conv in original_data['conversations']:
                if conv.get('from') == 'function_call':
                    # 解析JSON字符串为对象
                    try:
                        function_call_obj = json.loads(conv.get('value', '{}'))
                        function_calls.append(function_call_obj)
                    except json.JSONDecodeError as e:
                        logger.warning(f"解析function_call JSON时出错: {e}")
                        function_calls.append({})
                elif conv.get('from') == 'observation':
                    # 解析JSON字符串为对象
                    try:
                        observation_obj = json.loads(conv.get('value', '[]'))
                        observations.append(observation_obj)
                    except json.JSONDecodeError as e:
                        logger.warning(f"解析observation JSON时出错: {e}")
                        observations.append([])
        
        return {
            'function_calls': function_calls,
            'observations': observations
        }
    
    def compare_tool_call(self, server_call: Dict, original_call: Dict) -> Dict:
        """比较单个工具调用，检查name和arguments的匹配度"""
        try:
            # 检查name是否一致
            name_match = server_call.get('name') == original_call.get('name')
            name_score = 1.0 if name_match else 0.0
            
            # 检查arguments是否一致
            server_args = server_call.get('arguments', {})
            original_args = original_call.get('arguments', {})
            arguments_match = server_args == original_args
            arguments_score = 1.0 if arguments_match else 0.0
            
            return {
                'name_match': name_match,
                'name_score': name_score,
                'arguments_match': arguments_match,
                'arguments_score': arguments_score,
                'server_name': server_call.get('name', ''),
                'original_name': original_call.get('name', ''),
                'server_arguments': server_args,
                'original_arguments': original_args
            }
        except (KeyError, TypeError) as e:
            logger.warning(f"比较工具调用时出错: {e}")
            return {
                'name_match': False,
                'name_score': 0.0,
                'arguments_match': False,
                'arguments_score': 0.0,
                'server_name': '',
                'original_name': '',
                'server_arguments': {},
                'original_arguments': {},
                'error': str(e)
            }

    def compare_results(self, server_data: Dict[str, List], original_data: Dict[str, List]) -> Dict:
        """详细比较服务器返回结果和原始数据"""
        
        # 初始化比较结果结构
        comparison = {
            'tool_calls_comparison': {
                'server_count': len(server_data['tool_calls']),
                'original_count': len(original_data['function_calls']),
                'detailed_scores': [],
                'name_average_score': 0.0,
                'arguments_average_score': 0.0,
                'non_retrieval_name_average_score': 0.0,
                'non_retrieval_arguments_average_score': 0.0
            },
            'tool_responses_comparison': {
                'server_count': len(server_data['tool_responses']),
                'original_count': len(original_data['observations']),
                'detailed_scores': [],
                'average_score': 0.0
            },
            'overall_scores': {
                'tool_responses_avg': 0.0
            }
        }
        
        # 1. 比较工具调用 (tool_calls)
        tool_call_name_scores = []
        tool_call_arguments_scores = []
        non_retrieval_name_scores = []
        non_retrieval_arguments_scores = []
        max_tool_calls = max(len(server_data['tool_calls']), len(original_data['function_calls']))
        
        for i in range(max_tool_calls):
            server_call = server_data['tool_calls'][i] if i < len(server_data['tool_calls']) else None
            original_call = original_data['function_calls'][i] if i < len(original_data['function_calls']) else None
            
            if server_call is None:
                # 服务器缺少该调用
                score_detail = {
                    'index': i,
                    'server_present': False,
                    'original_present': True,
                    'name_score': 0.0,
                    'arguments_score': 0.0,
                    'original_call': original_call
                }
            elif original_call is None:
                # 原始数据缺少该调用
                score_detail = {
                    'index': i,
                    'server_present': True,
                    'original_present': False,
                    'name_score': 0.0,
                    'arguments_score': 0.0,
                    'server_call': server_call
                }
            else:
                # 两者都存在，进行详细比较
                # 直接使用JSON对象
                call_comparison = self.compare_tool_call(server_call, original_call)
                score_detail = {
                    'index': i,
                    'server_present': True,
                    'original_present': True,
                    'name_score': call_comparison['name_score'],
                    'arguments_score': call_comparison['arguments_score'],
                    'name_match': call_comparison['name_match'],
                    'arguments_match': call_comparison['arguments_match'],
                    'server_name': call_comparison['server_name'],
                    'original_name': call_comparison['original_name'],
                    'server_call': server_call,  # 直接存储JSON对象
                    'original_call': original_call  # 直接存储JSON对象
                }
                if 'error' in call_comparison:
                    score_detail['error'] = call_comparison['error']
            
            comparison['tool_calls_comparison']['detailed_scores'].append(score_detail)
            tool_call_name_scores.append(score_detail['name_score'])
            tool_call_arguments_scores.append(score_detail['arguments_score'])
            
            # 收集非retrieval_tool的评分
            if server_call and original_call:
                server_name = server_call.get('name', '')
                original_name = original_call.get('name', '')
                # 只有当两个都不是retrieval_tool时才计入非retrieval评分
                if server_name != 'retrieval_tool' and original_name != 'retrieval_tool':
                    non_retrieval_name_scores.append(score_detail['name_score'])
                    non_retrieval_arguments_scores.append(score_detail['arguments_score'])
        
        # 计算工具调用name和arguments分别的平均分
        comparison['tool_calls_comparison']['name_average_score'] = (
            sum(tool_call_name_scores) / len(tool_call_name_scores) if tool_call_name_scores else 0.0
        )
        comparison['tool_calls_comparison']['arguments_average_score'] = (
            sum(tool_call_arguments_scores) / len(tool_call_arguments_scores) if tool_call_arguments_scores else 0.0
        )
        
        # 计算非retrieval_tool的name和arguments分别的平均分
        comparison['tool_calls_comparison']['non_retrieval_name_average_score'] = (
            sum(non_retrieval_name_scores) / len(non_retrieval_name_scores) if non_retrieval_name_scores else 0.0
        )
        comparison['tool_calls_comparison']['non_retrieval_arguments_average_score'] = (
            sum(non_retrieval_arguments_scores) / len(non_retrieval_arguments_scores) if non_retrieval_arguments_scores else 0.0
        )
        
        # 2. 比较工具响应 (tool_responses)
        tool_response_scores = []
        max_tool_responses = max(len(server_data['tool_responses']), len(original_data['observations']))
        
        for i in range(max_tool_responses):
            server_response = server_data['tool_responses'][i] if i < len(server_data['tool_responses']) else None
            original_response = original_data['observations'][i] if i < len(original_data['observations']) else None
            
            if server_response is None:
                # 服务器缺少该响应
                score_detail = {
                    'index': i,
                    'server_present': False,
                    'original_present': True,
                    'match_score': 0.0,
                    'original_response': original_response
                }
            elif original_response is None:
                # 原始数据缺少该响应
                score_detail = {
                    'index': i,
                    'server_present': True,
                    'original_present': False,
                    'match_score': 0.0,
                    'server_response': server_response
                }
            else:
                # 两者都存在，比较完全一致性
                responses_match = server_response == original_response
                match_score = 1.0 if responses_match else 0.0
                
                score_detail = {
                    'index': i,
                    'server_present': True,
                    'original_present': True,
                    'match_score': match_score,
                    'responses_match': responses_match,
                    'server_response': server_response,  # 直接存储JSON对象
                    'original_response': original_response  # 直接存储JSON对象
                }
            
            comparison['tool_responses_comparison']['detailed_scores'].append(score_detail)
            tool_response_scores.append(score_detail['match_score'])
        
        # 计算工具响应平均分
        comparison['tool_responses_comparison']['average_score'] = (
            sum(tool_response_scores) / len(tool_response_scores) if tool_response_scores else 0.0
        )
        
        # 3. 计算总体评分
        comparison['overall_scores']['tool_responses_avg'] = comparison['tool_responses_comparison']['average_score']
        
        # 保持向后兼容性
        comparison['tool_calls_match'] = (
            comparison['tool_calls_comparison']['name_average_score'] == 1.0 and 
            comparison['tool_calls_comparison']['arguments_average_score'] == 1.0
        )
        comparison['tool_responses_match'] = comparison['overall_scores']['tool_responses_avg'] == 1.0
        
        return comparison
    
    def calculate_global_scores(self, results: List[Dict]) -> Dict:
        """计算多个结果的全局评分"""
        if not results:
            return {
                'global_tool_responses_avg': 0.0,
                'global_tool_calls_name_avg': 0.0,
                'global_tool_calls_arguments_avg': 0.0,
                'global_non_retrieval_name_avg': 0.0,
                'global_non_retrieval_arguments_avg': 0.0,
                'total_queries': 0
            }
        
        # 收集所有评分
        all_tool_responses_scores = []
        all_tool_calls_name_scores = []
        all_tool_calls_arguments_scores = []
        all_non_retrieval_name_scores = []
        all_non_retrieval_arguments_scores = []
        
        for result in results:
            comparison = result.get('comparison', {})
            overall_scores = comparison.get('overall_scores', {})
            tool_calls_comparison = comparison.get('tool_calls_comparison', {})
            
            tool_responses_avg = overall_scores.get('tool_responses_avg', 0.0)
            
            # 收集每个工具调用的name和arguments分数
            detailed_scores = tool_calls_comparison.get('detailed_scores', [])
            for score_detail in detailed_scores:
                if score_detail.get('server_present') and score_detail.get('original_present'):
                    all_tool_calls_name_scores.append(score_detail.get('name_score', 0.0))
                    all_tool_calls_arguments_scores.append(score_detail.get('arguments_score', 0.0))
                    
                    # 收集非retrieval_tool的评分
                    server_call = score_detail.get('server_call', {})
                    original_call = score_detail.get('original_call', {})
                    if server_call and original_call:
                        server_name = server_call.get('name', '')
                        original_name = original_call.get('name', '')
                        # 只有当两个都不是retrieval_tool时才计入非retrieval评分
                        if server_name != 'retrieval_tool' and original_name != 'retrieval_tool':
                            all_non_retrieval_name_scores.append(score_detail.get('name_score', 0.0))
                            all_non_retrieval_arguments_scores.append(score_detail.get('arguments_score', 0.0))
            
            all_tool_responses_scores.append(tool_responses_avg)
        
        # 计算全局平均分
        global_tool_responses_avg = sum(all_tool_responses_scores) / len(all_tool_responses_scores) if all_tool_responses_scores else 0.0
        global_tool_calls_name_avg = sum(all_tool_calls_name_scores) / len(all_tool_calls_name_scores) if all_tool_calls_name_scores else 0.0
        global_tool_calls_arguments_avg = sum(all_tool_calls_arguments_scores) / len(all_tool_calls_arguments_scores) if all_tool_calls_arguments_scores else 0.0
        global_non_retrieval_name_avg = sum(all_non_retrieval_name_scores) / len(all_non_retrieval_name_scores) if all_non_retrieval_name_scores else 0.0
        global_non_retrieval_arguments_avg = sum(all_non_retrieval_arguments_scores) / len(all_non_retrieval_arguments_scores) if all_non_retrieval_arguments_scores else 0.0
        
        return {
            'global_tool_responses_avg': global_tool_responses_avg,
            'global_tool_calls_name_avg': global_tool_calls_name_avg,
            'global_tool_calls_arguments_avg': global_tool_calls_arguments_avg,
            'global_non_retrieval_name_avg': global_non_retrieval_name_avg,
            'global_non_retrieval_arguments_avg': global_non_retrieval_arguments_avg,
            'total_queries': len(results)
        }
    
    def save_results(self, results: List[Dict], output_file: str):
        """Save evaluation results to file"""
        try:
            # 计算全局评分
            global_scores = self.calculate_global_scores(results)
            
            # 创建包含全局评分的完整结果
            complete_results = {
                'global_scores': global_scores,
                'results': results
            }
            
            with open(output_file, 'w', encoding='utf-8') as f:
                json.dump(complete_results, f, ensure_ascii=False, indent=2)
            logger.info(f"Results saved to: {output_file}")
        except Exception as e:
            logger.error(f"Save results failed: {e}")
    
    def save_checkpoint(self, results: List[Dict], checkpoint_file: str, processed_count: int, total_count: int):
        """保存检查点文件"""
        try:
            checkpoint_data = {
                'processed_count': processed_count,
                'total_count': total_count,
                'results': results,
                'timestamp': time.strftime('%Y-%m-%d %H:%M:%S')
            }
            
            with open(checkpoint_file, 'w', encoding='utf-8') as f:
                json.dump(checkpoint_data, f, ensure_ascii=False, indent=2)
            logger.info(f"Checkpoint saved: {processed_count}/{total_count} processed")
        except Exception as e:
            logger.error(f"Save checkpoint failed: {e}")
    
    def load_checkpoint(self, checkpoint_file: str) -> Dict:
        """加载检查点文件"""
        try:
            if os.path.exists(checkpoint_file):
                with open(checkpoint_file, 'r', encoding='utf-8') as f:
                    checkpoint_data = json.load(f)
                logger.info(f"Checkpoint loaded: {checkpoint_data['processed_count']}/{checkpoint_data['total_count']} processed")
                return checkpoint_data
            else:
                logger.info("No checkpoint file found, starting from beginning")
                return None
        except Exception as e:
            logger.error(f"Load checkpoint failed: {e}")
            return None
    
    def print_progress(self, current: int, total: int, start_time: float):
        """打印进度信息"""
        if total == 0:
            return
        
        elapsed_time = time.time() - start_time
        progress_percent = (current / total) * 100
        
        if current > 0:
            avg_time_per_query = elapsed_time / current
            remaining_queries = total - current
            estimated_remaining_time = remaining_queries * avg_time_per_query
            
            logger.info(f"进度: {current}/{total} ({progress_percent:.1f}%) | "
                       f"成功: {self.success_count} | 错误: {self.error_count} | "
                       f"已用时间: {elapsed_time/60:.1f}分钟 | "
                       f"预计剩余: {estimated_remaining_time/60:.1f}分钟")
        else:
            logger.info(f"进度: {current}/{total} ({progress_percent:.1f}%) | "
                       f"成功: {self.success_count} | 错误: {self.error_count} | "
                       f"已用时间: {elapsed_time/60:.1f}分钟")
    
    def save_progress_report(self, output_file: str, current: int, total: int):
        """保存进度报告"""
        try:
            progress_data = {
                'timestamp': time.strftime('%Y-%m-%d %H:%M:%S'),
                'current_progress': current,
                'total_queries': total,
                'success_count': self.success_count,
                'error_count': self.error_count,
                'progress_percentage': (current / total * 100) if total > 0 else 0,
                'elapsed_time_minutes': (time.time() - self.start_time) / 60 if self.start_time else 0
            }
            
            progress_file = f"{output_file}.progress"
            with open(progress_file, 'w', encoding='utf-8') as f:
                json.dump(progress_data, f, ensure_ascii=False, indent=2)
            
        except Exception as e:
            logger.error(f"Save progress report failed: {e}")
    
    def generate_interruption_report(self, output_file: str, processed_count: int, total_queries: int, error_message: str):
        """生成中断报告"""
        try:
            total_time = time.time() - self.start_time if self.start_time else 0
            
            interruption_report = {
                'interruption_type': 'server_connection_failure',
                'timestamp': time.strftime('%Y-%m-%d %H:%M:%S'),
                'processed_count': processed_count,
                'total_queries': total_queries,
                'success_count': self.success_count,
                'error_count': self.error_count,
                'progress_percentage': (processed_count / total_queries * 100) if total_queries > 0 else 0,
                'elapsed_time_minutes': total_time / 60,
                'error_message': error_message,
                'resume_instructions': {
                    'checkpoint_file': f"{output_file}.checkpoint",
                    'command': f"await evaluator.evaluate(input_file='data/9.17_evaluate_data_top5_final.json', output_file='{output_file}', resume=True)",
                    'note': '使用 resume=True 参数从检查点恢复评估'
                }
            }
            
            interruption_file = f"{output_file}.interruption_report"
            with open(interruption_file, 'w', encoding='utf-8') as f:
                json.dump(interruption_report, f, ensure_ascii=False, indent=2)
            
            logger.info(f"中断报告已保存到: {interruption_file}")
            
        except Exception as e:
            logger.error(f"Generate interruption report failed: {e}")
    
    async def evaluate(self, input_file: str, output_file: str = "evaluation_results.json", 
                      batch_size: int = 50, start_index: int = 0, max_queries: int = None,
                      checkpoint_file: str = None, resume: bool = True):
        """Execute complete evaluation process with batch processing and checkpoint support"""
        logger.info("Start model evaluation...")
        
        # 初始化时间跟踪
        self.start_time = time.time()
        self.error_count = 0
        self.success_count = 0
        
        # 设置检查点文件
        if checkpoint_file is None:
            checkpoint_file = f"{output_file}.checkpoint"
        
        # 尝试加载检查点
        checkpoint_data = None
        if resume:
            checkpoint_data = self.load_checkpoint(checkpoint_file)
            if checkpoint_data:
                self.results = checkpoint_data.get('results', [])
                start_index = checkpoint_data.get('processed_count', 0)
                self.success_count = len(self.results)  # 假设已处理的结果都是成功的
                logger.info(f"Resuming from checkpoint: {start_index} queries already processed")
        
        # 1. Load data
        data = self.load_data(input_file)
        if not data:
            logger.error("Cannot load data, evaluation terminated")
            return
        
        # 2. Extract queries
        queries = self.extract_human_queries(data)
        if not queries:
            logger.error("No valid queries found, evaluation terminated")
            return
        
        # 3. Apply limits and offsets
        if max_queries:
            queries = queries[:max_queries]
        
        if start_index > 0:
            queries = queries[start_index:]
            logger.info(f"Starting from index {start_index}, processing {len(queries)} queries")
        
        # 4. Process queries in batches
        total_queries = len(queries)
        processed_count = len(self.results)  # 从已有结果开始计数
        
        for batch_start in range(0, total_queries, batch_size):
            batch_end = min(batch_start + batch_size, total_queries)
            batch_queries = queries[batch_start:batch_end]
            
            logger.info(f"Processing batch {batch_start//batch_size + 1}: queries {batch_start + 1}-{batch_end} of {total_queries}")
            
            # Process each query in the current batch
            for i, query_data in enumerate(batch_queries):
                global_index = batch_start + i
                logger.info(f"Process {global_index + 1}/{total_queries} query: {query_data['query'][:50]}...")
                
                try:
                    # 异步调用服务器
                    events = await self.call_server(query_data['query'])
                    
                    # Extract tool calls and responses from server
                    server_data = self.extract_tool_calls_and_observations(events)
                    
                    # Extract function_call and observation from original data
                    original_data = self.extract_original_data(query_data['original_data'])
                    
                    # Compare results
                    comparison = self.compare_results(server_data, original_data)
                    
                    # Save results
                    result = {
                        'index': query_data['index'],
                        'query': query_data['query'],
                        'server_events_count': len(events),
                        'server_tool_calls': server_data['tool_calls'],
                        'server_tool_responses': server_data['tool_responses'],
                        'original_function_calls': original_data['function_calls'],
                        'original_observations': original_data['observations'],
                        'comparison': comparison,
                        'timestamp': time.strftime('%Y-%m-%d %H:%M:%S')
                    }
                    
                    self.results.append(result)
                    processed_count += 1
                    self.success_count += 1
                    
                    # 每处理10个查询保存一次检查点和进度报告
                    if processed_count % 10 == 0:
                        self.save_checkpoint(self.results, checkpoint_file, processed_count, total_queries)
                        self.save_progress_report(output_file, processed_count, total_queries)
                        self.print_progress(processed_count, total_queries, self.start_time)
                    
                    # Add delay to avoid server pressure
                    await asyncio.sleep(1)
                    
                except Exception as e:
                    logger.error(f"Error processing query {global_index + 1}: {e}")
                    self.error_count += 1
                    
                    # 检查是否是服务器连接失败（重试后仍然失败）
                    if "Server connection failed after" in str(e) or "Server timeout after" in str(e) or "Unexpected error after" in str(e):
                        logger.error(f"🚨 服务器连接失败，保存检查点并结束评估")
                        logger.error(f"失败查询: {query_data['query'][:50]}...")
                        
                        # 保存当前进度到检查点
                        self.save_checkpoint(self.results, checkpoint_file, processed_count, total_queries)
                        self.save_progress_report(output_file, processed_count, total_queries)
                        
                        # 生成中断报告
                        self.generate_interruption_report(output_file, processed_count, total_queries, str(e))
                        
                        logger.error(f"评估因服务器连接失败而中断")
                        logger.error(f"已处理 {processed_count}/{total_queries} 个查询")
                        logger.error(f"检查点已保存到: {checkpoint_file}")
                        logger.error(f"可以稍后使用 resume=True 从检查点恢复")
                        
                        return  # 直接结束评估
                    else:
                        # 其他类型的错误，继续处理下一个查询
                        logger.warning(f"查询处理失败，继续处理下一个查询: {e}")
                        continue
            
            # Save intermediate results after each batch
            batch_output_file = f"{output_file}.batch_{batch_start//batch_size + 1}"
            self.save_results(self.results, batch_output_file)
            logger.info(f"Batch {batch_start//batch_size + 1} completed, results saved to {batch_output_file}")
            
            # 保存检查点和进度报告
            self.save_checkpoint(self.results, checkpoint_file, processed_count, total_queries)
            self.save_progress_report(output_file, processed_count, total_queries)
            self.print_progress(processed_count, total_queries, self.start_time)
        
        # 5. Save final results
        self.save_results(self.results, output_file)
        
        # 6. 删除检查点文件（处理完成）
        if os.path.exists(checkpoint_file):
            os.remove(checkpoint_file)
            logger.info("Checkpoint file removed after successful completion")
        
        # 7. Generate summary report
        self.generate_summary_report()
        
        # 8. 最终进度报告
        total_time = time.time() - self.start_time
        logger.info(f"=== 评估完成 ===")
        logger.info(f"总查询数: {total_queries}")
        logger.info(f"成功处理: {self.success_count}")
        logger.info(f"处理失败: {self.error_count}")
        logger.info(f"总用时: {total_time/60:.1f}分钟")
        logger.info(f"平均每查询用时: {total_time/total_queries:.1f}秒")
    
    def generate_summary_report(self):
        """生成详细的评估摘要报告"""
        if not self.results:
            return
        
        total_queries = len(self.results)
        
        # 计算全局评分
        global_scores = self.calculate_global_scores(self.results)
        
        # 收集详细评分信息
        query_details = []
        
        for i, result in enumerate(self.results):
            comparison = result['comparison']
            overall_scores = comparison.get('overall_scores', {})
            tool_calls_comparison = comparison.get('tool_calls_comparison', {})
            
            tool_responses_avg = overall_scores.get('tool_responses_avg', 0.0)
            tool_calls_name_avg = tool_calls_comparison.get('name_average_score', 0.0)
            tool_calls_arguments_avg = tool_calls_comparison.get('arguments_average_score', 0.0)
            non_retrieval_name_avg = tool_calls_comparison.get('non_retrieval_name_average_score', 0.0)
            non_retrieval_arguments_avg = tool_calls_comparison.get('non_retrieval_arguments_average_score', 0.0)
            
            query_details.append({
                'index': i,
                'query': result['query'][:50] + '...' if len(result['query']) > 50 else result['query'],
                'tool_calls_name_score': tool_calls_name_avg,
                'tool_calls_arguments_score': tool_calls_arguments_avg,
                'non_retrieval_name_score': non_retrieval_name_avg,
                'non_retrieval_arguments_score': non_retrieval_arguments_avg,
                'tool_responses_score': tool_responses_avg
            })
        
        # 兼容性统计（完全匹配）
        tool_calls_perfect_matches = sum(1 for r in self.results if r['comparison']['tool_calls_match'])
        tool_responses_perfect_matches = sum(1 for r in self.results if r['comparison']['tool_responses_match'])
        
        # 生成报告
        report = f"""
=== 模型评估详细摘要报告 ===

【整体统计】
总查询数: {total_queries}
工具调用完全匹配数: {tool_calls_perfect_matches} ({tool_calls_perfect_matches/total_queries*100:.1f}%)
工具响应完全匹配数: {tool_responses_perfect_matches} ({tool_responses_perfect_matches/total_queries*100:.1f}%)

【全局平均评分】
工具名称匹配平均分: {global_scores['global_tool_calls_name_avg']:.3f}
工具参数匹配平均分: {global_scores['global_tool_calls_arguments_avg']:.3f}
非retrieval工具名称匹配平均分: {global_scores['global_non_retrieval_name_avg']:.3f}
非retrieval工具参数匹配平均分: {global_scores['global_non_retrieval_arguments_avg']:.3f}
工具响应全局平均分: {global_scores['global_tool_responses_avg']:.3f}

【各查询详细评分】"""
        
        for detail in query_details:
            report += f"""
Query {detail['index']}: {detail['query']}
  - 工具名称评分: {detail['tool_calls_name_score']:.3f}
  - 工具参数评分: {detail['tool_calls_arguments_score']:.3f}
  - 非retrieval工具名称评分: {detail['non_retrieval_name_score']:.3f}
  - 非retrieval工具参数评分: {detail['non_retrieval_arguments_score']:.3f}
  - 工具响应评分: {detail['tool_responses_score']:.3f}"""
        
        report += f"""

【评分说明】
- 工具名称匹配分: 工具名称完全一致为1分，否则为0分
- 工具参数匹配分: 工具参数完全一致为1分，否则为0分
- 非retrieval工具名称匹配分: 排除retrieval_tool后，工具名称完全一致为1分，否则为0分
- 非retrieval工具参数匹配分: 排除retrieval_tool后，工具参数完全一致为1分，否则为0分
- 工具响应评分: 完全一致为1分，否则为0分

详细结果请查看 evaluation_results.json 文件
"""
        
        print(report)
        logger.info("详细摘要报告已生成")

    def test_sse_parsing(self):
        """测试SSE解析功能"""
        test_data_tool_call = """id: 3
event: tool_call.created
data: {"conversation_id": "c_9c5b3617", "message_id": "m_1248", "sequence": 3, "role": "assistant", "timestamp": "2025-09-18T13:01:34.230464Z", "content": "", "tool_call": {"name": "intelligent_route_analysis", "arguments": {"access": "1"}}}
"""
        test_data_tool_response = """
id: 4
event: tool_response.completed
data: {"conversation_id": "c_9c5b3617", "message_id": "m_1248", "sequence": 4, "role": "tool", "timestamp": "2025-09-18T13:01:34.358678Z", "tool_call_id": "tool_2", "result_delta": {"chat_log_id": 1234, "content": "", "markdown": "智能路由分析结果\\n\\n访问链接: [上传派团单](https://testai.compassaihz.com/#/$&!upload \\"成功匹配到对应页面\\")\\n\\n", "result": {"success": true, "url": "https://testai.compassaihz.com/#/$&!upload", "message": "成功匹配到对应页面"}, "ambulance": "", "potential_tools": [{"api": "/intelligent_route_analysis", "api_cn": "页面跳转工具", "queryData": [{"type": "select", "key": "access", "label": "数字访问码", "value": {"options": [{"label": "1", "value": "1"}, {"label": "2", "value": "2"}, {"label": "3", "value": "3"}, {"label": "4", "value": "4"}, {"label": "5", "value": "5"}, {"label": "6", "value": "6"}, {"label": "7", "value": "7"}, {"label": "8", "value": "8"}, {"label": "9", "value": "9"}, {"label": "10", "value": "10"}]}, "default": "1", "multiple": false}], "description": "智能页面路由工具，通过输入1-10的数字快速跳转到对应的业务功能页面。业务功能包括：上传派团单，新增资源，新增产品，新增协议，新增协议模版，离线上传协议，离线上传价格政策，新增价格政策，新增报表，新增审批流"}], "tool_calling_chain": [{"role": "function", "tool_call": {"name": "intelligent_route_analysis", "arguments": {"access": "1"}}, "tool_response": {"success": true, "url": "https://testai.compassaihz.com/#/$&!upload", "message": "成功匹配到对应页面"}}], "api_Info": {"api": "/intelligent_route_analysis", "api_cn": "页面跳转工具", "queryData": [{"type": "select", "key": "access", "label": "数字访问码", "value": {"options": [{"label": "1", "value": "1"}, {"label": "2", "value": "2"}, {"label": "3", "value": "3"}, {"label": "4", "value": "4"}, {"label": "5", "value": "5"}, {"label": "6", "value": "6"}, {"label": "7", "value": "7"}, {"label": "8", "value": "8"}, {"label": "9", "value": "9"}, {"label": "10", "value": "10"}]}, "default": "1", "multiple": false}], "description": "智能页面路由工具，通过输入1-10的数字快速跳转到对应的业务功能页面。业务功能包括：上传派团单，新增资源，新增产品，新增协议，新增协议模版，离线上传协议，离线上传价格政策，新增价格政策，新增报表，新增审批流"}}, "success": true, "execution_time": 0.0}

"""
        
        logger.info("=== 开始测试SSE解析功能 ===")
        
        # 合并测试数据
        combined_test_data = test_data_tool_call + test_data_tool_response
        
        # 使用封装的方法解析SSE内容
        events = self.parse_sse_events(
            combined_test_data, 
            filter_events=['tool_call.created', 'tool_response.completed']
        )
        
        logger.info(f"=== 测试解析结果总结 ===")
        logger.info(f"总共解析到 {len(events)} 个事件")
        for event in events:
            logger.info(f"事件: {event['event']}, ID: {event['id']}")
            logger.info(f"   数据摘要: {str(event['data'])[:100]}...")
        
        # 测试提取功能
        extracted = self.extract_tool_calls_and_observations(events)
        logger.info(f"Extraction results from one tool calling: {extracted}")
        
        return events

async def main():
    """Main function"""
    evaluator = ModelEvaluator()
    
    # 首先测试SSE解析功能
    # logger.info("Test SSE parsing function...")
    # evaluator.test_sse_parsing()
    
    # Use the JSON file in the current directory
    input_file = "data/9.17_evaluate_data_top5_final.json"
    output_file = "eval_results/evaluation_results.json"
    
    # 使用新的参数进行评估
    # batch_size: 每批处理50个查询
    # max_queries: 可以限制处理的查询数量（用于测试）
    # resume: 支持断点续传
    await evaluator.evaluate(
        input_file=input_file, 
        output_file=output_file,
        batch_size=50,  # 每批50个查询
        max_queries=None,  # 处理所有查询，可以设置为较小数字进行测试
        checkpoint_file="eval_results/evaluation_results.json.checkpoint",
        resume=True  # 支持断点续传
    )

if __name__ == "__main__":
    asyncio.run(main())