refactor(core_text): split BenchDatasetEvaluatorQuestion.run and fix row indexing (#514)

* Split BenchDatasetEvaluatorQuestion.run into helpers and fix row indexing

run() carried both evaluation modes inline at ~90 lines. Splitting it
into per-mode helpers made two latent problems visible:

- Match mode indexed the answer Series by position. That is a label
  lookup on a pandas Series, so any frame with a non-default index --
  what an upstream filter produces -- raised KeyError. Now iterates
  dataframe.index.
- Column validation ran after 'answer_match_result' had already been
  added, so a frame failing validation came back mutated.

Also drops the unused numpy and json imports and the dead result_mask.

* fix(core_text): validate semantic response count

---------

Co-authored-by: Sunnyhaze <mxch1122@126.com>

Author: MBemera

dataflow/operators/core_text/eval/bench_dataset_evaluator_question.py

@@ -11,11 +11,9 @@
 from dataflow import get_logger
 from typing import Literal, Union
 import pandas as pd
-import numpy as np
 import time
-import os  # 添加os模块导入
+import os
 import re
-import json
 import json5
 
 @prompt_restrict(
@@ -32,7 +30,7 @@ def __init__(self,
                 llm_serving: LLMServingABC = None,
                 prompt_template: Union[AnswerJudgePromptQuestion, AnswerJudgeMultipleQuestionsPrompt, DIYPromptABC] = AnswerJudgePromptQuestion,
                 support_subquestions: bool = False,
-                keep_all_samples: bool = False
+                keep_all_samples: bool = False,
                 ):
 
         if eval_result_path is None:
@@ -44,7 +42,7 @@ def __init__(self,
         self.empty_responses_count = 0  # 添加空响应计数器
         self.keep_all_samples = keep_all_samples
         self.support_subquestions = support_subquestions
-
+        
         if compare_method == "match":
             self.compare = self.math_verify_compare
             unit_manager = UnitTextManager()
@@ -203,7 +201,136 @@ def statistic(self, file_name_prefix: str, dataframe: pd.DataFrame, compare_meth
         self.logger.success(f"Statistics saved to {self.eval_result_path}")
 
         return stats_df
-        
+
+    def _get_required_columns(
+        self,
+        input_test_answer_key: str,
+        input_gt_answer_key: str,
+        input_question_key: str,
+    ) -> list[str]:
+        required_columns = [input_test_answer_key, input_gt_answer_key]
+        if self.compare_method == "semantic":
+            required_columns.append(input_question_key)
+        return required_columns
+
+    def _run_match_evaluation(
+        self,
+        storage: DataFlowStorage,
+        dataframe: pd.DataFrame,
+        required_columns: list[str],
+    ) -> list[str]:
+        for row_index in dataframe.index:
+            answer = dataframe.at[row_index, self.test_answer_key]
+            ground_truth = dataframe.at[row_index, self.gt_answer_key]
+            final_answer = self.answer_extractor.extract_answer(answer, None)
+            dataframe.at[row_index, "answer_match_result"] = self.compare(
+                final_answer,
+                ground_truth,
+            )
+
+        storage.write(dataframe)
+        self.statistic(storage.file_name_prefix, dataframe, self.compare_method)
+        return required_columns + ["answer_match_result"]
+
+    def _build_semantic_inputs(self, valid_rows: pd.DataFrame) -> list[str]:
+        return [
+            self.prompt_template.build_prompt(
+                question=row[self.question_key],
+                answer=row[self.test_answer_key],
+                reference_answer=row[self.gt_answer_key],
+            )
+            for _, row in valid_rows.iterrows()
+        ]
+
+    def _handle_missing_reference_answers(
+        self,
+        storage: DataFlowStorage,
+        dataframe: pd.DataFrame,
+        required_columns: list[str],
+        skipped_count: int,
+    ) -> list[str]:
+        self.logger.warning(
+            "No valid samples with reference answers found. All samples skipped."
+        )
+        output_dataframe = (
+            dataframe if self.keep_all_samples else dataframe.iloc[0:0].copy()
+        )
+        output_file = storage.write(output_dataframe)
+        self.logger.info(
+            f"Dataframe saved to {output_file}. Skipped {skipped_count} samples "
+            "due to missing reference answers."
+        )
+        return required_columns + ["answer_match_result"]
+
+    def _apply_subquestion_results(
+        self,
+        dataframe: pd.DataFrame,
+        valid_rows: pd.DataFrame,
+        responses: list[str],
+        results: list[str],
+    ) -> None:
+        for row_index, response, result in zip(valid_rows.index, responses, results):
+            correct_answer_count, total_subquestions = map(int, result.split("/"))
+            dataframe.at[row_index, "correct_answer_num"] = correct_answer_count
+            dataframe.at[row_index, "total_subquestions"] = total_subquestions
+            dataframe.at[row_index, "answer_match_result"] = (
+                correct_answer_count == total_subquestions and total_subquestions > 0
+            )
+            dataframe.at[row_index, "response_evaluation"] = response
+
+    def _apply_semantic_results(
+        self,
+        dataframe: pd.DataFrame,
+        valid_rows: pd.DataFrame,
+        responses: list[str],
+        results: list,
+    ) -> None:
+        if self.support_subquestions:
+            self._apply_subquestion_results(dataframe, valid_rows, responses, results)
+            return
+
+        for row_index, result in zip(valid_rows.index, results):
+            dataframe.at[row_index, "answer_match_result"] = result
+
+    def _run_semantic_evaluation(
+        self,
+        storage: DataFlowStorage,
+        dataframe: pd.DataFrame,
+        required_columns: list[str],
+    ) -> list[str]:
+        empty_reference_mask = dataframe[self.gt_answer_key].isna() | (
+            dataframe[self.gt_answer_key] == ""
+        )
+        valid_rows = dataframe[~empty_reference_mask]
+        skipped_count = int(empty_reference_mask.sum())
+
+        if valid_rows.empty:
+            return self._handle_missing_reference_answers(
+                storage,
+                dataframe,
+                required_columns,
+                skipped_count,
+            )
+
+        inputs = self._build_semantic_inputs(valid_rows)
+        responses = self.llm_serving.generate_from_input(
+            user_inputs=inputs,
+            system_prompt=self.system_prompt,
+        )
+        response_count = 0 if responses is None else len(responses)
+        if response_count != len(inputs):
+            raise RuntimeError(
+                "LLM serving returned an unexpected number of responses: "
+                f"expected {len(inputs)}, got {response_count}."
+            )
+
+        results = [self.ResolveResponse(response) for response in responses]
+        self._apply_semantic_results(dataframe, valid_rows, responses, results)
+        storage.write(dataframe)
+        self.statistic(storage.file_name_prefix, dataframe, self.compare_method)
+        self.empty_responses_count = 0
+        return required_columns + ["answer_match_result"]
+
     def run(
             self,
             storage:DataFlowStorage,
@@ -217,93 +344,15 @@ def run(
         self.question_key = input_question_key
 
         dataframe = storage.read("dataframe")
-        dataframe['answer_match_result'] = False
-        answers = dataframe[self.test_answer_key]
-        ground_truths = dataframe[self.gt_answer_key]
-    
-        if self.compare_method == "match":
-            required_columns = [input_test_answer_key, input_gt_answer_key]
-            if self.check_column(
-                required_columns=required_columns,
-                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:
-            required_columns = [input_test_answer_key, input_gt_answer_key, input_question_key]
-            if self.check_column(
-                required_columns=required_columns,
-                dataframe=dataframe
-            ) is False:
-                return required_columns
-            
-            empty_reference_mask = dataframe[input_gt_answer_key].isna() | (dataframe[input_gt_answer_key] == '')
-            skipped_rows = dataframe[empty_reference_mask]
-            valid_rows = dataframe[~empty_reference_mask]
-            skipped_count = len(skipped_rows)
-            
-            if len(valid_rows) == 0:
-                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']
+        required_columns = self._get_required_columns(
+            input_test_answer_key,
+            input_gt_answer_key,
+            input_question_key,
+        )
+        if not self.check_column(required_columns, dataframe):
+            return required_columns
 
-        
+        dataframe["answer_match_result"] = False
+        if self.compare_method == "match":
+            return self._run_match_evaluation(storage, dataframe, required_columns)
+        return self._run_semantic_evaluation(storage, dataframe, required_columns)

test/cpu_only/test_undefined_name_crashes.py

@@ -156,6 +156,24 @@ def test_bench_evaluator_handles_missing_reference_answers(
     assert len(storage.dataframe) == expected_rows
 
 
+def test_bench_evaluator_validates_columns_before_reading_them():
+    """Missing input columns should be reported instead of raising KeyError."""
+    from dataflow.prompts.model_evaluation.general import AnswerJudgePromptQuestion
+    from dataflow.operators.core_text import BenchDatasetEvaluatorQuestion
+
+    evaluator = BenchDatasetEvaluatorQuestion(
+        compare_method="semantic",
+        llm_serving=object(),
+        prompt_template=AnswerJudgePromptQuestion(),
+    )
+    storage = InMemoryStorage(pd.DataFrame({"question": ["a"]}))
+
+    returned_keys = evaluator.run(storage=storage, input_question_key="question")
+
+    assert returned_keys == ["generated_cot", "golden_answer", "question"]
+    assert "answer_match_result" not in storage.dataframe.columns
+
+
 @pytest.mark.parametrize("support_subquestions", [False, True])
 def test_bench_match_mode_initializes_subquestion_setting(
     support_subquestions, tmp_path, monkeypatch
@@ -190,3 +208,73 @@ def test_bench_match_mode_initializes_subquestion_setting(
         "answer_match_result",
     ]
     assert storage.dataframe["answer_match_result"].tolist() == [True]
+
+
+@pytest.mark.parametrize("compare_method", ["match", "semantic"])
+def test_bench_evaluator_handles_non_default_index(compare_method, tmp_path, monkeypatch):
+    """Upstream filters hand downstream a sliced frame whose index is not 0..n-1.
+
+    Row lookups used to index the Series by position, which is a label lookup
+    on a pandas Series, so a gapped index raised KeyError.
+    """
+    from dataflow.prompts.model_evaluation.general import AnswerJudgePromptQuestion
+    from dataflow.operators.core_text import BenchDatasetEvaluatorQuestion
+
+    class StubLLMServing:
+        def generate_from_input(self, user_inputs, system_prompt=None):
+            return ['{"judgement_result": true}', '{"judgement_result": false}']
+
+    evaluator = BenchDatasetEvaluatorQuestion(
+        compare_method=compare_method,
+        eval_result_path=str(tmp_path / "statistics.json"),
+        llm_serving=StubLLMServing(),
+        prompt_template=AnswerJudgePromptQuestion(),
+    )
+    if compare_method == "match":
+        monkeypatch.setattr(
+            evaluator.answer_extractor, "extract_answer", lambda answer, _: answer
+        )
+        monkeypatch.setattr(
+            evaluator, "compare", lambda answer, expected: answer == expected
+        )
+    sliced_frame = pd.DataFrame(
+        {
+            "question": ["q1", "q2"],
+            "generated_cot": ["42", "7"],
+            "golden_answer": ["42", "9"],
+        },
+        index=[3, 7],
+    )
+    storage = InMemoryStorage(sliced_frame)
+
+    evaluator.run(storage=storage, input_question_key="question")
+
+    assert storage.dataframe["answer_match_result"].tolist() == [True, False]
+
+
+def test_bench_evaluator_rejects_misaligned_llm_responses():
+    """Every semantic-evaluation prompt must receive exactly one response."""
+    from dataflow.prompts.model_evaluation.general import AnswerJudgePromptQuestion
+    from dataflow.operators.core_text import BenchDatasetEvaluatorQuestion
+
+    class StubLLMServing:
+        def generate_from_input(self, user_inputs, system_prompt=None):
+            return ['{"judgement_result": true}']
+
+    evaluator = BenchDatasetEvaluatorQuestion(
+        compare_method="semantic",
+        llm_serving=StubLLMServing(),
+        prompt_template=AnswerJudgePromptQuestion(),
+    )
+    storage = InMemoryStorage(
+        pd.DataFrame(
+            {
+                "question": ["q1", "q2"],
+                "generated_cot": ["42", "7"],
+                "golden_answer": ["42", "9"],
+            }
+        )
+    )
+
+    with pytest.raises(RuntimeError, match="expected 2, got 1"):
+        evaluator.run(storage=storage, input_question_key="question")