distributed_eval.py

import json
import argparse
import os
from fusion_bench.tasks.flan_t5_text_generation.glue_prompt_templates import glue_prompt_templates
import numpy as np
import tqdm
import re
from fractions import Fraction
from scipy.stats import spearmanr
import subprocess
import shutil
from eval_mbpp import get_mbpp_score
from human_eval.evaluation import evaluate_functional_correctness
from fusion_bench.method.pso_merging.path_utils import get_xfinder_path, get_alpaca_eval_path, get_human_eval_path, get_infer_res_path

parser = argparse.ArgumentParser()
parser.add_argument("--model_name", type=str, default='none')
parser.add_argument("--task_name", type=str, default='mbpp')
args = parser.parse_args()

def is_number(s):
    try:
        float(s)
        return True
    except ValueError:
        pass
    try:
        import unicodedata
        unicodedata.numeric(s)
        return True
    except (TypeError, ValueError):
        pass
    return False


def extract_answer_number(completion):
    text = completion.split('The answer is: ')
    if len(text) > 1:
        extract_ans = text[-1].strip()
        match = re.search(r'[\-+]?\d*[\.,/]?\d+', extract_ans)
        if match:
            if '/' in match.group():
                denominator = match.group().split('/')[1]
                numerator = match.group().split('/')[0]
                if is_number(denominator) == True and is_number(numerator) == True:
                    if denominator == '0':
                        return round(float(numerator.replace(',', '')))
                    else:
                        frac = Fraction(match.group().replace(',', ''))
                        num_numerator = frac.numerator
                        num_denominator = frac.denominator
                        return round(float(num_numerator / num_denominator))
                else:
                    return None
            else:
                if float(match.group().replace(',', '')) == float('inf'):
                    return None
                return round(float(match.group().replace(',', '')))
        else:
            return None
    else:
        return None

def test_gsm8k_org(all_data):
    acc = 0
    tot = 0
    invalid_outputs = []
    for data in all_data:
        y_pred = extract_answer_number(data['output'])
        if y_pred != None:
            acc += (float(y_pred) == float(data['label']))
        else:
            invalid_outputs.append(data)
        tot += 1
    with open("invalid_outputs.json", "w") as f:
        f.write(json.dumps(invalid_outputs) + '\n')
    return acc / tot

def test_gsm8k(all_data):
    to_write = []
    for data in all_data:
        to_append = {
            "key_answer_type": "math",
            "question": data['input'],
            "llm_output": data['output'],
            "correct_answer": data['label'],
            "standard_answer_range": "a(n) number / set / vector / matrix / interval / expression / function / equation / inequality",
        }
        to_write.append(to_append)
    data_path = "gsm8k_to_eval.json"
    # with open(data_path, "w") as f:
    #     f.write(json.dumps(to_write))
    from xfinder.eval import Evaluator
    xfinder_paths = get_xfinder_path()
    evaluator = Evaluator(
        model_name="xFinder-qwen1505",   # Model name
        inference_mode="local",            # Inference mode, 'local' or 'api'
        model_path_or_url=xfinder_paths["model"],
    )
    # accuracy = evaluator.evaluate(data_path)
    # os.remove(data_path)
    acc = 0
    tot = 0
    for data in tqdm.tqdm(to_write):
        question = data['question']
        llm_output = data['llm_output']
        standard_answer_range = data['standard_answer_range']
        key_answer_type = data['key_answer_type']
        correct_answer = data['correct_answer']
        result = evaluator.evaluate_single_example(
            question,
            llm_output,
            standard_answer_range,
            key_answer_type,
            correct_answer
        )
        if result == 'Correct':
            acc += 1
        tot += 1
        data['result'] = result
    accuracy = acc / tot
    # with open(data_path, "w") as f:
    #     f.write(json.dumps(to_write))
    return accuracy

def test_sciq(all_data):
    to_write = []
    for data in all_data:
        to_append = {
            "key_answer_type": "short_text",
            "question": data['input'],
            "llm_output": data['output'],
            "correct_answer": data['label'],
            "standard_answer_range": ' / '.join(data['choices']),
        }
        to_write.append(to_append)
    from xfinder.eval import Evaluator
    evaluator = Evaluator(
        model_name="xFinder-qwen1505",   # Model name
        inference_mode="local",            # Inference mode, 'local' or 'api'
        model_path_or_url="/data/zhangkehao/xFinder-qwen1505",
    )
    acc = 0
    tot = 0
    for data in tqdm.tqdm(to_write):
        question = data['question']
        llm_output = data['llm_output']
        standard_answer_range = data['standard_answer_range']
        key_answer_type = data['key_answer_type']
        correct_answer = data['correct_answer']
        result = evaluator.evaluate_single_example(
            question,
            llm_output,
            standard_answer_range,
            key_answer_type,
            correct_answer
        )
        if result == 'Correct':
            acc += 1
        tot += 1
        data['result'] = result
    accuracy = acc / tot
    return accuracy

def test_mbpp(all_data):
    outputs = [[] for _ in range(500)]
    for data in all_data:
        completion = data['output'].split("### Response:")[-1]
        completion = completion.replace('\t', '    ')
        completion = completion.strip()
        if '```python' in completion:
            print("completion matches ```python")
            def_line = completion.index('```python')
            completion = completion[def_line:].strip()
            completion = completion.replace('```python', '')
            try:
                next_line = completion.index('\n```')
                completion = completion[:next_line].strip()
            except:
                print("wrong completion")
        if "__name__ == \"__main__\"" in completion:
            print("completion matches __name__ == \"__main__\"")
            try:
                next_line = completion.index('if __name__ == "__main__":')
                completion = completion[:next_line].strip()
            except:
                print("wrong completion")
        if "# Example usage" in completion:
            print("completion matches # Example usage")
            next_line = completion.index('# Example usage')
            completion = completion[:next_line].strip()
        if "# Test examples" in completion:
            print("completion matches # Test examples")
            next_line = completion.index('# Test examples')
            completion = completion[:next_line].strip()
        # the following codes are used to deal with the outputs of code-alpaca
        if "The solution is:" in completion:
            print("completion matches The solution is:")
            def_line = completion.index("The solution is:")
            completion = completion[def_line:].strip()
            completion = completion.replace('The solution is:', '')
            try:
                next_line = completion.index('\n\nThe answer is:')
                completion = completion[:next_line].strip()
            except:
                completion = completion.strip()
                print("maybe wrong completion")
        if "The answer is:" in completion:
            print("completion matches The answer is:")
            def_line = completion.index("The answer is:")
            completion = completion[def_line:].strip()
            completion = completion.replace('The answer is:', '')
            try:
                next_line = completion.index('\n\nThe answer is:')
                completion = completion[:next_line].strip()
            except:
                completion = completion.strip()
                print("maybe wrong completion")
        task_id = data['idx']
        outputs[task_id - 11].append(completion)
        # data['output'] = completion
    json.dump(outputs, open("mbpp_processed.json", 'w'))
    return get_mbpp_score()

def test_other_glue(task_name, all_data):
    labels = list(glue_prompt_templates[task_name]['target_text'].values())
    acc = 0
    tot = 0
    for data in all_data:
        split_output = [x.strip().replace('.','').lower() for x in data["output"].split()]
        if labels[data['label']].lower() in split_output and not any([x.lower() in split_output for x in labels if x != labels[data['label']]]):
            acc += 1
        tot += 1
    return acc / tot

def test_stsb(all_data):
    all_preds = []
    all_labels = []
    for data in all_data:
        split_output = [x.strip().replace('.','').lower() for x in data["output"].split()]
        first_num = 0.0
        for x in split_output:
            try:
                first_num = float(x)
                break
            except:
                continue
        all_preds.append(first_num)
        all_labels.append(float(data['label']))
    all_preds = np.array(all_preds)
    all_labels = np.array(all_labels)
    rho = spearmanr(all_preds, all_labels)[0]
    if np.isnan(rho):
        rho = 0
    return rho

def test_alpaca_eval(all_data):
    alpaca_paths = get_alpaca_eval_path()
    infer_res_path = get_infer_res_path("alpaca_eval.json")
    json.dump(all_data, open(infer_res_path, 'w', encoding='utf-8'), ensure_ascii=False)
    success = False
    while not success:
        print(f'Running alpaca_eval')
        env = os.environ.copy()
        env['IS_ALPACA_EVAL_2'] = 'False'
        env['HF_DATASETS_OFFLINE'] = '1'
        process = subprocess.Popen(['alpaca_eval', '--model_outputs', infer_res_path, '--annotators_config', 'alpaca_eval_ollama_llama3_70b_fn'], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, bufsize=1, env=env)
        stdout, stderr = process.communicate()
        return_code = process.returncode
        if return_code == 0:
            print('脚本执行成功')
            success = True
        else:
            print('脚本执行失败，返回码:', return_code, 'stdout:', stdout, 'stderr:', stderr)
    with open(alpaca_paths["leaderboard"], 'r') as file:
        win_rate = float(file.readlines()[1].strip().split(',')[1])
    # 删除文件
    file_path = alpaca_paths["annotations"]
    if os.path.exists(file_path):
        os.remove(file_path)
        print(f"Deleted file: {file_path}")
    else:
        print(f"File not found: {file_path}")

    # 删除文件夹及其内容
    folder_path = alpaca_paths["weights"]
    if os.path.exists(folder_path):
        shutil.rmtree(folder_path)
        print(f"Deleted folder and its contents: {folder_path}")
    else:
        print(f"Folder not found: {folder_path}")
    return win_rate / 100

def test_human(all_data):
    human_eval_path = get_human_eval_path()
    infer_res_path = get_infer_res_path("human_eval.jsonl")
    # json.dump(all_data, open(infer_res_path, 'w', encoding='utf-8'), ensure_ascii=False)
    with open(infer_res_path, 'w', encoding='utf-8') as f:
        for data in all_data:
            f.write(json.dumps(data, ensure_ascii=False) + '\n')
    sample_file: str = infer_res_path
    k = list(map(int, "1,10,100".split(",")))
    n_workers: int = 4
    timeout: float = 3.0
    problem_file: str = human_eval_path
    result = evaluate_functional_correctness(sample_file, k, n_workers, timeout, problem_file)
    return result['pass@1']

def get_score(task_name, all_data):
    if task_name == "gsm8k":
        return test_gsm8k(all_data)
    elif task_name == "mbpp":
        return test_mbpp(all_data)
    elif task_name == "human_eval":
        return test_human(all_data)
    elif task_name == "alpaca_eval":
        return test_alpaca_eval(all_data)
    elif task_name == "stsb":
        return test_stsb(all_data)
    elif task_name == "sciq":
        return test_sciq(all_data)
    elif task_name in glue_prompt_templates:
        return test_other_glue(task_name, all_data)
    else:
        raise NotImplementedError(f"Task {task_name} not implemented")

if __name__ == '__main__':
    # outfile = f'./{args.model_name}_{args.task_name}.jsonl'
    all_data = [json.loads(line) for line in open(f"infer_res_{args.task_name}.jsonl", 'r')]
    # json.dump(all_data, open(outfile, 'w'))
    score = get_score(args.task_name, all_data)
    # with open("outputs/t5_base_mtl_res/final_scores.json", "a") as f:
    #     f.write(json.dumps({args.task_name: score}) + '\n')
    print(score)