Interviewlary: Designing Role-Specific Voice Interview Practice

Building a voice-first AI interview practice platform presents a brutal engineering challenge: human conversational turn-taking requires an end-to-end response latency under 600 milliseconds. When candidates pause to think, the system must not interrupt prematurely; when they finish speaking, the system must acknowledge, evaluate, and stream speech response with zero noticeable delay.

At Interviewlary, we engineered a dedicated real-time audio pipeline combining bi-directional WebSockets, streaming speech-to-text (STT), low-latency LLM generation, and neural text-to-speech (TTS) streaming.

1. The Real-Time Audio Waterfall Breakdown

Traditional request-response REST architectures create cumulative latency bottlenecks. Below is our optimized streaming waterfall:

2. Production Code: WebSocket Audio Pipeline Handler

Below is our core FastAPI WebSocket event loop that handles binary audio frame buffering, speech interruption detection, and sentence-boundary TTS streaming:

# interviewlary/streaming/audio_bridge.py
from fastapi import WebSocket
import asyncio
from typing import AsyncGenerator

class InterviewAudioBridge:
    def __init__(self, websocket: WebSocket, session_id: str):
        self.ws = websocket
        self.session_id = session_id
        self.is_speaking = False
        self.audio_buffer = bytearray()

    async def handle_candidate_audio_stream(self):
        await self.ws.accept()
        try:
            while True:
                data = await self.ws.receive()
                if "bytes" in data:
                    pcm_chunk = data["bytes"]
                    # Interruption detection: cancel ongoing AI speech playback
                    if self.is_speaking and self._detect_barge_in(pcm_chunk):
                        await self.ws.send_json({"type": "INTERRUPT_PLAYBACK"})
                        self.is_speaking = False

                    # Stream to real-time STT engine
                    text_delta = await self.stt_client.push_audio(pcm_chunk)
                    if text_delta.is_final:
                        asyncio.create_task(self.dispatch_interviewer_response(text_delta.transcript))

        except Exception as e:
            await self.cleanup_session()

    async def dispatch_interviewer_response(self, candidate_transcript: str):
        self.is_speaking = True
        sentence_buffer = ""
        
        async for token in self.llm_client.stream_evaluation(candidate_transcript):
            sentence_buffer += token
            # Synthesize audio immediately upon reaching sentence boundary punctuation
            if any(sentence_buffer.endswith(punct) for punct in [".", "!", "?", "\n"]):
                audio_bytes = await self.tts_client.synthesize_stream(sentence_buffer.strip())
                await self.ws.send_bytes(audio_bytes)
                sentence_buffer = ""

3. The Multi-Metric Candidate Scoring Engine

During the interview, candidate responses are evaluated asynchronously against 5 rubric dimensions without introducing latency into the live conversational flow:

• Structural Coherence (STAR Method): Situation, Task, Action, Result completeness check.
• Domain Competency & Tech Depth: Accuracy of algorithmic complexity, architectural patterns, and terminology.
• Communication & Tone: Filler word frequency, pacing (words per minute), and conciseness.
• Actionable Feedback Generator: Concrete recommendations for re-phrasing responses to target senior-level expectations.