An enterprise-grade technical and behavioral interview platform. Ingests candidate resumes, shallow clones and inspects project Git codebases (AST, manifests, and architecture), grounds technical drilling in the candidate's actual source code, and conducts full-duplex spoken dialogue with real-time 11-dimension evaluation scorecards.
Notice: Talkative is currently in active prototype testing. When tested locally over public WAN endpoints, minute latency gaps (~450ms–750ms) occur between Speech-to-Text and Text-to-Speech roundtrips due to transatlantic WebSocket packet serialization and AudioWorklet ring-buffer alignment.
Upload candidate resume. The ingestion service extracts skills, experience history, and verifiable metrics.
Shallow clones and scans package manifests (package.json,
requirements.txt) to ground technical questions in actual code.
Synthesizes candidate resume achievements with GitHub AST code patterns to generate probing interview questions.
Probes the candidate's implementation of decoupled_session.py, specifically how the
AudioWorklet ring-buffer prevents audio underruns.
"You implemented faster-whisper on CUDA FP16. How did you handle DLL path resolution on Windows without crashing the ASGI loop?"
Evaluates Situation, Task, Action, and Result when dealing with Gemini Live API rate limits and network jitter.
"Tell me about a time when an external WebSocket API dropped connection mid-session. How did your telemetry tracker handle state recovery?"
Direct Gemini Live full-duplex audio vs Decoupled GPU Whisper + Edge-TTS latency comparison.
"Why did you decouple STT from Gemini Live rather than streaming raw audio straight to the cloud? What was the latency trade-off?"
Comprehensive multi-axial grading compiled from spoken transcripts, code defenses, and STAR communication.
"We implemented an AudioWorkletProcessor in JavaScript running on a dedicated audio rendering thread. When the Python HearingVAD detects candidate speech, an interruption packet is pushed over WebSocket, flushing the audio queue in under 250 milliseconds."
Precise understanding of the browser audio thread architecture. Demonstrated deep engineering
competency regarding thread isolation and WebAssembly ring buffer management. Matches repository
source code in audio_handler.js.
Overcoming direct Gemini Live audio limitations through a dedicated 3-tier local GPU architecture.
Runs faster-whisper on local NVIDIA CUDA (FP16). Captures verbatim candidate
speech including domain acronyms and filler words (*"um", "uh"*).
Maintains strict multi-turn conversational history while feeding resume facts and Git AST snippets to Gemini 2.5 Flash.
Streams high-fidelity neural voices via edge-tts, converted to 24kHz Linear
PCM with sub-250ms barge-in interruption capability.
Inspect verified Python and JavaScript audio orchestration components powering the platform.
import asyncio
import time
from faster_whisper import WhisperModel
import edge_tts
from google import genai
# Global singleton for GPU WhisperModel ensuring zero cold-load latency
def get_whisper_model() -> WhisperModel:
global _WHISPER_GPU_MODEL
if _WHISPER_GPU_MODEL is None:
_WHISPER_GPU_MODEL = WhisperModel(
"base.en",
device="cuda",
compute_type="float16"
)
return _WHISPER_GPU_MODEL
async def process_candidate_turn(pcm_audio_bytes: bytes, context: dict):
model = get_whisper_model()
# 1. Local GPU Transcription with technical vocabulary biasing
segments, info = model.transcribe(
pcm_audio_bytes,
beam_size=5,
initial_prompt="FastAPI, LangGraph, DuckDB, Pyodide, SDXL LoRA, AudioWorklet"
)
transcript = " ".join([s.text for s in segments])
# 2. Feed grounded turn into Gemini Flash with full candidate AST history
response = await client.aio.models.generate_content(
model="gemini-2.5-flash",
contents=[*context["history"], transcript]
)
# 3. Stream neural audio back to candidate speakers
communicate = edge_tts.Communicate(response.text, voice="en-US-AndrewMultilingualNeural")
async for chunk in communicate.stream():
if chunk["type"] == "audio":
await send_pcm_to_client(chunk["data"])class HearingEcosystem:
def __init__(self, energy_threshold=0.015, barge_in_window_ms=250):
self.threshold = energy_threshold
self.window_ms = barge_in_window_ms
self.consecutive_speech_frames = 0
self.is_interviewer_speaking = False
def evaluate_frame(self, pcm_chunk: bytes) -> bool:
"""Calculates RMS audio energy to identify instantaneous candidate interruption."""
samples = np.frombuffer(pcm_chunk, dtype=np.int16) / 32768.0
rms_energy = np.sqrt(np.mean(samples ** 2))
if rms_energy > self.threshold:
self.consecutive_speech_frames += 1
# Candidate interrupted while AI was speaking → Flush immediately
if self.is_interviewer_speaking and self.consecutive_speech_frames >= 3:
return True # Trigger BARGE_IN event
else:
self.consecutive_speech_frames = 0
return Falseclass TurnTelemetry(BaseModel):
session_id: str
turn_index: int
speaker: str
input_type: str = "voice"
ttfb_ms: Optional[float] = None # Time to first audio token
total_latency_ms: Optional[float] = None # Complete round-trip turn
audio_chunks_in: int = 0
audio_bytes_out: int = 0
speed_rating: str = "optimal" # ultra_fast (<400ms)
def finalize(self, status="success"):
if self.total_latency_ms < 400:
self.speed_rating = "ultra_fast"
elif self.total_latency_ms < 800:
self.speed_rating = "fast"
else:
self.speed_rating = "standard"class AudioStreamProcessor extends AudioWorkletProcessor {
constructor() {
super();
self.buffer = new Float32Array(2048);
self.bufferIndex = 0;
}
process(inputs, outputs, parameters) {
const input = inputs[0];
if (!input || !input[0]) return true;
const inputChannel = input[0];
// Downsample candidate mic stream to 16kHz Linear PCM
for (let i = 0; i < inputChannel.length; i++) {
const pcmSample = Math.max(-1, Math.min(1, inputChannel[i]));
const int16 = pcmSample < 0 ? pcmSample * 0x8000 : pcmSample * 0x7fff;
this.port.postMessage({ type: 'PCM_CHUNK', sample: int16 });
}
return true;
}
}
registerProcessor('audio-stream-processor', AudioStreamProcessor);