""" H2HMem Leaderboard 评估模块 包含两种评估方式： 1. 词法指标（Lexical Metrics）：Precision, Recall, F1, BLEU-1 2. LLM-as-judge：使用大语言模型评估答案质量 支持按 main_type (Memory Recall/Reasoning/Application) 和 sub_type 分类统计 """ import json import re import logging import os from pathlib import Path from typing import Dict, List, Any, Set, Optional, Tuple from collections import defaultdict import numpy as np # 英文 NLP 工具 try: import nltk from nltk.tokenize import word_tokenize from nltk.corpus import stopwords from nltk.translate.bleu_score import sentence_bleu, SmoothingFunction NLTK_AVAILABLE = True except ImportError: NLTK_AVAILABLE = False raise ImportError("请安装 nltk: pip install nltk") # LLM-as-judge 依赖 try: from openai import OpenAI import tenacity OPENAI_AVAILABLE = True except ImportError: OPENAI_AVAILABLE = False print("警告: openai 未安装，LLM-as-judge 功能不可用") # 下载 nltk 数据 try: nltk.data.find('tokenizers/punkt') except LookupError: nltk.download('punkt', quiet=True) try: nltk.data.find('corpora/stopwords') except LookupError: nltk.download('stopwords', quiet=True) # 配置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) # ========================================================= # 常量定义：类型映射 # ========================================================= # Main type 到类别的映射 MAIN_TYPE_MAPPING = { "Memory Recall": "recall", "Memory Reasoning": "reasoning", "Memory Application": "application" } # Sub type 全称到简称的映射 SUB_TYPE_TO_ABBR = { # Memory Recall 子类型 "Unimodal Precise Recall": "UPR", "Cross-modal Related Retrieval": "CRR", "Knowledge Resolution": "KR", # Memory Reasoning 子类型 "Temporal Reasoning": "TR", "Multimodal Causal Inference": "MCR", "Reference & Evolution Tracking": "RET", # Memory Application 子类型 "Test-Time Learning": "TTL", "Conflict Detection": "CD", "Answer Refusal": "AR" } # 指标列表 METRICS = ["LLM", "P", "R", "F1", "BLEU-1"] # 数据集列表 DATASETS = ["D", "M", "D&M"] # ========================================================= # 第一部分：词法指标计算器 # ========================================================= class EnglishTextProcessor: """英文文本处理器""" def __init__(self, use_stopwords: bool = True): self.use_stopwords = use_stopwords self.stopwords = self._load_stopwords() if use_stopwords else set() def _load_stopwords(self) -> Set[str]: stop_words = set(stopwords.words('english')) extra = {'.', ',', '!', '?', ';', ':', '"', "'", '(', ')', '[', ']', '{', '}', '-', '–', '—', '...', '..'} stop_words.update(extra) return stop_words def tokenize(self, text: str, remove_stopwords: bool = True) -> List[str]: if not text: return [] text = text.lower() text = re.sub(r'\s+', ' ', text).strip() try: tokens = word_tokenize(text) except Exception: tokens = text.split() filtered = [] for t in tokens: t = t.strip() if not t: continue if remove_stopwords and self.use_stopwords and t in self.stopwords: continue if all(ch in "!\"#$%&'()*+,-./:;<=>?@[\\]^_`{|}~" for ch in t): continue filtered.append(t) return filtered class LexicalMetricsCalculator: """词法指标计算器""" def __init__(self): self.text_processor = EnglishTextProcessor() def calculate(self, prediction: str, reference: str) -> Dict[str, float]: return { 'precision': self._precision(prediction, reference), 'recall': self._recall(prediction, reference), 'f1': self._f1(prediction, reference), 'bleu1': self._bleu1(prediction, reference), } def _precision(self, pred: str, ref: str) -> float: p = set(self.text_processor.tokenize(pred)) r = set(self.text_processor.tokenize(ref)) if not p: return 0.0 inter = p & r return len(inter) / len(p) def _recall(self, pred: str, ref: str) -> float: p = set(self.text_processor.tokenize(pred)) r = set(self.text_processor.tokenize(ref)) if not r: return 0.0 inter = p & r return len(inter) / len(r) def _f1(self, pred: str, ref: str) -> float: p = set(self.text_processor.tokenize(pred)) r = set(self.text_processor.tokenize(ref)) if not p or not r: return 0.0 inter = p & r if not inter: return 0.0 prec = len(inter) / len(p) rec = len(inter) / len(r) return 2 * prec * rec / (prec + rec) if (prec + rec) > 0 else 0.0 def _bleu1(self, prediction: str, reference: str, **kwargs) -> float: pred_tokens = self.text_processor.tokenize(prediction, remove_stopwords=False) ref_tokens = self.text_processor.tokenize(reference, remove_stopwords=False) if not pred_tokens or not ref_tokens: return 0.0 if pred_tokens == ref_tokens: return 1.0 try: smoothie = SmoothingFunction().method4 bleu_score = sentence_bleu( [ref_tokens], pred_tokens, weights=(1.0, 0, 0, 0), smoothing_function=smoothie ) return float(bleu_score) except Exception: return 0.0 # ========================================================= # 第二部分：LLM-as-judge 评估器（支持批量评估） # ========================================================= class LLMJudgeEvaluator: """LLM-as-judge 评估器 - 支持批量评估""" SCORE_RUBRIC = """ **Score 0 (Incorrect / Miss):** - The answer contradicts the Ground Truth. - For Yes/No questions: The answer has the wrong polarity. **Score 0.25 (Poor / Tangential):** - Touches on topic but misses core entity. - Contains significant hallucinations. **Score 0.5 (Partial / Vague):** - Technically correct but incomplete. - Captures main entity but misses details. **Score 0.75 (Good / Minor Imperfection):** - Largely accurate, misses only minor details. **Score 1 (Correct / Exact):** - Accurate, precise, and complete. """ def __init__(self, config: Optional[Dict[str, Any]] = None): self.config = config or {} self.api_key = self.config.get('api_key') or os.getenv('OPENAI_API_KEY') self.base_url = self.config.get('base_url') or os.getenv('OPENAI_BASE_URL', 'https://api.openai.com/v1') self.model_name = self.config.get('model_name', 'gpt-4o-mini') self.temperature = self.config.get('temperature', 0) self.batch_size = self.config.get('batch_size', 5) self._client = None def _get_client(self): if self._client is None: if not self.api_key: raise ValueError("未设置OPENAI_API_KEY") self._client = OpenAI(api_key=self.api_key, base_url=self.base_url) return self._client def _build_batch_prompt(self, qa_pairs: List[Dict[str, str]]) -> str: """构建批量评估的提示词，包含临时 ID""" prompt = f"""You are an impartial judge evaluating an AI assistant's answers. You will evaluate {len(qa_pairs)} questions and provide scores for each. ### Scoring Rubric {self.SCORE_RUBRIC} ### Questions and Answers """ for item in qa_pairs: prompt += f""" --- Question ID: {item['temp_id']} --- Question: {item['question']} Ground Truth: {item['ground_truth']} Assistant Answer: {item['assistant_answer'] if item['assistant_answer'] else "(No answer provided)"} """ prompt += """ ### Output Format Output strictly in JSON format as a list of objects, each containing the question_id and score: [ {"question_id": 0, "score": 0.75, "reasoning": ""}, {"question_id": 1, "score": 0.5, "reasoning": ""}, ... ] Each score must be one of: 0, 0.25, 0.5, 0.75, or 1. The number of objects must equal the number of questions. The question_id must match the ID provided above. """ return prompt def _parse_batch_response(self, response_text: str, qa_pairs: List[Dict[str, str]]) -> List[Dict[str, Any]]: """解析批量评估的响应，按临时 ID 匹配""" expected_count = len(qa_pairs) # 创建 ID 到结果的映射 result_map = {item['temp_id']: None for item in qa_pairs} try: # 尝试解析 JSON 数组 json_match = re.search(r'\[.*\]', response_text, re.DOTALL) if json_match: parsed_results = json.loads(json_match.group()) if isinstance(parsed_results, list): for r in parsed_results: if isinstance(r, dict) and 'question_id' in r: qid = r['question_id'] score = float(r.get('score', 0)) valid_scores = [0, 0.25, 0.5, 0.75, 1] if score not in valid_scores: score = min(valid_scores, key=lambda x: abs(x - score)) result_map[qid] = { 'score': score, 'reasoning': r.get('reasoning', ''), 'success': True } # 为每个问题返回结果（未匹配的使用默认值） results = [] for item in qa_pairs: qid = item['temp_id'] if result_map[qid] is not None: results.append(result_map[qid]) else: logger.warning(f"问题 ID {qid} 未在响应中找到，使用默认值") results.append({'score': 0, 'reasoning': 'Not found in response', 'success': False}) return results except Exception as e: logger.error(f"解析批量响应失败: {e}") return [{'score': 0, 'reasoning': str(e), 'success': False} for _ in range(expected_count)] @tenacity.retry( stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential(multiplier=1, min=1, max=10), retry=tenacity.retry_if_exception_type(Exception), ) def evaluate_batch(self, qa_pairs: List[Dict[str, str]]) -> List[Dict[str, Any]]: """批量评估多个问答对""" if not qa_pairs: return [] try: client = self._get_client() prompt = self._build_batch_prompt(qa_pairs) response = client.chat.completions.create( model=self.model_name, messages=[{"role": "user", "content": prompt}], temperature=self.temperature, timeout=120 ) response_text = response.choices[0].message.content.strip() return self._parse_batch_response(response_text, qa_pairs) except Exception as e: logger.error(f"批量LLM评估失败: {e}") return [{'score': 0, 'reasoning': str(e), 'success': False} for _ in range(len(qa_pairs))] def evaluate_single(self, question: str, ground_truth: str, assistant_answer: str) -> Dict[str, Any]: """单个评估（向后兼容）""" result = self.evaluate_batch([{ 'temp_id': 0, 'question': question, 'ground_truth': ground_truth, 'assistant_answer': assistant_answer }]) return result[0] if result else {'score': 0, 'reasoning': 'Failed', 'success': False} # ========================================================= # 第三部分：评估结果收集器 # ========================================================= class ScoreCollector: """分数收集器，用于按类别统计加权平均""" def __init__(self): self.scores = defaultdict(list) self.weights = defaultdict(int) def add(self, category: str, score: float): self.scores[category].append(score) self.weights[category] += 1 def get_weighted_average(self, category: str = None) -> float: if category: scores = self.scores.get(category, []) return np.mean(scores) if scores else 0.0 total_score = 0 total_weight = 0 for cat, scores in self.scores.items(): total_score += sum(scores) total_weight += len(scores) return total_score / total_weight if total_weight > 0 else 0.0 def get_all_averages(self) -> Dict[str, float]: return {cat: np.mean(scores) if scores else 0.0 for cat, scores in self.scores.items()} # ========================================================= # 第四部分：主评估器 # ========================================================= class SubmissionEvaluator: """提交文件评估器""" def __init__(self, llm_config: Optional[Dict[str, Any]] = None, enable_llm_judge: bool = True): self.lexical_calculator = LexicalMetricsCalculator() self.enable_llm_judge = enable_llm_judge and OPENAI_AVAILABLE if self.enable_llm_judge and llm_config: try: self.llm_judge = LLMJudgeEvaluator(llm_config) logger.info(f"LLM-as-judge 已启用，批量大小: {self.llm_judge.batch_size}") except Exception as e: logger.warning(f"LLM-as-judge 初始化失败: {e}") self.enable_llm_judge = False self.llm_judge = None else: self.llm_judge = None def load_predictions(self, file_path: str) -> List[Dict]: with open(file_path, 'r', encoding='utf-8') as f: data = json.load(f) if 'predictions' not in data: raise ValueError(f"文件格式错误: 缺少 'predictions' 字段") return data['predictions'] def evaluate_single_prediction(self, pred: Dict) -> Dict[str, float]: """评估单个预测的词法指标（不包含 LLM）""" system_answer = pred.get('system_answer', '').strip() original_answer = pred.get('original_answer', '').strip() lexical = self.lexical_calculator.calculate(system_answer, original_answer) return { 'P': lexical['precision'], 'R': lexical['recall'], 'F1': lexical['f1'], 'BLEU-1': lexical['bleu1'] } def evaluate_dataset(self, predictions: List[Dict], dataset_name: str, enable_llm: bool = True) -> Dict[str, Any]: """ 评估单个数据集，按 main_type 和 sub_type 分类统计 """ # 存储每个问题的词法指标 all_lexical_metrics = [] category_lexical_metrics = {cat: [] for cat in SUB_TYPE_TO_ABBR.values()} # 按 main_type 和 sub_type 分类的收集器（词法指标） main_type_lexical = {mt: [] for mt in ['recall', 'reasoning', 'application']} sub_type_lexical = {st: [] for st in SUB_TYPE_TO_ABBR.values()} # 存储每个问题的完整信息（用于后续分类） questions_info = [] # 第一遍：计算词法指标，收集问题信息 for idx, pred in enumerate(predictions): system_answer = pred.get('system_answer', '').strip() original_answer = pred.get('original_answer', '').strip() question_text = pred.get('question_text', '') # 获取类型信息 main_type_full = pred.get('question_type', {}).get('main_type', '') sub_type_full = pred.get('question_type', {}).get('sub_type', '') main_type = MAIN_TYPE_MAPPING.get(main_type_full, '') sub_type_abbr = SUB_TYPE_TO_ABBR.get(sub_type_full, '') # 计算词法指标 lexical_scores = self.evaluate_single_prediction(pred) # 存储信息 questions_info.append({ 'index': idx, 'main_type': main_type, 'sub_type': sub_type_abbr, 'lexical_scores': lexical_scores, 'question_text': question_text, 'original_answer': original_answer, 'system_answer': system_answer }) # 收集词法指标 all_lexical_metrics.append(lexical_scores) # 按类别收集词法指标 if sub_type_abbr and sub_type_abbr in category_lexical_metrics: category_lexical_metrics[sub_type_abbr].append(lexical_scores) # 按 main_type 收集 if main_type and main_type in main_type_lexical: main_type_lexical[main_type].append(lexical_scores) # 按 sub_type 收集 if sub_type_abbr and sub_type_abbr in sub_type_lexical: sub_type_lexical[sub_type_abbr].append(lexical_scores) # ===================================================== # LLM-as-judge 批量评估 # ===================================================== llm_scores_by_index = {} # 索引 -> LLM分数 if self.enable_llm_judge and enable_llm and self.llm_judge: # 准备批量评估数据 batch_size = self.llm_judge.batch_size total_questions = len(questions_info) total_batches = (total_questions + batch_size - 1) // batch_size logger.info(f"开始批量LLM评估，共 {total_questions} 个问题，分成 {total_batches} 批（每批最多 {batch_size} 个）") for batch_start in range(0, total_questions, batch_size): batch_end = min(batch_start + batch_size, total_questions) batch_questions = questions_info[batch_start:batch_end] current_batch_num = batch_start // batch_size + 1 logger.info(f"评估第 {current_batch_num}/{total_batches} 批，包含 {len(batch_questions)} 个问题...") # 准备批量数据，添加临时 ID batch_data = [] for i, q in enumerate(batch_questions): batch_data.append({ 'temp_id': batch_start + i, # 使用原始索引作为临时 ID 'question': q['question_text'], 'ground_truth': q['original_answer'], 'assistant_answer': q['system_answer'] }) # 批量调用 LLM batch_results = self.llm_judge.evaluate_batch(batch_data) # 存储结果 for i, result in enumerate(batch_results): original_idx = batch_start + i if result['success']: llm_scores_by_index[original_idx] = result['score'] else: llm_scores_by_index[original_idx] = 0.0 success_count = len([r for r in batch_results if r['success']]) logger.info(f"第 {current_batch_num} 批完成，成功 {success_count}/{len(batch_questions)} 个") logger.info(f"LLM批量评估完成，总成功 {len(llm_scores_by_index)}/{total_questions} 个") # ===================================================== # 收集完整的分数（词法 + LLM） # ===================================================== # 用于收集按类别的 LLM 分数 main_type_llm = {mt: [] for mt in ['recall', 'reasoning', 'application']} sub_type_llm = {st: [] for st in SUB_TYPE_TO_ABBR.values()} for q in questions_info: idx = q['index'] llm_score = llm_scores_by_index.get(idx, 0.0) # 按 main_type 收集 LLM 分数 if q['main_type'] and q['main_type'] in main_type_llm: main_type_llm[q['main_type']].append(llm_score) # 按 sub_type 收集 LLM 分数 if q['sub_type'] and q['sub_type'] in sub_type_llm: sub_type_llm[q['sub_type']].append(llm_score) # ===================================================== # 计算各类别的平均分 # ===================================================== def calc_avg(scores_list): return np.mean(scores_list) if scores_list else 0.0 # 计算整体词法平均 overall_lexical = { 'P': calc_avg([m['P'] for m in all_lexical_metrics]), 'R': calc_avg([m['R'] for m in all_lexical_metrics]), 'F1': calc_avg([m['F1'] for m in all_lexical_metrics]), 'BLEU-1': calc_avg([m['BLEU-1'] for m in all_lexical_metrics]) } # 计算整体 LLM 平均 overall_llm = calc_avg(list(llm_scores_by_index.values())) # 构建结果 result = { 'total': len(predictions), 'main_type_counts': {mt: len(main_type_lexical[mt]) for mt in main_type_lexical}, 'sub_type_counts': {st: len(sub_type_lexical[st]) for st in sub_type_lexical} } # 为每个指标构建结果 for metric in METRICS: result[metric] = { 'overall': 0.0, 'recall': {'score': 0.0, 'sub_types': {}}, 'reasoning': {'score': 0.0, 'sub_types': {}}, 'application': {'score': 0.0, 'sub_types': {}} } if metric == 'LLM': # LLM 指标 result[metric]['overall'] = overall_llm for mt in ['recall', 'reasoning', 'application']: result[metric][mt]['score'] = calc_avg(main_type_llm.get(mt, [])) for st, abbr in SUB_TYPE_TO_ABBR.items(): if abbr in ['UPR', 'CRR', 'KR']: result[metric]['recall']['sub_types'][abbr] = calc_avg(sub_type_llm.get(abbr, [])) elif abbr in ['TR', 'MCR', 'RET']: result[metric]['reasoning']['sub_types'][abbr] = calc_avg(sub_type_llm.get(abbr, [])) elif abbr in ['TTL', 'CD', 'AR']: result[metric]['application']['sub_types'][abbr] = calc_avg(sub_type_llm.get(abbr, [])) else: # 词法指标 metric_key = metric if metric != 'BLEU-1' else 'BLEU-1' result[metric]['overall'] = overall_lexical.get(metric_key, 0.0) for mt in ['recall', 'reasoning', 'application']: scores = [m[metric_key] for m in main_type_lexical.get(mt, [])] result[metric][mt]['score'] = calc_avg(scores) for st, abbr in SUB_TYPE_TO_ABBR.items(): scores = [m[metric_key] for m in sub_type_lexical.get(abbr, [])] if abbr in ['UPR', 'CRR', 'KR']: result[metric]['recall']['sub_types'][abbr] = calc_avg(scores) elif abbr in ['TR', 'MCR', 'RET']: result[metric]['reasoning']['sub_types'][abbr] = calc_avg(scores) elif abbr in ['TTL', 'CD', 'AR']: result[metric]['application']['sub_types'][abbr] = calc_avg(scores) return result def evaluate_submission(self, dyadic_path: str, multiparty_path: str, method_name: str = "", backbone: str = "", category: str = "", organization: str = "", paper_url: str = "", code_url: str = "") -> Dict[str, Any]: """ 评估完整的提交，返回 pending_leaderboard 格式的数据 """ try: # 加载预测文件 dyadic_preds = self.load_predictions(dyadic_path) multiparty_preds = self.load_predictions(multiparty_path) logger.info(f"Dyadic: {len(dyadic_preds)} 个问题") logger.info(f"Multiparty: {len(multiparty_preds)} 个问题") # 分别评估两种对话类型 dyadic_results = self.evaluate_dataset(dyadic_preds, "D") multiparty_results = self.evaluate_dataset(multiparty_preds, "M") # 计算 D&M 的加权平均（按问题数量） total_d = dyadic_results['total'] total_m = multiparty_results['total'] total_all = total_d + total_m dm_results = {'total': total_all} # 对于每个指标，计算加权平均 for metric in METRICS: dm_results[metric] = {'overall': 0.0, 'recall': {'score': 0.0, 'sub_types': {}}, 'reasoning': {'score': 0.0, 'sub_types': {}}, 'application': {'score': 0.0, 'sub_types': {}}} if total_all > 0: # overall 加权平均 dm_results[metric]['overall'] = ( dyadic_results[metric]['overall'] * total_d + multiparty_results[metric]['overall'] * total_m ) / total_all # recall 加权平均 dm_results[metric]['recall']['score'] = ( dyadic_results[metric]['recall']['score'] * total_d + multiparty_results[metric]['recall']['score'] * total_m ) / total_all # reasoning 加权平均 dm_results[metric]['reasoning']['score'] = ( dyadic_results[metric]['reasoning']['score'] * total_d + multiparty_results[metric]['reasoning']['score'] * total_m ) / total_all # application 加权平均 dm_results[metric]['application']['score'] = ( dyadic_results[metric]['application']['score'] * total_d + multiparty_results[metric]['application']['score'] * total_m ) / total_all # 子类型加权平均 for sub in ['UPR', 'CRR', 'KR']: dm_results[metric]['recall']['sub_types'][sub] = ( dyadic_results[metric]['recall']['sub_types'].get(sub, 0) * total_d + multiparty_results[metric]['recall']['sub_types'].get(sub, 0) * total_m ) / total_all for sub in ['TR', 'MCR', 'RET']: dm_results[metric]['reasoning']['sub_types'][sub] = ( dyadic_results[metric]['reasoning']['sub_types'].get(sub, 0) * total_d + multiparty_results[metric]['reasoning']['sub_types'].get(sub, 0) * total_m ) / total_all for sub in ['TTL', 'CD', 'AR']: dm_results[metric]['application']['sub_types'][sub] = ( dyadic_results[metric]['application']['sub_types'].get(sub, 0) * total_d + multiparty_results[metric]['application']['sub_types'].get(sub, 0) * total_m ) / total_all # 构建 pending_leaderboard 条目 pending_entry = { "method": { "name": method_name, "backbone": backbone, "category": category, "organization": organization, "paper_url": paper_url, "code_url": code_url }, "results": { "D": {}, "M": {}, "D&M": {} } } # 填充 D 结果 for metric in METRICS: pending_entry["results"]["D"][metric] = { "overall": dyadic_results[metric]['overall'], "recall": dyadic_results[metric]['recall'], "reasoning": dyadic_results[metric]['reasoning'], "application": dyadic_results[metric]['application'] } # 填充 M 结果 for metric in METRICS: pending_entry["results"]["M"][metric] = { "overall": multiparty_results[metric]['overall'], "recall": multiparty_results[metric]['recall'], "reasoning": multiparty_results[metric]['reasoning'], "application": multiparty_results[metric]['application'] } # 填充 D&M 结果 for metric in METRICS: pending_entry["results"]["D&M"][metric] = { "overall": dm_results[metric]['overall'], "recall": dm_results[metric]['recall'], "reasoning": dm_results[metric]['reasoning'], "application": dm_results[metric]['application'] } # 构建 dataset_scores（用于 pending_submissions 表格） dataset_scores = { "D": {}, "M": {}, "D&M": {} } for dataset in ["D", "M", "D&M"]: for metric in METRICS: dataset_scores[dataset][metric] = pending_entry["results"][dataset][metric]["overall"] return { 'pending_leaderboard_entry': pending_entry, 'dataset_scores': dataset_scores } except Exception as e: logger.error(f"评估失败: {e}") import traceback traceback.print_exc() # 返回空结果 empty_entry = { "method": {"name": method_name, "backbone": backbone, "category": category, "organization": organization, "paper_url": paper_url, "code_url": code_url}, "results": {} } for dataset in DATASETS: empty_entry["results"][dataset] = {} for metric in METRICS: empty_entry["results"][dataset][metric] = { "overall": 0.0, "recall": {"score": 0.0, "sub_types": {"UPR": 0, "CRR": 0, "KR": 0}}, "reasoning": {"score": 0.0, "sub_types": {"TR": 0, "MCR": 0, "RET": 0}}, "application": {"score": 0.0, "sub_types": {"TTL": 0, "CD": 0, "AR": 0}} } empty_scores = {d: {m: 0.0 for m in METRICS} for d in DATASETS} return { 'pending_leaderboard_entry': empty_entry, 'dataset_scores': empty_scores } # ========================================================= # 全局函数 # ========================================================= _evaluator = None def evaluate_submission(dyadic_path: str, multiparty_path: str, method_name: str = "", backbone: str = "", category: str = "", organization: str = "", paper_url: str = "", code_url: str = "") -> Dict[str, Any]: """ 评估提交的函数入口 Returns: 包含 pending_leaderboard_entry 和 dataset_scores 的字典 """ global _evaluator if _evaluator is None: llm_config = { 'api_key': os.getenv('OPENAI_API_KEY'), 'base_url': os.getenv('OPENAI_BASE_URL', 'https://api.openai.com/v1'), 'model_name': os.getenv('LLM_JUDGE_MODEL', 'gpt-4o-mini'), 'temperature': 0, 'batch_size': 5 # 每批最多5个问题 } _evaluator = SubmissionEvaluator(llm_config=llm_config, enable_llm_judge=True) return _evaluator.evaluate_submission( dyadic_path, multiparty_path, method_name, backbone, category, organization, paper_url, code_url )