Introduction to Voice Assistant Backends with LlamaIndex
Voice assistants have evolved from simple command-response systems into sophisticated conversational agents capable of understanding context, retrieving relevant information, and generating natural responses. At the heart of these systems lies a robust backend that orchestrates speech-to-text, language understanding, knowledge retrieval, and text-to-speech. LlamaIndex, a leading data framework for LLM applications, provides the perfect foundation for building such backends.
In this tutorial, you'll learn how to build a production-ready voice assistant backend using LlamaIndex. We'll cover everything from basic architecture to advanced retrieval strategies, real-time streaming, and deployment considerations. By the end, you'll have a complete system that can power voice-driven applications across web, mobile, and IoT platforms.
What Is a Voice Assistant Backend?
A voice assistant backend is the server-side infrastructure that processes voice inputs and generates intelligent responses. Unlike text-based chatbots, voice assistants must handle additional complexities: audio processing, latency constraints, and the ambiguity of spoken language. The backend typically consists of several interconnected components working in a pipeline.
Core Components
- Speech-to-Text (STT): Converts spoken audio into text transcriptions
- Natural Language Understanding (NLU): Extracts intent, entities, and context from transcribed text
- Knowledge Retrieval: Fetches relevant information from indexed data sources
- Response Generation: Produces natural language responses using LLMs
- Text-to-Speech (TTS): Converts generated text responses back into audio
- Session Management: Maintains conversation state and user context
Why LlamaIndex?
LlamaIndex excels at the knowledge retrieval and response generation layers. It provides powerful abstractions for connecting LLMs to your private data, whether that's documents, databases, APIs, or structured knowledge graphs. For voice assistants specifically, LlamaIndex offers several advantages: low-latency retrieval through optimized indexing, built-in support for streaming responses (critical for voice), conversation memory management, and a rich ecosystem of data connectors.
Architecture Overview
Before diving into code, let's understand the architecture we'll build. The system follows a pipeline pattern where each stage processes the output of the previous one. The key design principle is minimizing latency — voice users expect responses within seconds, not minutes.
┌─────────────┐ ┌──────────────┐ ┌─────────────────┐
│ Audio Input │───▶│ STT Engine │───▶│ LlamaIndex │
│ (WebSocket) │ │ (Whisper) │ │ Query Engine │
└─────────────┘ └──────────────┘ └────────┬────────┘
│
┌────────────────────────┘
▼
┌─────────────────┐ ┌──────────────┐ ┌─────────────┐
│ LLM Response │───▶│ TTS Engine │───▶│ Audio Output │
│ (Streaming) │ │ (ElevenLabs) │ │ (WebSocket) │
└─────────────────┘ └──────────────┘ └─────────────┘
Prerequisites and Setup
Let's start by setting up the project environment. You'll need Python 3.10 or later, and several API keys for the services we'll use.
Installing Dependencies
# Create a virtual environment
python -m venv voice-assistant-env
source voice-assistant-env/bin/activate # On Windows: voice-assistant-env\Scripts\activate
# Install core dependencies
pip install llama-index llama-index-core
pip install llama-index-llms-openai
pip install llama-index-embeddings-openai
pip install fastapi uvicorn
pip install python-dotenv
pip install openai-whisper
pip install websockets
pip install pydantic
# For document processing
pip install llama-index-readers-file
Environment Configuration
Create a .env file to manage your API keys securely:
# .env
OPENAI_API_KEY=sk-your-openai-api-key-here
ELEVENLABS_API_KEY=your-elevenlabs-api-key-here
VOICE_ID=your-preferred-voice-id
Project Structure
voice-assistant/
├── .env
├── requirements.txt
├── main.py # FastAPI application entry point
├── config.py # Configuration management
├── knowledge_base.py # LlamaIndex index setup
├── stt_engine.py # Speech-to-text module
├── tts_engine.py # Text-to-speech module
├── assistant.py # Core assistant logic
├── session_manager.py # Conversation state management
└── data/ # Knowledge base documents
├── faq.md
├── product_docs.txt
└── policies.pdf
Building the Knowledge Base with LlamaIndex
The knowledge base is the brain of your voice assistant. LlamaIndex makes it straightforward to ingest documents, create embeddings, and build query engines optimized for different use cases. Let's build a comprehensive knowledge base that supports both precise fact retrieval and broader conversational queries.
Basic Index Setup
# knowledge_base.py
import os
from llama_index.core import (
VectorStoreIndex,
SimpleDirectoryReader,
StorageContext,
load_index_from_storage,
Settings,
)
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.llms.openai import OpenAI
from llama_index.core.node_parser import SentenceSplitter
class KnowledgeBase:
def __init__(self, data_dir: str = "./data", persist_dir: str = "./storage"):
self.data_dir = data_dir
self.persist_dir = persist_dir
self._configure_settings()
self.index = self._load_or_build_index()
def _configure_settings(self):
"""Configure global LlamaIndex settings."""
Settings.llm = OpenAI(
model="gpt-4o-mini",
temperature=0.3,
max_tokens=512,
)
Settings.embed_model = OpenAIEmbedding(
model="text-embedding-3-small",
)
Settings.node_parser = SentenceSplitter(
chunk_size=512,
chunk_overlap=50,
)
def _load_or_build_index(self) -> VectorStoreIndex:
"""Load existing index from disk or build a new one."""
if os.path.exists(self.persist_dir):
print("Loading existing index from storage...")
storage_context = StorageContext.from_defaults(
persist_dir=self.persist_dir
)
return load_index_from_storage(storage_context)
print("Building new index from documents...")
documents = SimpleDirectoryReader(self.data_dir).load_data()
index = VectorStoreIndex.from_documents(documents)
index.storage_context.persist(persist_dir=self.persist_dir)
return index
def get_query_engine(self, **kwargs):
"""Create a query engine with custom configuration."""
return self.index.as_query_engine(
similarity_top_k=kwargs.get("similarity_top_k", 3),
streaming=kwargs.get("streaming", True),
**kwargs
)
def get_chat_engine(self, **kwargs):
"""Create a chat engine for conversational interactions."""
return self.index.as_chat_engine(
chat_mode=kwargs.get("chat_mode", "context"),
similarity_top_k=kwargs.get("similarity_top_k", 3),
streaming=kwargs.get("streaming", True),
**kwargs
)
Adding a Custom Knowledge Source
For a voice assistant, you often need structured data alongside unstructured documents. Let's add support for a custom data source — a product catalog stored as structured data.
# knowledge_base.py (continued)
from llama_index.core import Document
from llama_index.core.tools import QueryEngineTool, ToolMetadata
from llama_index.core.query_engine import SubQuestionQueryEngine
import json
class KnowledgeBase:
# ... previous code ...
def _load_product_catalog(self, catalog_path: str) -> list:
"""Load structured product data as LlamaIndex documents."""
with open(catalog_path, 'r') as f:
products = json.load(f)
documents = []
for product in products:
content = (
f"Product: {product['name']}\n"
f"Category: {product['category']}\n"
f"Price: ${product['price']}\n"
f"Description: {product['description']}\n"
f"Features: {', '.join(product.get('features', []))}\n"
f"In Stock: {product.get('in_stock', True)}\n"
)
documents.append(Document(text=content, metadata={
"source": "product_catalog",
"product_id": product.get("id"),
}))
return documents
def build_multi_source_index(self, catalog_path: str = None):
"""Build an index combining documents and structured data."""
documents = SimpleDirectoryReader(self.data_dir).load_data()
if catalog_path and os.path.exists(catalog_path):
catalog_docs = self._load_product_catalog(catalog_path)
documents.extend(catalog_docs)
print(f"Added {len(catalog_docs)} product documents")
index = VectorStoreIndex.from_documents(documents)
index.storage_context.persist(persist_dir=self.persist_dir)
return index
def get_router_query_engine(self):
"""Create a router query engine for multi-source queries."""
doc_engine = self.index.as_query_engine(similarity_top_k=3)
# Create a summary engine for high-level questions
summary_engine = self.index.as_query_engine(
response_mode="tree_summarize",
use_async=True,
)
router_engine = self.index.as_query_engine(
similarity_top_k=3,
response_mode="compact",
)
return router_engine
Implementing Speech-to-Text
Speech-to-text is the entry point of your voice assistant pipeline. We'll use OpenAI's Whisper model, which offers excellent accuracy across multiple languages and handles various accents well. For production, you might use cloud-based STT services for better scalability, but Whisper provides a solid local option.
# stt_engine.py
import whisper
import torch
import io
import wave
from typing import Optional
class STTEngine:
def __init__(self, model_size: str = "base"):
"""
Initialize the STT engine.
Args:
model_size: Whisper model size ('tiny', 'base', 'small', 'medium', 'large')
Smaller models are faster but less accurate.
"""
print(f"Loading Whisper model: {model_size}")
self.model = whisper.load_model(model_size)
self.device = "cuda" if torch.cuda.is_available() else "cpu"
print(f"Using device: {self.device}")
def transcribe(
self,
audio_bytes: bytes,
sample_rate: int = 16000,
language: Optional[str] = None
) -> dict:
"""
Transcribe audio bytes to text.
Args:
audio_bytes: Raw audio data in bytes
sample_rate: Audio sample rate (default 16000 Hz)
language: Language code (e.g., 'en', 'es'). None for auto-detection.
Returns:
Dictionary with 'text', 'language', and 'segments'
"""
# Convert bytes to numpy array
audio_array = self._bytes_to_array(audio_bytes, sample_rate)
# Transcribe with Whisper
options = {
"fp16": self.device == "cuda",
"language": language,
"task": "transcribe",
}
# Remove None values
options = {k: v for k, v in options.items() if v is not None}
result = self.model.transcribe(audio_array, **options)
return {
"text": result["text"].strip(),
"language": result.get("language", "unknown"),
"segments": result.get("segments", []),
"confidence": self._calculate_confidence(result),
}
def _bytes_to_array(self, audio_bytes: bytes, sample_rate: int):
"""Convert raw audio bytes to numpy array for Whisper."""
import numpy as np
# If it's a WAV file, parse it
if audio_bytes[:4] == b'RIFF':
with io.BytesIO(audio_bytes) as wav_buffer:
with wave.open(wav_buffer, 'rb') as wav_file:
frames = wav_file.readframes(wav_file.getnframes())
audio_array = np.frombuffer(frames, dtype=np.int16)
# Normalize to [-1, 1] range for Whisper
audio_array = audio_array.astype(np.float32) / 32768.0
return audio_array
# Otherwise, treat as raw 16-bit PCM
audio_array = np.frombuffer(audio_bytes, dtype=np.int16)
audio_array = audio_array.astype(np.float32) / 32768.0
return audio_array
def _calculate_confidence(self, result: dict) -> float:
"""Calculate average confidence from segments."""
segments = result.get("segments", [])
if not segments:
return 0.0
total_logprob = sum(s.get("avg_logprob", 0) for s in segments)
avg_logprob = total_logprob / len(segments)
# Convert log probability to approximate confidence
return float(min(max(2 ** (avg_logprob * 1.5), 0), 1))
Implementing Text-to-Speech
Text-to-speech converts your assistant's text responses back into natural-sounding audio. We'll use ElevenLabs for its high-quality neural voices, but we'll also include a fallback option using OpenAI's TTS API.
# tts_engine.py
import requests
import os
from typing import Optional, AsyncGenerator
import asyncio
class TTSEngine:
def __init__(self, provider: str = "elevenlabs"):
"""
Initialize the TTS engine.
Args:
provider: 'elevenlabs' or 'openai'
"""
self.provider = provider
self.elevenlabs_api_key = os.getenv("ELEVENLABS_API_KEY")
self.openai_api_key = os.getenv("OPENAI_API_KEY")
self.voice_id = os.getenv("VOICE_ID", "21m00Tcm4TlvDq8ikWAM")
# ElevenLabs API endpoints
self.elevenlabs_base = "https://api.elevenlabs.io/v1"
def synthesize(
self,
text: str,
voice_id: Optional[str] = None,
stability: float = 0.5,
clarity: float = 0.75
) -> bytes:
"""
Convert text to speech audio bytes.
Args:
text: Text to convert to speech
voice_id: ElevenLabs voice ID (uses default if None)
stability: Voice stability (0-1)
clarity: Voice clarity/similarity (0-1)
Returns:
Audio bytes in MP3 format
"""
if self.provider == "elevenlabs":
return self._synthesize_elevenlabs(
text, voice_id, stability, clarity
)
else:
return self._synthesize_openai(text)
def _synthesize_elevenlabs(
self,
text: str,
voice_id: Optional[str],
stability: float,
clarity: float
) -> bytes:
"""Synthesize speech using ElevenLabs API."""
voice = voice_id or self.voice_id
url = f"{self.elevenlabs_base}/text-to-speech/{voice}"
headers = {
"xi-api-key": self.elevenlabs_api_key,
"Content-Type": "application/json",
}
payload = {
"text": text,
"model_id": "eleven_turbo_v2",
"voice_settings": {
"stability": stability,
"similarity_boost": clarity,
"style": 0.0,
"use_speaker_boost": True,
},
}
response = requests.post(url, json=payload, headers=headers)
response.raise_for_status()
return response.content
def _synthesize_openai(self, text: str) -> bytes:
"""Synthesize speech using OpenAI TTS API as fallback."""
from openai import OpenAI
client = OpenAI(api_key=self.openai_api_key)
response = client.audio.speech.create(
model="tts-1",
voice="nova",
input=text,
response_format="mp3",
)
return response.content
async def stream_synthesize(
self,
text_chunks: AsyncGenerator[str, None],
voice_id: Optional[str] = None
) -> AsyncGenerator[bytes, None]:
"""
Stream TTS synthesis for real-time audio output.
Accumulates text chunks and synthesizes in segments.
"""
buffer = ""
sentence_endings = {'.', '!', '?'}
async for chunk in text_chunks:
buffer += chunk
# Synthesize when we have a complete sentence
while any(ending in buffer for ending in sentence_endings):
# Find the earliest sentence ending
positions = [
buffer.index(e) for e in sentence_endings if e in buffer
]
end_pos = min(positions) + 1
sentence = buffer[:end_pos].strip()
buffer = buffer[end_pos:]
if sentence:
audio = self.synthesize(sentence, voice_id)
yield audio
# Synthesize any remaining text
if buffer.strip():
audio = self.synthesize(buffer.strip(), voice_id)
yield audio
Building the Core Assistant Logic
Now we'll tie everything together. The assistant module orchestrates the STT, knowledge retrieval, LLM response generation, and TTS components. It also handles conversation context and streaming responses for low-latency voice interactions.
# assistant.py
from llama_index.core.chat_engine import ContextChatEngine
from llama_index.core.memory import ChatMemoryBuffer
from llama_index.core.prompts import ChatPromptTemplate
from typing import AsyncGenerator, Optional
import asyncio
# Custom system prompt for voice assistant behavior
VOICE_ASSISTANT_PROMPT = """\
You are a helpful voice assistant. Follow these guidelines:
1. Keep responses concise and conversational (2-4 sentences max).
2. Avoid markdown formatting, bullet points, or special characters.
3. Speak naturally as if talking to a person on the phone.
4. If you don't know something, say so clearly and offer alternatives.
5. Use context from previous turns to maintain conversation flow.
6. For numbers, spell them out when they might be confused (e.g., "two" vs "to").
7. Avoid abbreviations that don't sound natural when spoken.
Context information from knowledge base:
{context_str}
Current conversation:
{chat_history}
User: {query_str}
Assistant:"""
class VoiceAssistant:
def __init__(self, knowledge_base, stt_engine, tts_engine):
self.kb = knowledge_base
self.stt = stt_engine
self.tts = tts_engine
self.sessions = {} # session_id -> ChatEngine
def get_or_create_session(self, session_id: str) -> ContextChatEngine:
"""Get existing chat session or create a new one."""
if session_id not in self.sessions:
memory = ChatMemoryBuffer.from_defaults(
token_limit=3000 # Keep context manageable for voice
)
chat_prompt = ChatPromptTemplate.from_messages([
("system", VOICE_ASSISTANT_PROMPT),
])
chat_engine = self.kb.index.as_chat_engine(
chat_mode="context",
memory=memory,
system_prompt=VOICE_ASSISTANT_PROMPT,
similarity_top_k=3,
streaming=True,
verbose=False,
)
self.sessions[session_id] = chat_engine
return self.sessions[session_id]
async def process_voice_input(
self,
audio_bytes: bytes,
session_id: str,
language: Optional[str] = None
) -> dict:
"""
Process voice input end-to-end.
Args:
audio_bytes: Raw audio input
session_id: Unique session identifier
language: Optional language hint for STT
Returns:
Dictionary with transcription, response text, and audio
"""
# Step 1: Speech to Text
loop = asyncio.get_event_loop()
stt_result = await loop.run_in_executor(
None,
lambda: self.stt.transcribe(audio_bytes, language=language)
)
if not stt_result["text"]:
return {
"transcription": "",
"response": "I didn't catch that. Could you please repeat?",
"audio": self.tts.synthesize(
"I didn't catch that. Could you please repeat?"
),
"confidence": 0.0,
}
# Step 2: Generate response using LlamaIndex chat engine
chat_engine = self.get_or_create_session(session_id)
# Use streaming for faster first-token response
response_stream = chat_engine.stream_chat(stt_result["text"])
response_text = str(response_stream)
# Step 3: Text to Speech
audio_response = await loop.run_in_executor(
None,
lambda: self.tts.synthesize(response_text)
)
return {
"transcription": stt_result["text"],
"response": response_text,
"audio": audio_response,
"confidence": stt_result["confidence"],
"language": stt_result["language"],
}
async def process_voice_input_streaming(
self,
audio_bytes: bytes,
session_id: str,
language: Optional[str] = None
) -> AsyncGenerator[dict, None]:
"""
Process voice input with streaming response for minimal latency.
Yields intermediate results as they become available.
"""
loop = asyncio.get_event_loop()
# Stream STT result
stt_result = await loop.run_in_executor(
None,
lambda: self.stt.transcribe(audio_bytes, language=language)
)
yield {
"type": "transcription",
"text": stt_result["text"],
"confidence": stt_result["confidence"],
}
if not stt_result["text"]:
yield {
"type": "response_text",
"text": "I didn't catch that. Could you please repeat?",
}
audio = await loop.run_in_executor(
None,
lambda: self.tts.synthesize(
"I didn't catch that. Could you please repeat?"
)
)
yield {"type": "audio", "data": audio}
return
# Stream LLM response chunks
chat_engine = self.get_or_create_session(session_id)
response_stream = chat_engine.stream_chat(stt_result["text"])
async def text_generator():
for chunk in response_stream:
yield str(chunk)
await asyncio.sleep(0) # Yield control
# Stream TTS audio in parallel with text
async for audio_chunk in self.tts.stream_synthesize(text_generator()):
yield {"type": "audio", "data": audio_chunk}
def clear_session(self, session_id: str):
"""Clear conversation history for a session."""
if session_id in self.sessions:
del self.sessions[session_id]
def get_session_history(self, session_id: str) -> list:
"""Retrieve conversation history for a session."""
if session_id not in self.sessions:
return []
chat_engine = self.sessions[session_id]
memory = chat_engine.memory
messages = memory.get_all()
return [
{"role": msg.role, "content": str(msg.content)}
for msg in messages
]
Session Management
For a voice assistant, session management is critical. Users may pause, resume, or switch topics mid-conversation. A robust session manager tracks active sessions, handles timeouts, and persists conversation state.
# session_manager.py
import time
import uuid
from typing import Optional, Dict
from dataclasses import dataclass, field
from datetime import datetime, timedelta
import asyncio
@dataclass
class Session:
session_id: str
user_id: Optional[str] = None
created_at: datetime = field(default_factory=datetime.now)
last_active: datetime = field(default_factory=datetime.now)
metadata: dict = field(default_factory=dict)
is_active: bool = True
def touch(self):
"""Update last active timestamp."""
self.last_active = datetime.now()
def is_expired(self, timeout_minutes: int = 30) -> bool:
"""Check if session has expired."""
expiry = self.last_active + timedelta(minutes=timeout_minutes)
return datetime.now() > expiry
class SessionManager:
def __init__(self, timeout_minutes: int = 30):
self.sessions: Dict[str, Session] = {}
self.timeout_minutes = timeout_minutes
self._cleanup_task = None
def create_session(
self,
user_id: Optional[str] = None,
metadata: Optional[dict] = None
) -> Session:
"""Create a new session."""
session_id = str(uuid.uuid4())
session = Session(
session_id=session_id,
user_id=user_id,
metadata=metadata or {},
)
self.sessions[session_id] = session
return session
def get_session(self, session_id: str) -> Optional[Session]:
"""Get a session by ID, touching it if found."""
session = self.sessions.get(session_id)
if session:
if session.is_expired(self.timeout_minutes):
self.end_session(session_id)
return None
session.touch()
return session
def end_session(self, session_id: str):
"""End and remove a session."""
if session_id in self.sessions:
self.sessions[session_id].is_active = False
del self.sessions[session_id]
def get_active_sessions(self) -> list:
"""Get all active, non-expired sessions."""
return [
s for s in self.sessions.values()
if s.is_active and not s.is_expired(self.timeout_minutes)
]
async def start_cleanup_loop(self, interval_seconds: int = 300):
"""Background task to clean up expired sessions."""
while True:
expired = [
sid for sid, session in self.sessions.items()
if session.is_expired(self.timeout_minutes)
]
for sid in expired:
self.end_session(sid)
print(f"Cleaned up expired session: {sid}")
await asyncio.sleep(interval_seconds)
Building the FastAPI Server
Now let's create the FastAPI application that exposes our voice assistant through HTTP and WebSocket endpoints. The WebSocket endpoint is particularly important for voice applications, as it enables real-time bidirectional audio streaming.
# main.py
import os
import json
import asyncio
from fastapi import FastAPI, WebSocket, WebSocketDisconnect, HTTPException
from fastapi.responses import JSONResponse, StreamingResponse
from pydantic import BaseModel
from typing import Optional
import uvicorn
from config import Config
from knowledge_base import KnowledgeBase
from stt_engine import STTEngine
from tts_engine import TTSEngine
from assistant import VoiceAssistant
from session_manager import SessionManager
# Load configuration
config = Config()
# Initialize components
print("Initializing voice assistant backend...")
knowledge_base = KnowledgeBase(
data_dir=config.DATA_DIR,
persist_dir=config.STORAGE_DIR,
)
stt_engine = STTEngine(model_size=config.WHISPER_MODEL)
tts_engine = TTSEngine(provider=config.TTS_PROVIDER)
assistant = VoiceAssistant(knowledge_base, stt_engine, tts_engine)
session_manager = SessionManager(timeout_minutes=config.SESSION_TIMEOUT_MINUTES)
# Create FastAPI app
app = FastAPI(
title="Voice Assistant API",
description="Backend API for LlamaIndex-powered voice assistant",
version="1.0.0",
)
# Request/Response models
class TextQueryRequest(BaseModel):
query: str
session_id: Optional[str] = None
user_id: Optional[str] = None
class TextQueryResponse(BaseModel):
session_id: str
response: str
sources: list = []
class SessionResponse(BaseModel):
session_id: str
created_at: str
is_active: bool
# --- HTTP Endpoints ---
@app.on_event("startup")
async def startup_event():
"""Start background tasks on startup."""
asyncio.create_task(session_manager.start_cleanup_loop())
@app.get("/health")
async def health_check():
"""Health check endpoint."""
return {"status": "healthy", "active_sessions": len(session_manager.sessions)}
@app.post("/sessions", response_model=SessionResponse)
async def create_session(user_id: Optional[str] = None):
"""Create a new conversation session."""
session = session_manager.create_session(user_id=user_id)
return SessionResponse(
session_id=session.session_id,
created_at=session.created_at.isoformat(),
is_active=session.is_active,
)
@app.delete("/sessions/{session_id}")
async def end_session(session_id: str):
"""End a conversation session."""
if not session_manager.get_session(session_id):
raise HTTPException(status_code=404, detail="Session not found")
session_manager.end_session(session_id)
assistant.clear_session(session_id)
return {"message": "Session ended successfully"}
@app.post("/query", response_model=TextQueryResponse)
async def text_query(request: TextQueryRequest):
"""Process a text query (non-voice endpoint for testing)."""
# Get or create session
if request.session_id:
session = session_manager.get_session(request.session_id)
if not session:
raise HTTPException(status_code=404, detail="Session not found")
else:
session = session_manager.create_session(user_id=request.user_id)
# Process query through assistant
chat_engine = assistant.get_or_create_session(session.session_id)
response = chat_engine.chat(request.query)
# Extract source nodes if available
sources = []
if hasattr(response, "source_nodes"):
for node in response.source_nodes:
sources.append({
"text": node.node.text[:200] + "..." if len(node.node.text) > 200 else node.node.text,
"score": node.score if node.score else 0,
"metadata": node.node.metadata,
})
return TextQueryResponse(
session_id=session.session_id,
response=str(response),
sources=sources,
)
@app.post("/voice")
async def voice_query(
audio_data: bytes,
session_id: Optional[str] = None,
language: Optional[str] = None,
):
"""
Process a voice query via HTTP.
Accepts raw audio bytes and returns audio response.
"""
# Get or create session
if session_id:
session = session_manager.get_session(session_id)
if not session:
raise HTTPException(status_code=404, detail="Session not found")
else:
session = session_manager.create_session()
# Process voice input
result = await assistant.process_voice_input(
audio_data,
session.session_id,
language=language,
)
# Return audio as streaming response
return StreamingResponse(
iter([result["audio"]]),
media_type="audio/mpeg",
headers={
"X-Transcription": result["transcription"],
"X-Response-Text": result["response"],
"X-Confidence": str(result["confidence"]),
"X-Session-Id": session.session_id,
},
)
# --- WebSocket Endpoint ---
@app.websocket("/ws/voice")
async def voice_websocket(websocket: WebSocket):
"""
WebSocket endpoint for real-time voice interaction.
Message protocol:
- Client sends: {"type": "audio", "data": ""}
- Client sends: {"type": "start_session", "user_id": "..."}
- Server sends: {"type": "transcription", "text": "..."}
- Server sends: {"type": "response_text", "text": "..."}
- Server sends: {"type": "audio", "data": ""}
- Server sends: {"type": "error", "message": "..."}
"""
await websocket.accept()
session = None
try:
while True:
# Receive message
data = await websocket.receive_text()
message = json.loads(data)
if message["type"] == "start_session":
session = session_manager.create_session(
user_id=message.get("user_id")
)
await websocket.send_text(json.dumps({
"type": "session_started",
"session_id": session.session_id,
}))
continue
if message["type"] == "end_session":
if session:
session_manager.end_session(session.session_id)
assistant.clear_session(session.session_id)
await websocket.send_text(json.dumps({
"type": "session_ended",
}))
break
if message["type"] == "audio":
if not session:
session = session_manager.create_session()
import base64
audio_bytes = base64.b64decode(message["data"])
language = message.get("language")
# Process with streaming response
try:
async for result in assistant.process_voice_input_streaming(
audio_bytes,
session.session_id,
language=language,
):
if result["type"] == "transcription":
await websocket.send_text(json.dumps({
"type": "transcription",
"text": result["text"],
"confidence": result["confidence"],
}))
elif result["type"] == "response_text":
await websocket.send_text(json.dumps({
"type": "response_text",
"text": result["text"],
}))
elif result["type"] == "audio":
audio_b64 = base64.b64encode(result["data"]).decode()
await websocket.send_text(json.dumps({
"type": "audio",
"data": audio_b64,
}))
await websocket.send_text(json.dumps({
"type": "response_complete",
}))
except Exception as e:
await websocket.send_text(json.dumps({
"type": "error",
"message": str(e),
}))
except WebSocketDisconnect:
print(f"WebSocket disconnected for session: {session.session_id if session else 'unknown'}")
if session:
session_manager.end_session(session.session_id)
assistant.clear_session(session.session_id)
if __name__ == "__main__":
uvicorn.run(
"main:app",
host=config.HOST,
port=config.PORT,
reload=config.DEBUG,
log_level="info",
)
Configuration Module
# config.py
import os
from dotenv import load_dotenv
from dataclasses import dataclass
load_dotenv()
@dataclass
class Config:
# Server settings
HOST: str = os.getenv("HOST", "0.0.0.0")
PORT: int = int(os.getenv("PORT", "8000"))
DEBUG: bool = os.getenv("DEBUG", "false").lower() == "true"
# Data paths
DATA_DIR: str = os.getenv("DATA_DIR", "./data")
STORAGE_DIR: str = os.getenv("STORAGE_DIR", "./storage")
# Model settings
WHISPER_MODEL: str = os.getenv("WHISPER_MODEL", "base")
TTS_PROVIDER: str = os.getenv("TTS_PROVIDER", "elevenlabs")
# Session settings
SESSION_TIMEOUT_MINUTES: int = int(
os.getenv("SESSION_TIMEOUT_MINUTES", "30")
)
# API keys (loaded from .env)
OPENAI_API_KEY: str = os.getenv("OPENAI_API_KEY", "")
ELEVENLABS_API_KEY: str = os.getenv("ELEVENLABS_API_KEY", "")
Adding Advanced Retrieval Features
Basic vector search is a good starting point, but production voice assistants benefit from more sophisticated retrieval strategies. Let's enhance our knowledge base with hybrid search, re-ranking, and query transformations that improve answer quality.
Hybrid Search with Re-ranking
# knowledge_base.py (advanced features)
from llama_index.core import VectorStoreIndex
from llama_index.core.retrievers import QueryFusionRetriever
from llama_index.retrievers.bm25 import BM25Retriever
from llama_index.core.postprocessor import (
SentenceTransformerRerank,
SimilarityPostprocessor,
)
from llama_index.core.query_engine import RetrieverQueryEngine
class AdvancedKnowledgeBase(KnowledgeBase):
def get_hybrid_query_engine(self, **kwargs):
"""
Create a hybrid query engine combining vector and BM25 retrieval
with re-ranking for improved accuracy.
"""
# Vector retriever for semantic search
vector_retriever = self.index.as_retriever(