interviewer / api /llm.py
IliaLarchenko's picture
Added errors handling
44800eb
raw
history blame
No virus
5.07 kB
import os
from openai import OpenAI
from utils.errors import APIError
class LLMManager:
def __init__(self, config, prompts):
self.config = config
self.client = OpenAI(base_url=config.llm.url, api_key=config.llm.key)
self.prompts = prompts
def test_connection(self):
try:
response = self.client.chat.completions.create(
model=self.config.llm.name,
messages=[
{"role": "system", "content": "You just help me test the connection."},
{"role": "user", "content": "Hi!"},
{"role": "user", "content": "Ping!"},
],
)
if not response.choices:
raise APIError("LLM Test Connection Error", details="No choices in response")
return response.choices[0].message.content.strip()
except Exception as e:
raise APIError(f"LLM Test Connection Error: Unexpected error: {e}")
def init_bot(self, problem=""):
system_prompt = self.prompts["coding_interviewer_prompt"]
if os.getenv("IS_DEMO"):
system_prompt += " Keep your responses very short and simple, no more than 100 words."
return [
{"role": "system", "content": system_prompt},
{"role": "system", "content": f"The candidate is solving the following problem: {problem}"},
]
def get_problem(self, requirements, difficulty, topic):
full_prompt = (
f"Create a {difficulty} {topic} coding problem. "
f"Additional requirements: {requirements}. "
"The problem should be clearly stated, well-formatted, and solvable within 30 minutes. "
"Ensure the problem varies each time to provide a wide range of challenges."
)
if os.getenv("IS_DEMO"):
full_prompt += " Keep your response very short and simple, no more than 200 words."
try:
response = self.client.chat.completions.create(
model=self.config.llm.name,
messages=[
{"role": "system", "content": self.prompts["problem_generation_prompt"]},
{"role": "user", "content": full_prompt},
],
temperature=1.0,
)
if not response.choices:
raise APIError("LLM Problem Generation Error", details="No choices in response")
question = response.choices[0].message.content.strip()
except Exception as e:
raise APIError(f"LLM Problem Generation Error: Unexpected error: {e}")
chat_history = self.init_bot(question)
return question, chat_history
def send_request(self, code, previous_code, message, chat_history, chat_display):
if code != previous_code:
chat_history.append({"role": "user", "content": f"My latest code:\n{code}"})
chat_history.append({"role": "user", "content": message})
try:
response = self.client.chat.completions.create(model=self.config.llm.name, messages=chat_history)
if not response.choices:
raise APIError("LLM Send Request Error", details="No choices in response")
reply = response.choices[0].message.content.strip()
except Exception as e:
raise APIError(f"LLM Send Request Error: Unexpected error: {e}")
chat_history.append({"role": "assistant", "content": reply})
if chat_display:
chat_display[-1][1] = reply
else:
chat_display.append([message, reply])
return chat_history, chat_display, "", code
def end_interview(self, problem_description, chat_history):
if not chat_history or len(chat_history) <= 2:
return "No interview content available to review."
transcript = [f"{message['role'].capitalize()}: {message['content']}" for message in chat_history[1:]]
system_prompt = self.prompts["grading_feedback_prompt"]
if os.getenv("IS_DEMO"):
system_prompt += " Keep your response very short and simple, no more than 200 words."
try:
response = self.client.chat.completions.create(
model=self.config.llm.name,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": f"The original problem to solve: {problem_description}"},
{"role": "user", "content": "\n\n".join(transcript)},
{"role": "user", "content": "Grade the interview based on the transcript provided and give feedback."},
],
temperature=0.5,
)
if not response.choices:
raise APIError("LLM End Interview Error", details="No choices in response")
feedback = response.choices[0].message.content.strip()
except Exception as e:
raise APIError(f"LLM End Interview Error: Unexpected error: {e}")
return feedback