The Ultimate Guide to Real-Time Voice AI
Master the pipeline of modern Voice Assistants. This documentation covers a safe, efficient, and hallucination-free architecture from microphone input to speaker output.
#FastAPI
#Whisper
#Llama3
#Python
1 Architecture Overview
User Voice Input → Microphone
↓
Speech-to-Text (Faster-Whisper)
↓
LLM Context Synthesis (Llama-3.3)
↓
Text-to-Speech (Tacotron2)
↓
Speaker Output (Low Latency)
This pipeline is designed for production efficiency, ensuring zero-hallucination and rapid response times.
The Tech Stack
- ✅ FastAPI – High-performance Async API
- ✅ Faster-Whisper – SOTA Speech-to-Text
- ✅ Groq / Llama-3 – Ultra-fast LLM Response
- ✅ Coqui TTS – Natural Voice Synthesis
Project Structure
app/ ├── frontend/ │ └── index.html ├── backend/ │ ├── server.py │ ├── stt.py │ ├── llm.py │ └── tts.py └── README.md
Frontend Implementation
The client-side handles audio recording and WebSocket communication.
frontend/index.html
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8" />
<meta name="viewport" content="width=device-width, initial-scale=1.0" />
<title>Realtime Voice AI</title>
<style>
body { background: #0f172a; color: white; font-family: Arial, sans-serif; margin: 0; }
.container { max-width: 800px; margin: 40px auto; padding: 24px; }
button { padding: 16px 30px; background: #06b6d4; border: none; color: white; border-radius: 12px; font-size: 18px; cursor: pointer; transition: 0.3s; }
button:hover { background: #0891b2; transform: scale(1.05); }
#messages { margin-top: 30px; white-space: pre-wrap; background: #111827; padding: 16px; border-radius: 12px; min-height: 200px; border: 1px solid #1e293b; }
</style>
</head>
<body>
<div class="container">
<h1>🎙️ Realtime Voice AI</h1>
<p>Speak for 4 seconds and get an AI voice response.</p>
<button id="startBtn">Start Talking</button>
<pre id="messages"></pre>
</div>
<script>
const ws = new WebSocket("wss://your-ngrok-url.app/ws");
const startBtn = document.getElementById("startBtn");
const messages = document.getElementById("messages");
let mediaRecorder;
function log(msg) {
messages.textContent += msg + "\n";
messages.scrollTop = messages.scrollHeight;
}
ws.onopen = () => log("Connected to backend.");
ws.onmessage = async (event) => {
if (typeof event.data === "string") {
if (event.data.startsWith("AI_TOKEN:")) {
const token = event.data.replace("AI_TOKEN:", "");
messages.textContent += token;
} else { log(event.data); }
} else {
const audioBlob = new Blob([event.data], { type: "audio/wav" });
const url = URL.createObjectURL(audioBlob);
const audio = new Audio(url);
await audio.play();
}
};
startBtn.onclick = async () => {
const stream = await navigator.mediaDevices.getUserMedia({ audio: true });
mediaRecorder = new MediaRecorder(stream, { mimeType: "audio/webm;codecs=opus" });
let chunks = [];
mediaRecorder.ondataavailable = (e) => chunks.push(e.data);
mediaRecorder.onstop = async () => {
const blob = new Blob(chunks, { type: "audio/webm;codecs=opus" });
chunks = [];
const buffer = await blob.arrayBuffer();
ws.send(buffer);
log("\nListening complete. Processing...\n");
};
mediaRecorder.start();
log("Recording for 4 seconds...");
setTimeout(() => {
mediaRecorder.stop();
stream.getTracks().forEach(track => track.stop());
}, 4000);
};
</script>
</body>
</html>Backend: FastAPI WebSocket Server
backend/server.py
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.middleware.cors import CORSMiddleware
from stt import transcribe_audio
from llm import stream_llm
from tts import generate_tts
app = FastAPI(title="Realtime Voice AI")
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
@app.websocket("/ws")
async def websocket_endpoint(ws: WebSocket):
await ws.accept()
print("Client connected")
try:
while True:
audio_bytes = await ws.receive_bytes()
print("Transcribing...")
text = await transcribe_audio(audio_bytes)
if not text:
await ws.send_text("USER: [No speech detected]")
continue
print("USER:", text)
await ws.send_text(f"USER: {text}")
llm_text = ""
async for token in stream_llm(text):
llm_text += token
await ws.send_text(f"AI_TOKEN:{token}")
await ws.send_text("AI_DONE")
print("Generating TTS...")
audio = await generate_tts(llm_text)
await ws.send_bytes(audio)
except WebSocketDisconnect:
print("Client disconnected")
except Exception as e:
print("Error:", e)
try: await ws.send_text(f"ERROR: {str(e)}")
except: passSpeech-to-Text Logic
stt.pyfrom faster_whisper import WhisperModel
import tempfile
import subprocess
import os
try:
import torch
DEVICE = "cuda" if torch.cuda.is_available() else "cpu"
COMPUTE_TYPE = "float16" if DEVICE == "cuda" else "int8"
except Exception:
DEVICE = "cpu"
COMPUTE_TYPE = "int8"
model = WhisperModel("base", device=DEVICE, compute_type=COMPUTE_TYPE)
async def transcribe_audio(audio_bytes: bytes) -> str:
with tempfile.NamedTemporaryFile(delete=False, suffix=".webm") as f:
f.write(audio_bytes)
input_path = f.name
wav_path = input_path.replace(".webm", ".wav")
try:
subprocess.run([
"ffmpeg", "-y", "-i", input_path,
"-ar", "16000", "-ac", "1", wav_path
], check=True, capture_output=True)
segments, _ = model.transcribe(wav_path, beam_size=1, vad_filter=True)
text = "".join(seg.text for seg in segments)
return text.strip()
finally:
for p in [input_path, wav_path]:
if os.path.exists(p): os.remove(p)LLM Synthesis (Groq Llama-3)
from openai import OpenAI
client = OpenAI(
base_url="https://api.groq.com/openai/v1",
api_key="your_api_key"
)
async def stream_llm(text: str):
response = client.chat.completions.create(
model="llama-3.3-70b-versatile",
stream=True,
messages=[
{"role": "system", "content": "You are a realtime AI voice assistant."},
{"role": "user", "content": text}
]
)
for chunk in response:
delta = chunk.choices[0].delta.content
if delta:
yield deltaVoice Synthesis (TTS)
from TTS.api import TTS
import tempfile
import os
import logging
import asyncio
logging.basicConfig(level=logging.INFO)
tts = TTS("tts_models/en/ljspeech/tacotron2-DDC")
async def generate_tts(text: str) -> bytes:
if not text.strip(): raise ValueError("Empty text")
with tempfile.NamedTemporaryFile(delete=False, suffix=".wav") as f:
path = f.name
try:
logging.info("Generating TTS")
await asyncio.to_thread(tts.tts_to_file, text=text, file_path=path)
with open(path, "rb") as audio:
return audio.read()
except Exception as e:
logging.exception("TTS failed")
raise e
finally:
if os.path.exists(path): os.remove(path)Elevating with RAG
Retrieval-Augmented Generation (RAG) is the gold standard for voice bots. It prevents hallucinations and allows your bot to answer based on your own documentation.
Pros & Cons of Non-RAG
- 🟢 Natural conversation style.
- 🟢 Fully self-contained.
- 🔴 Low coverage of specific data.
- 🔴 Requires retraining for updates.
- 🔴 Hallucinations on niche topics.
The RAG Advantage
- ✅ Never makes up answers outside content.
- ✅ Add new data in seconds (No Retraining).
- ✅ Handles complex, paraphrased queries.
Suggested Transition
- Store your knowledge (FAQs/Docs) in a Vector Database (Pinecone/Chroma).
- Use a Retriever to find the most relevant document before calling the LLM.
- Set FAQ content as High Priority in the system prompt.
- Combine RAG with fine-tuning only if you need a very specific "persona" or voice tone.