Huge refactoring

api/audio.py

@@ -7,6 +7,7 @@ import requests
 from openai import OpenAI
 
 from utils.errors import APIError, AudioConversionError
+from typing import List, Dict, Optional, Generator, Tuple
 
 
 class STTManager:
@@ -20,7 +21,13 @@ class STTManager:
         self.status = self.test_stt()
         self.streaming = self.test_streaming()
 
-    def numpy_audio_to_bytes(self, audio_data):
+    def numpy_audio_to_bytes(self, audio_data: np.ndarray) -> bytes:
+        """
+        Convert a numpy array of audio data to bytes.
+
+        :param audio_data: Numpy array containing audio data.
+        :return: Bytes representation of the audio data.
+        """
         num_channels = 1
         sampwidth = 2
 
@@ -35,20 +42,34 @@ class STTManager:
             raise AudioConversionError(f"Error converting numpy array to audio bytes: {e}")
         return buffer.getvalue()
 
-    def process_audio_chunk(self, audio, audio_buffer, transcript):
-        """Process streamed audio data to accumulate and transcribe with overlapping segments."""
+    def process_audio_chunk(
+        self, audio: Tuple[int, np.ndarray], audio_buffer: np.ndarray, transcript: Dict
+    ) -> Tuple[Dict, np.ndarray, str]:
+        """
+        Process streamed audio data to accumulate and transcribe with overlapping segments.
+
+        :param audio: Tuple containing the sample rate and audio data as numpy array.
+        :param audio_buffer: Current audio buffer as numpy array.
+        :param transcript: Current transcript dictionary.
+        :return: Updated transcript, updated audio buffer, and transcript text.
+        """
         audio_buffer = np.concatenate((audio_buffer, audio[1]))
 
         if len(audio_buffer) >= self.SAMPLE_RATE * self.CHUNK_LENGTH or len(audio_buffer) % (self.SAMPLE_RATE // 2) != 0:
             audio_bytes = self.numpy_audio_to_bytes(audio_buffer[: self.SAMPLE_RATE * self.CHUNK_LENGTH])
             audio_buffer = audio_buffer[self.SAMPLE_RATE * self.STEP_LENGTH :]
-
             new_transcript = self.speech_to_text_stream(audio_bytes)
             transcript = self.merge_transcript(transcript, new_transcript)
 
         return transcript, audio_buffer, transcript["text"]
 
-    def speech_to_text_stream(self, audio):
+    def speech_to_text_stream(self, audio: bytes) -> List[Dict[str, str]]:
+        """
+        Convert speech to text from a byte stream using streaming.
+
+        :param audio: Bytes representation of audio data.
+        :return: List of dictionaries containing transcribed words and their timestamps.
+        """
         if self.config.stt.type == "HF_API":
             raise APIError("STT Error: Streaming not supported for this STT type")
         try:
@@ -57,18 +78,24 @@ class STTManager:
             transcription = client.audio.transcriptions.create(
                 model=self.config.stt.name, file=data, response_format="verbose_json", timestamp_granularities=["word"]
             )
-        except APIError as e:
+        except APIError:
             raise
         except Exception as e:
             raise APIError(f"STT Error: Unexpected error: {e}")
         return transcription.words
 
-    def merge_transcript(self, transcript, new_transcript):
+    def merge_transcript(self, transcript: Dict, new_transcript: List[Dict[str, str]]) -> Dict:
+        """
+        Merge new transcript data with the existing transcript.
+
+        :param transcript: Existing transcript dictionary.
+        :param new_transcript: New transcript data to merge.
+        :return: Updated transcript dictionary.
+        """
         cut_off = transcript["last_cutoff"]
         transcript["last_cutoff"] = self.MAX_RELIABILITY_CUTOFF - self.STEP_LENGTH
 
         transcript["words"] = transcript["words"][: len(transcript["words"]) - transcript["not_confirmed"]]
-
         transcript["not_confirmed"] = 0
         first_word = True
 
@@ -85,40 +112,55 @@ class STTManager:
                     transcript["last_cutoff"] = max(1.0, word_dict["end"] - self.STEP_LENGTH)
 
         transcript["text"] = " ".join(transcript["words"])
-
         return transcript
 
-    def speech_to_text_full(self, audio):
-        audio = self.numpy_audio_to_bytes(audio[1])
+    def speech_to_text_full(self, audio: Tuple[int, np.ndarray]) -> str:
+        """
+        Convert speech to text from a full audio segment.
+
+        :param audio: Tuple containing the sample rate and audio data as numpy array.
+        :return: Transcribed text.
+        """
+        audio_bytes = self.numpy_audio_to_bytes(audio[1])
         try:
             if self.config.stt.type == "OPENAI_API":
-                data = ("temp.wav", audio, "audio/wav")
+                data = ("temp.wav", audio_bytes, "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}
-                response = requests.post(self.config.stt.url, headers=headers, data=audio)
+                response = requests.post(self.config.stt.url, headers=headers, data=audio_bytes)
                 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:
+        except APIError:
             raise
         except Exception as e:
             raise APIError(f"STT Error: Unexpected error: {e}")
 
         return transcription
 
-    def test_stt(self):
+    def test_stt(self) -> bool:
+        """
+        Test if the STT service is working correctly.
+
+        :return: True if the STT service is working, False otherwise.
+        """
         try:
             self.speech_to_text_full((48000, np.zeros(10000)))
             return True
         except:
             return False
 
-    def test_streaming(self):
+    def test_streaming(self) -> bool:
+        """
+        Test if the STT streaming service is working correctly.
+
+        :return: True if the STT streaming service is working, False otherwise.
+        """
         try:
             self.speech_to_text_stream(self.numpy_audio_to_bytes(np.zeros(10000)))
             return True
@@ -127,14 +169,30 @@ class STTManager:
 
 
 class TTSManager:
-    def test_tts(self):
+    def __init__(self, config):
+        self.config = config
+        self.status = self.test_tts()
+        self.streaming = self.test_tts_stream() if self.status else False
+        self.read_last_message = self.rlm_stream if self.streaming else self.rlm
+
+    def test_tts(self) -> bool:
+        """
+        Test if the TTS service is working correctly.
+
+        :return: True if the TTS service is working, False otherwise.
+        """
         try:
             self.read_text("Handshake")
             return True
         except:
             return False
 
-    def test_tts_stream(self):
+    def test_tts_stream(self) -> bool:
+        """
+        Test if the TTS streaming service is working correctly.
+
+        :return: True if the TTS streaming service is working, False otherwise.
+        """
         try:
             for _ in self.read_text_stream("Handshake"):
                 pass
@@ -142,19 +200,13 @@ class TTSManager:
         except:
             return False
 
-    def __init__(self, config):
-        self.config = config
-        self.status = self.test_tts()
-        if self.status:
-            self.streaming = self.test_tts_stream()
-        else:
-            self.streaming = False
-        if self.streaming:
-            self.read_last_message = self.rlm_stream
-        else:
-            self.read_last_message = self.rlm
-
-    def read_text(self, text):
+    def read_text(self, text: str) -> bytes:
+        """
+        Convert text to speech and return the audio bytes.
+
+        :param text: Text to convert to speech.
+        :return: Bytes representation of the audio.
+        """
         headers = {"Authorization": "Bearer " + self.config.tts.key}
         try:
             if self.config.tts.type == "OPENAI_API":
@@ -165,15 +217,21 @@ class TTSManager:
             if response.status_code != 200:
                 error_details = response.json().get("error", "No error message provided")
                 raise APIError(f"TTS Error: {self.config.tts.type} error", status_code=response.status_code, details=error_details)
-        except APIError as e:
+        except APIError:
             raise
         except Exception as e:
             raise APIError(f"TTS Error: Unexpected error: {e}")
 
         return response.content
 
-    def read_text_stream(self, text):
-        if self.config.tts.type not in ["OPENAI_API"]:
+    def read_text_stream(self, text: str) -> Generator[bytes, None, None]:
+        """
+        Convert text to speech using streaming and return the audio bytes.
+
+        :param text: Text to convert to speech.
+        :return: Generator yielding chunks of audio bytes.
+        """
+        if self.config.tts.type != "OPENAI_API":
             raise APIError("TTS Error: Streaming not supported for this TTS type")
         headers = {"Authorization": "Bearer " + self.config.tts.key}
         data = {"model": self.config.tts.name, "input": text, "voice": "alloy", "response_format": "opus"}
@@ -187,15 +245,27 @@ class TTSManager:
                     yield from response.iter_content(chunk_size=1024)
         except StopIteration:
             pass
-        except APIError as e:
+        except APIError:
             raise
         except Exception as e:
             raise APIError(f"TTS Error: Unexpected error: {e}")
 
-    def rlm(self, chat_history):
+    def rlm(self, chat_history: List[List[Optional[str]]]) -> bytes:
+        """
+        Read the last message in the chat history and convert it to speech.
+
+        :param chat_history: List of chat messages.
+        :return: Bytes representation of the audio.
+        """
         if len(chat_history) > 0 and chat_history[-1][1]:
             return self.read_text(chat_history[-1][1])
 
-    def rlm_stream(self, chat_history):
+    def rlm_stream(self, chat_history: List[List[Optional[str]]]) -> Generator[bytes, None, None]:
+        """
+        Read the last message in the chat history and convert it to speech using streaming.
+
+        :param chat_history: List of chat messages.
+        :return: Generator yielding chunks of audio bytes.
+        """
         if len(chat_history) > 0 and chat_history[-1][1]:
             yield from self.read_text_stream(chat_history[-1][1])

api/llm.py

@@ -1,40 +1,38 @@
 import os
-
 from openai import OpenAI
-
 from utils.errors import APIError
+from typing import List, Dict, Generator, Optional, Tuple
 
 
 class PromptManager:
-    def __init__(self, prompts):
+    def __init__(self, prompts: Dict[str, str]):
         self.prompts = prompts
         self.limit = os.getenv("DEMO_WORD_LIMIT")
 
-    def add_limit(self, prompt):
+    def add_limit(self, prompt: str) -> str:
         if self.limit:
             prompt += f" Keep your responses very short and simple, no more than {self.limit} words."
         return prompt
 
-    def get_system_prompt(self, key):
+    def get_system_prompt(self, key: str) -> str:
         prompt = self.prompts[key]
         return self.add_limit(prompt)
 
-    def get_problem_requirements_prompt(self, type, difficulty=None, topic=None, requirements=None):
-        prompt = f"Create a {type} problem. Difficulty: {difficulty}. Topic: {topic} " f"Additional requirements: {requirements}. "
+    def get_problem_requirements_prompt(
+        self, type: str, difficulty: Optional[str] = None, topic: Optional[str] = None, requirements: Optional[str] = None
+    ) -> str:
+        prompt = f"Create a {type} problem. Difficulty: {difficulty}. Topic: {topic} Additional requirements: {requirements}. "
         return self.add_limit(prompt)
 
 
 class LLMManager:
-    def __init__(self, config, prompts):
+    def __init__(self, config, prompts: Dict[str, str]):
         self.config = config
         self.client = OpenAI(base_url=config.llm.url, api_key=config.llm.key)
         self.prompt_manager = PromptManager(prompts)
 
         self.status = self.test_llm()
-        if self.status:
-            self.streaming = self.test_llm_stream()
-        else:
-            self.streaming = False
+        self.streaming = self.test_llm_stream() if self.status else False
 
         if self.streaming:
             self.end_interview = self.end_interview_stream
@@ -45,19 +43,7 @@ class LLMManager:
             self.get_problem = self.get_problem_full
             self.send_request = self.send_request_full
 
-    def text_processor(self):
-        def ans_full(response):
-            return response
-
-        def ans_stream(response):
-            yield from response
-
-        if self.streaming:
-            return ans_full
-        else:
-            return ans_stream
-
-    def get_text(self, messages):
+    def get_text(self, messages: List[Dict[str, str]]) -> str:
         try:
             response = self.client.chat.completions.create(model=self.config.llm.name, messages=messages, temperature=1, max_tokens=2000)
             if not response.choices:
@@ -66,14 +52,10 @@ class LLMManager:
         except Exception as e:
             raise APIError(f"LLM Get Text Error: Unexpected error: {e}")
 
-    def get_text_stream(self, messages):
+    def get_text_stream(self, messages: List[Dict[str, str]]) -> Generator[str, None, None]:
         try:
             response = self.client.chat.completions.create(
-                model=self.config.llm.name,
-                messages=messages,
-                temperature=1,
-                stream=True,
-                max_tokens=2000,
+                model=self.config.llm.name, messages=messages, temperature=1, stream=True, max_tokens=2000
             )
         except Exception as e:
             raise APIError(f"LLM End Interview Error: Unexpected error: {e}")
@@ -83,110 +65,107 @@ class LLMManager:
                 text += chunk.choices[0].delta.content
             yield text
 
-    test_messages = [
-        {"role": "system", "content": "You just help me test the connection."},
-        {"role": "user", "content": "Hi!"},
-        {"role": "user", "content": "Ping!"},
-    ]
-
-    def test_llm(self):
+    def test_llm(self) -> bool:
         try:
-            self.get_text(self.test_messages)
+            self.get_text(
+                [
+                    {"role": "system", "content": "You just help me test the connection."},
+                    {"role": "user", "content": "Hi!"},
+                    {"role": "user", "content": "Ping!"},
+                ]
+            )
             return True
         except:
             return False
 
-    def test_llm_stream(self):
+    def test_llm_stream(self) -> bool:
         try:
-            for _ in self.get_text_stream(self.test_messages):
+            for _ in self.get_text_stream(
+                [
+                    {"role": "system", "content": "You just help me test the connection."},
+                    {"role": "user", "content": "Hi!"},
+                    {"role": "user", "content": "Ping!"},
+                ]
+            ):
                 pass
             return True
         except:
             return False
 
-    def init_bot(self, problem, interview_type="coding"):
+    def init_bot(self, problem: str, interview_type: str = "coding") -> List[Dict[str, str]]:
         system_prompt = self.prompt_manager.get_system_prompt(f"{interview_type}_interviewer_prompt")
+        return [{"role": "system", "content": f"{system_prompt}\nThe candidate is solving the following problem:\n {problem}"}]
 
-        return [
-            {"role": "system", "content": system_prompt + f"\nThe candidate is solving the following problem:\n {problem}"},
-        ]
-
-    def get_problem_prepare_messages(self, requirements, difficulty, topic, interview_type):
+    def get_problem_prepare_messages(self, requirements: str, difficulty: str, topic: str, interview_type: str) -> List[Dict[str, str]]:
         system_prompt = self.prompt_manager.get_system_prompt(f"{interview_type}_problem_generation_prompt")
         full_prompt = self.prompt_manager.get_problem_requirements_prompt(interview_type, difficulty, topic, requirements)
-
-        messages = [
+        return [
             {"role": "system", "content": system_prompt},
             {"role": "user", "content": full_prompt},
         ]
 
-        return messages
-
-    def get_problem_full(self, requirements, difficulty, topic, interview_type="coding"):
+    def get_problem_full(self, requirements: str, difficulty: str, topic: str, interview_type: str = "coding") -> str:
         messages = self.get_problem_prepare_messages(requirements, difficulty, topic, interview_type)
         return self.get_text(messages)
 
-    def get_problem_stream(self, requirements, difficulty, topic, interview_type="coding"):
+    def get_problem_stream(
+        self, requirements: str, difficulty: str, topic: str, interview_type: str = "coding"
+    ) -> Generator[str, None, None]:
         messages = self.get_problem_prepare_messages(requirements, difficulty, topic, interview_type)
         yield from self.get_text_stream(messages)
 
-    def update_chat_history(self, code, previous_code, chat_history, chat_display):
+    def update_chat_history(
+        self, code: str, previous_code: str, chat_history: List[Dict[str, str]], chat_display: List[List[Optional[str]]]
+    ) -> List[Dict[str, str]]:
         message = chat_display[-1][0]
-
         if code != previous_code:
-            message += "\nMY NOTES AND CODE:\n"
-            message += code
-
+            message += "\nMY NOTES AND CODE:\n" + code
         chat_history.append({"role": "user", "content": message})
-
         return chat_history
 
-    def send_request_full(self, code, previous_code, chat_history, chat_display):
+    def send_request_full(
+        self, code: str, previous_code: str, chat_history: List[Dict[str, str]], chat_display: List[List[Optional[str]]]
+    ) -> Tuple[List[Dict[str, str]], List[List[Optional[str]]], str]:
         chat_history = self.update_chat_history(code, previous_code, chat_history, chat_display)
-
         reply = self.get_text(chat_history)
         chat_display.append([None, reply.split("#NOTES#")[0].strip()])
         chat_history.append({"role": "assistant", "content": reply})
-
         return chat_history, chat_display, code
 
-    def send_request_stream(self, code, previous_code, chat_history, chat_display):
+    def send_request_stream(
+        self, code: str, previous_code: str, chat_history: List[Dict[str, str]], chat_display: List[List[Optional[str]]]
+    ) -> Generator[Tuple[List[Dict[str, str]], List[List[Optional[str]]], str], None, None]:
         chat_history = self.update_chat_history(code, previous_code, chat_history, chat_display)
-
         chat_display.append([None, ""])
         chat_history.append({"role": "assistant", "content": ""})
-
         reply = self.get_text_stream(chat_history)
         for message in reply:
             chat_display[-1][1] = message.split("#NOTES#")[0].strip()
             chat_history[-1]["content"] = message
-
             yield chat_history, chat_display, code
 
-    def end_interview_prepare_messages(self, problem_description, chat_history, interview_type):
+    def end_interview_prepare_messages(
+        self, problem_description: str, chat_history: List[Dict[str, str]], interview_type: str
+    ) -> List[Dict[str, str]]:
         transcript = [f"{message['role'].capitalize()}: {message['content']}" for message in chat_history[1:]]
-
         system_prompt = self.prompt_manager.get_system_prompt(f"{interview_type}_grading_feedback_prompt")
-
-        messages = [
+        return [
             {"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."},
         ]
 
-        return messages
-
-    def end_interview_full(self, problem_description, chat_history, interview_type="coding"):
+    def end_interview_full(self, problem_description: str, chat_history: List[Dict[str, str]], interview_type: str = "coding") -> str:
         if len(chat_history) <= 2:
             return "No interview history available"
-        else:
-            messages = self.end_interview_prepare_messages(problem_description, chat_history, interview_type)
-            return self.get_text(messages)
+        messages = self.end_interview_prepare_messages(problem_description, chat_history, interview_type)
+        return self.get_text(messages)
 
-    def end_interview_stream(self, problem_description, chat_history, interview_type="coding"):
+    def end_interview_stream(
+        self, problem_description: str, chat_history: List[Dict[str, str]], interview_type: str = "coding"
+    ) -> Generator[str, None, None]:
         if len(chat_history) <= 2:
             yield "No interview history available"
-        else:
-            messages = self.end_interview_prepare_messages(problem_description, chat_history, interview_type)
-            yield from self.get_text_stream(messages)
+        messages = self.end_interview_prepare_messages(problem_description, chat_history, interview_type)
+        yield from self.get_text_stream(messages)

app.py

@@ -1,5 +1,4 @@
 import os
-
 import gradio as gr
 
 from api.audio import STTManager, TTSManager
@@ -10,33 +9,44 @@ from ui.coding import get_problem_solving_ui
 from ui.instructions import get_instructions_ui
 from utils.params import default_audio_params
 
-config = Config()
-llm = LLMManager(config, prompts)
-tts = TTSManager(config)
-stt = STTManager(config)
-
-default_audio_params["streaming"] = stt.streaming
-
-if os.getenv("SILENT", False):
-    tts.read_last_message = lambda x: None
-
-# Interface
-
-with gr.Blocks(title="AI Interviewer") as demo:
-    audio_output = gr.Audio(label="Play audio", autoplay=True, visible=os.environ.get("DEBUG", False), streaming=tts.streaming)
-    tabs = [
-        get_instructions_ui(llm, tts, stt, default_audio_params),
-        get_problem_solving_ui(llm, tts, stt, default_audio_params, audio_output, name="Coding", interview_type="coding"),
-        get_problem_solving_ui(llm, tts, stt, default_audio_params, audio_output, name="ML Design (Beta)", interview_type="ml_design"),
-        get_problem_solving_ui(llm, tts, stt, default_audio_params, audio_output, name="ML Theory (Beta)", interview_type="ml_theory"),
-        get_problem_solving_ui(
-            llm, tts, stt, default_audio_params, audio_output, name="System Design (Beta)", interview_type="system_design"
-        ),
-        get_problem_solving_ui(llm, tts, stt, default_audio_params, audio_output, name="Math (Beta)", interview_type="math"),
-        get_problem_solving_ui(llm, tts, stt, default_audio_params, audio_output, name="SQL (Beta)", interview_type="sql"),
-    ]
-
-    for tab in tabs:
-        tab.render()
 
-demo.launch(show_api=False)
+def initialize_services():
+    """Initialize configuration, LLM, TTS, and STT services."""
+    config = Config()
+    llm = LLMManager(config, prompts)
+    tts = TTSManager(config)
+    stt = STTManager(config)
+    default_audio_params["streaming"] = stt.streaming
+    if os.getenv("SILENT", False):
+        tts.read_last_message = lambda x: None
+    return config, llm, tts, stt
+
+
+def create_interface(llm, tts, stt, audio_params):
+    """Create and configure the Gradio interface."""
+    with gr.Blocks(title="AI Interviewer") as demo:
+        audio_output = gr.Audio(label="Play audio", autoplay=True, visible=os.environ.get("DEBUG", False), streaming=tts.streaming)
+        tabs = [
+            get_instructions_ui(llm, tts, stt, audio_params),
+            get_problem_solving_ui(llm, tts, stt, audio_params, audio_output, name="Coding", interview_type="coding"),
+            get_problem_solving_ui(llm, tts, stt, audio_params, audio_output, name="ML Design (Beta)", interview_type="ml_design"),
+            get_problem_solving_ui(llm, tts, stt, audio_params, audio_output, name="ML Theory (Beta)", interview_type="ml_theory"),
+            get_problem_solving_ui(llm, tts, stt, audio_params, audio_output, name="System Design (Beta)", interview_type="system_design"),
+            get_problem_solving_ui(llm, tts, stt, audio_params, audio_output, name="Math (Beta)", interview_type="math"),
+            get_problem_solving_ui(llm, tts, stt, audio_params, audio_output, name="SQL (Beta)", interview_type="sql"),
+        ]
+
+        for tab in tabs:
+            tab.render()
+    return demo
+
+
+def main():
+    """Main function to initialize services and launch the Gradio interface."""
+    config, llm, tts, stt = initialize_services()
+    demo = create_interface(llm, tts, stt, default_audio_params)
+    demo.launch(show_api=False)
+
+
+if __name__ == "__main__":
+    main()

tests/analysis.py

@@ -15,6 +15,7 @@ from IPython.display import Markdown, display
 from openai import OpenAI
 from tests.testing_prompts import feedback_analyzer
 from resources.prompts import prompts, base_prompts
+from typing import List, Dict, Any, Tuple, Optional
 
 criteria_list = {
     "problem_statement",
@@ -65,7 +66,15 @@ criteria_list = {
 }
 
 
-def grade_attempt(file_path, grader_model, attempt_index):
+def grade_attempt(file_path: str, grader_model: str, attempt_index: int) -> Optional[Dict[str, Any]]:
+    """
+    Grade an interview attempt using the specified grader model.
+
+    :param file_path: Path to the JSON file containing interview data.
+    :param grader_model: Grader model to use for grading.
+    :param attempt_index: Index of the grading attempt.
+    :return: Feedback dictionary or None if grading fails.
+    """
     for retry in range(3):  # Retry mechanism
         try:
             feedback = grade(file_path, grader_model, str(attempt_index))
@@ -76,26 +85,36 @@ def grade_attempt(file_path, grader_model, attempt_index):
     return None
 
 
-def complete_and_grade(interview_params, exp_name, grader_models, candidate_model):
-    interview_type, attempt_num, llm_config = interview_params
+def complete_and_grade(
+    interview_params: Tuple[str, int, Any], exp_name: str, grader_models: List[str], candidate_model: str
+) -> List[Dict[str, Any]]:
+    """
+    Complete an interview and grade it using specified grader models.
 
+    :param interview_params: Tuple containing interview type, attempt number, and LLM config.
+    :param exp_name: Experiment name.
+    :param grader_models: List of grader models.
+    :param candidate_model: Candidate model name.
+    :return: List of feedback dictionaries.
+    """
+    interview_type, attempt_num, llm_config = interview_params
     feedback_list = []
-    attempt_successful = False
-    for attempt in range(3):  # Retry up to 3 times
+
+    # Attempt interview completion with retries
+    for attempt in range(3):
         try:
             file_path, _ = complete_interview(interview_type, exp_name, llm_config, model=candidate_model, pause=attempt * 5)
             print(
                 f"Attempt {attempt_num + 1}, retry {attempt + 1} interview simulation of {interview_type} by {llm_config.name} completed successfully"
             )
-            attempt_successful = True
             break
         except Exception as e:
             print(f"Retry {attempt + 1} for attempt {attempt_num + 1} of {interview_type} by {llm_config.name} failed with error: {e}")
-
-    if not attempt_successful:
+    else:
         print(f"All retries failed for attempt {attempt_num + 1} of {interview_type} by {llm_config.name}")
         return feedback_list
 
+    # Grade the interview
     try:
         for i, grader_model in enumerate(grader_models):
             feedback = grade_attempt(file_path, grader_model, i)
@@ -103,19 +122,36 @@ def complete_and_grade(interview_params, exp_name, grader_models, candidate_mode
                 feedback_list.append(feedback)
                 print(f"Attempt {attempt_num + 1} of {interview_type} by {llm_config.name} graded by {grader_model} successfully")
                 print(f"Overall score: {feedback['overall_score']}")
-
     except Exception as e:
         print(f"Grading for attempt {attempt_num + 1} of {interview_type} by {llm_config.name} failed with error: {e}")
 
-    if len(feedback_list) == 0:
+    if not feedback_list:
         print(f"Attempt {attempt_num + 1} of {interview_type} by {llm_config.name} returned an empty list")
 
     return feedback_list
 
 
 def run_evaluation(
-    exp_name, num_attempts=5, interview_types=None, grader_models=None, llm_configs=None, candidate_model="gpt-3.5-turbo", num_workers=3
-):
+    exp_name: str,
+    num_attempts: int = 5,
+    interview_types: Optional[List[str]] = None,
+    grader_models: Optional[List[str]] = None,
+    llm_configs: Optional[List[Any]] = None,
+    candidate_model: str = "gpt-3.5-turbo",
+    num_workers: int = 3,
+) -> str:
+    """
+    Run the evaluation by completing and grading interviews.
+
+    :param exp_name: Experiment name.
+    :param num_attempts: Number of attempts per interview type.
+    :param interview_types: List of interview types.
+    :param grader_models: List of grader models.
+    :param llm_configs: List of LLM configurations.
+    :param candidate_model: Candidate model name.
+    :param num_workers: Number of workers for concurrent execution.
+    :return: Experiment name.
+    """
     if interview_types is None:
         interview_types = ["ml_design", "math", "ml_theory", "system_design", "sql", "coding"]
     if grader_models is None:
@@ -143,12 +179,25 @@ def run_evaluation(
     return exp_name
 
 
-def highlight_color(val):
-    color = "red" if val < 0.7 else "orange" if val < 0.9 else "lightgreen" if val < 0.95 else "green"
+def highlight_color(val: float) -> str:
+    """
+    Highlight the cell color based on the value.
+
+    :param val: The value to determine the color.
+    :return: The color style string.
+    """
+    color_map = {val < 0.7: "red", 0.7 <= val < 0.9: "orange", 0.9 <= val < 0.95: "lightgreen", val >= 0.95: "green"}
+    color = next(color for condition, color in color_map.items() if condition)
     return f"color: {color}"
 
 
-def generate_and_display_tables(df):
+def generate_and_display_tables(df: pd.DataFrame) -> Dict[str, Any]:
+    """
+    Generate and display various tables for analysis.
+
+    :param df: DataFrame containing the data.
+    :return: Dictionary of styled tables.
+    """
     # Grouping by prefix
     prefixes = ["problem", "interviewer", "feedback"]
     prefix_columns = [col for col in df.columns if any(col.startswith(prefix) for prefix in prefixes)]
@@ -235,15 +284,19 @@ def generate_and_display_tables(df):
     return tables_dict
 
 
-def filter_df(df, prefixes=["problem", "interviewer", "feedback"]):
-    # Identify all columns starting with any of the prefixes
+def filter_df(df: pd.DataFrame, prefixes: List[str] = ["problem", "interviewer", "feedback"]) -> pd.DataFrame:
+    """
+    Filter the DataFrame to keep only rows with valid values in specified columns.
+
+    :param df: DataFrame to filter.
+    :param prefixes: List of prefixes to identify columns to check.
+    :return: Filtered DataFrame.
+    """
     columns_to_check = [col for col in df.columns if any(col.startswith(prefix) for prefix in prefixes)]
 
-    # Function to check if a value is a boolean, None, or string representations of boolean types
     def is_valid_value(val):
         return isinstance(val, bool) or val is None or val is np.nan or val in {"True", "False", "None", "NaN"}
 
-    # Function to convert string representations to actual booleans
     def to_bool(val):
         if val == "True":
             return True
@@ -253,25 +306,17 @@ def filter_df(df, prefixes=["problem", "interviewer", "feedback"]):
             return None
         return val
 
-    # Check if all values in the specified columns are valid
     def all_values_valid(row):
         return all(is_valid_value(row[col]) for col in columns_to_check)
 
-    # Apply filtering to keep only rows with valid values
     valid_df = df[df.apply(all_values_valid, axis=1)].copy()
-
-    # Convert string representations to booleans
     for col in columns_to_check:
         valid_df[col] = valid_df[col].apply(to_bool)
 
-    # Identify removed rows
     removed_rows = df[~df.index.isin(valid_df.index)]
-
-    # Print the number of rows removed
     num_removed = len(removed_rows)
     print(f"Number of rows removed: {num_removed}")
 
-    # Print the value from the "file_name" column for each removed row, or `None` if not present
     if "file_name" in removed_rows.columns:
         for value in removed_rows["file_name"].tolist():
             print(f"Removed row file_name: {value}")
@@ -281,26 +326,30 @@ def filter_df(df, prefixes=["problem", "interviewer", "feedback"]):
     return valid_df
 
 
-def generate_analysis_report(df, folder, focus=None, model="gpt-4o"):
+def generate_analysis_report(df: pd.DataFrame, folder: Optional[str], focus: Optional[str] = None, model: str = "gpt-4o") -> str:
+    """
+    Generate an analysis report based on the feedback data.
 
+    :param df: DataFrame containing the feedback data.
+    :param folder: Folder to save the analysis report.
+    :param focus: Specific focus area for the analysis.
+    :param model: Model used for generating the analysis.
+    :return: Analysis report content.
+    """
     client = OpenAI(base_url="https://api.openai.com/v1")
 
     all_comments = "\n\n".join([f"Interview type: {t}. Feedback: {str(f)}" for t, f in zip(df["type"].values, df["comments"].values)])
 
-    messages = [
-        {"role": "system", "content": feedback_analyzer},
-        {"role": "user", "content": f"Interview feedback: {all_comments}"},
-    ]
+    messages = [{"role": "system", "content": feedback_analyzer}, {"role": "user", "content": f"Interview feedback: {all_comments}"}]
 
     if focus:
         messages.append({"role": "user", "content": f"Focus only on comments about {focus} part of the interview"})
 
     response = client.chat.completions.create(model=model, messages=messages, temperature=1)
-
     comments_analysis = response.choices[0].message.content
     display(Markdown(comments_analysis))
 
-    if folder is not None:
+    if folder:
         with open(os.path.join(folder, "analysis.md"), "w") as f:
             f.write(comments_analysis)
             f.write("\n\n")
@@ -308,16 +357,18 @@ def generate_analysis_report(df, folder, focus=None, model="gpt-4o"):
                 f.write(f"Type: {t}\n")
                 f.write(df[[c for c in df.columns if c != "comments"]][df["type"] == t].T.to_markdown())
                 f.write("\n\n")
-            f.write(f"Type: all\n")
-            f.write("\n\n")
-            f.write("Feedback:\n")
-            f.write(all_comments)
+            f.write(f"Type: all\n\nFeedback:\n{all_comments}")
 
     return comments_analysis
 
 
-def analyze_and_improve_segment(df, segment_to_improve=None):
+def analyze_and_improve_segment(df: pd.DataFrame, segment_to_improve: Optional[str] = None) -> None:
+    """
+    Analyze and improve a specific segment of the interview process.
 
+    :param df: DataFrame containing the data.
+    :param segment_to_improve: Segment to focus on for improvement.
+    """
     sorted_stages = df[["problem", "interviewer", "feedback"]].mean().sort_values()
     if not segment_to_improve:
         segment_to_improve = sorted_stages.index[0]
@@ -326,43 +377,25 @@ def analyze_and_improve_segment(df, segment_to_improve=None):
     print(f"Let's try to improve {segment_to_improve}")
     print(f"Quality threshold {th_score}")
 
-    # Identifying types that need improvement
     type_stage_scores = df.groupby("type")[segment_to_improve].mean()
-    types_to_improve = []
-    for t, s in type_stage_scores.items():
-        if s < th_score:
-            types_to_improve.append(t)
-
+    types_to_improve = [t for t, s in type_stage_scores.items() if s < th_score]
     print(f"We will focus on {types_to_improve}")
 
-    # Filtering DataFrame based on identified types and scoring criteria
     filtered_df = df[df["type"].apply(lambda x: x in types_to_improve)]
     prefix_columns = [col for col in df.columns if col.startswith(segment_to_improve)]
     filtered_df = filtered_df[filtered_df[prefix_columns].mean(axis=1) < th_score]
 
-    # Generating an analysis report
     comments_analysis = generate_analysis_report(filtered_df, None, focus=segment_to_improve, model="gpt-4o")
 
-    # Constructing improvement prompt
-    improvement_prompt = """You want to improve the prompts for LLM interviewer.
-Below you will see some of the prompts that are used right now. 
-As well as a summary of mistakes that interviewer make.
-You can add 1-3 lines to each of prompts if needed, but you can't change or remove anything.
-"""
+    improvement_prompt = "You want to improve the prompts for LLM interviewer. Below you will see some of the prompts that are used right now. As well as a summary of mistakes that interviewer make. You can add 1-3 lines to each of prompts if needed, but you can't change or remove anything."
 
-    # Selecting the base prompt for the segment to improve
     base_prompt = base_prompts.get(f"base_{segment_to_improve}", "Base prompt not found for the segment")
 
-    # Constructing the current prompts display
-    current_prompts = "The current prompts are below. \n"
-    current_prompts += "BASE PROMPT (applied to all interview types): \n"
-    current_prompts += base_prompt + "\n"
-
+    current_prompts = f"The current prompts are below. \nBASE PROMPT (applied to all interview types): \n{base_prompt}\n"
     for k, v in prompts.items():
         if segment_to_improve in k:
             current_prompts += f"{k}: {v[len(base_prompt):]} \n\n"
 
-    # Making API call to OpenAI
     client = OpenAI(base_url="https://api.openai.com/v1")
     model = "gpt-4o"
     messages = [

tests/candidate.py

@@ -3,11 +3,10 @@ import os
 import random
 import string
 import time
-
 from collections import defaultdict
+from typing import Dict, Optional, Tuple
 
 from openai import OpenAI
-
 from api.llm import LLMManager
 from utils.config import Config
 from resources.data import fixed_messages, topic_lists
@@ -15,14 +14,36 @@ from resources.prompts import prompts
 from tests.testing_prompts import candidate_prompt
 
 
-def complete_interview(interview_type, exp_name, llm_config=None, requirements="", difficulty="", topic="", model="gpt-3.5-turbo", pause=0):
+def complete_interview(
+    interview_type: str,
+    exp_name: str,
+    llm_config: Optional[Config] = None,
+    requirements: str = "",
+    difficulty: str = "",
+    topic: str = "",
+    model: str = "gpt-3.5-turbo",
+    pause: int = 0,
+) -> Tuple[str, Dict]:
+    """
+    Complete an interview and record the results.
+
+    :param interview_type: Type of interview to complete.
+    :param exp_name: Experiment name for file saving.
+    :param llm_config: Optional LLM configuration.
+    :param requirements: Additional requirements for the interview.
+    :param difficulty: Difficulty level for the interview.
+    :param topic: Topic for the interview.
+    :param model: Model to use for the candidate.
+    :param pause: Pause duration between requests to prevent rate limits.
+    :return: Tuple containing the file path and interview data.
+    """
     client = OpenAI(base_url="https://api.openai.com/v1")
     config = Config()
     if llm_config:
         config.llm = llm_config
     llm = LLMManager(config, prompts)
     llm_name = config.llm.name
-    print(f"Starting evaluation interviewer LLM: {llm_name}, candidate_LLM: {model} interview_type: {interview_type}")
+    print(f"Starting evaluation interviewer LLM: {llm_name}, candidate LLM: {model}, interview type: {interview_type}")
     # Select a random topic or difficulty if not provided
     topic = topic or random.choice(topic_lists[interview_type])
     difficulty = difficulty or random.choice(["easy", "medium", "hard"])
@@ -46,7 +67,6 @@ def complete_interview(interview_type, exp_name, llm_config=None, requirements="
             "average_response_time_seconds": 0,
         },
     )
-
     # Initialize interviewer and candidate messages
     messages_interviewer = llm.init_bot(problem_statement_text, interview_type)
     chat_display = [[None, fixed_messages["start"]]]
@@ -82,7 +102,6 @@ def complete_interview(interview_type, exp_name, llm_config=None, requirements="
 
         chat_display.append([candidate_message, None])
 
-        # Check if the interview should finish
         if response_json.get("finished") and not response_json.get("question"):
             break
 

tests/grader.py

@@ -1,27 +1,23 @@
 import json
-
+from typing import Dict, Any, List
 from openai import OpenAI
-
 from tests.testing_prompts import grader_prompt
 
 
-def grade(json_file_path, model="gpt-4o", suffix=""):
-    client = OpenAI(base_url="https://api.openai.com/v1")
+def grade(json_file_path: str, model: str = "gpt-4o", suffix: str = "") -> Dict[str, Any]:
+    """
+    Grade the interview data and provide feedback.
 
+    :param json_file_path: Path to the JSON file containing interview data.
+    :param model: Model to use for grading.
+    :param suffix: Suffix to add to the feedback file name.
+    :return: Feedback dictionary.
+    """
+    client = OpenAI(base_url="https://api.openai.com/v1")
     with open(json_file_path) as file:
         interview_data = json.load(file)
 
-    interview_summary_list = []
-    interview_summary_list.append(f"Interview type: {interview_data['inputs']['interview_type']}")
-    interview_summary_list.append(f"Interview difficulty: {interview_data['inputs']['difficulty']}")
-    interview_summary_list.append(f"Interview topic: {interview_data['inputs']['topic']}")
-    if interview_data["inputs"]["requirements"] != "":
-        interview_summary_list.append(f"Interview requirements: {interview_data['inputs']['requirements']}")
-    interview_summary_list.append(f"Problem statement proposed by interviewer: {interview_data['problem_statement']}")
-    interview_summary_list.append(f"\nTranscript of the whole interview below:")
-    interview_summary_list += interview_data["transcript"]
-    interview_summary_list.append(f"\nTHE MAIN PART OF THE INTERVIEW ENDED HERE.")
-    interview_summary_list.append(f"Feedback provided by interviewer: {interview_data['feedback']}")
+    interview_summary_list = generate_interview_summary(interview_data)
 
     messages = [
         {"role": "system", "content": grader_prompt},
@@ -31,25 +27,81 @@ def grade(json_file_path, model="gpt-4o", suffix=""):
     response = client.chat.completions.create(model=model, messages=messages, temperature=0, response_format={"type": "json_object"})
     feedback = json.loads(response.choices[0].message.content)
 
-    feedback["file_name"] = json_file_path
-    feedback["agent_llm"] = interview_data["interviewer_llm"]
-    feedback["candidate_llm"] = interview_data["candidate_llm"]
-    feedback["grader_model"] = model
-    feedback["type"] = interview_data["inputs"]["interview_type"]
-    feedback["difficulty"] = interview_data["inputs"]["difficulty"]
-    feedback["topic"] = interview_data["inputs"]["topic"]
-    feedback["average_response_time_seconds"] = interview_data["average_response_time_seconds"]
-    feedback["number_of_messages"] = len(interview_data["transcript"])
+    populate_feedback_metadata(feedback, json_file_path, interview_data, model)
+    calculate_overall_score(feedback)
+
+    save_feedback(json_file_path, feedback, suffix)
+
+    return feedback
+
+
+def generate_interview_summary(interview_data: Dict[str, Any]) -> List[str]:
+    """
+    Generate a summary of the interview data.
+
+    :param interview_data: Dictionary containing interview data.
+    :return: List of summary strings.
+    """
+    summary = [
+        f"Interview type: {interview_data['inputs']['interview_type']}",
+        f"Interview difficulty: {interview_data['inputs']['difficulty']}",
+        f"Interview topic: {interview_data['inputs']['topic']}",
+    ]
+    if interview_data["inputs"]["requirements"]:
+        summary.append(f"Interview requirements: {interview_data['inputs']['requirements']}")
+    summary.append(f"Problem statement proposed by interviewer: {interview_data['problem_statement']}")
+    summary.append(f"\nTranscript of the whole interview below:")
+    summary += interview_data["transcript"]
+    summary.append(f"\nTHE MAIN PART OF THE INTERVIEW ENDED HERE.")
+    summary.append(f"Feedback provided by interviewer: {interview_data['feedback']}")
+    return summary
+
+
+def populate_feedback_metadata(feedback: Dict[str, Any], json_file_path: str, interview_data: Dict[str, Any], model: str) -> None:
+    """
+    Populate feedback metadata with interview details.
+
+    :param feedback: Feedback dictionary to populate.
+    :param json_file_path: Path to the JSON file containing interview data.
+    :param interview_data: Dictionary containing interview data.
+    :param model: Model used for grading.
+    """
+    feedback.update(
+        {
+            "file_name": json_file_path,
+            "agent_llm": interview_data["interviewer_llm"],
+            "candidate_llm": interview_data["candidate_llm"],
+            "grader_model": model,
+            "type": interview_data["inputs"]["interview_type"],
+            "difficulty": interview_data["inputs"]["difficulty"],
+            "topic": interview_data["inputs"]["topic"],
+            "average_response_time_seconds": interview_data["average_response_time_seconds"],
+            "number_of_messages": len(interview_data["transcript"]),
+        }
+    )
+
 
+def calculate_overall_score(feedback: Dict[str, Any]) -> None:
+    """
+    Calculate the overall score from the feedback.
+
+    :param feedback: Feedback dictionary containing scores.
+    """
     scores = [
-        feedback[x]
-        for x in feedback
-        if (x.startswith("interviewer_") or x.startswith("feedback_") or x.startswith("problem_")) and feedback[x] is not None
+        feedback[key]
+        for key in feedback
+        if (key.startswith("interviewer_") or key.startswith("feedback_") or key.startswith("problem_")) and feedback[key] is not None
     ]
     feedback["overall_score"] = sum(scores) / len(scores)
 
-    # save results to json file in the same folder as the interview data
+
+def save_feedback(json_file_path: str, feedback: Dict[str, Any], suffix: str) -> None:
+    """
+    Save the feedback to a JSON file.
+
+    :param json_file_path: Path to the original JSON file.
+    :param feedback: Feedback dictionary to save.
+    :param suffix: Suffix to add to the feedback file name.
+    """
     with open(json_file_path.replace(".json", f"_feedback_{suffix}.json"), "w") as file:
         json.dump(feedback, file, indent=4)
-
-    return feedback

tests/test_e2e.py

@@ -2,21 +2,31 @@ from tests.candidate import complete_interview
 from tests.grader import grade
 from concurrent.futures import ThreadPoolExecutor
 
+from typing import List
 
-def complete_and_grade_interview(interview_type):
+
+def complete_and_grade_interview(interview_type: str) -> float:
+    """
+    Complete an interview and return the overall score.
+
+    :param interview_type: Type of the interview.
+    :return: Overall score of the interview.
+    """
     file_path, _ = complete_interview(interview_type, "test", model="gpt-3.5-turbo")
     feedback = grade(file_path, model="gpt-4o")
     assert feedback["overall_score"] > 0.4
     return feedback["overall_score"]
 
 
-def test_complete_interview():
+def test_complete_interview() -> None:
+    """
+    Test the complete interview process for various interview types.
+    """
     interview_types = ["ml_design", "math", "ml_theory", "system_design", "sql", "coding"]
-    scores = []
+    scores: List[float] = []
 
     with ThreadPoolExecutor(max_workers=3) as executor:
         futures = [executor.submit(complete_and_grade_interview, it) for it in interview_types]
-
         for future in futures:
             score = future.result()
             scores.append(score)

tests/test_models.py

@@ -13,15 +13,23 @@ def app_config():
     return Config()
 
 
-def test_llm_connection(app_config):
+def test_llm_connection(app_config: Config):
+    """
+    Test the connection and streaming capability of the LLM.
+
+    :param app_config: Configuration object.
+    """
     llm = LLMManager(app_config, {})
-    status = llm.status
-    streaming = llm.streaming
-    assert status, "LLM connection failed - status check failed"
-    assert streaming, "LLM streaming failed - streaming check failed"
+    assert llm.status, "LLM connection failed - status check failed"
+    assert llm.streaming, "LLM streaming failed - streaming check failed"
+
 
+def test_stt_connection(app_config: Config):
+    """
+    Test the connection and streaming capability of the STT.
 
-def test_stt_connection(app_config):
+    :param app_config: Configuration object.
+    """
     stt = STTManager(app_config)
     status = stt.status
     streaming = stt.streaming
@@ -29,7 +37,12 @@ def test_stt_connection(app_config):
     assert streaming, "STT streaming failed - streaming check failed"
 
 
-def test_tts_connection(app_config):
+def test_tts_connection(app_config: Config):
+    """
+    Test the connection and streaming capability of the TTS.
+
+    :param app_config: Configuration object.
+    """
     tts = TTSManager(app_config)
     status = tts.status
     streaming = tts.streaming

ui/coding.py

@@ -93,6 +93,7 @@ def get_problem_solving_ui(llm, tts, stt, default_audio_params, audio_output, na
         with gr.Accordion("Feedback", open=True) as feedback_acc:
             feedback = gr.Markdown(elem_id=f"{interview_type}_feedback")
 
+        # Start button click action chain
         start_btn.click(fn=add_interviewer_message(fixed_messages["start"]), inputs=[chat], outputs=[chat]).success(
             fn=lambda: True, outputs=[started_coding]
         ).success(fn=tts.read_last_message, inputs=[chat], outputs=[audio_output]).success(

utils/config.py

@@ -1,19 +1,29 @@
-import os
-
 from dotenv import load_dotenv
+import os
+from typing import Optional
 
 
 class ServiceConfig:
-    def __init__(self, url_var, type_var, name_var):
-        self.url = os.getenv(url_var)
-        self.type = os.getenv(type_var)
-        self.name = os.getenv(name_var)
-        self.key = os.getenv(f"{self.type}_KEY")
+    def __init__(self, url_var: str, type_var: str, name_var: str):
+        """
+        Initialize the ServiceConfig with environment variables.
+
+        :param url_var: Environment variable for the service URL.
+        :param type_var: Environment variable for the service type.
+        :param name_var: Environment variable for the service name.
+        """
+        self.url: Optional[str] = os.getenv(url_var)
+        self.type: Optional[str] = os.getenv(type_var)
+        self.name: Optional[str] = os.getenv(name_var)
+        self.key: Optional[str] = os.getenv(f"{self.type}_KEY")
 
 
 class Config:
     def __init__(self):
+        """
+        Load environment variables and initialize service configurations.
+        """
         load_dotenv(override=True)
-        self.llm = ServiceConfig("LLM_URL", "LLM_TYPE", "LLM_NAME")
-        self.stt = ServiceConfig("STT_URL", "STT_TYPE", "STT_NAME")
-        self.tts = ServiceConfig("TTS_URL", "TTS_TYPE", "TTS_NAME")
+        self.llm: ServiceConfig = ServiceConfig("LLM_URL", "LLM_TYPE", "LLM_NAME")
+        self.stt: ServiceConfig = ServiceConfig("STT_URL", "STT_TYPE", "STT_NAME")
+        self.tts: ServiceConfig = ServiceConfig("TTS_URL", "TTS_TYPE", "TTS_NAME")