Khi xây dựng một voice assistant production-grade, việc kết hợp Whisper v4 để nhận diện giọng nói với TTS (Text-to-Speech) để tổng hợp âm thanh phản hồi là nền tảng cốt lõi. Bài viết này sẽ hướng dẫn bạn xây dựng một pipeline hoàn chỉnh với khả năng xử lý đồng thời cao, độ trễ thấp và chi phí tối ưu nhất.
1. Tổng quan kiến trúc hệ thống
Kiến trúc voice assistant gồm 4 thành phần chính: Audio Input → Speech Recognition (Whisper) → LLM Processing → Speech Synthesis (TTS). Pipeline này cần đảm bảo end-to-end latency dưới 2 giây cho các tác vụ đơn giản.
- Audio Capture: WebRTC hoặc Web Audio API, 16kHz mono PCM
- Speech-to-Text: Whisper v4 với optimized inference
- LLM Processing: Xử lý ngôn ngữ tự nhiên với context management
- Text-to-Speech: Streaming audio synthesis với voice cloning tùy chọn
2. Cài đặt môi trường và dependencies
# requirements.txt cho Voice Assistant production
openai-whisper==20240930
torch>=2.1.0
torchaudio>=2.1.0
numpy>=1.24.0
scipy>=1.11.0
fastapi==0.109.0
uvicorn[standard]==0.27.0
websockets==12.0
python-multipart==0.0.6
pydantic==2.5.0
redis==5.0.1
aiohttp==3.9.1
# Cài đặt với GPU support (NVIDIA CUDA 12.1)
pip install torch torchaudio --index-url https://download.pytorch.org/whl/cu121
pip install openai-whisper
Kiểm tra GPU availability
python -c "import torch; print(f'CUDA: {torch.cuda.is_available()}, Device: {torch.cuda.get_device_name(0)}')"
Output mong đợi: CUDA: True, Device: NVIDIA RTX 4090
3. Production code: Whisper v4 Integration
Với kinh nghiệm triển khai nhiều hệ thống voice assistant, tôi nhận thấy việc tối ưu Whisper inference là yếu tố quyết định latency. Dưới đây là implementation với batching và streaming support.
# whisper_client.py
import whisper
import numpy as np
import torch
from typing import Optional, AsyncGenerator
import time
import io
class WhisperProcessor:
"""
Production-grade Whisper v4 processor với:
- Streaming support
- Automatic language detection
- Timestamp generation
- Batch processing optimization
"""
def __init__(
self,
model_name: str = "base",
device: str = "cuda" if torch.cuda.is_available() else "cpu",
fp16: bool = True
):
self.device = device
self.fp16 = device == "cuda" and fp16
print(f"Loading Whisper {model_name} on {device}...")
start = time.perf_counter()
self.model = whisper.load_model(model_name, device=device)
load_time = (time.perf_counter() - start) * 1000
print(f"Model loaded in {load_time:.2f}ms")
# Warmup
self._warmup()
def _warmup(self):
"""Warmup GPU kernel để avoid first-call overhead"""
dummy_audio = np.random.randn(16000).astype(np.float32)
_ = self.model.transcribe(dummy_audio, fp16=self.fp16)
async def transcribe_stream(
self,
audio_chunk: bytes,
sample_rate: int = 16000,
language: Optional[str] = None
) -> dict:
"""
Transcribe một audio chunk với timing metrics
Returns: dict với text, language, timestamps, và inference_time_ms
"""
# Convert bytes to numpy array
audio_np = np.frombuffer(audio_chunk, dtype=np.int16).astype(np.float32) / 32768.0
# Resample nếu cần
if sample_rate != 16000:
import librosa
audio_np = librosa.resample(audio_np, orig_sr=sample_rate, target_sr=16000)
start = time.perf_counter()
result = self.model.transcribe(
audio_np,
fp16=self.fp16,
language=language,
task="transcribe",
verbose=False
)
inference_time = (time.perf_counter() - start) * 1000
return {
"text": result["text"].strip(),
"language": result.get("language", "unknown"),
"segments": result.get("segments", []),
"inference_time_ms": round(inference_time, 2),
"audio_duration_sec": len(audio_np) / 16000,
"rtf": round(inference_time / (len(audio_np) / 16000 * 1000), 4)
}
Benchmark function
async def benchmark_whisper():
processor = WhisperProcessor(model_name="base")
# Generate test audio (5 seconds)
test_audio = np.random.randn(80000).astype(np.float32)
results = []
for i in range(10):
result = await processor.transcribe_stream(
(test_audio * 32768).astype(np.int16).tobytes()
)
results.append(result["inference_time_ms"])
avg_time = sum(results) / len(results)
print(f"Whisper base - Average inference: {avg_time:.2f}ms")
print(f"Whisper base - RTF: {avg_time / 5000:.4f}")
4. TTS Integration với HolySheep AI
HolySheep AI cung cấp API TTS với latency trung bình dưới 50ms, hỗ trợ multi-language và voice cloning. Với tỷ giá cạnh tranh (so sánh: GPT-4.1 $8/MTok, trong khi các giải pháp TTS chuyên dụng thường $15-30/MTok), đây là lựa chọn tối ưu cho production. Đăng ký tại đây để nhận tín dụng miễn phí khi bắt đầu.
# tts_client.py
import aiohttp
import asyncio
import base64
import json
from typing import Optional, AsyncGenerator
class HolySheepTTS:
"""
HolySheep AI TTS Client với streaming audio support
Base URL: https://api.holysheep.ai/v1
"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self._session: Optional[aiohttp.ClientSession] = None
async def _get_session(self) -> aiohttp.ClientSession:
if self._session is None or self._session.closed:
self._session = aiohttp.ClientSession(
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
)
return self._session
async def synthesize_streaming(
self,
text: str,
voice: str = "alloy",
model: str = "tts-1",
speed: float = 1.0
) -> AsyncGenerator[bytes, None]:
"""
Stream TTS audio với chunk-by-chunk delivery
Giảm perceived latency đáng kể so với waiting for full response
"""
session = await self._get_session()
payload = {
"model": model,
"input": text,
"voice": voice,
"speed": speed,
"stream": True
}
async with session.post(
f"{self.BASE_URL}/audio/speech",
json=payload
) as response:
if response.status != 200:
error = await response.text()
raise RuntimeError(f"TTS API error: {response.status} - {error}")
async for chunk in response.content.iter_chunked(4096):
if chunk:
yield chunk
async def synthesize(
self,
text: str,
voice: str = "alloy",
model: str = "tts-1"
) -> bytes:
"""Non-streaming TTS - trả về complete audio"""
chunks = []
async for chunk in self.synthesize_streaming(text, voice, model):
chunks.append(chunk)
return b"".join(chunks)
async def synthesize_with_timestamps(
self,
text: str,
voice: str = "alloy"
) -> dict:
"""
TTS với word-level timestamps cho subtitle generation
"""
session = await self._get_session()
payload = {
"model": "tts-1-hd",
"input": text,
"voice": voice,
"response_format": "verbose_json",
"timestamp_granularity": "word"
}
async with session.post(
f"{self.BASE_URL}/audio/speech",
json=payload
) as response:
result = await response.json()
return result
async def close(self):
if self._session and not self._session.closed:
await self._session.close()
Demo usage
async def demo_tts():
tts = HolySheepTTS("YOUR_HOLYSHEEP_API_KEY")
# Streaming synthesis
print("Streaming TTS demo...")
audio_chunks = []
async for chunk in tts.synthesize_streaming(
"Xin chào, đây là demo voice assistant với Whisper và HolySheep TTS.",
voice="alloy"
):
audio_chunks.append(chunk)
print(f"Received chunk: {len(chunk)} bytes")
total_audio = b"".join(audio_chunks)
print(f"Total audio size: {len(total_audio)} bytes")
await tts.close()
5. LLM Integration cho Intent Recognition
Sau khi có text từ Whisper, cần LLM để phân tích intent và tạo response. HolySheep AI hỗ trợ đa model với chi phí cực kỳ cạnh tranh: DeepSeek V3.2 chỉ $0.42/MTok so với $8 cho GPT-4.1.
# llm_client.py
import aiohttp
import json
from typing import Optional, List, Dict, Any
class HolySheepLLM:
"""
HolySheep AI LLM Client cho voice assistant
Hỗ trợ streaming response và function calling
"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str):
self.api_key = api_key
async def chat_completion(
self,
messages: List[Dict[str, str]],
model: str = "gpt-4.1",
temperature: float = 0.7,
max_tokens: int = 500,
stream: bool = False
) -> Dict[str, Any]:
"""
Gửi chat completion request đến HolySheep API
"""
async with aiohttp.ClientSession(
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
) as session:
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens,
"stream": stream
}
async with session.post(
f"{self.BASE_URL}/chat/completions",
json=payload
) as response:
if response.status != 200:
error = await response.text()
raise RuntimeError(f"LLM API error: {response.status} - {error}")
result = await response.json()
return result
async def stream_chat(
self,
messages: List[Dict[str, str]],
model: str = "gpt-4.1"
) -> AsyncGenerator[str, None]:
"""
Streaming chat completion cho real-time response
"""
async with aiohttp.ClientSession(
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
) as session:
payload = {
"model": model,
"messages": messages,
"stream": True
}
async with session.post(
f"{self.BASE_URL}/chat/completions",
json=payload
) as response:
async for line in response.content:
if line:
line = line.decode("utf-8").strip()
if line.startswith("data: "):
if line == "data: [DONE]":
break
data = json.loads(line[6:])
delta = data.get("choices", [{}])[0].get("delta", {})
if "content" in delta:
yield delta["content"]
Voice assistant system prompt
VOICE_ASSISTANT_SYSTEM = """Bạn là một voice assistant thông minh, trả lời ngắn gọn, tự nhiên.
- Trả lời trong tiếng Việt
- Giọng văn thân thiện, gần gũi
- Câu trả lời ngắn (dưới 50 từ) để phù hợp với TTS
- Không sử dụng markdown formatting
"""
async def process_user_input(user_text: str, api_key: str) -> str:
"""
Full pipeline: user speech → LLM → response
"""
llm = HolySheepLLM(api_key)
messages = [
{"role": "system", "content": VOICE_ASSISTANT_SYSTEM},
{"role": "user", "content": user_text}
]
result = await llm.chat_completion(messages)
response_text = result["choices"][0]["message"]["content"]
return response_text
6. Kiến trúc xử lý đồng thời với asyncio
Để handle nhiều concurrent users, cần implement connection pooling và queue management. Dưới đây là architecture với Redis cho distributed processing.
# voice_assistant_server.py
import asyncio
import uuid
import time
from dataclasses import dataclass, field
from typing import Dict, Optional, AsyncGenerator
from collections import deque
import redis.asyncio as redis
import json
@dataclass
class VoiceSession:
"""Quản lý state cho mỗi voice session"""
session_id: str
user_id: str
created_at: float = field(default_factory=time.time)
last_activity: float = field(default_factory=time.time)
message_history: list = field(default_factory=list)
audio_buffer: bytes = b""
def is_expired(self, ttl: int = 300) -> bool:
"""Session expires sau 5 phút không activity"""
return (time.time() - self.last_activity) > ttl
class VoiceAssistantServer:
"""
Production voice assistant server với:
- Concurrent session management
- Redis-backed state storage
- Auto-scaling support
"""
def __init__(
self,
redis_url: str = "redis://localhost:6379",
max_concurrent: int = 100,
audio_queue_size: int = 50
):
self.redis_url = redis_url
self.max_concurrent = max_concurrent
self.sessions: Dict[str, VoiceSession] = {}
self.session_lock = asyncio.Lock()
self.audio_queue: asyncio.Queue = asyncio.Queue(maxsize=audio_queue_size)
self.active_connections = 0
async def start(self):
"""Initialize server components"""
self.redis_client = await redis.from_url(
self.redis_url,
encoding="utf-8",
decode_responses=True
)
# Start audio processor workers
for i in range(4): # 4 worker processes
asyncio.create_task(self._audio_worker(i))
async def _audio_worker(self, worker_id: int):
"""Worker process audio từ queue"""
print(f"Audio worker {worker_id} started")
while True:
try:
audio_data, session_id = await asyncio.wait_for(
self.audio_queue.get(),
timeout=5.0
)
# Process audio
await self._process_audio_chunk(audio_data, session_id)
self.audio_queue.task_done()
except asyncio.TimeoutError:
continue
except Exception as e:
print(f"Worker {worker_id} error: {e}")
async def _process_audio_chunk(self, audio: bytes, session_id: str):
"""Process audio chunk - Whisper → LLM → TTS pipeline"""
# Transcribe
whisper_result = await self.whisper.transcribe_stream(audio)
if not whisper_result["text"]:
return
# Get session
async with self.session_lock:
session