Spaces:
Sleeping
Sleeping
File size: 2,412 Bytes
cd63cf6 103de27 cd63cf6 103de27 cd63cf6 |
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 |
import time
from logging import getLogger
from dotenv import load_dotenv
from utils_groupclassification.openai_utils import init_openai
import logging
load_dotenv(".env.dev")
client = init_openai(api_type="open_ai")
logger = getLogger(__name__)
# Function to print messages from a thread
def print_messages_from_thread(thread_id):
messages = client.beta.threads.messages.list(thread_id=thread_id)
for msg in messages:
# logger.info(
# f"PRINT MESSAGE FROM THREAD || {msg.role}: {msg.content[0].text.value}"
# )
print("PRINT MESSAGE FROM THREAD ||", msg.role, msg.content[0].text.value)
# Function to log messages from a thread
def log_messages_from_thread(thread_id):
messages = client.beta.threads.messages.list(thread_id=thread_id)
for msg in messages:
# logger.info(
# f"PRINT MESSAGE FROM THREAD || {msg.role}: {msg.content[0].text.value}"
# )
print("PRINT MESSAGE FROM THREAD ||", msg.role, msg.content[0].text.value)
# Function to wait for a run to complete
def wait_for_run_completion(thread_id, run_id):
while True:
time.sleep(1)
run = client.beta.threads.runs.retrieve(thread_id=thread_id, run_id=run_id)
print(f"WAIT FOR RUN COMPLETION || Current run status: {run.status}")
if run.status in ["completed", "failed", "requires_action"]:
return run
# Function to create an assistant
def create_assistant(instructions, model, tools):
assistant = client.beta.assistants.create(
instructions=instructions,
model=model,
tools=tools,
)
return assistant
# Function to create a thread
def create_thread():
thread = client.beta.threads.create()
return thread
# Function to create a message
def create_message(thread_id, role, content):
message = client.beta.threads.messages.create(
thread_id=thread_id,
role=role,
content=content,
)
return message
# Function to create a run
def create_run(thread_id, assistant_id):
run = client.beta.threads.runs.create(
thread_id=thread_id,
assistant_id=assistant_id,
)
return run
def create_run_with_instruction(thread_id, assistant_id, instruction):
run = client.beta.threads.runs.create(
thread_id=thread_id,
assistant_id=assistant_id,
instructions=instruction,
)
return run
|