File size: 2,609 Bytes
084fe8e acb3380 084fe8e acb3380 084fe8e acb3380 084fe8e acb3380 084fe8e acb3380 084fe8e acb3380 084fe8e acb3380 084fe8e acb3380 084fe8e acb3380 084fe8e |
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 |
from typing import Any, Optional
from openai import OpenAI
from ..utils import info_exponential_backoff, score_exponential_backoff
from .supervisor_base import BaseSupervisor
@BaseSupervisor.register_supervisor("gpt4_supervisor")
class GPT4Supervisor(BaseSupervisor):
def __init__(self, *args: Any, **kwargs: Any) -> None:
self.init_supervisor()
def init_supervisor(self) -> None:
self.model = OpenAI()
@info_exponential_backoff(retries=5, base_wait_time=1)
def ask_info(self, query: str, context: Optional[str] = None) -> str | Any:
responses = self.model.chat.completions.create(
model="gpt-4-turbo-preview",
messages=[
{
"role": "user",
"content": f"The following is detailed information on the topic: {context}. Based on this information, answer the question: {query}. Answer with a few words:",
}
],
max_tokens=300,
n=1,
)
answer = (
responses.choices[0].message.content
if responses.choices[0].message.content
else "FAILED"
)
return answer
@score_exponential_backoff(retries=5, base_wait_time=1)
def ask_score(
self,
query: str,
gist: str,
verbose: bool = False,
*args: Any,
**kwargs: Any,
) -> float:
max_attempts = 5
for attempt in range(max_attempts):
try:
response = self.model.chat.completions.create(
model="gpt-4-0125-preview",
messages=[
{
"role": "user",
"content": f"How related is the information ({gist}) with the query ({query})? We want to make sure that the information includes a person's name as the answer. Answer with a number from 0 to 5 and do not add any other thing.",
},
],
max_tokens=50,
)
score = (
float(response.choices[0].message.content.strip()) / 5
if response.choices[0].message.content
else 0.0
)
return score
except Exception as e:
print(f"Attempt {attempt + 1} failed: {e}")
if attempt < max_attempts - 1:
print("Retrying...")
else:
print("Max attempts reached. Returning default score.")
return 0.0
|