diff --git a/.env.example b/.env.example index 9847a1d..12138c1 100644 --- a/.env.example +++ b/.env.example @@ -1 +1,12 @@ -OPENAI_API_KEY= \ No newline at end of file +# Provider: "groq" (default) or "openai" +LLM_PROVIDER=groq + +# Groq — https://console.groq.com/keys (free tier available) +GROQ_API_KEY=your_groq_api_key_here + +# OpenAI — only needed if LLM_PROVIDER=openai +# OPENAI_API_KEY=your_openai_api_key_here + +# Audio source default: mic | system | both +# Use "system" to capture Teams/Zoom interviewer audio (requires Stereo Mix or VB-Cable) +AUDIO_SOURCE=system diff --git a/README.md b/README.md index 82aecba..4c67f96 100644 --- a/README.md +++ b/README.md @@ -11,15 +11,16 @@ Hack Interview application is a tool designed to assist in job interviews using ## Features - **Real-Time Audio Processing**: Records and transcribes audio seamlessly. -- **Voice Recognition**: Uses OpenAI's Whisper model for accurate voice recognition. -- **Intelligent Response Generation**: Leverages OpenAI's GPT models for generating concise and relevant answers. +- **Voice Recognition**: Uses Groq Whisper (or OpenAI Whisper) for accurate voice recognition. +- **Intelligent Response Generation**: Uses Groq Llama models (or OpenAI GPT) for concise and relevant answers. - **Cross-Platform Functionality**: Designed to work on various operating systems. - **User-Friendly Interface**: Simple, intuitive and hideous GUI for easy interaction. ## Requirements - **Python 3.10+**: Ensure Python is installed on your system. -- **OpenAI API Key**: To use OpenAI's GPT models, you will need an API key. +- **Groq API Key** (recommended): Free tier at [console.groq.com](https://console.groq.com/keys). Used for Whisper transcription and Llama answers by default. +- **OpenAI API Key** (optional): Only if you set `LLM_PROVIDER=openai` in `.env`. - **BlackHole for MacOS**: An essential tool for recording your computer's audio output (e.g. from Zoom calls or browser tabs). Microphone Fallback: If BlackHole isn't installed or properly configured, the application can still function by recording your microphone input. ## Installation @@ -40,15 +41,47 @@ Hack Interview application is a tool designed to assist in job interviews using 3. **BlackHole**: If using MacOS, install [BlackHole](https://github.com/ExistentialAudio/BlackHole) and set up a [Multi Output Device](https://github.com/ExistentialAudio/BlackHole/wiki/Multi-Output-Device) 4. **Environment Setup**: - - Add your OpenAI API key to the `.env` file. If you don't have one, you can get it [here](https://platform.openai.com/api-keys). + - Copy `.env.example` to `.env`. + - Add your Groq API key (`GROQ_API_KEY`) from [console.groq.com](https://console.groq.com/keys). + - Default provider is Groq (`LLM_PROVIDER=groq`). For OpenAI instead, set `LLM_PROVIDER=openai` and add `OPENAI_API_KEY`. ## Usage - **Starting the Application**: Run `python main.py` to launch the GUI. -- *(optional)* **Setup**: You can choose the OpenAI model to use for response generation and the position you are being interviewed for. The default settings are set in the `src/config.py` file. -- **Recording**: Press `R` or click the big red toggle button to start/stop audio recording. It will create a `recording.wav` file in the project directory. -- **Transcription and Response Generation**: Press `A` or click the 'Analyze' button to transcribe the recorded audio and generate answers. -- **Viewing Responses**: Responses are displayed in the GUI, offering both a quick and detailed answer. +- **Audio source**: In the GUI, set **Audio** to **System (Teams/Zoom)** for live calls (default). Use **Microphone** only when testing alone. +- **Recording**: Press `R` to start, speak or let the interviewer ask a question, press `R` again to stop. Saves `record.wav` in the project folder. +- **Transcription**: Press `A` to transcribe and generate short/full answers. + +## Capture Microsoft Teams / Zoom audio (Windows) + +Your **microphone only hears you**, not the interviewer. Teams plays their voice through **speakers/headphones** (system audio). Use one of these setups: + +### Option A — Stereo Mix (free, if your PC supports it) + +1. Right-click the **speaker icon** → **Sound settings** → **More sound settings**. +2. Open the **Recording** tab → right-click empty area → check **Show disabled devices**. +3. Enable **Stereo Mix** → Set as **Default Device** (or note its name). +4. In the app, set **Audio** to **System (Teams/Zoom)**. +5. In Teams, use your normal **speakers or headset** for call audio. + +### Option B — VB-Audio Virtual Cable (works on most PCs) + +1. Install [VB-Audio Virtual Cable](https://vb-audio.com/Cable/) (free). +2. In **Teams** → Settings → **Devices** → set **Speaker** to **CABLE Input**. +3. In **Windows Sound** → **Recording**, enable **CABLE Output** as default (or the app will detect it automatically). +4. Listen on your headset via Teams **Test call** or use **Listen to this device** on CABLE Output if you need to hear the call. + +### During the interview + +1. Set **Audio** → **System (Teams/Zoom)**. +2. When the interviewer asks a question, press **R** (record) → wait for the question → press **R** (stop). +3. Press **A** to transcribe and get answers. + +Use **Both** if you want system audio plus your mic in one recording. + +### macOS + +Install [BlackHole](https://github.com/ExistentialAudio/BlackHole), create a **Multi-Output Device** that includes BlackHole + your headphones, and set Teams output to that device. Set **Audio** to **System (Teams/Zoom)**. ## Contributions diff --git a/audio.py b/audio.py new file mode 100644 index 0000000..f7b3d83 --- /dev/null +++ b/audio.py @@ -0,0 +1,268 @@ +import sys +import threading +from typing import Any, Dict, List, Optional, Tuple + +import numpy as np +import PySimpleGUI as sg +import sounddevice as sd +import soundfile as sf +from loguru import logger + +from src.config import OUTPUT_FILE_NAME, SAMPLE_RATE + +_PREFERRED_HOSTAPIS = ("MME", "Windows DirectSound", "Windows WASAPI") + +# Devices that capture speaker / call audio (not a physical mic). +_SYSTEM_DEVICE_KEYWORDS = ( + "blackhole", + "stereo mix", + "stereomix", + "what u hear", + "wave out", + "loopback", + "cable output", + "vb-audio", + "vb audio", + "virtual cable", + "soundflower", +) + + +def _hostapi_name(device: Dict[str, Any]) -> str: + return sd.query_hostapis()[device["hostapi"]]["name"] + + +def _is_system_device(name: str) -> bool: + lower = name.lower() + return any(keyword in lower for keyword in _SYSTEM_DEVICE_KEYWORDS) + + +def _device_priority(device_id: int) -> int: + """Lower is better. Prefer stable host APIs on Windows.""" + api = _hostapi_name(sd.query_devices(device_id)) + if sys.platform == "win32": + order = {name: i for i, name in enumerate(_PREFERRED_HOSTAPIS)} + return order.get(api, len(_PREFERRED_HOSTAPIS)) + return 0 + + +def find_system_device_id() -> Optional[int]: + """ + Find a device that records system/call audio (Teams, Zoom, browser). + + Windows: enable Stereo Mix, or install VB-Audio Cable (see README). + macOS: install BlackHole and route call audio to it. + """ + candidates: List[Tuple[int, int]] = [] + for device_id, device in enumerate(sd.query_devices()): + if device["max_input_channels"] < 1: + continue + if _is_system_device(device["name"]): + candidates.append((device_id, _device_priority(device_id))) + + if not candidates: + return None + + device_id = min(candidates, key=lambda item: item[1])[0] + device = sd.query_devices(device_id) + logger.debug( + f"Using system audio device: {device['name']} ({_hostapi_name(device)})" + ) + return device_id + + +def find_microphone_device_id() -> Optional[int]: + """Find a normal microphone, excluding virtual system-capture devices.""" + system_id = find_system_device_id() + candidates: List[Tuple[int, int]] = [] + + for device_id, device in enumerate(sd.query_devices()): + if device["max_input_channels"] < 1: + continue + if device_id == system_id: + continue + if _is_system_device(device["name"]): + continue + candidates.append((device_id, _device_priority(device_id))) + + if candidates: + device_id = min(candidates, key=lambda item: item[1])[0] + device = sd.query_devices(device_id) + logger.debug( + f"Using microphone: {device['name']} ({_hostapi_name(device)})" + ) + return device_id + + default_input = sd.default.device[0] + if default_input is not None and default_input >= 0: + device = sd.query_devices(default_input) + if not _is_system_device(device["name"]): + logger.debug(f"Using default input device: {device['name']}") + return default_input + + return None + + +def resolve_device_ids(audio_source: str) -> List[int]: + """Return device id(s) to record for the given source mode.""" + source = audio_source.lower() + mic_id = find_microphone_device_id() + system_id = find_system_device_id() + + if source == "mic": + return [mic_id] if mic_id is not None else [] + if source == "system": + return [system_id] if system_id is not None else [] + if source == "both": + ids = [] + if system_id is not None: + ids.append(system_id) + if mic_id is not None and mic_id not in ids: + ids.append(mic_id) + return ids + + logger.warning(f"Unknown audio source '{audio_source}', using microphone.") + return [mic_id] if mic_id is not None else [] + + +def _device_sample_rate(device_id: int) -> int: + info = sd.query_devices(device_id, "input") + rate = int(info["default_samplerate"]) + return rate if rate > 0 else SAMPLE_RATE + + +def _to_mono(audio_data: np.ndarray) -> np.ndarray: + if audio_data.ndim == 1: + return audio_data + return audio_data.mean(axis=1) + + +def _mix_tracks(tracks: List[np.ndarray]) -> np.ndarray: + if len(tracks) == 1: + return tracks[0] + min_len = min(track.shape[0] for track in tracks) + mono_tracks = [_to_mono(track[:min_len]) for track in tracks] + mixed = np.sum(mono_tracks, axis=0) + peak = np.max(np.abs(mixed)) + if peak > 1.0: + mixed = mixed / peak + return mixed + + +def _record_device( + device_id: int, + button: sg.Element, + frames: List[np.ndarray], + lock: threading.Lock, +) -> None: + samplerate = _device_sample_rate(device_id) + device_info = sd.query_devices(device_id, "input") + channels = min(int(device_info["max_input_channels"]), 2) + + def callback( + indata: np.ndarray, + frame_count: int, + time_info: Any, + status: sd.CallbackFlags, + ) -> None: + if status: + logger.warning(f"Audio stream status: {status}") + if button.metadata.state: + with lock: + frames.append(indata.copy()) + + with sd.InputStream( + samplerate=samplerate, + device=device_id, + channels=channels, + dtype="float32", + callback=callback, + blocksize=int(samplerate * 0.1), + ): + while button.metadata.state: + sd.sleep(100) + + +def record(button: sg.Element, audio_source: str = "system") -> None: + """ + Record audio while the record button is active. + + Args: + button: The record toggle button. + audio_source: "mic", "system", or "both". + """ + logger.debug(f"Recording (source={audio_source})...") + device_ids = resolve_device_ids(audio_source) + + if not device_ids: + if audio_source.lower() in ("system", "both"): + logger.error( + "No system audio device found. Enable Stereo Mix in Windows Sound " + "settings, or install VB-Audio Virtual Cable (see README)." + ) + else: + logger.error("No microphone found.") + return + + if audio_source.lower() in ("system", "both") and find_system_device_id() is None: + logger.warning( + "System audio device not found; only microphone will be used. " + "See README to capture Teams/Zoom audio." + ) + + all_frames: List[List[np.ndarray]] = [[] for _ in device_ids] + locks = [threading.Lock() for _ in device_ids] + errors: List[str] = [] + + def worker(index: int, device_id: int) -> None: + try: + _record_device(device_id, button, all_frames[index], locks[index]) + except Exception as e: + errors.append(str(e)) + logger.error(f"Recording error on device {device_id}: {e}") + + threads = [ + threading.Thread(target=worker, args=(i, device_id), daemon=True) + for i, device_id in enumerate(device_ids) + ] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + + if errors and not any(all_frames): + return + + tracks = [] + for frames in all_frames: + if frames: + tracks.append(np.concatenate(frames, axis=0)) + + if not tracks: + logger.warning("No audio recorded.") + return + + audio_data = _mix_tracks(tracks) + samplerate = _device_sample_rate(device_ids[0]) + save_audio_file(audio_data, samplerate=samplerate) + + +def save_audio_file( + audio_data: np.ndarray, + output_file_name: str = OUTPUT_FILE_NAME, + samplerate: int = SAMPLE_RATE, +) -> None: + """Save audio data to a WAV file.""" + if audio_data.ndim == 1: + data = audio_data + else: + data = audio_data + + sf.write( + file=output_file_name, + data=data, + samplerate=samplerate, + format="WAV", + subtype="PCM_16", + ) + logger.debug(f"Audio saved to: {output_file_name}...") diff --git a/config.py b/config.py new file mode 100644 index 0000000..4907c70 --- /dev/null +++ b/config.py @@ -0,0 +1,45 @@ +import os +from pathlib import Path + +APPLICATION_WIDTH = 85 +THEME = "DarkGray12" + +PROJECT_ROOT = Path(__file__).resolve().parent.parent +OUTPUT_FILE_NAME = str(PROJECT_ROOT / "record.wav") +SAMPLE_RATE = 48000 + +# "groq" or "openai" — set LLM_PROVIDER in .env +LLM_PROVIDER = os.getenv("LLM_PROVIDER", "groq").lower() + +OPENAI_MODELS = ["gpt-4o-mini", "gpt-4o", "gpt-4-turbo", "gpt-3.5-turbo"] +GROQ_MODELS = [ + "llama-3.3-70b-versatile", + "llama-3.1-8b-instant", + "mixtral-8x7b-32768", + "gemma2-9b-it", +] + +MODELS = GROQ_MODELS if LLM_PROVIDER == "groq" else OPENAI_MODELS +DEFAULT_MODEL = MODELS[0] + +OPENAI_WHISPER_MODEL = "whisper-1" +GROQ_WHISPER_MODEL = "whisper-large-v3-turbo" + +DEFAULT_POSITION = "Python Developer" + +# Audio: mic | system | both — system captures Teams/Zoom from speakers +AUDIO_SOURCE_OPTIONS = { + "Microphone": "mic", + "System (Teams/Zoom)": "system", + "Both": "both", +} +_env_source = os.getenv("AUDIO_SOURCE", "system").lower() +DEFAULT_AUDIO_SOURCE = ( + _env_source + if _env_source in ("mic", "system", "both") + else "system" +) +DEFAULT_AUDIO_SOURCE_LABEL = next( + (label for label, key in AUDIO_SOURCE_OPTIONS.items() if key == DEFAULT_AUDIO_SOURCE), + "System (Teams/Zoom)", +) diff --git a/gpt_query.py b/gpt_query.py new file mode 100644 index 0000000..f05f4dd --- /dev/null +++ b/gpt_query.py @@ -0,0 +1,126 @@ +from pathlib import Path +from typing import Any + +from dotenv import load_dotenv +from loguru import logger + +from src.config import ( + DEFAULT_MODEL, + DEFAULT_POSITION, + GROQ_WHISPER_MODEL, + LLM_PROVIDER, + OPENAI_WHISPER_MODEL, + OUTPUT_FILE_NAME, +) + +SYS_PREFIX: str = "You are interviewing for a " +SYS_SUFFIX: str = """ position. +You will receive an audio transcription of the question. It may not be complete. You need to understand the question and write an answer to it.\n +""" + +SHORT_INSTRUCTION: str = "Concisely respond, limiting your answer to 50 words." +LONG_INSTRUCTION: str = "Before answering, take a deep breath and think one step at a time. Believe the answer in no more than 150 words." + +load_dotenv() + +_client: Any = None + + +def _get_client() -> Any: + global _client + if _client is None: + if LLM_PROVIDER == "groq": + from groq import Groq + + _client = Groq() + else: + from openai import OpenAI + + _client = OpenAI() + return _client + + +def transcribe_audio(path_to_file: str = OUTPUT_FILE_NAME) -> str: + """ + Transcribe audio using Groq or OpenAI Whisper API. + + Args: + path_to_file (str, optional): Path to the audio file. Defaults to OUTPUT_FILE_NAME. + + Returns: + str: The audio transcription or an error message. + """ + whisper_model = ( + GROQ_WHISPER_MODEL if LLM_PROVIDER == "groq" else OPENAI_WHISPER_MODEL + ) + logger.debug( + f"Transcribing audio ({LLM_PROVIDER}, {whisper_model}) from: {path_to_file}..." + ) + + if not Path(path_to_file).is_file(): + message = ( + "No recording found. Press R to start recording, speak your question, " + "press R again to stop, then press A to transcribe." + ) + logger.error(message) + return message + + try: + with open(path_to_file, "rb") as audio_file: + transcript: str = _get_client().audio.transcriptions.create( + model=whisper_model, + file=audio_file, + response_format="text", + ) + except Exception as error: + message = f"Transcription failed: {error}" + logger.error(message) + return message + + logger.debug("Audio transcribed.") + print("Transcription:", transcript) + + return transcript + + +def generate_answer( + transcript: str, + short_answer: bool = True, + temperature: float = 0.7, + model: str = DEFAULT_MODEL, + position: str = DEFAULT_POSITION, +) -> str: + """ + Generate an answer using Groq or OpenAI chat completions. + + Args: + transcript (str): The audio transcription. + short_answer (bool, optional): Whether to generate a short answer. Defaults to True. + temperature (float, optional): The temperature to use. Defaults to 0.7. + model (str, optional): The model to use. Defaults to DEFAULT_MODEL. + position (str, optional): The position to use. Defaults to DEFAULT_POSITION. + + Returns: + str: The generated answer or an error message. + """ + system_prompt: str = SYS_PREFIX + position + SYS_SUFFIX + if short_answer: + system_prompt += SHORT_INSTRUCTION + else: + system_prompt += LONG_INSTRUCTION + + try: + response = _get_client().chat.completions.create( + model=model, + temperature=temperature, + messages=[ + {"role": "system", "content": system_prompt}, + {"role": "user", "content": transcript}, + ], + ) + except Exception as error: + message = f"Answer generation failed: {error}" + logger.error(message) + return message + + return response.choices[0].message.content diff --git a/gui.py b/gui.py new file mode 100644 index 0000000..7d1b1db --- /dev/null +++ b/gui.py @@ -0,0 +1,314 @@ +from typing import List, Optional, Tuple, Union + +import PySimpleGUI as sg + +from src.button import GREY_BUTTON, OFF_IMAGE +from src.config import ( + APPLICATION_WIDTH, + AUDIO_SOURCE_OPTIONS, + DEFAULT_AUDIO_SOURCE_LABEL, + DEFAULT_MODEL, + MODELS, + THEME, +) + + +class BtnInfo: + """ + A class to store the state of a button. + """ + + def __init__(self, state: bool = False) -> None: + self.state: bool = state + + +def create_button( + key: str, + tooltip: str, + text: str = "", + image_data: str = None, + subsample: int = 1, + standard: bool = False, +) -> sg.Button: + """ + Create a button element with the given parameters. + + Args: + key (str): The key of the button. + tooltip (str): The tooltip of the button. + text (str, optional): The text of the button. Defaults to "". + image_data (str, optional): The image data of the button. Defaults to None. + subsample (int, optional): The subsample of the image. Defaults to 1. + standard (bool, optional): Whether to use the standard theme. Defaults to False. + + Returns: + sg.Button: The button element. + """ + if not standard: + theme_bg_color: str = sg.theme_background_color() + color = (theme_bg_color, theme_bg_color) + else: + color = None + + return sg.Button( + image_data=image_data, + key=key, + image_subsample=subsample, + border_width=0, + tooltip=tooltip, + button_color=color, + disabled_button_color=color, + metadata=BtnInfo(), + button_text=text, + ) + + +def create_text_area( + text: str = "", + size: Optional[Tuple[int, int]] = None, + key: str = "", + text_color: str = None, +) -> sg.Text: + """ + Create a text area element with the given parameters. + + Args: + text (str, optional): The text of the text area. Defaults to "". + size (Optional[Tuple[int, int]], optional): The size of the text area. Defaults to None. + key (str, optional): The key of the text area. Defaults to "". + text_color (str, optional): The color of the text. Defaults to None. + + Returns: + sg.Text: The text area element. + """ + return sg.Text( + text=text, + size=size, + key=key, + background_color=sg.theme_background_color(), + text_color=text_color, + expand_x=True, + expand_y=True, + ) + + +def name(name: str) -> sg.Text: + """ + Create a text element with spaces to the right. + + Args: + name (str): The name of the text element. + + Returns: + sg.Text: The text element. + """ + spaces: int = 15 - len(name) - 2 + return sg.Text( + name + " " * spaces, + ) + + +def create_frame( + layout: List[List[Union[sg.Element, sg.Element]]] = [[]], + title: str = "", + key: str = "", + border: int = 0, +) -> sg.Frame: + """ + Create a frame element with the given parameters. + + Args: + layout (List[List[Union[sg.Element, sg.ContainerElement]]], optional): The layout of the frame. Defaults to [[]]. + title (str, optional): The title of the frame. Defaults to "". + key (str, optional): The key of the frame. Defaults to "". + border (int, optional): The border width of the frame. Defaults to 0. + + Returns: + sg.Frame: The frame element. + """ + return sg.Frame( + title=title, + layout=layout, + key=key, + border_width=border, + expand_x=True, + expand_y=True, + ) + + +def create_column( + layout: List[List[Union[sg.Element, sg.Element]]] = [[]], key: str = "" +) -> sg.Column: + """ + Create a column element with the given parameters. + + Args: + layout (List[List[Union[sg.Element, sg.ContainerElement]]], optional): The layout of the column. Defaults to [[]]. + key (str, optional): The key of the column. Defaults to "". + + Returns: + sg.Column: The column element. + """ + return sg.Column( + layout=layout, + key=key, + expand_x=True, + expand_y=True, + ) + + +def build_layout() -> ( + List[List[Union[sg.Text, sg.Button, sg.Frame, sg.Combo, sg.Input]]] +): + """ + Build the layout of the application. + + Returns: + List[List[Union[sg.Text, sg.Button, sg.Frame, sg.Combo, sg.Input]]]: The layout of the application. + """ + # Create elements + record_button: sg.Button = create_button( + image_data=OFF_IMAGE, + tooltip="Start/Stop Recording", + key="-RECORD_BUTTON-", + ) + analyze_button: sg.Button = create_button( + image_data=GREY_BUTTON, + text="Analyze", + tooltip="Transcribe and Analyze", + key="-ANALYZE_BUTTON-", + subsample=2, + ) + close_button: sg.Button = create_button( + image_data=GREY_BUTTON, + text="Close", + tooltip="Exit the application", + key="-CLOSE_BUTTON-", + subsample=2, + ) + + transcribed_text: sg.Text = create_text_area( + size=(APPLICATION_WIDTH, 3), key="-TRANSCRIBED_TEXT-", text_color="white" + ) + quick_answer: sg.Text = create_text_area( + size=(APPLICATION_WIDTH, 7), key="-QUICK_ANSWER-", text_color="white" + ) + full_answer: sg.Text = create_text_area( + size=(APPLICATION_WIDTH, 20), key="-FULL_ANSWER-", text_color="white" + ) + + instructions: sg.Text = create_text_area( + size=(int(APPLICATION_WIDTH * 0.7), 2), + key="-INSTRUCTIONS-", + text="Press 'R' to start recording\nPress 'A' to transcribe the recording and provide answers", + ) + + model = sg.Combo( + MODELS, + default_value=DEFAULT_MODEL, + readonly=True, + k="-MODEL_COMBO-", + s=28, + tooltip="Select the model to use", + ) + position = sg.Input( + default_text="Python Developer", + k="-POSITION_INPUT-", + s=30, + tooltip="Enter the position you are applying for", + focus=False, + ) + audio_source = sg.Combo( + list(AUDIO_SOURCE_OPTIONS.keys()), + default_value=DEFAULT_AUDIO_SOURCE_LABEL, + readonly=True, + k="-AUDIO_SOURCE_COMBO-", + s=28, + tooltip="System = Teams/Zoom call audio. Mic = your microphone only.", + ) + + # Create frames + top_frame = create_frame( + layout=[ + [name("Model"), model], + [name("Position"), position], + [name("Audio"), audio_source], + ], + key="-TOP_FRAME-", + ) + instructions_frame = create_frame( + title="", + layout=[[instructions]], + key="-INSTRUCTIONS_FRAME-", + ) + buttons_frame = create_frame( + layout=[[record_button], [analyze_button]], + key="-BUTTONS_FRAME-", + ) + question_frame = create_frame( + title="Transcribed Question", + layout=[[transcribed_text]], + key="-QUESTION_FRAME-", + border=1, + ) + short_answer_frame = create_frame( + title="Short Answer", + layout=[[quick_answer]], + key="-SHORT_ANSWER_FRAME-", + border=1, + ) + full_answer_frame = create_frame( + title="Full Answer", layout=[[full_answer]], key="-FULL_ANSWER_FRAME-", border=1 + ) + close_button_frame = create_frame( + title="", + layout=[[close_button]], + key="-CLOSE_BUTTON_FRAME-", + ) + + # Create columns + col1 = create_column( + layout=[[instructions_frame], [top_frame]], + key="-COL1-", + ) + + col2 = create_column( + layout=[[buttons_frame]], + key="-COL2-", + ) + + col3 = create_column( + layout=[[question_frame], [short_answer_frame], [full_answer_frame]], + key="-COL3-", + ) + + col4 = create_column( + layout=[[close_button_frame]], + key="-COL4-", + ) + + layout = [[col1, col2], [col3], [col4]] + + return layout + + +def initialize_window() -> sg.Window: + """ + Initialize the application window. + + Returns: + sg.Window: The application window. + """ + sg.theme(THEME) + + layout: List[ + List[Union[sg.Text, sg.Button, sg.Frame, sg.Combo, sg.Input]] + ] = build_layout() + + return sg.Window( + "Interview", + layout, + return_keyboard_events=True, + use_default_focus=False, + resizable=True, + ) diff --git a/handlers.py b/handlers.py new file mode 100644 index 0000000..b397be6 --- /dev/null +++ b/handlers.py @@ -0,0 +1,169 @@ +from pathlib import Path +from typing import Any, Dict + +import PySimpleGUI as sg +from loguru import logger + +from src import audio, gpt_query +from src.button import OFF_IMAGE, ON_IMAGE +from src.config import AUDIO_SOURCE_OPTIONS, OUTPUT_FILE_NAME + + +def handle_events(window: sg.Window, event: str, values: Dict[str, Any]) -> None: + """ + Handle the events. Record audio, transcribe audio, generate quick and full answers. + + Args: + window (sg.Window): The window element. + event (str): The event. + values (Dict[str, Any]): The values of the window. + """ + # If the user is not focused on the position input, process the events + focused_element: sg.Element = window.find_element_with_focus() + if not focused_element or focused_element.Key != "-POSITION_INPUT-": + if event in ("r", "R", "-RECORD_BUTTON-"): + recording_event(window) + elif event in ("a", "A", "-ANALYZE_BUTTON-"): + transcribe_event(window) + + # If the user is focused on the position input + if event[:6] in ("Return", "Escape"): + window["-ANALYZE_BUTTON-"].set_focus() + + elif event == "-RECORDED-": + _recording_finished(window) + + # When the transcription is ready + elif event == "-WHISPER-": + answer_events(window, values) + + # When the quick answer is ready + elif event == "-QUICK_ANSWER-": + logger.debug("Quick answer generated.") + print("Quick answer:", values["-QUICK_ANSWER-"]) + window["-QUICK_ANSWER-"].update(values["-QUICK_ANSWER-"]) + + # When the full answer is ready + elif event == "-FULL_ANSWER-": + logger.debug("Full answer generated.") + print("Full answer:", values["-FULL_ANSWER-"]) + window["-FULL_ANSWER-"].update(values["-FULL_ANSWER-"]) + + +def _recording_finished(window: sg.Window) -> None: + """Notify user if recording failed (common when system audio is not configured).""" + if Path(OUTPUT_FILE_NAME).is_file(): + return + + label = window["-AUDIO_SOURCE_COMBO-"].get() + source = AUDIO_SOURCE_OPTIONS.get(label, "system") + transcribed_text: sg.Element = window["-TRANSCRIBED_TEXT-"] + + if source in ("system", "both"): + transcribed_text.update( + "No system audio captured. Enable Stereo Mix in Windows Sound settings, " + "or install VB-Audio Virtual Cable (see README). Then set Audio to " + "System (Teams/Zoom) and try again." + ) + else: + transcribed_text.update( + "No audio recorded. Check your microphone and try again." + ) + + +def recording_event(window: sg.Window) -> None: + """ + Handle the recording event. Record audio and update the record button. + + Args: + window (sg.Window): The window element. + """ + button: sg.Element = window["-RECORD_BUTTON-"] + button.metadata.state = not button.metadata.state + button.update(image_data=ON_IMAGE if button.metadata.state else OFF_IMAGE) + + # Record audio + if button.metadata.state: + label = window["-AUDIO_SOURCE_COMBO-"].get() + source = AUDIO_SOURCE_OPTIONS.get(label, "system") + window.perform_long_operation( + lambda: audio.record(button, source), "-RECORDED-" + ) + + +def transcribe_event(window: sg.Window) -> None: + """ + Handle the transcribe event. Transcribe audio and update the text area. + + Args: + window (sg.Window): The window element. + """ + transcribed_text: sg.Element = window["-TRANSCRIBED_TEXT-"] + record_button: sg.Element = window["-RECORD_BUTTON-"] + + if record_button.metadata.state: + transcribed_text.update( + "Still recording. Press R again to stop recording, then press A." + ) + return + + if not Path(OUTPUT_FILE_NAME).is_file(): + transcribed_text.update( + "No recording found. Press R to record, R again to stop, then press A." + ) + return + + transcribed_text.update("Transcribing audio...") + window.perform_long_operation(gpt_query.transcribe_audio, "-WHISPER-") + + +def answer_events(window: sg.Window, values: Dict[str, Any]) -> None: + """ + Handle the answer events. Generate quick and full answers and update the text areas. + + Args: + window (sg.Window): The window element. + values (Dict[str, Any]): The values of the window. + """ + transcribed_text: sg.Element = window["-TRANSCRIBED_TEXT-"] + quick_answer: sg.Element = window["-QUICK_ANSWER-"] + full_answer: sg.Element = window["-FULL_ANSWER-"] + + # Get audio transcript and update text area + audio_transcript: str = values["-WHISPER-"] + transcribed_text.update(audio_transcript) + + if audio_transcript.startswith(("No recording found", "Transcription failed")): + return + + # Get model and position + model: str = values["-MODEL_COMBO-"] + position: str = values["-POSITION_INPUT-"] + + # Generate quick answer + logger.debug("Generating quick answer...") + quick_answer.update("Generating quick answer...") + window.perform_long_operation( + lambda: gpt_query.generate_answer( + audio_transcript, + short_answer=True, + temperature=0, + model=model, + position=position, + ), + "-QUICK_ANSWER-", + ) + + # Generate full answer + logger.debug("Generating full answer...") + full_answer.update("Generating full answer...") + window.perform_long_operation( + lambda: gpt_query.generate_answer( + audio_transcript, + short_answer=False, + temperature=0.7, + model=model, + position=position, + ), + "-FULL_ANSWER-", + ) diff --git a/pyproject.toml b/pyproject.toml index 78df5b3..f7a312f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -10,6 +10,7 @@ packages = [ { include = "src" } ] [tool.poetry.dependencies] python = "^3.10" numpy = "^1.26.2" +groq = "^0.30.0" openai = "^1.2.4" loguru = "^0.7.2" sounddevice = "^0.4.6" diff --git a/requirements.txt b/requirements.txt index a7588a7..6560dad 100644 --- a/requirements.txt +++ b/requirements.txt @@ -11,15 +11,16 @@ httpx==0.25.1 ; python_version >= "3.10" and python_version < "4.0" idna==3.4 ; python_version >= "3.10" and python_version < "4.0" loguru==0.7.2 ; python_version >= "3.10" and python_version < "4.0" numpy==1.26.2 ; python_version >= "3.10" and python_version < "4.0" +groq>=0.30.0,<1.0.0 ; python_version >= "3.10" and python_version < "4.0" openai==1.3.3 ; python_version >= "3.10" and python_version < "4.0" pycparser==2.21 ; python_version >= "3.10" and python_version < "4.0" pydantic-core==2.14.3 ; python_version >= "3.10" and python_version < "4.0" pydantic==2.5.1 ; python_version >= "3.10" and python_version < "4.0" -pysimplegui==4.60.5 ; python_version >= "3.10" and python_version < "4.0" +pysimplegui==4.60.5.1 ; python_version >= "3.10" and python_version < "4.0" python-dotenv==1.0.0 ; python_version >= "3.10" and python_version < "4.0" sniffio==1.3.0 ; python_version >= "3.10" and python_version < "4.0" sounddevice==0.4.6 ; python_version >= "3.10" and python_version < "4.0" soundfile==0.12.1 ; python_version >= "3.10" and python_version < "4.0" tqdm==4.66.1 ; python_version >= "3.10" and python_version < "4.0" -typing-extensions==4.8.0 ; python_version >= "3.10" and python_version < "4.0" +typing-extensions>=4.10.0 ; python_version >= "3.10" and python_version < "4.0" win32-setctime==1.1.0 ; python_version >= "3.10" and python_version < "4.0" and sys_platform == "win32" diff --git a/src/config.py b/src/config.py index 6b000f0..79821a2 100644 --- a/src/config.py +++ b/src/config.py @@ -1,10 +1,37 @@ +import os + +from dotenv import load_dotenv + +load_dotenv() + APPLICATION_WIDTH = 85 THEME = "DarkGray12" OUTPUT_FILE_NAME = "record.wav" SAMPLE_RATE = 48000 -MODELS = ["gpt-4o-mini", "gpt-4o", "gpt-4-turbo", "gpt-3.5-turbo"] +GROQ_BASE_URL = "https://api.groq.com/openai/v1" + +_provider = os.getenv("LLM_PROVIDER", "groq").strip().lower() +LLM_PROVIDER = _provider if _provider in {"groq", "openai"} else "groq" + +PROVIDER_MODELS = { + "groq": [ + "llama-3.1-8b-instant", + "llama-3.3-70b-versatile", + "llama3-8b-8192", + "llama3-70b-8192", + ], + "openai": ["gpt-4o-mini", "gpt-4o", "gpt-4-turbo", "gpt-3.5-turbo"], +} + +PROVIDER_TRANSCRIPTION_MODELS = { + "groq": "whisper-large-v3-turbo", + "openai": "whisper-1", +} + +MODELS = PROVIDER_MODELS[LLM_PROVIDER] DEFAULT_MODEL = MODELS[0] +TRANSCRIPTION_MODEL = PROVIDER_TRANSCRIPTION_MODELS[LLM_PROVIDER] DEFAULT_POSITION = "Python Developer" diff --git a/src/gpt_query.py b/src/gpt_query.py index 7ee1e48..c4a6056 100644 --- a/src/gpt_query.py +++ b/src/gpt_query.py @@ -1,8 +1,17 @@ -from dotenv import load_dotenv +import os +from functools import lru_cache + from loguru import logger -from openai import ChatCompletion, OpenAI +from openai import OpenAI -from src.config import DEFAULT_MODEL, DEFAULT_POSITION, OUTPUT_FILE_NAME +from src.config import ( + DEFAULT_MODEL, + DEFAULT_POSITION, + GROQ_BASE_URL, + LLM_PROVIDER, + OUTPUT_FILE_NAME, + TRANSCRIPTION_MODEL, +) SYS_PREFIX: str = "You are interviewing for a " SYS_SUFFIX: str = """ position. @@ -12,14 +21,30 @@ SHORT_INSTRUCTION: str = "Concisely respond, limiting your answer to 50 words." LONG_INSTRUCTION: str = "Before answering, take a deep breath and think one step at a time. Believe the answer in no more than 150 words." -load_dotenv() -client: OpenAI = OpenAI() +def _get_required_env(name: str) -> str: + value = os.getenv(name, "").strip() + if not value: + raise RuntimeError( + f"{name} is not set. Add it to your .env file or set it in your PowerShell session." + ) + return value + + +@lru_cache(maxsize=1) +def get_client() -> OpenAI: + if LLM_PROVIDER == "groq": + return OpenAI( + api_key=_get_required_env("GROQ_API_KEY"), + base_url=GROQ_BASE_URL, + ) + + return OpenAI(api_key=_get_required_env("OPENAI_API_KEY")) def transcribe_audio(path_to_file: str = OUTPUT_FILE_NAME) -> str: """ - Transcribe audio from a file using the OpenAI Whisper API. + Transcribe audio from a file using the configured provider. Args: path_to_file (str, optional): Path to the audio file. Defaults to OUTPUT_FILE_NAME. @@ -28,11 +53,14 @@ def transcribe_audio(path_to_file: str = OUTPUT_FILE_NAME) -> str: str: The audio transcription. """ logger.debug(f"Transcribing audio from: {path_to_file}...") + client = get_client() with open(path_to_file, "rb") as audio_file: try: transcript: str = client.audio.transcriptions.create( - model="whisper-1", file=audio_file, response_format="text" + model=TRANSCRIPTION_MODEL, + file=audio_file, + response_format="text", ) except Exception as error: logger.error(f"Can't transcribe audio: {error}") @@ -52,7 +80,7 @@ def generate_answer( position: str = DEFAULT_POSITION, ) -> str: """ - Generate an answer to the question using the OpenAI API. + Generate an answer to the question using the configured provider. Args: transcript (str): The audio transcription. @@ -65,6 +93,7 @@ def generate_answer( str: The generated answer. """ # Generate system prompt + client = get_client() system_prompt: str = SYS_PREFIX + position + SYS_SUFFIX if short_answer: system_prompt += SHORT_INSTRUCTION @@ -73,7 +102,7 @@ def generate_answer( # Generate answer try: - response: ChatCompletion = client.chat.completions.create( + response = client.chat.completions.create( model=model, temperature=temperature, messages=[