-
Notifications
You must be signed in to change notification settings - Fork 1
/
utils.py
145 lines (127 loc) · 4.41 KB
/
utils.py
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
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
import asyncio
import re
import string
import time
import dashscope
import openai
openai.api_key = "YOUR OPENAI KEY" # put your openai api key here
dashscope.api_key = "YOUR DASHSCOPE KEY"
def remove_prefix(text: str) -> str:
result = re.sub(r"^\d+\.\s", "", text)
return result
def single_run(messages, retry=3, model="gpt-3.5-turbo-0613", n=1, temperature=0.0):
for _ in range(retry):
try:
output = openai.ChatCompletion.create(
model=model,
messages=messages,
n=n,
temperature=temperature,
)
if n == 1:
return output.choices[0].message.content.strip()
else:
return [choice.message.content for choice in output.choices]
except:
time.sleep(20)
return None
def single_qwen_run(messages, model_name):
for _ in range(3):
try:
response = dashscope.Generation.call(
model=model_name,
messages=messages,
temperature=0.0,
result_format="message", # set the result to be "message" format.
)
return response["output"]["choices"][0]["message"]["content"]
except:
time.sleep(10)
return None
async def run_completion_api(prompts):
# Make api calls asynchronously
async def single_run(prompt, retry=3):
for _ in range(retry):
try:
output = openai.Completion.create(
model="gpt-judge/model/name",
prompt=prompt,
temperature=0,
max_tokens=1,
stop=None,
echo=False,
logprobs=2,
)
return output
except:
await asyncio.sleep(20)
return None
responses = [single_run(prompt) for prompt in prompts]
return await asyncio.gather(*responses)
async def run_api(messages, model="gpt-3.5-turbo-0613", retry=3, temperature=0.0):
# Make api calls asynchronously
async def single_run(message, model, retry=3, temperature=0.0):
for _ in range(retry):
try:
output = openai.ChatCompletion.create(
model=model,
messages=message,
temperature=temperature,
)
return output.choices[0].message.content.strip()
except:
await asyncio.sleep(20)
return None
responses = [single_run(message, model, retry, temperature) for message in messages]
return await asyncio.gather(*responses)
def is_supported(generated_answer):
generated_answer = generated_answer.lower()
if "true" in generated_answer or "false" in generated_answer:
if "true" in generated_answer and "false" not in generated_answer:
is_supported = True
elif "false" in generated_answer and "true" not in generated_answer:
is_supported = False
else:
is_supported = generated_answer.index("true") > generated_answer.index(
"false"
)
else:
is_supported = all(
[
keyword
not in generated_answer.lower()
.translate(str.maketrans("", "", string.punctuation))
.split()
for keyword in [
"not",
"cannot",
"unknown",
"information",
]
]
)
return is_supported
def is_supported_zh(generated_answer):
if "是" in generated_answer or "否" in generated_answer:
if "是" in generated_answer and "否" not in generated_answer:
is_supported = True
elif "否" in generated_answer and "是" not in generated_answer:
is_supported = False
else:
is_supported = generated_answer.index("是") > generated_answer.index("否")
else:
is_supported = all(
[
keyword
not in generated_answer.lower()
.translate(str.maketrans("", "", string.punctuation))
.split()
for keyword in [
"不",
"不能",
"未知",
"信息",
]
]
)
return is_supported