Build a Real-Time AI Voice Chatbot: FastAPI, Whisper & LLM + RAG

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: pass

Speech-to-Text Logic

stt.py

from 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 delta

Voice 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

1. Store your knowledge (FAQs/Docs) in a Vector Database (Pinecone/Chroma).
2. Use a Retriever to find the most relevant document before calling the LLM.
3. Set FAQ content as High Priority in the system prompt.
4. Combine RAG with fine-tuning only if you need a very specific "persona" or voice tone.

Visit Github →

Return to Machine Learning Main Page →

FastAPI Tutorial Main Page →