Posted inAI

Xây dựng Voice Agent thời gian thực với LiveKit Agents và Python: Hướng dẫn Conversational AI độ trễ siêu thấp

AI tutorial - IT technology blog
AI tutorial - IT technology blog

Thực trạng: Vì sao các Voice Bot truyền thống mang lại trải nghiệm rời rạc

Phần lớn các voice bot được phát triển trong thập kỷ qua đều phụ thuộc vào vòng lặp request-response vụng về. Phía client ghi âm cho đến khi người dùng ngừng nói, đóng gói thành file WAV rồi gửi một HTTP POST request. Từ đó, backend phải chờ phiên âm speech-to-text (STT), gửi truy vấn tới LLM, rồi mới tổng hợp âm thanh text-to-speech (TTS). Người gọi thường phải chờ đợi trong khoảng lặng kéo dài từ 3 đến 6 giây trước khi nghe thấy âm tiết đầu tiên.

Giao tiếp giữa con người không thể chấp nhận mức độ trễ đó. Trong một cuộc đối thoại tự nhiên, khoảng dừng khi đổi lượt nói (turn-taking) chỉ rơi vào khoảng 200 đến 300 mili-giây. Khi khoảng lặng vượt quá 600 mili-giây, cuộc trò chuyện lập tức trở nên ngượng ngập và máy móc.

Tệ hơn nữa, kiến trúc REST thông thường hoàn toàn bất lực khi người dùng ngắt lời. Nếu có ai đó chen ngang khi bot đang nói dở câu, client vẫn tiếp tục phát phần âm thanh cũ cho đến khi xả hết buffer stream. Để xây dựng voice agent đạt chuẩn production, bạn cần giải quyết hai bài toán hóc búa cốt lõi: cắt giảm buffer âm thanh tức thời (on the fly) và truyền tải raw audio qua giao thức có độ trễ cực thấp.

Các khái niệm cốt lõi: WebRTC, VAD và Framework LiveKit

Để đạt được giọng nói tương tác dưới một giây (sub-second), hệ thống bắt buộc phải streaming audio liên tục. Ba lựa chọn kiến trúc sau đây sẽ biến điều này thành hiện thực:

1. WebRTC đóng vai trò tầng truyền tải thời gian thực (Real-Time Transport)

REST và WebSocket hoạt động trên giao thức TCP. Khi xảy ra sự cố mạng, TCP sẽ tạm dừng để truyền lại các gói tin bị mất, gây ra hiện tượng giật lag âm thanh (jitter) rất rõ rệt. WebRTC loại bỏ cơ chế truyền lại của TCP để chuyển sang các kênh RTP và SRTP dựa trên UDP. Nhờ vào jitter buffer nguyên bản và khả năng streaming bitrate thích ứng (adaptive bitrate), độ trễ khứ hồi (round-trip) của âm thanh luôn được duy trì dưới 120 mili-giây trên đường truyền ổn định.

2. Pipeline xử lý giọng nói dạng Streaming

Hãy quên cách xử lý batch tuần tự đi. Các voice engine hiệu năng cao sẽ stream từng chunk âm thanh 20ms qua 4 giai đoạn pipeline:

  • Voice Activity Detection (VAD): Kiểm tra các khung âm thanh raw 16kHz PCM. Silero VAD chạy trực tiếp trên CPU, nhận diện giọng nói con người chỉ trong vòng 30 mili-giây.
  • Streaming Speech-to-Text (STT): Phiên âm các chunk âm thanh từng phần ngay khi người gọi vừa cất tiếng. Các nhà cung cấp như Deepgram Nova-2 trả về bản phiên âm từ tạm thời (interim transcript) trong khoảng 150 đến 200 mili-giây.
  • Streaming Large Language Model (LLM): Tiếp nhận các đoạn hội thoại hoàn chỉnh và stream token ra ngay lập tức. Những mô hình tốc độ cao như GPT-4o-mini trả về token đầu tiên chỉ trong 250 đến 400 mili-giây.
  • Streaming Text-to-Speech (TTS): Tổng hợp âm thanh theo từng cụm từ thay vì chờ đợi toàn bộ đoạn văn bản. Pipeline sẽ stream buffer âm thanh đầu tiên về client chỉ trong khoảng 180 mili-giây.

3. Kiến trúc LiveKit Agents

LiveKit cung cấp một WebRTC media server mã nguồn mở đi kèm với framework Python (livekit-agents). Bạn sẽ không phải tự tay viết code xử lý ICE negotiation, peer signaling hay track-mixing phức tạp. Thay vào đó, service Python của bạn sẽ đóng vai trò như một thành viên tự động tham gia vào phòng WebRTC. Khi người dùng kết nối, LiveKit server sẽ phân phối job tới một worker process. Worker này sẽ gắn các audio track, khởi động pipeline và hướng tới tổng thời gian phản hồi từ giọng nói đến giọng nói (voice-to-voice turnaround) dưới 800 mili-giây.

Thực hành: Xây dựng Voice Agent thời gian thực bằng Python

Bước 1: Thiết lập môi trường

Thiết lập một môi trường ảo sạch sử dụng Python 3.10 trở lên. Cài đặt LiveKit agent runtime cùng với các plugin chính thức từ các nhà cung cấp:

mkdir voice-agent && cd voice-agent
python3 -m venv .venv
source .venv/bin/activate

pip install livekit-agents livekit-plugins-openai livekit-plugins-silero livekit-plugins-deepgram python-dotenv

Bước 2: Cấu hình biến môi trường

Cấu hình thông tin xác thực cho LiveKit instance của bạn (cloud hoặc self-hosted) và các mô hình AI. Lưu các biến này vào file .env:

LIVEKIT_URL=wss://your-project.livekit.cloud
LIVEKIT_API_KEY=your_api_key
LIVEKIT_API_SECRET=your_api_secret

OPENAI_API_KEY=your_openai_key
DEEPGRAM_API_KEY=your_deepgram_key

Bước 3: Viết mã nguồn cho Agent Worker

Lưu đoạn mã sau vào file agent.py. Chúng ta sẽ sử dụng VoicePipelineAgent để điều phối Silero VAD, Deepgram STT và các mô hình OpenAI thành một vòng lặp có thể ngắt quãng (interruptible loop):

import os
from dotenv import load_dotenv
from livekit.agents import (
    AutoSubscribe,
    JobContext,
    JobProcess,
    WorkerOptions,
    cli,
    llm,
)
from livekit.agents.pipeline import VoicePipelineAgent
from livekit.plugins import deepgram, openai, silero

load_dotenv()

def prewarm_process(proc: JobProcess):
    # Tải trước trọng số VAD vào bộ nhớ
    # để loại bỏ độ trễ cold-start khi người gọi kết nối
    proc.userdata["vad"] = silero.VAD.load()

async def entrypoint(ctx: JobContext):
    # Tham gia phòng và thu audio track của người gọi
    await ctx.connect(auto_subscribe=AutoSubscribe.AUDIO_ONLY)

    initial_chat_ctx = llm.ChatContext().append(
        role="system",
        text=(
            "Bạn là trợ lý giọng nói hỗ trợ kỹ thuật tự động cho một đội ngũ hạ tầng. "
            "Hãy giữ câu trả lời ngắn gọn, rõ ràng và trực tiếp. Tránh dùng markdown, danh sách hoặc emoji, "
            "vì nội dung sẽ được phát trực tiếp qua âm thanh."
        ),
    )

    agent = VoicePipelineAgent(
        vad=ctx.proc.userdata["vad"],
        stt=deepgram.STT(model="nova-2"),
        llm=openai.LLM(model="gpt-4o-mini"),
        tts=openai.TTS(voice="alloy"),
        chat_ctx=initial_chat_ctx,
        allow_interruptions=True,
        interrupt_speech_duration=0.5,
    )

    agent.start(ctx.room)
    await agent.say("Xin chào! Tôi đã kết nối và sẵn sàng. Hôm nay bạn cần hỗ trợ xử lý sự cố cho hệ thống nào?")

if __name__ == "__main__":
    cli.run_app(WorkerOptions(entrypoint_fnc=entrypoint, prewarm_fnc=prewarm_process))

Bước 4: Chạy và kiểm thử Agent

Chạy agent ở môi trường local trong chế độ phát triển (dev mode):

python agent.py dev

Bạn không cần phải tự dựng frontend để kiểm thử. Hãy truy cập agents-playground.livekit.io, dán URL và thông tin xác thực API của bạn vào rồi bấm kết nối.

Hãy nói vào tai nghe và quan sát pipeline hoạt động:

  1. Deepgram in văn bản nhận diện được ra terminal của bạn chỉ trong vòng 200ms.
  2. Silero VAD phát hiện các khoảng ngắt trong cuộc trò chuyện sau 500ms im lặng.
  3. OpenAI sinh các token, và âm thanh tổng hợp được phát lại trong khoảng 650 đến 800ms.
  4. Hãy thử nói khi bot đang trả lời. Agent sẽ nhận diện giọng nói của bạn, ngắt âm thanh đầu ra trong chưa đầy 100ms và bắt đầu lắng nghe.

Bước 5: Thêm Function Calling cho các tác vụ trong môi trường Production

Voice bot thực sự phát huy sức mạnh khi có thể truy vấn các hệ thống đang hoạt động ngay giữa cuộc trò chuyện. Tương tự như việc chuẩn hóa tích hợp công cụ cho AI Agent, hãy định nghĩa các công cụ Python với decorator @llm.ai_callable:

class ServerOps(llm.FunctionContext):
    @llm.ai_callable(description="Kiểm tra mức sử dụng CPU và bộ nhớ của một production host")
    def get_server_metrics(self, hostname: str) -> str:
        if "db" in hostname.lower():
            return f"{hostname} báo cáo mức sử dụng 88% CPU và 92% bộ nhớ."
        return f"{hostname} báo cáo mức sử dụng 14% CPU và 35% bộ nhớ."

# Truyền fnc_ctx vào VoicePipelineAgent:
# agent = VoicePipelineAgent(..., fnc_ctx=ServerOps())

Hãy thử hỏi: “Bạn có thể kiểm tra tình trạng của db-node-01 không?” Agent sẽ tạm dừng tổng hợp âm thanh, kích hoạt get_server_metrics, phân tích kết quả trả về và đọc lại các chỉ số một cách tự nhiên. Đây là nền tảng lý tưởng cho các giải pháp AIOps phân tích cảnh báo và tự động hóa vận hành hạ tầng.

Kết luận: Chuyển từ Prototype sang Hạ tầng Voice bền bỉ

Streaming media qua WebRTC đã thay đổi hoàn toàn cục diện của Conversational AI. Bằng cách ủy thác việc truyền tải media và phục hồi gói tin mạng cho LiveKit, bạn có thể duy trì độ trễ hội thoại khứ hồi thấp hơn nhiều so với ngưỡng phản hồi tự nhiên của con người. Đây cũng là bước đệm để mở rộng sang các hệ thống AI đa tác vụ (Multi-Agent) linh hoạt và mạnh mẽ hơn.

Trước khi triển khai lên môi trường production, hãy lưu ý 3 biến số vận hành sau:

  • Worker Prewarming: Tải trước trọng số VAD và tokenizer trong prewarm_fnc. Nếu bỏ qua bước này, kết nối đầu tiên sẽ phải chịu độ trễ từ 1,2 đến 2 giây.
  • Cloud Colocation: Đặt các worker node cùng khu vực với endpoint của mô hình AI sao cho độ trễ ping dưới 20ms (ví dụ: cùng đặt tại us-east-1 cho OpenAI và Deepgram) nhằm loại bỏ độ trễ do truyền dữ liệu liên vùng (cross-region).
  • Capacity Sizing: Việc xử lý khung âm thanh tiêu tốn lượng CPU ổn định. Một worker node tiêu chuẩn với 2 vCPU và 4GB RAM có thể phục vụ ổn định từ 15 đến 25 phòng thoại đồng thời trước khi cần kích hoạt autoscaling theo chiều ngang.
Share:
Tags: