utils_groupclassification/assistants_utils.py

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