Added errors handling

.env.huggingface.example

@@ -24,7 +24,7 @@ STT_NAME=whisper-tiny.en
 
 # You can use compatible TTS model from HuggingFace
 # For example you can try public Inference API endpoint for Facebook MMS-TTS model
-# Im my experience OS TTS models from HF sound much more robotic than OpenAI TTS models
+# In my experience OS TTS models from HF sound much more robotic than OpenAI TTS models
 TTS_URL=https://api-inference.huggingface.co/models/facebook/mms-tts-eng
 TTS_TYPE=HF_API
 TTS_NAME=Facebook-mms-tts-eng

api/audio.py

@@ -5,6 +5,8 @@ import requests
 
 from openai import OpenAI
 
+from utils.errors import APIError, AudioConversionError
+
 
 def numpy_audio_to_bytes(audio_data):
     sample_rate = 44100
@@ -12,11 +14,14 @@ def numpy_audio_to_bytes(audio_data):
     sampwidth = 2
 
     buffer = io.BytesIO()
-    with wave.open(buffer, "wb") as wf:
-        wf.setnchannels(num_channels)
-        wf.setsampwidth(sampwidth)
-        wf.setframerate(sample_rate)
-        wf.writeframes(audio_data.tobytes())
+    try:
+        with wave.open(buffer, "wb") as wf:
+            wf.setnchannels(num_channels)
+            wf.setsampwidth(sampwidth)
+            wf.setframerate(sample_rate)
+            wf.writeframes(audio_data.tobytes())
+    except Exception as e:
+        raise AudioConversionError(f"Error converting numpy array to audio bytes: {e}")
     return buffer.getvalue()
 
 
@@ -28,14 +33,31 @@ class STTManager:
         if convert_to_bytes:
             audio = numpy_audio_to_bytes(audio[1])
 
-        if self.config.stt.type == "OPENAI_API":
-            data = ("temp.wav", audio, "audio/wav")
-            client = OpenAI(base_url=self.config.stt.url, api_key=self.config.stt.key)
-            transcription = client.audio.transcriptions.create(model=self.config.stt.name, file=data, response_format="text")
-        elif self.config.stt.type == "HF_API":
-            headers = {"Authorization": "Bearer " + self.config.stt.key}
-            transcription = requests.post(self.config.stt.url, headers=headers, data=audio)
-            transcription = transcription.json()["text"]
+        try:
+            if self.config.stt.type == "OPENAI_API":
+                data = ("temp.wav", audio, "audio/wav")
+                client = OpenAI(base_url=self.config.stt.url, api_key=self.config.stt.key)
+                response = client.audio.transcriptions.create(model=self.config.stt.name, file=data, response_format="text")
+                if not response.success:
+                    raise APIError(
+                        "STT Error: OpenAI API error",
+                        status_code=response.status_code,
+                        details=response.error.get("message", "No error message provided"),
+                    )
+                transcription = response.data
+            elif self.config.stt.type == "HF_API":
+                headers = {"Authorization": "Bearer " + self.config.stt.key}
+                response = requests.post(self.config.stt.url, headers=headers, data=audio)
+                if response.status_code != 200:
+                    error_details = response.json().get("error", "No error message provided")
+                    raise APIError("STT Error: HF API error", status_code=response.status_code, details=error_details)
+                transcription = response.json().get("text", None)
+                if transcription is None:
+                    raise APIError("STT Error: No transcription returned by HF API")
+        except APIError as e:
+            raise
+        except Exception as e:
+            raise APIError(f"STT Error: Unexpected error: {e}")
 
         return transcription
 
@@ -45,18 +67,32 @@ class TTSManager:
         self.config = config
 
     def text_to_speech(self, text):
-        if self.config.tts.type == "OPENAI_API":
-            client = OpenAI(base_url=self.config.tts.url, api_key=self.config.tts.key)
-            response = client.audio.speech.create(model=self.config.tts.name, voice="alloy", response_format="opus", input=text)
-        elif self.config.tts.type == "HF_API":
-            headers = {"Authorization": "Bearer " + self.config.tts.key}
-            response = requests.post(self.config.tts.url, headers=headers)
+        try:
+            if self.config.tts.type == "OPENAI_API":
+                client = OpenAI(base_url=self.config.tts.url, api_key=self.config.tts.key)
+                response = client.audio.speech.create(model=self.config.tts.name, voice="alloy", response_format="opus", input=text)
+                if not response.success:
+                    raise APIError(
+                        "TTS Error: OpenAI API error",
+                        status_code=response.status_code,
+                        details=response.error.get("message", "No error message provided"),
+                    )
+            elif self.config.tts.type == "HF_API":
+                headers = {"Authorization": "Bearer " + self.config.tts.key}
+                response = requests.post(self.config.tts.url, headers=headers, json={"inputs": text})
+                if response.status_code != 200:
+                    error_details = response.json().get("error", "No error message provided")
+                    raise APIError("TTS Error: HF API error", status_code=response.status_code, details=error_details)
+        except APIError as e:
+            raise
+        except Exception as e:
+            raise APIError(f"TTS Error: Unexpected error: {e}")
 
         return response.content
 
     def read_last_message(self, chat_display):
         if chat_display:
-            last_message = chat_display[-1][1]  # Assuming the message is stored at index 1 of the last tuple/list in chat_display
+            last_message = chat_display[-1][1]
             if last_message is not None:
                 return self.text_to_speech(last_message)
         return None

api/llm.py

@@ -2,6 +2,8 @@ import os
 
 from openai import OpenAI
 
+from utils.errors import APIError
+
 
 class LLMManager:
     def __init__(self, config, prompts):
@@ -10,27 +12,30 @@ class LLMManager:
         self.prompts = prompts
 
     def test_connection(self):
-        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!"},
-            ],
-        )
-        return response.choices[0].message.content.strip()
+        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."
 
-        chat_history = [
+        return [
             {"role": "system", "content": system_prompt},
             {"role": "system", "content": f"The candidate is solving the following problem: {problem}"},
         ]
-        return chat_history
 
     def get_problem(self, requirements, difficulty, topic):
         full_prompt = (
@@ -43,61 +48,71 @@ class LLMManager:
         if os.getenv("IS_DEMO"):
             full_prompt += " Keep your response very short and simple, no more than 200 words."
 
-        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,
-        )
-        question = response.choices[0].message.content.strip()
+        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):
-        # Update chat history if code has changed
         if code != previous_code:
             chat_history.append({"role": "user", "content": f"My latest code:\n{code}"})
         chat_history.append({"role": "user", "content": message})
 
-        # Process the updated chat history with the language model
-        response = self.client.chat.completions.create(model=self.config.llm.name, messages=chat_history)
-        reply = response.choices[0].message.content.strip()
+        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})
 
-        # Update chat display with the new reply
         if chat_display:
             chat_display[-1][1] = reply
         else:
             chat_display.append([message, reply])
 
-        # Return updated chat history, chat display, an empty string placeholder, and the unchanged code
         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 = []
-        for message in chat_history[1:]:
-            role = message["role"]
-            content = f"{role.capitalize()}: {message['content']}"
-            transcript.append(content)
+        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."
 
-        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,
-        )
-        feedback = response.choices[0].message.content.strip()
+        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

docs/instruction.py

@@ -3,7 +3,7 @@
 instruction = {
     "demo": """
 <span style="color: red;">
-    This is a demo version utilizing free API access with strict request limits. As a result, the experience may be slow, occasionally buggy, and not of the highest quality. If a model is unavailable, please wait for a minute before retrying. Persistent unavailability may indicate that the request limit has been reached, making the demo temporarily inaccessible.
+    This is a demo version utilizing free API access with strict request limits. As a result, the experience may be slow, occasionally buggy, and not of the highest quality (e.g. robotic voice and very short problem and feedback). If a model is unavailable, please wait for a minute before retrying. Persistent unavailability may indicate that the request limit has been reached, making the demo temporarily inaccessible.
     For a significantly better experience, please run the service locally and use your own OpenAI key or HuggingFace models.
 </span>
 

utils/errors.py

@@ -0,0 +1,15 @@
+class AudioConversionError(Exception):
+    """Exception raised for errors in the audio conversion process."""
+
+    pass
+
+
+class APIError(Exception):
+    """Custom exception for API error handling."""
+
+    def __init__(self, message, status_code=None, details=None):
+        if details:
+            super().__init__(f"{message} - Details: {details}")
+        else:
+            super().__init__(message)
+        self.status_code = status_code