services/utils.py

import json
import re

# Suppress all UserWarnings
from openai import OpenAI

from config import get_settings

LLM_CLIENT = OpenAI(
    api_key=get_settings().openai_api_key,
    timeout=20,
)


def parse_response(response: str) -> tuple[str, list[dict] | None]:
    answer_match = re.search(
        r"\[ANSWER\]\n(.+?)(?=\n\[(.*?)\])", response, re.DOTALL
    )
    related_docs_match = re.search(r"\[REFERENCE\]\n(.+)", response, re.DOTALL)

    if answer_match:
        answer = answer_match.group(1).strip()
        related_docs = None

        if related_docs_match:
            related_docs_str = (
                related_docs_match.group(1).strip().replace("```", "")
            )
            try:
                related_docs = json.loads(related_docs_str)
            except Exception:
                print(f"failed to parse related docs: {related_docs_str}")

        return answer, related_docs

    raise ValueError(f"failed to parse response: {response}")


def parse_histories(histories: list[str, str]):
    conversational_histories = []
    for history in histories:
        user_message, assistant_message = history
        conversational_histories.extend(
            [
                {"role": "user", "content": user_message},
                {"role": "assistant", "content": assistant_message},
            ]
        )

    return conversational_histories


def get_chat_model_response(
    model: str,
    messages: list[str],
    temperature: float = 0,
    tools: list | None = None,
):
    if tools is not None:
        chat_model_response = LLM_CLIENT.chat.completions.create(
            model=model,
            messages=messages,
            temperature=temperature,
            tools=tools,
        )
    else:
        chat_model_response = LLM_CLIENT.chat.completions.create(
            model=model,
            messages=messages,
            temperature=temperature,
        )

    return chat_model_response


def parse_tools_message(response) -> dict:
    response_choices = response.choices
    if len(response_choices) == 0:
        return None

    tool_calls_response = response_choices[0].message.tool_calls
    if not tool_calls_response:
        return None

    response_choice_argument = tool_calls_response[0].function.arguments
    response_choice_argument_dict = json.loads(response_choice_argument)
    return response_choice_argument_dict


def parse_message(response) -> str | None:
    response_choices = response.choices
    if len(response.choices) == 0:
        return None

    return response_choices[0].message.content