Introduction to Voice Assistant Backends with vLLM
Voice assistants have evolved from simple command-and-control systems into sophisticated conversational agents capable of reasoning, context retention, and natural dialogue. At the heart of this transformation lies the Large Language Model (LLM) that powers the assistant's understanding and response generation. vLLM — a high-throughput, memory-efficient inference engine — has emerged as one of the most effective ways to serve these models in production.
This tutorial walks you through building a complete voice assistant backend using vLLM as the LLM serving layer. We'll cover the architecture, set up the inference server, integrate speech-to-text (STT) and text-to-speech (TTS) components, manage conversation state, and deploy the system for real-time interaction.
What Is vLLM and Why It Matters for Voice Assistants
vLLM is an open-source inference engine developed at UC Berkeley that dramatically improves LLM serving performance. Its key innovation is PagedAttention, a technique that manages the KV cache in a way similar to how operating systems handle virtual memory paging. This allows vLLM to achieve throughput improvements of 2-4x compared to naive implementations.
Why vLLM Is Ideal for Voice Assistants
Voice assistants have unique requirements that make vLLM particularly well-suited:
- Low latency: Voice interactions demand sub-second response times. vLLM's optimized inference pipeline minimizes time-to-first-token.
- Concurrent sessions: Multiple users may interact simultaneously. vLLM's continuous batching handles this efficiently.
- Memory efficiency: PagedAttention reduces memory waste, allowing more concurrent conversations on the same hardware.
- Streaming support: vLLM supports token streaming, which is essential for sending partial responses to TTS as they're generated.
- OpenAI-compatible API: Drop-in compatibility with the OpenAI API format simplifies integration with existing tooling.
Architecture Overview
A voice assistant backend built with vLLM typically consists of four main components:
- Speech-to-Text (STT): Converts user audio input into text. Common choices include Whisper, Deepgram, or AssemblyAI.
- LLM Inference (vLLM): Processes the transcribed text and generates a conversational response.
- Text-to-Speech (TTS): Converts the LLM's text response back into audio. Options include Coqui XTTS, ElevenLabs, or Azure TTS.
- Orchestration Layer: A FastAPI (or similar) service that ties everything together, manages session state, and handles WebSocket connections for real-time streaming.
The data flow is straightforward: audio in → STT → text → vLLM → response text → TTS → audio out. The orchestration layer ensures this pipeline runs smoothly with minimal latency.
Prerequisites and Environment Setup
Before building, ensure you have the following prerequisites:
- A machine with an NVIDIA GPU (at least 16GB VRAM for a 7B model, more for larger models)
- Python 3.10 or later
- CUDA 12.1 or later
- Docker (optional, for containerized deployment)
Start by creating a virtual environment and installing the core dependencies:
python -m venv voice-assistant-env
source voice-assistant-env/bin/activate
pip install vllm fastapi uvicorn websockets pydantic
pip install openai-whisper torch torchaudio
pip install python-multipart aiohttp
For TTS, we'll use Coqui XTTS in this tutorial, but you can substitute any TTS engine:
pip install TTS
Setting Up the vLLM Inference Server
vLLM can be run as a standalone server or embedded directly in your Python application. For a voice assistant backend, embedding it gives you more control over the pipeline. However, running it as a server allows you to scale the LLM layer independently.
Option 1: Running vLLM as a Server
The simplest approach is to launch vLLM's OpenAI-compatible server:
python -m vllm.entrypoints.openai.api_server \
--model meta-llama/Llama-3.1-8B-Instruct \
--port 8000 \
--tensor-parallel-size 1 \
--gpu-memory-utilization 0.85 \
--max-model-len 4096 \
--enable-auto-tool-choice \
--quantization awq
This launches an OpenAI-compatible API at http://localhost:8000. You can then interact with it using the standard OpenAI Python client.
Option 2: Embedding vLLM in Your Application
For tighter integration, embed vLLM directly:
from vllm import LLM, SamplingParams
llm = LLM(
model="meta-llama/Llama-3.1-8B-Instruct",
tensor_parallel_size=1,
gpu_memory_utilization=0.85,
max_model_len=4096,
)
sampling_params = SamplingParams(
temperature=0.7,
top_p=0.9,
max_tokens=256,
)
response = llm.generate("Hello, how are you?", sampling_params)
print(response[0].outputs[0].text)
For this tutorial, we'll use the server approach since it provides a clean separation of concerns and allows independent scaling.
Building the Orchestration Layer with FastAPI
The orchestration layer is the backbone of your voice assistant. It receives audio, coordinates STT, vLLM, and TTS, and streams audio back to the client. We'll use FastAPI with WebSocket support for real-time bidirectional communication.
Project Structure
voice-assistant/
├── main.py # FastAPI app entry point
├── config.py # Configuration settings
├── stt/
│ └── whisper_stt.py # Speech-to-text module
├── llm/
│ └── vllm_client.py # vLLM integration
├── tts/
│ └── xtts_tts.py # Text-to-speech module
├── session/
│ └── manager.py # Conversation state management
└── requirements.txt
Configuration
Start with a central configuration file:
# config.py
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
# vLLM settings
vllm_base_url: str = "http://localhost:8000/v1"
vllm_model: str = "meta-llama/Llama-3.1-8B-Instruct"
vllm_api_key: str = "EMPTY" # vLLM doesn't require a real key by default
# LLM generation settings
temperature: float = 0.7
top_p: float = 0.9
max_tokens: int = 256
# STT settings
whisper_model: str = "base" # tiny, base, small, medium, large
whisper_device: str = "cuda"
# TTS settings
tts_model: str = "tts_models/multilingual/multi-dataset/xtts_v2"
tts_language: str = "en"
# Server settings
host: str = "0.0.0.0"
port: int = 8080
# Session settings
max_history_turns: int = 10
class Config:
env_file = ".env"
settings = Settings()
Session Management
Voice assistants need to maintain conversation context across turns. Here's a simple in-memory session manager:
# session/manager.py
from collections import defaultdict
from datetime import datetime
from typing import List, Dict
import uuid
class SessionManager:
def __init__(self, max_history_turns: int = 10):
self.sessions: Dict[str, dict] = defaultdict(dict)
self.max_history_turns = max_history_turns
def create_session(self) -> str:
session_id = str(uuid.uuid4())
self.sessions[session_id] = {
"messages": [],
"created_at": datetime.utcnow(),
"last_active": datetime.utcnow(),
}
return session_id
def add_message(self, session_id: str, role: str, content: str):
if session_id not in self.sessions:
self.create_session()
self.sessions[session_id]["messages"].append({
"role": role,
"content": content,
})
self.sessions[session_id]["last_active"] = datetime.utcnow()
# Trim history to prevent context overflow
if len(self.sessions[session_id]["messages"]) > self.max_history_turns * 2:
# Keep the system prompt and last N turns
system_msgs = [m for m in self.sessions[session_id]["messages"] if m["role"] == "system"]
conversation_msgs = [m for m in self.sessions[session_id]["messages"] if m["role"] != "system"]
self.sessions[session_id]["messages"] = system_msgs + conversation_msgs[-(self.max_history_turns * 2):]
def get_messages(self, session_id: str) -> List[dict]:
return self.sessions.get(session_id, {}).get("messages", [])
def set_system_prompt(self, session_id: str, prompt: str):
if session_id not in self.sessions:
self.create_session()
# Remove existing system messages
self.sessions[session_id]["messages"] = [
m for m in self.sessions[session_id]["messages"] if m["role"] != "system"
]
# Prepend new system prompt
self.sessions[session_id]["messages"].insert(0, {
"role": "system",
"content": prompt,
})
def delete_session(self, session_id: str):
self.sessions.pop(session_id, None)
Integrating Speech-to-Text with Whisper
For STT, we'll use OpenAI's Whisper model running locally. This keeps latency low and data private:
# stt/whisper_stt.py
import whisper
import torch
import io
import numpy as np
from config import settings
class WhisperSTT:
def __init__(self):
self.model = whisper.load_model(
settings.whisper_model,
device=settings.whisper_device
)
def transcribe(self, audio_bytes: bytes, sample_rate: int = 16000) -> str:
"""
Transcribe audio bytes to text.
Expects raw PCM 16-bit audio.
"""
# Convert bytes to numpy array
audio_np = np.frombuffer(audio_bytes, dtype=np.int16).astype(np.float32)
audio_np = audio_np / 32768.0 # Normalize to [-1, 1]
# Whisper expects 16kHz audio
result = self.model.transcribe(
audio_np,
language="en",
task="transcribe",
fp16=torch.cuda.is_available(),
)
return result["text"].strip()
def transcribe_file(self, file_path: str) -> str:
"""Transcribe an audio file directly."""
result = self.model.transcribe(
file_path,
language="en",
task="transcribe",
fp16=torch.cuda.is_available(),
)
return result["text"].strip()
Connecting to vLLM for LLM Inference
Since vLLM exposes an OpenAI-compatible API, we can use the OpenAI Python client to communicate with it. This abstraction makes it easy to swap vLLM for other backends if needed:
# llm/vllm_client.py
from openai import AsyncOpenAI
from typing import AsyncGenerator, List, Dict
from config import settings
class VLLMClient:
def __init__(self):
self.client = AsyncOpenAI(
base_url=settings.vllm_base_url,
api_key=settings.vllm_api_key,
)
self.model = settings.vllm_model
async def generate(
self,
messages: List[Dict[str, str]],
temperature: float = None,
top_p: float = None,
max_tokens: int = None,
) -> str:
"""Generate a complete response (non-streaming)."""
response = await self.client.chat.completions.create(
model=self.model,
messages=messages,
temperature=temperature or settings.temperature,
top_p=top_p or settings.top_p,
max_tokens=max_tokens or settings.max_tokens,
)
return response.choices[0].message.content
async def generate_stream(
self,
messages: List[Dict[str, str]],
temperature: float = None,
top_p: float = None,
max_tokens: int = None,
) -> AsyncGenerator[str, None]:
"""Stream tokens as they are generated."""
stream = await self.client.chat.completions.create(
model=self.model,
messages=messages,
temperature=temperature or settings.temperature,
top_p=top_p or settings.top_p,
max_tokens=max_tokens or settings.max_tokens,
stream=True,
)
async for chunk in stream:
if chunk.choices[0].delta.content is not None:
yield chunk.choices[0].delta.content
Integrating Text-to-Speech with XTTS
For TTS, we'll use Coqui XTTS v2, which supports voice cloning and multiple languages:
# tts/xtts_tts.py
import torch
import io
import wave
from TTS.api import TTS
from config import settings
class XTTSEngine:
def __init__(self):
self.tts = TTS(settings.tts_model)
# Use GPU if available
self.tts.to("cuda" if torch.cuda.is_available() else "cpu")
def synthesize(self, text: str, speaker_wav: str = None) -> bytes:
"""
Convert text to speech and return WAV audio bytes.
speaker_wav: path to a reference audio file for voice cloning.
"""
buffer = io.BytesIO()
with wave.open(buffer, "wb") as wav_file:
wav_file.setnchannels(1)
wav_file.setsampwidth(2)
wav_file.setframerate(24000)
# Generate audio
if speaker_wav:
wav_data = self.tts.tts(
text=text,
speaker_wav=speaker_wav,
language=settings.tts_language,
)
else:
wav_data = self.tts.tts(
text=text,
language=settings.tts_language,
)
# Convert to bytes
import numpy as np
audio_np = np.array(wav_data, dtype=np.float32)
audio_int16 = (audio_np * 32767).astype(np.int16)
return audio_int16.tobytes()
def synthesize_stream(self, text: str, speaker_wav: str = None):
"""
For streaming, we synthesize sentence by sentence.
Returns a generator yielding audio chunks.
"""
import re
# Split text into sentences for streaming TTS
sentences = re.split(r'(?<=[.!?])\s+', text)
for sentence in sentences:
if sentence.strip():
yield self.synthesize(sentence.strip(), speaker_wav)
Putting It All Together: The Main Application
Now let's assemble everything into a FastAPI application with WebSocket support for real-time voice interaction:
# main.py
import json
import asyncio
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.responses import JSONResponse
from config import settings
from session.manager import SessionManager
from stt.whisper_stt import WhisperSTT
from llm.vllm_client import VLLMClient
from tts.xtts_tts import XTTSEngine
app = FastAPI(title="Voice Assistant Backend")
# Initialize components
session_manager = SessionManager(max_history_turns=settings.max_history_turns)
stt_engine = WhisperSTT()
llm_client = VLLMClient()
tts_engine = XTTSEngine()
SYSTEM_PROMPT = """You are a helpful voice assistant. Keep your responses concise and conversational.
Avoid markdown formatting, code blocks, or special characters since your responses will be spoken aloud.
Aim for natural, spoken-language responses that are typically 1-3 sentences long unless the user
asks for more detail."""
@app.get("/health")
async def health_check():
return {"status": "healthy", "model": settings.vllm_model}
@app.post("/session/create")
async def create_session():
session_id = session_manager.create_session()
session_manager.set_system_prompt(session_id, SYSTEM_PROMPT)
return {"session_id": session_id}
@app.delete("/session/{session_id}")
async def delete_session(session_id: str):
session_manager.delete_session(session_id)
return {"status": "deleted"}
@app.websocket("/ws/voice/{session_id}")
async def voice_websocket(websocket: WebSocket, session_id: str):
"""
WebSocket endpoint for real-time voice interaction.
Client sends:
- {"type": "audio", "data": ""}
- {"type": "text", "data": "Hello assistant"}
Server responds:
- {"type": "transcription", "text": "..."}
- {"type": "token", "text": "..."} (streamed LLM tokens)
- {"type": "audio", "data": ""}
- {"type": "done"}
"""
await websocket.accept()
# Ensure session exists
if session_id not in session_manager.sessions:
session_manager.create_session()
session_manager.set_system_prompt(session_id, SYSTEM_PROMPT)
try:
while True:
message = await websocket.receive_text()
data = json.loads(message)
if data["type"] == "audio":
import base64
audio_bytes = base64.b64decode(data["data"])
# Step 1: Speech to Text
transcription = stt_engine.transcribe(audio_bytes)
await websocket.send_json({
"type": "transcription",
"text": transcription
})
if not transcription:
await websocket.send_json({"type": "error", "message": "Could not transcribe audio"})
continue
# Step 2: Add user message to session
session_manager.add_message(session_id, "user", transcription)
# Step 3: Stream LLM response
messages = session_manager.get_messages(session_id)
full_response = ""
async for token in llm_client.generate_stream(messages):
full_response += token
await websocket.send_json({"type": "token", "text": token})
# Step 4: Add assistant response to session
session_manager.add_message(session_id, "assistant", full_response)
# Step 5: Text to Speech
speaker_wav = data.get("speaker_wav")
for audio_chunk in tts_engine.synthesize_stream(full_response, speaker_wav):
audio_b64 = base64.b64encode(audio_chunk).decode("utf-8")
await websocket.send_json({
"type": "audio",
"data": audio_b64
})
await websocket.send_json({"type": "done"})
elif data["type"] == "text":
# Handle text input directly (skip STT)
user_text = data["data"]
session_manager.add_message(session_id, "user", user_text)
messages = session_manager.get_messages(session_id)
full_response = ""
async for token in llm_client.generate_stream(messages):
full_response += token
await websocket.send_json({"type": "token", "text": token})
session_manager.add_message(session_id, "assistant", full_response)
# Generate audio response
import base64
speaker_wav = data.get("speaker_wav")
for audio_chunk in tts_engine.synthesize_stream(full_response, speaker_wav):
audio_b64 = base64.b64encode(audio_chunk).decode("utf-8")
await websocket.send_json({
"type": "audio",
"data": audio_b64
})
await websocket.send_json({"type": "done"})
except WebSocketDisconnect:
print(f"Client disconnected from session {session_id}")
except Exception as e:
await websocket.send_json({"type": "error", "message": str(e)})
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host=settings.host, port=settings.port)
Creating a Test Client
To test your voice assistant backend, create a simple Python client that records audio and sends it over WebSocket:
# test_client.py
import asyncio
import json
import base64
import websockets
import pyaudio
import wave
import io
WEBSOCKET_URL = "ws://localhost:8080/ws/voice"
async def test_text_interaction():
"""Test with text input (no microphone needed)."""
async with websockets.connect(f"{WEBSOCKET_URL}/test-session") as ws:
# Send text message
message = json.dumps({
"type": "text",
"data": "What's the weather like today?"
})
await ws.send(message)
# Receive responses
while True:
response = await ws.recv()
data = json.loads(response)
if data["type"] == "token":
print(data["text"], end="", flush=True)
elif data["type"] == "audio":
print(f"\n[Received audio chunk: {len(data['data'])} bytes]")
elif data["type"] == "done":
print("\n[Done]")
break
elif data["type"] == "error":
print(f"\n[Error: {data['message']}]")
break
async def test_voice_interaction():
"""Test with actual microphone input."""
# Record audio
chunk = 1024
sample_format = pyaudio.paInt16
channels = 1
fs = 16000
seconds = 5
p = pyaudio.PyAudio()
print("Recording for 5 seconds...")
stream = p.open(
format=sample_format,
channels=channels,
rate=fs,
frames_per_buffer=chunk,
input=True
)
frames = []
for _ in range(0, int(fs / chunk * seconds)):
data = stream.read(chunk)
frames.append(data)
stream.stop_stream()
stream.close()
p.terminate()
print("Recording complete.")
# Combine audio frames
audio_bytes = b"".join(frames)
audio_b64 = base64.b64encode(audio_bytes).decode("utf-8")
async with websockets.connect(f"{WEBSOCKET_URL}/voice-session") as ws:
message = json.dumps({
"type": "audio",
"data": audio_b64
})
await ws.send(message)
while True:
response = await ws.recv()
data = json.loads(response)
if data["type"] == "transcription":
print(f"You said: {data['text']}")
elif data["type"] == "token":
print(data["text"], end="", flush=True)
elif data["type"] == "audio":
pass # Would play audio here
elif data["type"] == "done":
print("\n[Done]")
break
if __name__ == "__main__":
asyncio.run(test_text_interaction())
Optimizing for Low Latency
Latency is the most critical metric for voice assistants. Users expect responses within 500-800ms. Here are several optimization strategies:
1. Use a Smaller, Faster Model
While larger models produce better responses, they're slower. For voice assistants, a 7B or 8B parameter model often provides the best balance. Consider quantized models (AWQ or GPTQ) to reduce memory usage and increase throughput:
# Launch vLLM with an AWQ-quantized model
python -m vllm.entrypoints.openai.api_server \
--model TheBloke/Llama-3.1-8B-Instruct-AWQ \
--quantization awq \
--port 8000 \
--gpu-memory-utilization 0.85
2. Pipeline STT, LLM, and TTS
Instead of waiting for the full LLM response before starting TTS, pipeline the stages. Generate TTS for each sentence as soon as it's complete from the LLM stream. The synthesize_stream method in our TTS module already implements this approach.
3. Use Speculative Decoding
vLLM supports speculative decoding, which uses a smaller draft model to predict tokens that the larger model then verifies. This can reduce latency by 1.5-2x:
python -m vllm.entrypoints.openai.api_server \
--model meta-llama/Llama-3.1-70B-Instruct \
--speculative-model meta-llama/Llama-3.2-1B-Instruct \
--num-speculative-tokens 5 \
--port 8000
4. Optimize Whisper for Speed
Use a smaller Whisper model or apply optimizations like faster-whisper, which uses CTranslate2 for faster inference:
pip install faster-whisper
# stt/fast_whisper_stt.py
from faster_whisper import WhisperModel
class FastWhisperSTT:
def __init__(self):
self.model = WhisperModel(
"base",
device="cuda",
compute_type="float16"
)
def transcribe(self, audio_bytes: bytes, sample_rate: int = 16000) -> str:
import numpy as np
audio_np = np.frombuffer(audio_bytes, dtype=np.int16).astype(np.float32)
audio_np = audio_np / 32768.0
segments, _ = self.model.transcribe(
audio_np,
language="en",
beam_size=1, # Faster, slightly less accurate
)
return " ".join([segment.text for segment in segments]).strip()
Best Practices for Production
Session Persistence
The in-memory session manager works for development, but production deployments need persistent storage. Use Redis or a database to store conversation history:
# session/redis_manager.py
import redis
import json
from config import settings
class RedisSessionManager:
def __init__(self):
self.redis = redis.Redis(host="localhost", port=6379, db=0)
self.ttl = 3600 # 1 hour session timeout
def create_session(self) -> str:
import uuid
session_id = str(uuid.uuid4())
self.redis.setex(
f"session:{session_id}",
self.ttl,
json.dumps({"messages": []})
)
return session_id
def add_message(self, session_id: str, role: str, content: str):
key = f"session:{session_id}"
data = json.loads(self.redis.get(key) or '{"messages": []}')
data["messages"].append({"role": role, "content": content})
self.redis.setex(key, self.ttl, json.dumps(data))
def get_messages(self, session_id: str) -> list:
key = f"session:{session_id}"
data = json.loads(self.redis.get(key) or '{"messages": []}')
return data["messages"]
Error Handling and Fallbacks
Production systems need robust error handling. If STT fails, the assistant should inform the user. If the LLM times out, provide a fallback response. If TTS fails, at least return the text:
async def safe_pipeline(websocket, session_id, audio_bytes):
try:
transcription = stt_engine.transcribe(audio_bytes)
except Exception as e:
await websocket.send_json({
"type": "error",
"message": "I couldn't understand that. Could you repeat?"
})
return
if not transcription:
await websocket.send_json({
"type": "error",
"message": "I didn't catch that. Please try again."
})
return
await websocket.send_json({"type": "transcription", "text": transcription})
session_manager.add_message(session_id, "user", transcription)
messages = session_manager.get_messages(session_id)
full_response = ""
try:
async for token in llm_client.generate_stream(messages):
full_response += token
await websocket.send_json({"type": "token", "text": token})
except Exception as e:
full_response = "I'm having trouble processing that right now."
await websocket.send_json({"type": "token", "text": full_response})
session_manager.add_message(session_id, "assistant", full_response)
try:
for audio_chunk in tts_engine.synthesize_stream(full_response):
audio_b64 = base64.b64encode(audio_chunk).decode("utf-8")
await websocket.send_json({"type": "audio", "data": audio_b64})
except Exception as e:
print(f"TTS error: {e}")
await websocket.send_json({"type": "done"})
Monitoring and Metrics
Track key metrics to identify bottlenecks. Add timing instrumentation to each pipeline stage:
import time
from contextlib import asynccontextmanager
@asynccontextmanager
async def timed_stage(name: str):
start = time.perf_counter()
try:
yield
finally:
elapsed = (time.perf_counter() - start) * 1000
print(f"[METRIC] {name}: {elapsed:.1f}ms")
# Usage in the WebSocket handler:
async with timed_stage("stt"):
transcription = stt_engine.transcribe(audio_bytes)
async with timed_stage("llm_stream"):
async for token in llm_client.generate_stream(messages):
# ...
async with timed_stage("tts"):
for audio_chunk in tts_engine.synthesize_stream(full_response):
# ...
Containerization with Docker
For deployment, containerize your application. Here's a Dockerfile for the voice assistant backend:
# Dockerfile
FROM nvidia/cuda:12.1.0-runtime-ubuntu22.04
RUN apt-get update && apt-get install -y \
python3.10 python3-pip python3.10-dev \
ffmpeg libsndfile1 \
&& rm -rf /var/lib/apt/lists/*
WORKDIR /app
COPY requirements.txt .
RUN pip3 install --no-cache-dir -r requirements.txt
COPY . .
EXPOSE 8080
CMD ["python3", "main.py"]
And a docker-compose.yml that runs both vLLM and your application:
# docker-compose.yml
version: "3.9"
services:
vllm:
image: vllm/vllm-openai:latest
runtime: nvidia
environment:
- NVIDIA_VISIBLE_DEVICES=all
command:
- --model=meta-llama/Llama-3.1-8B-Instruct
- --port=8000
- --gpu-memory-utilization=0.85
ports:
- "8000:8000"
volumes:
- huggingface-cache:/root/.cache/huggingface
deploy:
resources:
reservations:
devices:
- driver: nvidia
count: 1
capabilities: [gpu]
voice-assistant:
build: .
runtime: nvidia
environment:
- VLLM_BASE_URL=http://vllm:8000/v1
- NVIDIA_VISIBLE_DEVICES=all
ports:
- "8080:8080"
depends_on:
- vllm
deploy:
resources:
reservations:
devices:
- driver: nvidia
count: 1
capabilities: [gpu]
redis:
image: redis:7-alpine
ports:
- "6379:6379"
volumes:
huggingface-cache:
Advanced Features
Function Calling and Tool Use
vLLM supports function calling, which lets your assistant perform actions like checking the weather, setting reminders, or querying databases. Enable it in the vLLM server and define tools in your client:
tools = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "Get current weather for a location",
"parameters": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "City name"
}
},
"required": ["location"]
}
}
}
]
async def generate_with_tools(self, messages, tools):
response = await self.client.chat.completions.create(
model=self.model,
messages=messages,
tools=tools,
tool_choice="auto",
)
return response.choices[0].message
Voice Activity Detection (VAD)
Instead of fixed-length recording, use VAD to detect when the user starts and stops speaking. This creates a more natural interaction:
pip install silero-vad
# vad/silero_vad.py
import torch
from collections import deque
class VoiceActivityDetector:
def __init__(self, threshold: float = 0.5, sample_rate: int = 16000):
self.model, _ = torch.hub.load(
"snakers4/silero-vad",
"model",
trust_repo=True
)
self.threshold = threshold
self.sample_rate = sample_rate
def detect_speech_segments(self, audio_bytes: bytes):
import numpy as np
audio = np.frombuffer(audio_bytes, dtype=np.int16).astype(np.float32)
audio = torch.from_numpy(audio)
# Process in 512-sample windows
segments = []
is_speaking = False
current_segment = []
for i in range(0, len(audio) - 512, 512):
chunk = audio[i:i + 512]
prob = self.model(chunk, self.sample_rate).item()
if prob > self.threshold and not is_speaking:
is_speaking = True
current_segment = [chunk]
elif prob > self.threshold and is_speaking:
current_segment.append(chunk)
elif prob < self.threshold and is_speaking:
is_speaking = False
if len(current_segment) > 10: # Min segment length
segments.append(torch.cat(current_segment))
current_segment = []
return segments
Multi-Language Support
Both Whisper and XTTS support multiple languages. To build a multilingual assistant, detect the language from the STT output and pass it through the pipeline:
# Detect language in Whisper
result = self.model.transcribe(audio_np, task="transcribe")
detected_language = result.get("language", "en")
# Pass language to TTS
tts_engine.synthesize(text, language=detected_language)
# Adjust system prompt for language-aware responses
MULTILINGUAL_PROMPT = """You are a helpful voice assistant.
Respond in the same language the user speaks.
Keep responses concise and natural for spoken conversation."""
Conclusion
Building a voice assistant backend with vLLM gives you a powerful, production-ready foundation for real-time conversational AI. By leveraging vLLM's high-throughput inference, streaming token generation, and OpenAI-compatible API, you can create a system that responds in under a second while handling multiple concurrent users. The modular architecture — with separate STT, LLM, and TTS components — lets you swap individual pieces as better models become available. Start with the basic pipeline described here, then iterate on latency optimizations, add function calling for real-world