Spaces:
Runtime error
Runtime error
File size: 4,134 Bytes
0e519fe 88c54be 0e519fe 88c54be 0e519fe 88c54be 0e519fe 88c54be 0e519fe |
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 |
import codecs
import json
import time
import openai
import os
from dotenv import load_dotenv
load_dotenv()
class Llm:
def __init__(self):
print("### Step 1: Get the Assistant's ID ###")
# OpenAI API Key settings
self.client = openai.OpenAI()
openai.api_key = os.getenv('OPENAI_API_KEY')
self.assistant_id = os.getenv('OPENAI_ASSISTANT_ID')
self.assistant = self.client.beta.assistants.retrieve(self.assistant_id)
print(self.assistant)
assistant_description = self.assistant.description
assistant_model = self.assistant.model
if assistant_description is None:
assistant_description = f"This assistant OpenAI API {assistant_model} was created using."
def setup(self):
load_dotenv()
self.api_key = os.getenv('OPENAI_API_KEY')
self.assistant_id = os.getenv('OPENAI_ASSISTANT_ID')
self.client = openai.OpenAI()
def retrieve_assistant(self):
self.assistant = self.client.beta.assistants.retrieve(self.assistant_id)
return self.assistant
# def openai_completion(self, prompt):
# import random
# import csv
# with open('flagged/log.csv', 'r') as file:
# reader = csv.DictReader(file)
# logs = [row for row in reader]
# random_log = random.choice(logs)
# answer = random_log['output']
# return answer
def chatcompletion(self, prompt):
try:
"### Step 2: Create a Thread ###"
empty_thread = self.client.beta.threads.create()
thread_id = empty_thread.id
print(empty_thread)
print("### Step 3: Add a Message to the Thread ###")
thread = self.client.beta.threads.retrieve(thread_id)
print(thread)
print("### Step 4: Add a Message to the Thread ###")
thread_message = self.client.beta.threads.messages.create(
thread_id,
role="user",
content=prompt,
)
message_id = thread_message.id
print(thread_message)
print("### Step 5: Retrieve the Message ###")
message = self.client.beta.threads.messages.retrieve(
message_id=message_id,
thread_id=thread_id,
)
print(message)
print("### Step 6: Run the Assistant ###")
run = self.client.beta.threads.runs.create(
thread_id=thread.id,
assistant_id=self.assistant.id,
)
print("### Step 7: Wait for the Assistant to Finish ###")
def wait_on_run(run, thread):
while run.status == "queued" or run.status == "in_progress":
run = self.client.beta.threads.runs.retrieve(
thread_id=thread.id,
run_id=run.id,
)
time.sleep(0.5)
return run
run = wait_on_run(run, thread)
print(run)
print("### Step 8: Retrieve the Messages ###")
messages = self.client.beta.threads.messages.list(
thread_id=thread.id
)
messages_str = json.dumps(messages.dict(), indent=2)
print(codecs.decode(messages_str, 'unicode-escape'))
print("### Step 9: Retrieve the Assistant's Response ###")
answers = []
for message in messages.data:
if message.role == "assistant":
if message.content[0].type == "text":
answers.append(message.content[0].text.value + "\n\n")
else:
answers.append("Content is not text.\n\n")
elif message.role == "user":
break
# answers reverse order
answers.reverse()
return "".join(answers)
except Exception as e:
print(f"An error occurred while creating thread:{e}")
raise |