import os import sys # 添加父一级目录到 sys.path（上一级） parent_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), '..')) if parent_dir not in sys.path: sys.path.insert(0, parent_dir) from dataflow.utils.reasoning.AnswerExtraction import StringCleaner, UnitTextManager, AnswerExtractor from prompts.bench_evaluate import AnswerJudgePromptQuestion, AnswerJudgeMultipleQuestionsPrompt from dataflow.core.prompt import DIYPromptABC from dataflow.utils.registry import OPERATOR_REGISTRY from dataflow.utils.storage import DataFlowStorage from dataflow.core import LLMServingABC from dataflow.core import OperatorABC from math_verify import parse, verify from dataflow import get_logger from typing import Literal import pandas as pd import numpy as np import time import re import json import json5 @OPERATOR_REGISTRY.register() class BenchDatasetEvaluatorQuestion(OperatorABC): def __init__(self, eval_result_path: str = None, compare_method: Literal["match", "semantic"] = "match", system_prompt: str = "You are a helpful assistant specialized in evaluating answer correctness.", llm_serving: LLMServingABC = None, prompt_template: DIYPromptABC = None, support_subquestions: bool = False, skip_true: bool = False, # 是否跳过已经验证过为True的样本 ): if eval_result_path is None: timestamp = int(time.time()) eval_result_path = f"result_bencheval/BenchDatasetEvaluator_result_{timestamp}.json" self.eval_result_path = eval_result_path self.compare_method = compare_method self.empty_responses_count = 0 # 添加空响应计数器 if compare_method == "match": self.compare = self.math_verify_compare unit_manager = UnitTextManager() string_cleaner = StringCleaner(unit_manager) self.answer_extractor = AnswerExtractor(string_cleaner) else: if prompt_template is None: prompt_template = AnswerJudgePromptQuestion() if not support_subquestions else AnswerJudgeMultipleQuestionsPrompt() self.prompt_template = prompt_template self.system_prompt = system_prompt self.llm_serving = llm_serving self.support_subquestions = support_subquestions self.skip_true = skip_true self.logger = get_logger() def math_verify_compare(self, answer, ground_truth): try: return verify(parse(str(ground_truth)), parse(str(answer))) except: try: return verify(parse(ground_truth), parse(answer)) except: return False def ResolveResponse(self, response): # 检查空响应 if not self.support_subquestions: if response is None or (isinstance(response, str) and response.strip() == ''): self.empty_responses_count += 1 return False try: pattern = re.compile(r'"judgement_result"\s*:\s*(true|false)', re.IGNORECASE) match = pattern.search(response) result_value = None if match: result_value = match.group(1).lower() else: # 备用解析逻辑，检查响应中是否包含true或false if "true" in response.lower(): result_value = "true" else: result_value = "false" if result_value == "true": return True else: return False except Exception as e: self.logger.error(f"Response format error: {response}. Error: {e}") return False if self.support_subquestions: # 如果支持子问题，假设response是一个列表, 返回正确的数量/总数 correct_num = 0 total_num = 0 try: response = json5.loads(response, strict=False) # 使用json5解析，允许更宽松的格式 judgement = response.get("judgement", []) except Exception as e: self.logger.error(f"Response JSON parse error: {response}. Error: {e}") self.empty_responses_count += 1 return "0/0" for resp in judgement: if isinstance(resp, bool): if resp is True: correct_num += 1 total_num += 1 elif resp is False: total_num += 1 elif resp.lower() == "empty": continue # 不计入总数 elif isinstance(resp, str): if resp.lower() == "true": correct_num += 1 total_num += 1 elif resp.lower() == "false": total_num += 1 elif resp.lower() == "empty": continue # 不计入总数 return f"{correct_num}/{total_num}" @staticmethod def get_desc(lang: str = "zh"): if lang == "zh": return ( "该算子用于对比预测答案与标准答案的匹配度，支持两种评估模式：\n\n" "1. 字符串匹配（match）：使用数学验证方法比较答案，适用于有明确答案的问题\n" "2. 语义匹配（semantic）：使用LLM评估答案的语义相似度，适用于开放性问题\n\n" "输入参数：\n" "- input_test_answer_key：预测答案字段名\n" "- input_gt_answer_key：标准答案字段名\n" "- input_question_key：问题字段名（语义匹配模式下必需）\n" "- compare_method：比较方法（match/semantic）\n\n" "输出参数：\n" "- answer_match_result：匹配结果（True/False）\n" "- 统计结果将保存到指定的eval_result_path路径\n" ) elif lang == "en": return ( "This operator compares predicted answers against ground truth using two evaluation modes:\n\n" "1. String Matching (match): Uses mathematical verification to compare answers, suitable for questions with definitive answers\n" "2. Semantic Matching (semantic): Uses LLM to evaluate semantic similarity, suitable for open-ended questions\n\n" "Input Parameters:\n" "- input_test_answer_key: Predicted answer field\n" "- input_gt_answer_key: Ground truth field\n" "- input_question_key: Question field (required for semantic mode)\n" "- compare_method: Comparison method (match/semantic)\n\n" "Output Parameters:\n" "- answer_match_result: Matching result (True/False)\n" "- Statistics will be saved to the specified eval_result_path\n" ) else: return "BenchEvaluator performs answer validation using string matching or semantic comparison" def check_column(self, required_columns: list[str], dataframe: pd.DataFrame): for column in required_columns: if column not in dataframe.columns: self.logger.error(f"Required column '{column}' not found in dataframe") return False return True def statistic(self, file_name_prefix: str, dataframe: pd.DataFrame, compare_method: Literal["match", "semantic"]): total_samples = len(dataframe) valid_samples = len(dataframe) - self.empty_responses_count matched_samples = sum(dataframe['answer_match_result']) accuracy = matched_samples / valid_samples if valid_samples > 0 else 0 # 创建统计信息字典 stats = { "bench_name_or_prefix": file_name_prefix, "total_samples": total_samples, "valid_samples": valid_samples, "matched_samples": matched_samples, "accuracy": float(accuracy), # 确保可以被JSON序列化 "empty_responses_count": self.empty_responses_count, "compare_method": compare_method } if self.support_subquestions: total_subquestions = dataframe['total_subquestions'].sum() correct_subquestions = dataframe['correct_answer_num'].sum() subquestion_accuracy = correct_subquestions / total_subquestions if total_subquestions > 0 else 0 stats.update({ "total_subquestions": int(total_subquestions), "correct_subquestions": int(correct_subquestions), "subquestion_accuracy": float(subquestion_accuracy) }) # 将字典转换为DataFrame stats_df = pd.DataFrame([stats]) # 直接将统计信息写入到self.eval_result_path os.makedirs(os.path.dirname(self.eval_result_path), exist_ok=True) stats_df.to_json(self.eval_result_path, orient="records", force_ascii=False, indent=2) self.logger.success(f"Statistics saved to {self.eval_result_path}") return stats_df def run( self, storage:DataFlowStorage, input_test_answer_key: str = "generated_cot", input_gt_answer_key: str = "golden_answer", input_question_key: str = None, ) -> list: self.test_answer_key = input_test_answer_key self.gt_answer_key = input_gt_answer_key self.question_key = input_question_key dataframe = storage.read("dataframe") if 'answer_match_result' not in dataframe.columns: dataframe['answer_match_result'] = False answers = dataframe[self.test_answer_key] ground_truths = dataframe[self.gt_answer_key] if self.compare_method == "match": if self.check_column( required_columns=[input_test_answer_key,input_gt_answer_key], dataframe=dataframe ) is False: return required_columns for i in range(len(answers)): final_answer = self.answer_extractor.extract_answer(answers[i], None) if self.compare(final_answer, ground_truths[i]): dataframe.at[i, 'answer_match_result'] = True else: dataframe.at[i, 'answer_match_result'] = False output_file = storage.write(dataframe) # 生成统计信息并直接写入JSON文件 stats = self.statistic(storage.file_name_prefix, dataframe, self.compare_method) return [self.test_answer_key, self.gt_answer_key, 'answer_match_result'] else: if self.check_column( required_columns=[input_test_answer_key,input_gt_answer_key, input_question_key], dataframe=dataframe ) is False: return required_columns empty_reference_mask = dataframe[input_gt_answer_key].isna() | (dataframe[input_gt_answer_key] == '') if self.skip_true: empty_reference_mask = empty_reference_mask | (dataframe['answer_match_result'] == True) skipped_rows = dataframe[empty_reference_mask] valid_rows = dataframe[~empty_reference_mask] skipped_count = len(skipped_rows) if len(valid_rows) == 0 and not self.skip_true: self.logger.warning("No valid samples with reference answers found. All samples skipped.") if self.keep_all_samples: output_file = storage.write(dataframe) # 保留所有行，但answer_match_result都为False else: output_file = storage.write(pd.DataFrame(columns=dataframe.columns)) # 不保留任何行 self.logger.info(f"Dataframe saved to {output_file}. Skipped {skipped_count} samples due to missing reference answers.") return required_columns + ['answer_match_result'] # 只对有参考答案的行构建提示词并调用LLM inputs = [self.prompt_template.build_prompt( question=row[input_question_key], answer=row[input_test_answer_key], reference_answer=row[input_gt_answer_key] ) for _, row in valid_rows.iterrows()] responses = self.llm_serving.generate_from_input(user_inputs=inputs, system_prompt=self.system_prompt) # if self.support_subquestions: # # 每个response是一个列表，连接一个长列表，比如[["true", "false"], ["true"]] -> ["true", "false", "true"] # responses = [item for sublist in responses for item in sublist] results = [self.ResolveResponse(response) for response in responses] # 创建结果掩码，与valid_rows长度相同 result_mask = np.array(results, dtype=bool) # 更新有效行的answer_match_result valid_indices = valid_rows.index if not self.support_subquestions: for i, idx in enumerate(valid_indices): dataframe.at[idx, 'answer_match_result'] = results[i] else: for i, idx in enumerate(valid_indices): correct_answer_num = int(results[i].split('/')[0]) total_subquestions = int(results[i].split('/')[1]) dataframe.at[idx, 'correct_answer_num'] = correct_answer_num dataframe.at[idx, 'total_subquestions'] = total_subquestions dataframe.at[idx, 'answer_match_result'] = (correct_answer_num == total_subquestions) and (total_subquestions > 0) # 全对为True，否则为False dataframe.at[idx, 'response_evaluation'] = responses[i] # 保存LLM的原始响应内容 output_file = storage.write(dataframe) # 生成统计信息并直接写入JSON文件 stats = self.statistic(storage.file_name_prefix, dataframe, self.compare_method) # 重置空响应计数器 self.empty_responses_count = 0 return [input_test_answer_key, input_gt_answer_key, input_question_key, 'answer_match_result']