|
import argparse
|
|
import json
|
|
import os
|
|
|
|
import openai
|
|
import tqdm
|
|
import ray
|
|
import time
|
|
|
|
NUM_SECONDS_TO_SLEEP = 3
|
|
|
|
@ray.remote(num_cpus=4)
|
|
def get_eval(content: str, max_tokens: int):
|
|
while True:
|
|
try:
|
|
response = openai.ChatCompletion.create(
|
|
model='gpt-4',
|
|
messages=[{
|
|
'role': 'system',
|
|
'content': 'You are a helpful and precise assistant for checking the quality of the answer.'
|
|
}, {
|
|
'role': 'user',
|
|
'content': content,
|
|
}],
|
|
temperature=0.2,
|
|
max_tokens=max_tokens,
|
|
)
|
|
break
|
|
except openai.error.RateLimitError:
|
|
pass
|
|
except Exception as e:
|
|
print(e)
|
|
time.sleep(NUM_SECONDS_TO_SLEEP)
|
|
|
|
print('success!')
|
|
return response['choices'][0]['message']['content']
|
|
|
|
|
|
def parse_score(review):
|
|
try:
|
|
score_pair = review.split('\n')[0]
|
|
score_pair = score_pair.replace(',', ' ')
|
|
sp = score_pair.split(' ')
|
|
if len(sp) == 2:
|
|
return [float(sp[0]), float(sp[1])]
|
|
else:
|
|
print('error', review)
|
|
return [-1, -1]
|
|
except Exception as e:
|
|
print(e)
|
|
print('error', review)
|
|
return [-1, -1]
|
|
|
|
|
|
if __name__ == '__main__':
|
|
parser = argparse.ArgumentParser(description='ChatGPT-based QA evaluation.')
|
|
parser.add_argument('-q', '--question')
|
|
|
|
parser.add_argument('-a', '--answer-list', nargs='+', default=[])
|
|
parser.add_argument('-r', '--rule')
|
|
parser.add_argument('-o', '--output')
|
|
parser.add_argument('--max-tokens', type=int, default=1024, help='maximum number of tokens produced in the output')
|
|
args = parser.parse_args()
|
|
|
|
ray.init()
|
|
|
|
f_q = open(os.path.expanduser(args.question))
|
|
f_ans1 = open(os.path.expanduser(args.answer_list[0]))
|
|
f_ans2 = open(os.path.expanduser(args.answer_list[1]))
|
|
rule_dict = json.load(open(os.path.expanduser(args.rule), 'r'))
|
|
|
|
review_file = open(f'{args.output}', 'w')
|
|
|
|
js_list = []
|
|
handles = []
|
|
idx = 0
|
|
for ques_js, ans1_js, ans2_js in zip(f_q, f_ans1, f_ans2):
|
|
|
|
|
|
|
|
ques = json.loads(ques_js)
|
|
ans1 = json.loads(ans1_js)
|
|
ans2 = json.loads(ans2_js)
|
|
|
|
category = json.loads(ques_js)['category']
|
|
if category in rule_dict:
|
|
rule = rule_dict[category]
|
|
else:
|
|
rule = rule_dict['default']
|
|
prompt = rule['prompt']
|
|
role = rule['role']
|
|
content = (f'[Question]\n{ques["text"]}\n\n'
|
|
f'[{role} 1]\n{ans1["text"]}\n\n[End of {role} 1]\n\n'
|
|
f'[{role} 2]\n{ans2["text"]}\n\n[End of {role} 2]\n\n'
|
|
f'[System]\n{prompt}\n\n')
|
|
js_list.append({
|
|
'id': idx+1,
|
|
'question_id': ques['question_id'],
|
|
'answer1_id': ans1['answer_id'],
|
|
'answer2_id': ans2['answer_id'],
|
|
'category': category})
|
|
idx += 1
|
|
handles.append(get_eval.remote(content, args.max_tokens))
|
|
|
|
time.sleep(NUM_SECONDS_TO_SLEEP)
|
|
|
|
reviews = ray.get(handles)
|
|
for idx, review in enumerate(reviews):
|
|
scores = parse_score(review)
|
|
js_list[idx]['content'] = review
|
|
js_list[idx]['tuple'] = scores
|
|
review_file.write(json.dumps(js_list[idx]) + '\n')
|
|
review_file.close()
|
|
|
