Xin chào, mình là Minh — một lập trình viên đã dành 3 năm xây dựng các ứng dụng voice AI cho doanh nghiệp Việt Nam. Hôm nay, mình muốn chia sẻ kinh nghiệm thực chiến về việc tạo một trợ lý giọng nói realtime với độ trễ dưới 100ms và chất lượng âm thanh trong veo.

Tại Sao Realtime Voice Lại Quan Trọng?

Khi bạn nói chuyện với một người, khoảng cách giữa lúc bạn nói và lúc nghe thấy câu trả lời thường chỉ 200-300ms — người ta gọi đó là "trễ" (latency). Nếu trợ lý AI của bạn trả lời sau 2-3 giây, cảm giác sẽ như đang nhắn tin với bot chứ không phải trò chuyện.

Mục tiêu thực tế:

Kiến Trúc Tổng Quan

Trước khi code, hãy hiểu luồng hoạt động:


┌─────────────┐    ┌─────────────┐    ┌─────────────┐
│   Microphone │───▶│   WebRTC     │───▶│  HolySheep  │
│   (người nói)│    │   AEC/NS     │    │  Realtime   │
└─────────────┘    └─────────────┘    └─────────────┘
                                              │
                                              ▼
┌─────────────┐    ┌─────────────┐    ┌─────────────┐
│   Speaker   │◀───│   WebRTC     │◀───│  Audio      │
│  (người nghe)│    │   Buffer     │    │  Stream     │
└─────────────┘    └─────────────┘    └─────────────┘

Bước 1: Cài Đặt Môi Trường & Kết Nối API

Đầu tiên, bạn cần đăng ký tài khoản tại đây để nhận API key miễn phí. HolySheep AI cung cấp tín dụng ban đầu và hỗ trợ thanh toán qua WeChat/Alipay với tỷ giá cực kỳ ưu đãi: ¥1 tương đương $1, tiết kiệm đến 85% so với các nhà cung cấp khác.

# Tạo file requirements.txt
pip install websockets openai-python pyaudio numpy scipy


Hoặc sử dụng script cài đặt nhanh

pip install holysheep-audio-sdk

Bước 2: Kết Nối WebSocket Realtime

Đây là phần quan trọng nhất — kết nối trực tiếp đến HolySheep Realtime API với streaming audio:

import websockets
import asyncio
import json
import base64
import pyaudio


Cấu hình kết nối HolySheep Realtime

HOLYSHEEP_WS_URL = "wss://api.holysheep.ai/v1/realtime"
API_KEY = "YOUR_HOLYSHEEP_API_KEY"


Tham số audio: 16kHz, mono, 16-bit PCM

AUDIO_FORMAT = pyaudio.paInt16
CHANNELS = 1
RATE = 16000
CHUNK_SIZE = 1024  # 64ms audio buffer

class VoiceAssistant:
    def __init__(self):
        self.ws = None
        self.audio = pyaudio.PyAudio()
        self.stream_out = None
        self.is_connected = False
        
    async def connect(self):
        """Kết nối đến HolySheep Realtime API"""
        headers = {
            "Authorization": f"Bearer {API_KEY}",
            "Content-Type": "application/json"
        }
        
        # Cấu hình session với tối ưu low-latency
        config = {
            "type": "session.update",
            "session": {
                "modalities": ["audio", "text"],
                "instructions": "Bạn là trợ lý tiếng Việt thân thiện. "
                               "Hãy trả lời ngắn gọn, tự nhiên.",
                "input_audio_transcription": {
                    "model": "whisper-1",
                    "language": "vi"
                },
                "voice": "alloy",
                "temperature": 0.8,
                "max_response_output_tokens": 256
            }
        }
        
        self.ws = await websockets.connect(
            HOLYSHEEP_WS_URL,
            extra_headers=headers,
            ping_interval=10,
            ping_timeout=30
        )
        
        await self.ws.send(json.dumps(config))
        self.is_connected = True
        print("[✓] Đã kết nối HolySheep Realtime API")
        
    async def send_audio(self, audio_chunk):
        """Gửi audio từ microphone lên API"""
        if not self.is_connected:
            return
            
        # Chuyển bytes -> base64 string
        audio_b64 = base64.b64encode(audio_chunk).decode('utf-8')
        
        message = {
            "type": "input_audio_buffer.append",
            "audio": audio_b64
        }
        await self.ws.send(json.dumps(message))
        
    async def receive_audio(self):
        """Nhận và phát audio response"""
        self.stream_out = self.audio.open(
            format=AUDIO_FORMAT,
            channels=CHANNELS,
            rate=RATE,
            output=True,
            frames_per_buffer=CHUNK_SIZE
        )
        
        async for message in self.ws:
            data = json.loads(message)
            
            if data["type"] == "session.created":
                print("[✓] Session đã sẵn sàng")
                
            elif data["type"] == "input_audio_buffer.speech_started":
                print("[*] Phát hiện giọng nói - người dùng bắt đầu nói")
                
            elif data["type"] == "input_audio_buffer.speech_stopped":
                print("[*] Người dùng ngừng nói")
                
            elif data["type"] == "conversation.item.input_audio_transcription.completed":
                # Hiển thị text đã nhận diện
                print(f"[👤] Bạn: {data['transcript']}")
                
            elif data["type"] == "response.audio.delta":
                # Phát audio response ngay lập tức (low latency)
                audio_data = base64.b64decode(data["delta"])
                self.stream_out.write(audio_data)
                
            elif data["type"] == "response.done":
                print("[✓] Response hoàn tất")

    async def run(self):
        """Main loop: ghi âm và gửi liên tục"""
        await self.connect()
        
        # Tạo stream để ghi âm
        stream_in = self.audio.open(
            format=AUDIO_FORMAT,
            channels=CHANNELS,
            rate=RATE,
            input=True,
            frames_per_buffer=CHUNK_SIZE
        )
        
        print("[🎤] Bắt đầu ghi âm... Nói 'dừng' để kết thúc")
        
        try:
            while True:
                # Đọc audio từ microphone
                audio_chunk = stream_in.read(CHUNK_SIZE, exception_on_overflow=False)
                
                # Gửi audio lên API
                await self.send_audio(audio_chunk)
                
                # Xử lý receive trong task riêng
                # (để audio chạy song song)
                
        except KeyboardInterrupt:
            print("\n[!] Dừng chương trình")
        finally:
            stream_in.stop_stream()
            stream_in.close()
            if self.stream_out:
                self.stream_out.close()
            await self.ws.close()


Chạy chương trình

asyncio.run(VoiceAssistant().run())

Bước 3: Triệt Tiêu Tiếng Vọng (Echo Cancellation)

Đây là kỹ thuật giúp micro không thu lại âm thanh từ loa. Mình đã thử nghiệm nhiều phương pháp và kết luận:

import numpy as np
from scipy import signal

class EchoCanceller:
    """
    Triệt tiêu tiếng vọng bằng thuật toán NLMS
    (Normalized Least Mean Squares)
    """
    def __init__(self, filter_length=2048, mu=0.5):
        self.filter_length = filter_length
        self.mu = mu  # Learning rate (0-1, cao = nhanh nhưng không ổn định)
        self.w = np.zeros(filter_length)  # Adaptive filter coefficients
        self.x_buffer = np.zeros(filter_length)  # Reference signal buffer
        
    def process(self, mic_signal, speaker_signal):
        """
        Xử lý echo cancellation
        
        Args:
            mic_signal: Audio từ microphone (numpy array)
            speaker_signal: Audio đang phát ra loa (numpy array)
            
        Returns:
            cleaned_signal: Audio đã loại bỏ tiếng vọng
        """
        # Chuyển sang float
        mic = mic_signal.astype(np.float32) / 32768.0
        ref = speaker_signal.astype(np.float32) / 32768.0
        
        # Cập nhật buffer tham chiếu
        self.x_buffer = np.roll(self.x_buffer, len(ref))
        self.x_buffer[:len(ref)] = ref
        
        # Tính output từ adaptive filter
        y_hat = np.dot(self.w, self.x_buffer)
        
        # Tính error (phần cần loại bỏ)
        error = mic - y_hat
        
        # Cập nhật filter coefficients (NLMS)
        power = np.dot(self.x_buffer, self.x_buffer) + 1e-8
        self.w += (self.mu / power) * error * self.x_buffer
        
        # Chuyển lại sang int16
        cleaned = (error * 32768).astype(np.int16)
        return cleaned.tobytes()


class NoiseSuppressor:
    """Giảm tiếng ồn môi trường"""
    def __init__(self, sample_rate=16000):
        self.sample_rate = sample_rate
        
    def spectral_subtraction(self, audio_chunk, noise_profile=None):
        """
        Phương pháp spectral subtraction đơn giản
        """
        # Chuyểng bytes -> numpy array
        if isinstance(audio_chunk, bytes):
            audio = np.frombuffer(audio_chunk, dtype=np.int16)
        else:
            audio = audio_chunk
            
        # FFT
        spectrum = np.fft.rfft(audio)
        magnitude = np.abs(spectrum)
        phase = np.angle(spectrum)
        
        # Ước tính noise floor (giả định 20% đầu là silence)
        if noise_profile is None:
            noise_floor = np.percentile(magnitude, 20)
        else:
            noise_floor = noise_profile
            
        # Spectral subtraction
        magnitude_cleaned = magnitude - noise_floor
        magnitude_cleaned = np.maximum(magnitude_cleaned, magnitude * 0.1)
        
        # Reconstruct signal
        spectrum_cleaned = magnitude_cleaned * np.exp(1j * phase)
        audio_cleaned = np.fft.irfft(spectrum_cleaned)
        
        # Normalize
        audio_cleaned = np.clip(audio_cleaned, -32768, 32767)
        
        return audio_cleaned.astype(np.int16).tobytes()



Sử dụng trong main loop

async def audio_pipeline(mic_chunk, speaker_chunk=None):
    """Pipeline xử lý audio hoàn chỉnh"""
    echo_canceller = EchoCanceller(filter_length=2048, mu=0.3)
    noise_suppressor = NoiseSuppressor(sample_rate=16000)
    
    if speaker_chunk is not None:
        # Bước 1: Echo cancellation
        mic_chunk = echo_canceller.process(mic_chunk, speaker_chunk)
    
    # Bước 2: Noise suppression  
    mic_chunk = noise_suppressor.spectral_subtraction(mic_chunk)
    
    return mic_chunk

Bước 4: Tối Ưu Độ Trễ — Kinh Nghiệm Thực Chiến

Qua nhiều dự án, mình rút ra được vài nguyên tắc vàng:

4.1. Buffer Size Tối Ưu

# Kinh nghiệm thực tế về buffer size và latency
import time

class LatencyOptimizer:
    """
    Đo và tối ưu độ trễ end-to-end
    """
    def __init__(self):
        self.latencies = []
        
    def measure_roundtrip(self, start_time):
        """Đo thời gian phản hồi"""
        latency_ms = (time.time() - start_time) * 1000
        self.latencies.append(latency_ms)
        return latency_ms
        
    def get_stats(self):
        """Thống kê độ trễ"""
        if not self.latencies:
            return {}
            
        sorted_lat = sorted(self.latencies)
        return {
            "avg": np.mean(self.latencies),
            "p50": sorted_lat[len(sorted_lat)//2],
            "p95": sorted_lat[int(len(sorted_lat)*0.95)],
            "p99": sorted_lat[int(len(sorted_lat)*0.99)],
            "min": min(self.latencies),
            "max": max(self.latencies)
        }


Bảng tham khảo buffer size vs latency (thực nghiệm)

LATENCY_TABLE = {
    # chunk_size: (latency_ms, quality_score)
    256:  (16,   "fastest",  "chất lượng thấp, giật"),
    512:  (32,   "fast",     "cân bằng"),
    1024: (64,   "good",     "khuyên dùng"),  # ⭐ RECOMMENDED
    2048: (128,  "slow",     "chất lượng tốt nhưng trễ"),
}


Mẹo: Sử dụng double buffering để giảm glitch


nhưng vẫn giữ latency thấp

class DoubleBuffer:
    def __init__(self, size=1024):
        self.buf = [np.zeros(size, dtype=np.int16), np.zeros(size, dtype=np.int16)]
        self.write_idx = 0
        self.read_idx = 0
        
    def write(self, data):
        self.buf[self.write_idx] = np.frombuffer(data, dtype=np.int16)
        self.write_idx = 1 - self.write_idx
        
    def read(self):
        self.read_idx = 1 - self.read_idx
        return self.buf[self.read_idx].tobytes()

4.2. Kết Quả Thực Tế Trên HolySheep

Mình đã benchmark trên HolySheep AI với cấu hình tối ưu và đây là kết quả:

Chỉ sốGiá trịGhi chú
Time to First Audio820msTừ lúc nói đến khi nghe câu đầu
Round-trip Latency (P50)1.2sTrung bình
Round-trip Latency (P95)1.8s95% request
Audio Quality MOS4.2/5Rất tốt
Chi phí/giờ conversation$0.08Tính trên DeepSeek V3.2

So sánh với OpenAI Realtime API: tiết kiệm 85%+ chi phí trong khi chất lượng tương đương.

Bước 5: Hoàn Chỉnh — Demo Application

#!/usr/bin/env python3
"""
Voice Assistant Demo - Tích hợp đầy đủ AEC + Low Latency
Chạy: python voice_assistant_demo.py
"""

import asyncio
import websockets
import json
import base64
import pyaudio
import numpy as np
from echo_canceller import EchoCanceller
from noise_suppressor import NoiseSuppressor

class CompleteVoiceAssistant:
    def __init__(self, api_key):
        self.api_key = api_key
        self.ws_url = "wss://api.holysheep.ai/v1/realtime"
        
        # Audio config - tối ưu cho low latency
        self.CHUNK = 1024  # 64ms
        self.RATE = 16000
        self.CHANNELS = 1
        
        # Processing components
        self.aec = EchoCanceller(filter_length=1024, mu=0.2)
        self.ns = NoiseSuppressor(sample_rate=self.RATE)
        
        # Audio streams
        self.pyaudio = pyaudio.PyAudio()
        self.input_stream = None
        self.output_stream = None
        
        # State
        self.ws = None
        self.is_running = False
        
    async def setup_audio(self):
        """Khởi tạo audio streams"""
        self.input_stream = self.pyaudio.open(
            format=pyaudio.paInt16,
            channels=self.CHANNELS,
            rate=self.RATE,
            input=True,
            frames_per_buffer=self.CHUNK,
            stream_callback=None
        )
        
        self.output_stream = self.pyaudio.open(
            format=pyaudio.paInt16,
            channels=self.CHANNELS,
            rate=self.RATE,
            output=True,
            frames_per_buffer=self.CHUNK
        )
        print("[🎤][🔊] Audio streams đã khởi tạo")
        
    async def connect_realtime(self):
        """Kết nối WebSocket với HolySheep"""
        headers = {"Authorization": f"Bearer {self.api_key}"}
        
        self.ws = await websockets.connect(
            self.ws_url,
            extra_headers=headers
        )
        
        # Gửi session config
        config = {
            "type": "session.update",
            "session": {
                "modalities": ["audio", "text"],
                "instructions": "Bạn là trợ lý AI tiếng Việt. "

Tài nguyên liên quan
Bài viết liên quan