bench-dataset-evaluator

BenchDatasetEvaluator Operator Reference

BenchDatasetEvaluator compares predicted answers against ground truth using two modes: match (math verification) or semantic (LLM-based).

1. Import

from dataflow.operators.core_text import BenchDatasetEvaluator

2. Match Mode

Constructor

BenchDatasetEvaluator(
    eval_result_path=None,
    compare_method="match",
)

Parameter	Required	Default	Description
eval_result_path	No	Auto-generated	Path to save evaluation statistics JSON file
compare_method	No	"match"	Must be "match"
system_prompt	No	"You are a helpful assistant..."	Not used in match mode
llm_serving	No	None	Not used in match mode
prompt_template	No	AnswerJudgePrompt	Not used in match mode

run() Signature

op.run(
    storage=self.storage.step(),
    input_test_answer_key="generated_cot",
    input_gt_answer_key="golden_answer",
)
# returns: [input_test_answer_key, input_gt_answer_key, 'answer_match_result']

Parameter	Required	Default	Description
storage	Yes	None	DataFlowStorage step object
input_test_answer_key	No	"generated_cot"	Column containing predicted answers
input_gt_answer_key	No	"golden_answer"	Column containing ground truth answers

Runtime Logic

1. Read DataFrame from storage.
2. Create answer_match_result column initialized to False.
3. For each row, extract answer using AnswerExtractor and compare with ground truth using math_verify_compare().
4. Write results to answer_match_result column.
5. Save statistics to eval_result_path.
6. Return column list.

Usage Example

from dataflow.operators.core_text import BenchDatasetEvaluator
from dataflow.utils.storage import FileStorage

class MyPipeline:
    def __init__(self):
        self.storage = FileStorage(
            first_entry_file_name="./data/bench.jsonl",
            cache_path="./cache",
            file_name_prefix="step",
            cache_type="jsonl"
        )

        self.evaluator = BenchDatasetEvaluator(
            compare_method="match",
            eval_result_path="./results/match_eval.json"
        )

    def forward(self):
        self.evaluator.run(
            storage=self.storage.step(),
            input_test_answer_key="predicted_answer",
            input_gt_answer_key="ground_truth"
        )

if __name__ == "__main__":
    pipeline = MyPipeline()
    pipeline.forward()