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