Spaces:
Configuration error
Configuration error
| #!/usr/bin/env python3 | |
| """ | |
| AI Mock Interviewer — 语音面试官 | |
| 功能: | |
| - 语音提问（OpenAI TTS） | |
| - 语音回答（Groq STT） | |
| - Claude 评估 + 追问 | |
| - 面试结束生成评分报告 | |
| 用法: | |
| python interviewer.py # 默认算法面试 | |
| python interviewer.py --type ml # ML面试 | |
| python interviewer.py --type system # 系统设计 | |
| python interviewer.py --type behavioral # 行为面试 | |
| python interviewer.py --lang zh # 中文面试 | |
| """ | |
| import anthropic | |
| import json | |
| import os | |
| import subprocess | |
| import sys | |
| import time | |
| import threading | |
| import tempfile | |
| import speech_recognition as sr | |
| # ── Config ── | |
| INTERVIEW_TYPES = { | |
| "algo": { | |
| "name": "Algorithm & Data Structures", | |
| "name_zh": "算法与数据结构", | |
| "system": """You are a senior technical interviewer at a top tech company (Google/Meta level). | |
| You are conducting a 30-minute algorithm and data structures interview. | |
| Rules: | |
| - Ask ONE question at a time | |
| - Start easy, increase difficulty based on candidate's performance | |
| - After the candidate answers, give brief feedback then ask a follow-up or new question | |
| - Ask about time/space complexity | |
| - If the candidate is stuck, give a hint | |
| - After 5-6 questions, wrap up with final feedback | |
| - Be professional but friendly | |
| - Keep responses concise (2-3 sentences max for feedback)""" | |
| }, | |
| "ml": { | |
| "name": "Machine Learning", | |
| "name_zh": "机器学习", | |
| "system": """You are a senior ML engineer conducting a machine learning interview. | |
| Cover: supervised/unsupervised learning, neural networks, transformers, training optimization, | |
| model evaluation, deployment at scale. | |
| Rules: | |
| - Ask ONE question at a time | |
| - Mix theory and practical experience questions | |
| - Ask follow-ups based on answers | |
| - After 5-6 questions, wrap up | |
| - Keep responses concise""" | |
| }, | |
| "system": { | |
| "name": "System Design", | |
| "name_zh": "系统设计", | |
| "system": """You are a senior architect conducting a system design interview. | |
| Pick a system to design (URL shortener, chat app, news feed, etc.) | |
| Rules: | |
| - Present the problem, let candidate drive the design | |
| - Ask clarifying questions about their choices | |
| - Probe on scalability, reliability, trade-offs | |
| - Discuss database choices, caching, load balancing | |
| - One main design problem for the full interview | |
| - Keep responses concise""" | |
| }, | |
| "behavioral": { | |
| "name": "Behavioral", | |
| "name_zh": "行为面试", | |
| "system": """You are an engineering manager conducting a behavioral interview. | |
| Use the STAR method to evaluate answers. | |
| Rules: | |
| - Ask about teamwork, conflict, leadership, failure, growth | |
| - Ask ONE question at a time | |
| - Follow up with "Tell me more about..." or "What would you do differently?" | |
| - After 5-6 questions, wrap up | |
| - Keep responses concise and encouraging""" | |
| }, | |
| } | |
| class AIInterviewer: | |
| def __init__(self, interview_type="algo", lang="en"): | |
| self.type = interview_type | |
| self.lang = lang | |
| self.config = INTERVIEW_TYPES[interview_type] | |
| self.conversation = [] | |
| self.scores = [] | |
| self.turn = 0 | |
| self.max_turns = 6 | |
| self.running = True | |
| # API clients | |
| self._init_clients() | |
| # Audio | |
| self.recognizer = sr.Recognizer() | |
| self.recognizer.energy_threshold = 300 | |
| self.recognizer.dynamic_energy_threshold = True | |
| self.recognizer.pause_threshold = 2.0 | |
| def _init_clients(self): | |
| """Initialize API clients.""" | |
| # Claude (via OAuth) | |
| token = self._read_keychain_token() | |
| if token: | |
| self.claude = anthropic.Anthropic( | |
| api_key=None, auth_token=token, | |
| default_headers={ | |
| "anthropic-beta": "claude-code-20250219,oauth-2025-04-20," | |
| "fine-grained-tool-streaming-2025-05-14," | |
| "interleaved-thinking-2025-05-14", | |
| "anthropic-dangerous-direct-browser-access": "true", | |
| "user-agent": "claude-cli/2.1.77", | |
| "x-app": "cli", | |
| }, | |
| ) | |
| else: | |
| self.claude = None | |
| # Groq (STT) | |
| self.groq_key = self._read_key("groq") | |
| # OpenAI (TTS) | |
| self.openai_key = self._read_key("openai") | |
| def _read_key(self, name): | |
| kf = os.path.expanduser(f"~/.{name}_key") | |
| if os.path.exists(kf): | |
| return open(kf).read().strip() | |
| return "" | |
| def _read_keychain_token(self): | |
| try: | |
| raw = subprocess.check_output( | |
| ["security", "find-generic-password", "-s", "Claude Code-credentials", "-w"], | |
| stderr=subprocess.DEVNULL, | |
| ).decode().strip() | |
| data = json.loads(raw) | |
| return data.get("claudeAiOauth", {}).get("accessToken", "") | |
| except Exception: | |
| return "" | |
| def speak(self, text): | |
| """TTS: speak the text aloud.""" | |
| if self.openai_key: | |
| self._speak_openai(text) | |
| else: | |
| self._speak_macos(text) | |
| def _speak_openai(self, text): | |
| """OpenAI TTS (most natural).""" | |
| try: | |
| import requests | |
| voice = "nova" if self.lang == "zh" else "shimmer" | |
| resp = requests.post( | |
| "https://api.openai.com/v1/audio/speech", | |
| headers={"Authorization": f"Bearer {self.openai_key}", "Content-Type": "application/json"}, | |
| json={"model": "tts-1-hd", "input": text, "voice": voice, "speed": 1.0, "response_format": "mp3"}, | |
| timeout=15, | |
| ) | |
| if resp.status_code == 200: | |
| tmp = tempfile.NamedTemporaryFile(suffix=".mp3", delete=False) | |
| tmp.write(resp.content) | |
| tmp.close() | |
| subprocess.run(["afplay", tmp.name], capture_output=True) | |
| os.unlink(tmp.name) | |
| return | |
| except Exception: | |
| pass | |
| self._speak_macos(text) | |
| def _speak_macos(self, text): | |
| """Fallback: macOS say command.""" | |
| voice = "Tingting" if self.lang == "zh" else "Samantha" | |
| subprocess.run(["say", "-v", voice, "-r", "190", text], capture_output=True) | |
| def listen(self) -> str: | |
| """STT: listen for user's spoken answer.""" | |
| print(" 🎙 Listening... (speak now, pause 2s to finish)") | |
| mic = sr.Microphone() | |
| with mic as source: | |
| self.recognizer.adjust_for_ambient_noise(source, duration=0.5) | |
| try: | |
| audio = self.recognizer.listen(source, timeout=30, phrase_time_limit=60) | |
| except sr.WaitTimeoutError: | |
| return "" | |
| # Transcribe with Groq (fastest) | |
| if self.groq_key: | |
| return self._transcribe_groq(audio) | |
| else: | |
| return self._transcribe_google(audio) | |
| def _transcribe_groq(self, audio) -> str: | |
| from groq import Groq | |
| client = Groq(api_key=self.groq_key) | |
| tmp = tempfile.NamedTemporaryFile(suffix=".wav", delete=False) | |
| tmp.write(audio.get_wav_data()) | |
| tmp.close() | |
| try: | |
| with open(tmp.name, "rb") as f: | |
| t = client.audio.transcriptions.create( | |
| file=("audio.wav", f), model="whisper-large-v3", | |
| response_format="text", temperature=0.0, | |
| ) | |
| return t.strip() if isinstance(t, str) else t.text.strip() | |
| finally: | |
| os.unlink(tmp.name) | |
| def _transcribe_google(self, audio) -> str: | |
| try: | |
| lang = "zh-CN" if self.lang == "zh" else "en-US" | |
| return self.recognizer.recognize_google(audio, language=lang) | |
| except Exception: | |
| return "" | |
| def ask_claude(self, user_message: str) -> str: | |
| """Send message to Claude and get response.""" | |
| if not self.claude: | |
| return "Error: Claude API not available" | |
| self.conversation.append({"role": "user", "content": user_message}) | |
| lang_instruction = "回答请用中文。" if self.lang == "zh" else "" | |
| system_blocks = [ | |
| {"type": "text", "text": "You are Claude Code, Anthropic's official CLI for Claude."}, | |
| {"type": "text", "text": self.config["system"] + "\n" + lang_instruction}, | |
| ] | |
| resp = self.claude.messages.create( | |
| model="claude-sonnet-4-6", | |
| max_tokens=500, | |
| system=system_blocks, | |
| messages=self.conversation, | |
| ) | |
| reply = resp.content[0].text | |
| self.conversation.append({"role": "assistant", "content": reply}) | |
| return reply | |
| def evaluate_answer(self, question: str, answer: str) -> dict: | |
| """Get Claude to evaluate the candidate's answer.""" | |
| eval_prompt = f"""Evaluate this interview answer on a scale of 1-10. | |
| Question: {question} | |
| Answer: {answer} | |
| Return JSON only: {{"score": N, "strengths": "...", "improvements": "..."}}""" | |
| self.conversation.append({"role": "user", "content": eval_prompt}) | |
| system_blocks = [ | |
| {"type": "text", "text": "You are Claude Code, Anthropic's official CLI for Claude."}, | |
| {"type": "text", "text": "You are an interview evaluator. Return only valid JSON."}, | |
| ] | |
| resp = self.claude.messages.create( | |
| model="claude-haiku-4-5", | |
| max_tokens=200, | |
| system=system_blocks, | |
| messages=[{"role": "user", "content": eval_prompt}], | |
| ) | |
| try: | |
| text = resp.content[0].text | |
| # Extract JSON from response | |
| if "{" in text: | |
| json_str = text[text.index("{"):text.rindex("}") + 1] | |
| return json.loads(json_str) | |
| except Exception: | |
| pass | |
| return {"score": 5, "strengths": "Good attempt", "improvements": "Could elaborate more"} | |
| def generate_report(self) -> str: | |
| """Generate final interview report.""" | |
| avg_score = sum(s["score"] for s in self.scores) / max(len(self.scores), 1) | |
| all_strengths = [s["strengths"] for s in self.scores] | |
| all_improvements = [s["improvements"] for s in self.scores] | |
| report = f""" | |
| {'='*60} | |
| INTERVIEW REPORT — {self.config['name']} | |
| {'='*60} | |
| Overall Score: {avg_score:.1f} / 10 | |
| Questions Asked: {len(self.scores)} | |
| Strengths: | |
| """ | |
| for i, s in enumerate(all_strengths, 1): | |
| report += f" {i}. {s}\n" | |
| report += "\n Areas for Improvement:\n" | |
| for i, s in enumerate(all_improvements, 1): | |
| report += f" {i}. {s}\n" | |
| report += f""" | |
| Individual Scores: | |
| """ | |
| for i, s in enumerate(self.scores, 1): | |
| report += f" Q{i}: {s['score']}/10\n" | |
| if avg_score >= 8: | |
| verdict = "STRONG HIRE ✅" | |
| elif avg_score >= 6: | |
| verdict = "HIRE (with reservations) ⚠️" | |
| elif avg_score >= 4: | |
| verdict = "NO HIRE (needs improvement) ❌" | |
| else: | |
| verdict = "STRONG NO HIRE ❌❌" | |
| report += f""" | |
| Verdict: {verdict} | |
| {'='*60} | |
| """ | |
| return report | |
| def run(self): | |
| """Main interview loop.""" | |
| type_name = self.config["name_zh"] if self.lang == "zh" else self.config["name"] | |
| print(f"\n{'='*60}") | |
| print(f" 🎤 AI Mock Interview — {type_name}") | |
| print(f" Language: {'中文' if self.lang == 'zh' else 'English'}") | |
| print(f" Questions: ~{self.max_turns}") | |
| print(f" TTS: {'OpenAI HD' if self.openai_key else 'macOS'}") | |
| print(f" STT: {'Groq' if self.groq_key else 'Google'}") | |
| print(f"{'='*60}\n") | |
| # Opening | |
| if self.lang == "zh": | |
| opening = "你好！欢迎参加今天的面试。我是你的面试官。准备好了吗？" | |
| else: | |
| opening = "Hello! Welcome to the interview. I'm your interviewer today. Are you ready to begin?" | |
| print(f" 🤖 Interviewer: {opening}") | |
| self.speak(opening) | |
| # Wait for ready | |
| print() | |
| ready = self.listen() | |
| print(f" 👤 You: {ready}") | |
| # Ask first question | |
| first_q = self.ask_claude("The candidate is ready. Ask your first interview question.") | |
| print(f"\n 🤖 Interviewer: {first_q}") | |
| self.speak(first_q) | |
| last_question = first_q | |
| # Interview loop | |
| while self.turn < self.max_turns and self.running: | |
| self.turn += 1 | |
| print(f"\n --- Question {self.turn}/{self.max_turns} ---") | |
| # Listen for answer | |
| answer = self.listen() | |
| if not answer: | |
| print(" (no response detected, skipping...)") | |
| continue | |
| print(f" 👤 You: {answer}") | |
| # Evaluate silently | |
| score = self.evaluate_answer(last_question, answer) | |
| self.scores.append(score) | |
| print(f" 📊 Score: {score['score']}/10") | |
| # Get next response from Claude (feedback + next question) | |
| response = self.ask_claude(answer) | |
| print(f"\n 🤖 Interviewer: {response}") | |
| self.speak(response) | |
| last_question = response | |
| # Wrap up | |
| if self.lang == "zh": | |
| closing = "面试到此结束。感谢你的参与！我会生成一份评估报告。" | |
| else: | |
| closing = "That concludes our interview. Thank you for your time! Let me generate your evaluation report." | |
| print(f"\n 🤖 Interviewer: {closing}") | |
| self.speak(closing) | |
| # Generate report | |
| report = self.generate_report() | |
| print(report) | |
| # Save report | |
| report_file = os.path.expanduser(f"~/ai-interviewer/reports/report_{int(time.time())}.txt") | |
| os.makedirs(os.path.dirname(report_file), exist_ok=True) | |
| with open(report_file, "w") as f: | |
| f.write(report) | |
| print(f" Report saved to: {report_file}") | |
| def main(): | |
| import argparse | |
| parser = argparse.ArgumentParser(description="AI Mock Interviewer") | |
| parser.add_argument("--type", choices=["algo", "ml", "system", "behavioral"], default="algo") | |
| parser.add_argument("--lang", choices=["en", "zh"], default="en") | |
| parser.add_argument("--turns", type=int, default=6) | |
| args = parser.parse_args() | |
| interviewer = AIInterviewer(interview_type=args.type, lang=args.lang) | |
| interviewer.max_turns = args.turns | |
| interviewer.run() | |
| if __name__ == "__main__": | |
| main() | |