DEMO/step2_eval.py at main · MozerWang/DEMO · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
import os
from src.utils.functions import load, read_jsonl, process_eval_json
from src.utils.generate import api_generate, LLMPredictor
from src.utils.eval_prompt import *
import json
import argparse
import uuid
import random

def build_da_prompt(item, num_dict):
    prompts = []

    def create_prompt(key, instruction, prediction, answer):
        prompt_template = {
            'persona': eval_persona,
            'goal': eval_goal,
            'scene': eval_scene,
            'utterance': eval_utterance
        }.get(key)
        if key == 'persona':
            instruction = instruction.split("\n```json")[0]
            input_prompt = prompt_template.format(instruction=instruction, prediction=prediction, answer=answer)
        else:
            input_prompt = prompt_template.format(prediction=prediction, answer=answer)

        return {
            'id': item['id'],
            'uuid': item['uuid'],
            'prompt': {'system': "", 'user': input_prompt},
            'answer': answer,
            'prediction': prediction,
            'key': key
        }

    prompts.append(create_prompt(item['type'], item['prompt']['user'], item['INFER'], item['answer']))
    num_dict[item['type']] += 1
    return prompts, num_dict


def fail_discover(prompts, output_path, tag, is_dg):
    existing_datas = read_jsonl(output_path)
    finish_ls = []
    finish_datas = []
    random.shuffle(existing_datas)
    for existing_data in existing_datas:
        unique_index = str(existing_data.get('uuid')) + str(existing_data.get('id'))

        if existing_data[tag] and process_eval_json(existing_data[tag]) and (unique_index not in finish_ls):
            finish_datas.append(existing_data)
            finish_ls.append(unique_index)

    with open(output_path, 'w', encoding='utf-8') as fw:
        for item in finish_datas:
            fw.write(json.dumps(item, ensure_ascii=False) + '\n')

    infer_prompts = []
    for item in prompts:
        unique_index = str(item.get('uuid')) + str(item.get('id'))
        if unique_index not in finish_ls:
            infer_prompts.append(item)
        else:
            continue
    return infer_prompts

def data2daeval_prompt(args):
    datas = read_jsonl(args.result_path)
    prompts = []
    num_dict = {"persona":0,"scene":0,"goal":0,"utterance":0}
    for item in datas:
        temp_prompts, num_dict= build_da_prompt(item, num_dict)
        prompts.extend(temp_prompts)
    print(json.dumps(num_dict, indent=4))
    return prompts

def data2dgeval_prompt(args):
    datas = read_jsonl(args.result_path)
    prompts = []
    for item in datas:
        prompt = eval_dg
        prompt_js = item.copy()
        try:
            item['familiarity'] = Familiar_map[str(item['familiarity'])][1] if args.en else Familiar_map[str(item['familiarity'])][0]
        except:
            item['familiarity'] = item['familiarity']
        prompt_js['prompt'] = {'system': "", 'user': eval_dg.format(result=item)}
        prompts.append(prompt_js)
    return prompts

def main(args):
    output_dir = os.path.join(args.output_path, args.data)
    output_path = os.path.join(output_dir, args.output_file)
    os.makedirs(output_dir, exist_ok=True)
    config = load(open(args.config_path, 'r'))
    print(json.dumps(config, indent=4))

    prompts = data2dgeval_prompt(args) if args.dg else data2daeval_prompt(args)
    prompts = fail_discover(prompts, output_path, args.tag, args.dg) if os.path.exists(output_path) else prompts
    if prompts:
        print(prompts[-1]['prompt'])
        api_generate(prompts, config, output_path, args.process_num, args.port, args.tag)
    print('Finish!!')

def parse_arguments():
    parser = argparse.ArgumentParser(description='args for step2_eval.py')
    parser.add_argument("--result_path", type=str, default="final_result/output/1104qwen2_7b_self/ca_en_infer_extract.jsonl")
    parser.add_argument("--config_path", type=str, default="config/gpt_4o.yaml")
    parser.add_argument("--data", type=str, default="eval/1104qwen2_7b_self")
    parser.add_argument("--output_path", type=str, default="result")
    parser.add_argument("--output_file", type=str, default="ca_en_eval.jsonl")
    parser.add_argument("--process_num", type=int, default=1)
    parser.add_argument("--tag", type=str, default="eval")
    parser.add_argument("--port", type=str, default="8000")
    parser.add_argument("--batch_size", type=int, default=2048)
    parser.add_argument("--dg", action="store_true", help="Dialogue generation")
    parser.add_argument("--en", action="store_true", help="Use English")
    args = parser.parse_args()
    print(json.dumps(vars(args), indent=4))
    return args

if __name__ == "__main__":
    args = parse_arguments()
    args.en = True
    main(args)