На основе имеющейся информации (к сожалению, ChatGPT оказался не слишком полезен) я создал следующий код, позволяющий взаимодействовать с API помощника. Однако мне все еще не нравится метод _wait_for_run_completion и цикл while. Есть ли лучший способ справиться с этим?
import os
import openai
from dotenv import load_dotenv
import time
class OpenAIChatAssistant:
def __init__(self, assistant_id, model="gpt-4o"):
self.assistant_id = assistant_id
self.model = model
if self.model != "just_copy":
load_dotenv()
openai.api_key = os.environ.get("OPENAI_API_KEY")
self.client = openai.OpenAI()
self._create_new_thread()
print('new instance started')
def _create_new_thread(self):
self.thread = self.client.beta.threads.create()
self.thread_id = self.thread.id
print(self.thread_id)
def reset_thread(self):
if self.model != "just_copy":
self._create_new_thread()
def set_model(self, model_name):
self.model = model_name
if self.model != "just_copy" and not hasattr(self, 'client'):
load_dotenv()
openai.api_key = os.environ.get("OPENAI_API_KEY")
self.client = openai.OpenAI()
self._create_new_thread()
def send_message(self, message):
if self.model == "just_copy":
return message
self.client.beta.threads.messages.create(
thread_id=self.thread_id, role="user", content=message
)
run = self.client.beta.threads.runs.create(
thread_id=self.thread_id,
assistant_id=self.assistant_id,
model=self.model
)
return self._wait_for_run_completion(run.id)
def _wait_for_run_completion(self, run_id, sleep_interval=1):
counter = 1
while True:
try:
run = self.client.beta.threads.runs.retrieve(thread_id=self.thread_id, run_id=run_id)
if run.completed_at:
messages = self.client.beta.threads.messages.list(thread_id=self.thread_id)
last_message = messages.data[0]
response = last_message.content[0].text.value
print(f'hello {counter}')
return response
except Exception as e:
raise RuntimeError(f"An error occurred while retrieving answer: {e}")
counter += 1
time.sleep(sleep_interval)
Подробнее здесь: https://stackoverflow.com/questions/785 ... erformance
