-
Notifications
You must be signed in to change notification settings - Fork 0
/
alpha.py
223 lines (159 loc) · 7.34 KB
/
alpha.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
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
import asyncio
import json
import os
from dataclasses import dataclass
import weave
import openai
import instructor
from pydantic import BaseModel, Field
import simple_parsing
# patch instructor retry with weave
from instructor import retry
retry.process_response = weave.op()(retry.process_response)
client = instructor.from_openai(openai.AsyncClient())
SLEEP_TIME = 0.2
@dataclass
class ScriptArgs:
model: str = "gpt-4o"
weave_project: str = "prompt-eng/connections"
file_path: str = "connections_prompts.jsonl"
max_tokens: int = 128
temperature: float = 0.7
num_samples: int = 5
N: int = 3
args = simple_parsing.parse(ScriptArgs)
@weave.op()
async def call_openai(messages, response_model=None, model=args.model):
response = await client.chat.completions.create(
model=model,
messages=messages,
response_model=response_model,
)
if response_model is None:
return response.choices[0].message.content
else:
return response
def load_jsonl(file_path):
data = []
with open(file_path, 'r') as file:
for line in file:
data.append(json.loads(line))
return data
class Reflection(BaseModel):
analysis: str
def __str__(self):
return self.analysis
class Group(BaseModel):
reason: str
words: list[str] = Field(max_length=4, description="A list of 4 words")
def __str__(self):
return "{" + f"{self.reason}: {list(self.words)}" + "}"
class Solution(BaseModel):
opinion: str = "A solution"
score: int = 7
groups: list[Group] = Field(max_length=4, description="A list of 4 groups of 4 words with their reasons")
def __str__(self):
return ("- [" + ",".join(str(group) for group in self.groups) + "]\n"
+ f" Opinion: {self.opinion}\n"
+ f" Score: {self.score}\n")
def validate(self, puzzle: list[str]):
total_words = []
for g in self.groups:
if not set(g.words).issubset(set(puzzle)):
return False
for w in g.words:
if w not in total_words:
total_words.append(w)
return len(total_words) == 16
def __eq__(self, other: 'Solution'):
solution_set = {frozenset(group.words) for group in self.groups}
other_set = {frozenset(group.words) for group in other.groups}
accuracy = len(solution_set.intersection(other_set))
return {"match": accuracy == 4, "accuracy": accuracy}
class PossibleSolutions(BaseModel):
solutions: list[Solution] = Field(max_length=100, description="A list of possible solutions to the puzzle")
def __str__(self):
return "\n".join(str(solution) for solution in self.solutions)
class AlphaModel(weave.Model):
system_prompt: str = (
"You are an expert puzzle solver. You understand literature and you are well versed on word play. "
"I want you to solve a daily word puzzle that finds commonalities between words.\n"
)
@weave.op()
async def analysis(self, messages, words):
user_prompt = (
"Here it's the puzzle:\n"
"- There are 16 words, which form 4 groups of 4 words. Each group has some common theme that links the words.\n"
"- You must use each of the 16 words, and use each word only once.\n"
"- Each group of 4 words are linked together in some way. \n"
"The connection between words can be simple.\n"
"""- An example of a simple connection would be {"reason":'types of fish', "words":["Bass", "Flounder", "Salmon", "Trout"]}. \n"""
"""- Categories can also be more complex, and require abstract or lateral thinking. An example of this type of connection would be {"reason": 'things that start with FIRE', "words": ['Ant', 'Drill', 'Island', 'Opal']}\n"""
f"Here are the starting 16 words:\n{words}\n"
"Give me an analysis of the puzzle and explore word relations. Does it lok hard? Are there multiple word connections?"
)
messages += [
{"role": "system", "content": self.system_prompt},
{"role": "user", "content": user_prompt}
]
reflections = await call_openai(messages, response_model=None)
messages += [{"role": "assistant", "content": str(reflections)},]
return messages
@weave.op()
async def generate_solutions(self, messages, words):
user_prompt = (
f"Ok, taking the previous analysis, let's try generating {args.N} different solutions to this problem. This means {args.N} times 4 groups of 4 words each. Remember to use all words and to not repeat words across groups.\n"
f"Score each solution from 1 to 7 and give me an opinion on why you picked that as a valid solution. If it's not a good solution, give me a reason why it's not a good solution."
)
messages += [
{"role": "user", "content": user_prompt}
]
possible_solutions = await call_openai(messages, response_model=PossibleSolutions)
# let's check they are valid and filter out the ones that are not
validated_possible_solutions = PossibleSolutions(
solutions=[s for s in possible_solutions.solutions if s.validate(words)]
)
messages += [
{"role": "assistant", "content": "Here are the possible solutions: \n\n" + str(validated_possible_solutions)},
]
return messages
@weave.op()
async def solutions_analysis(self, messages):
user_prompt2 = f"""Great, now we have to check if the solutions are correct.
- Inspect the solution so the relation between words makes sense.
- Verify that the words are related to each other by the reason you provided.
- Do the given scores match the quality of the solution?
- Reflect on each solution and give me an honest opinion on how plausible each solution is to be the real solution.
"""
messages += [
{"role": "user", "content": user_prompt2}
]
analysis_of_solutions = await call_openai(messages, response_model=None) #you the raw output from openai
messages += [
{"role": "assistant", "content": analysis_of_solutions},
]
return messages
@weave.op()
async def final_solution(self, messages):
user_prompt3 = f"""Now that you have analized the solutions, I want you to give me the best solution. Argument why you picked this one.
"""
messages += [
{"role": "user", "content": user_prompt3}
]
return await call_openai(messages, response_model=Solution)
@weave.op()
async def predict(self, words):
messages = []
messages = await self.analysis(messages, words)
messages = await self.generate_solutions(messages, words)
messages = await self.solutions_analysis(messages)
output = await self.final_solution(messages)
return output.dict()
@weave.op()
def check_final_solution(solution: dict, model_output: dict):
return Solution.model_validate(solution) == Solution.model_validate(model_output)
weave.init(args.weave_project)
ds = load_jsonl(args.file_path)
model = AlphaModel()
weave_eval = weave.Evaluation(dataset=ds[-args.num_samples:], scorers=[check_final_solution])
print(asyncio.run(weave_eval.evaluate(model)))