Real-time Voice AI 2025: Xây dựng Voice Assistant như Siri/Alexa với OpenAI Realtime API

Real-time Voice AI là công nghệ cho phép tương tác bằng giọng nói theo thời gian thực, tạo ra trải nghiệm tự nhiên như nói chuyện với con người.

PHẦN 1: REAL-TIME VOICE AI LÀ GÌ?

Định nghĩa:

Real-time Voice AI là hệ thống AI có khả năng:

• Nghe: Nhận diện giọng nói real-time (Speech-to-Text).
• Hiểu: Xử lý ngôn ngữ tự nhiên (NLU).
• Trả lời: Generate response (LLM).
• Nói: Chuyển text thành giọng nói (Text-to-Speech).

Tất cả trong < 1 giây!

Kiến trúc truyền thống (Multi-step):

User nói → Whisper (STT) → ChatGPT (LLM) → ElevenLabs (TTS) → Audio output
Latency: 2-5 giây

Kiến trúc Real-time (Single-step):

User nói → OpenAI Realtime API → Audio output
Latency: < 1 giây

---

PHẦN 2: OPENAI REALTIME API

Giới thiệu:

OpenAI Realtime API (ra mắt 10/2024) là API đầu tiên tích hợp:

• Speech-to-Speech: Giọng nói → Giọng nói (không qua text).
• Low latency: < 500ms.
• Interruption: Có thể ngắt lời AI.

Model:

• gpt-4o-realtime-preview: Model duy nhất hiện tại.
• Voice: 6 giọng có sẵn (alloy, echo, fable, onyx, nova, shimmer).

Pricing:

• Input audio: $100 / 1M tokens (~$0.06/phút).
• Output audio: $200 / 1M tokens (~$0.12/phút).
• Text input/output: $5 / 1M tokens.

Tổng: ~$0.18/phút (đắt hơn multi-step 3-4 lần, nhưng nhanh hơn).

---

PHẦN 3: HƯỚNG DẪN SỬ DỤNG REALTIME API

Setup:

pip install openai websockets

Code Example 1: Basic Voice Chat

import asyncio
import websockets
import json
import base64
import pyaudio

OPENAI_API_KEY = "your_api_key"

async def voice_chat():
    # Connect to Realtime API
    url = "wss://api.openai.com/v1/realtime?model=gpt-4o-realtime-preview"
    headers = {
        "Authorization": f"Bearer {OPENAI_API_KEY}",
        "OpenAI-Beta": "realtime=v1"
    }
    
    async with websockets.connect(url, extra_headers=headers) as ws:
        # Configure session
        await ws.send(json.dumps({
            "type": "session.update",
            "session": {
                "modalities": ["text", "audio"],
                "voice": "alloy",
                "instructions": "Bạn là trợ lý AI thân thiện"
            }
        }))
        
        # Start audio stream
        audio = pyaudio.PyAudio()
        stream = audio.open(
            format=pyaudio.paInt16,
            channels=1,
            rate=24000,
            input=True,
            frames_per_buffer=1024
        )
        
        # Send audio
        while True:
            data = stream.read(1024)
            audio_b64 = base64.b64encode(data).decode()
            await ws.send(json.dumps({
                "type": "input_audio_buffer.append",
                "audio": audio_b64
            }))
            
            # Receive response
            response = await ws.recv()
            event = json.loads(response)
            
            if event["type"] == "response.audio.delta":
                # Play audio
                audio_data = base64.b64decode(event["delta"])
                # ... play audio_data

asyncio.run(voice_chat())

Code Example 2: Function Calling

# Define tools
tools = [{
    "type": "function",
    "name": "get_weather",
    "description": "Lấy thông tin thời tiết",
    "parameters": {
        "type": "object",
        "properties": {
            "location": {"type": "string"}
        }
    }
}]

# Send session update
await ws.send(json.dumps({
    "type": "session.update",
    "session": {
        "tools": tools,
        "tool_choice": "auto"
    }
}))

# Handle function call
if event["type"] == "response.function_call_arguments.done":
    function_name = event["name"]
    arguments = json.loads(event["arguments"])
    
    # Execute function
    if function_name == "get_weather":
        result = get_weather(arguments["location"])
        
        # Send result back
        await ws.send(json.dumps({
            "type": "conversation.item.create",
            "item": {
                "type": "function_call_output",
                "call_id": event["call_id"],
                "output": json.dumps(result)
            }
        }))

---

PHẦN 4: ALTERNATIVE - BUILD CUSTOM PIPELINE

Nếu không muốn dùng Realtime API (đắt), bạn có thể build pipeline riêng:

Tech Stack:

1. Speech-to-Text:

• Whisper (local): Miễn phí, latency ~2s.
• Deepgram: $0.0043/phút, latency ~300ms.
• AssemblyAI: $0.00025/giây, latency ~500ms.

2. LLM:

• GPT-4 Turbo: $0.01/1K tokens.
• Claude 3 Haiku: $0.00025/1K tokens (nhanh nhất).
• Groq (Llama 3): Miễn phí, cực nhanh (500 tokens/s).

3. Text-to-Speech:

• ElevenLabs Turbo: $0.15/1K chars, latency ~500ms.
• Google TTS: $4/1M chars, latency ~300ms.
• Edge TTS: Miễn phí, latency ~200ms.

Code Example: Custom Pipeline

import asyncio
from faster_whisper import WhisperModel
from openai import OpenAI
from elevenlabs import generate, stream

# Initialize
whisper = WhisperModel("base")
openai_client = OpenAI()

async def voice_assistant(audio_file):
    # 1. STT (Whisper)
    segments, _ = whisper.transcribe(audio_file)
    text = " ".join([s.text for s in segments])
    print(f"User: {text}")
    
    # 2. LLM (GPT-4)
    response = openai_client.chat.completions.create(
        model="gpt-4-turbo",
        messages=[{"role": "user", "content": text}],
        stream=True
    )
    
    # 3. TTS (ElevenLabs streaming)
    full_text = ""
    for chunk in response:
        if chunk.choices[0].delta.content:
            full_text += chunk.choices[0].delta.content
    
    audio = generate(text=full_text, voice="Bella", stream=True)
    stream(audio)

asyncio.run(voice_assistant("user_audio.mp3"))

Latency: ~3-5 giây (chậm hơn Realtime API nhưng rẻ hơn 10x).

---

PHẦN 5: USE CASES THỰC TẾ

1. Voice Chatbot (Customer Support)

Tính năng:

• Trả lời câu hỏi khách hàng 24/7.
• Tự động xử lý 80% câu hỏi thường gặp.
• Chuyển sang human nếu phức tạp.

Tech Stack:

• Deepgram (STT) + GPT-4 + ElevenLabs (TTS).
• Twilio (tích hợp điện thoại).

2. Voice Assistant (Smart Home)

Tính năng:

• Điều khiển đèn, điều hòa, TV...
• Đặt lịch, nhắc nhở.
• Trả lời câu hỏi.

Tech Stack:

• Whisper (local) + Llama 3 (local) + Edge TTS.
• Home Assistant (smart home platform).

3. Language Learning App

Tính năng:

• Luyện nói tiếng Anh với AI.
• AI sửa phát âm, ngữ pháp.
• Đánh giá điểm số.

Tech Stack:

• Whisper + GPT-4 + ElevenLabs.
• Pronunciation assessment API.

4. Voice-to-Voice Translation

Tính năng:

• Nói tiếng Việt → AI nói tiếng Anh.
• Real-time, giữ nguyên giọng điệu.

Tech Stack:

• Whisper (STT) + GPT-4 (translate) + ElevenLabs Voice Clone.

---

PHẦN 6: SO SÁNH CÔNG CỤ

Tiêu chí	OpenAI Realtime	Custom Pipeline	Deepgram + GPT-4
Latency	< 500ms	3-5s	1-2s
Chi phí	$0.18/phút	$0.02/phút	$0.05/phút
Chất lượng	⭐⭐⭐⭐⭐	⭐⭐⭐⭐	⭐⭐⭐⭐⭐
Interruption	✅	❌	❌
Customization	❌	✅✅✅	✅✅

Kết luận:

• OpenAI Realtime: Tốt nhất cho UX, nhưng đắt.
• Custom Pipeline: Rẻ, linh hoạt, nhưng phức tạp.
• Deepgram + GPT-4: Cân bằng tốt.

---

PHẦN 7: CHALLENGES & SOLUTIONS

Challenge 1: Latency cao

Solutions:

• Dùng streaming (STT, LLM, TTS đều stream).
• Dùng model nhỏ, nhanh (Whisper base, Claude Haiku).
• Optimize network (WebSocket, HTTP/2).

Challenge 2: Chi phí cao

Solutions:

• Dùng open-source (Whisper local, Llama 3).
• Cache responses phổ biến.
• Limit usage (rate limiting).

Challenge 3: Interruption

Solutions:

• Dùng VAD (Voice Activity Detection) để detect khi user nói.
• Stop TTS ngay lập tức.
• Clear buffer.

Challenge 4: Accent/Noise

Solutions:

• Noise reduction (ffmpeg, noisereduce).
• Dùng model tốt (Whisper large, Deepgram Nova).
• Prompt engineering (gợi ý context).

---

PHẦN 8: BEST PRACTICES

1. UX Design:

• Visual feedback: Hiển thị waveform khi user nói.
• Interruption: Cho phép user ngắt lời AI.
• Fallback: Nếu không hiểu, hỏi lại thay vì trả lời sai.

2. Performance:

• Streaming: Stream STT, LLM, TTS để giảm latency.
• Caching: Cache responses phổ biến.
• CDN: Serve audio từ CDN.

3. Security:

• Authentication: Xác thực user trước khi cho phép voice access.
• Rate limiting: Giới hạn số request/phút.
• Content filtering: Lọc nội dung nhạy cảm.

4. Monitoring:

• Latency: Track latency từng component (STT, LLM, TTS).
• Error rate: Monitor lỗi (STT fail, LLM timeout...).
• Cost: Track chi phí real-time.

---

PHẦN 9: CODE EXAMPLE - FULL VOICE ASSISTANT

import asyncio
from faster_whisper import WhisperModel
from openai import OpenAI
from elevenlabs import generate, stream
import sounddevice as sd
import numpy as np

class VoiceAssistant:
    def __init__(self):
        self.whisper = WhisperModel("base")
        self.openai = OpenAI()
        self.conversation = []
    
    def record_audio(self, duration=5):
        """Record audio from microphone"""
        print("Listening...")
        audio = sd.rec(
            int(duration * 16000),
            samplerate=16000,
            channels=1
        )
        sd.wait()
        return audio
    
    def transcribe(self, audio):
        """STT"""
        segments, _ = self.whisper.transcribe(audio)
        return " ".join([s.text for s in segments])
    
    def chat(self, text):
        """LLM"""
        self.conversation.append({"role": "user", "content": text})
        response = self.openai.chat.completions.create(
            model="gpt-4-turbo",
            messages=self.conversation
        )
        reply = response.choices[0].message.content
        self.conversation.append({"role": "assistant", "content": reply})
        return reply
    
    def speak(self, text):
        """TTS"""
        audio = generate(text=text, voice="Bella")
        stream(audio)
    
    async def run(self):
        """Main loop"""
        while True:
            # Record
            audio = self.record_audio()
            
            # STT
            text = self.transcribe(audio)
            print(f"User: {text}")
            
            if "tạm biệt" in text.lower():
                break
            
            # LLM
            reply = self.chat(text)
            print(f"AI: {reply}")
            
            # TTS
            self.speak(reply)

# Run
assistant = VoiceAssistant()
asyncio.run(assistant.run())

---

KẾT LUẬN

Real-time Voice AI đang thay đổi cách chúng ta tương tác với máy tính. Với OpenAI Realtime API và các công cụ mã nguồn mở, việc build voice assistant giờ đây dễ dàng hơn bao giờ hết.

Lời khuyên:

• Prototype: Dùng OpenAI Realtime API để test nhanh.
• Production: Build custom pipeline để tiết kiệm chi phí.
• Optimize: Focus vào latency và UX.