Grok Voice Agent Builder: Hướng dẫn thực hành bằng Python

xAI đã ra mắt Voice Agent Builder, một bảng điều khiển để tạo tác tử giọng nói. Bạn mô tả luồng cuộc gọi, đính kèm tài liệu và công cụ, rồi chọn giọng đọc.

Khi tôi thử một bảng điều khiển tác tử giọng nói, tôi ít quan tâm đến ghi chú ra mắt mà tập trung vào các phần cần nối vào mã: cách cấu hình phiên WebSocket, cách luân chuyển âm thanh, nơi các lệnh gọi công cụ diễn ra, chi phí cuộc gọi là bao nhiêu, và một ứng dụng khác sẽ gọi workflow như thế nào.

Đoạn mã dưới đây dựng lại luồng đó trực tiếp qua Voice Agent API. Cụ thể, chúng ta sẽ dùng một trợ lý đặt lịch phòng khám kiểm tra khả dụng, phản hồi bằng giọng nói, theo dõi chi phí, xử lý lỗi công cụ và cung cấp một endpoint FastAPI.

Grok Voice Agent Builder là gì?

Voice Agent Builder là bảng điều khiển của xAI để tạo và triển khai tác tử giọng nói trên Grok Voice. Sản phẩm ra mắt bản beta vào ngày 1 tháng 7, 2026. Thay vì dùng riêng lẻ dịch vụ chuyển giọng nói thành văn bản, mô hình ngôn ngữ và chuyển văn bản thành giọng nói, nó đi theo một đường dẫn mô hình giọng nói duy nhất.

Bảng điều khiển bao gồm thoại qua mạng, truy xuất tài liệu, công cụ và connector, hàng rào an toàn, máy chủ MCP từ xa, cùng nhật ký cuộc gọi với bản ghi âm, bản chép và trace.

Âm thanh được tính phí theo phút. Vì bảng điều khiển vẫn là beta, chúng ta sẽ dùng trực tiếp API.

Cách Voice Agent API vận hành bên dưới Builder

Bên dưới bảng điều khiển là Voice Agent API, một API WebSocket thời gian thực cung cấp cùng runtime với Builder.

Builder nằm trên Voice API. Ảnh: Tác giả.

Mô hình dùng ở đây là grok-voice-think-fast-1.0. Bí danh grok-voice-latest trỏ đến mô hình mới nhất. Tôi dùng nó ở đây, nhưng với ứng dụng triển khai tôi sẽ cố định tên phiên bản. xAI báo điểm 67,3% cho mô hình này trên bảng xếp hạng τ-voice Bench; tôi xem đó là một dữ liệu tham khảo, không phải đảm bảo.

Lưu ý tương thích: API tương thích với OpenAI Realtime API. Nếu bạn có mã nói chuyện với endpoint realtime của OpenAI, chủ yếu chỉ cần đổi base URL và khóa.

Tổng quan dự án: Chúng ta sẽ xây gì

Trợ lý phòng khám nhận đầu vào giọng nói, trả lời bằng giọng tổng hợp, hỏi câu tiếp theo, kiểm tra khả dụng trước khi đề xuất khung giờ, và chuyển cho người thật khi cần. Ví dụ cốt lõi dùng một công cụ; bản demo Streamlit bổ sung hành động đặt lịch, chuyển máy và kết thúc cuộc gọi.

Hướng dẫn cốt lõi chia thành bốn tệp, mỗi tệp đảm nhiệm một việc:

• voice_client.py chứa client WebSocket, tiện ích âm thanh và theo dõi chi phí

• tools.py chứa check_availability, cùng các công cụ demo bổ sung dùng bởi Streamlit

• assistant.py chứa system prompt, cấu hình phiên và workflow

• app.py phục vụ toàn bộ qua FastAPI

Bốn tệp này chính là con đường đi xuyên bài viết. Repo cũng có app_streamlit.py cho bản demo trực quan và run.py làm trình khởi chạy Windows, nhưng chúng ta sẽ quay lại sau khi luồng cốt lõi chạy ổn.

Điều kiện tiên quyết

Trước khi chạy mã, bạn cần Python 3.10 trở lên, một tài khoản xAI, một API key từ console.x.ai, nạp trước tín dụng, và cơ bản quen với biến môi trường, JSON và WebSocket.

Thiết lập dự án

Tạo thư mục và môi trường ảo, rồi cài đặt các gói:

mkdir appointment-agent
cd appointment-agent
python -m venv .venv
.venv\Scripts\activate       # macOS/Linux: source .venv/bin/activate
pip install websockets python-dotenv fastapi uvicorn pydantic httpx numpy streamlit

Cố định phiên bản các gói này trong requirements.txt để mỗi lần checkout mới đều dùng cùng thiết lập.

Tạo tệp .env bên cạnh các tệp Python:

XAI_API_KEY=xai-your-key-here

Thêm .env vào .gitignore. API key nên ở lại trên máy chủ.

Xây dựng Voice Agent

Bắt đầu xây dựng thôi.

Kết nối tới Grok Voice Agent API qua WebSocket

Bước đầu là mở kết nối. Truyền mô hình dưới dạng tham số truy vấn và key của bạn làm bearer token trong bắt tay kết nối:

import asyncio
import json
import os
import websockets

async def voice_agent():
    url = "wss://api.x.ai/v1/realtime?model=grok-voice-latest"
    async with websockets.connect(
        url,
        additional_headers={"Authorization": f"Bearer {os.environ['XAI_API_KEY']}"},
    ) as ws:
        async for message in ws:
            print(json.loads(message)["type"])

asyncio.run(voice_agent())

Với key hợp lệ, sự kiện đầu tiên bạn thấy là session.created, nghĩa là socket đã mở và sẵn sàng cấu hình.

Sự kiện tạo phiên xác nhận kết nối. Ảnh: Tác giả.

Cấu hình phiên giọng nói

Một socket đang hoạt động chưa phải một tác tử đã cấu hình. Bạn định hình nó bằng cách gửi sự kiện session.update kèm đối tượng session.

Giọng đọc, định dạng âm thanh và hướng dẫn

Ba thiết lập bạn thường chỉnh là giọng, định dạng âm thanh và system prompt. API realtime cung cấp năm giọng đặt tên sẵn, eve, ara, rex, sal, và leo, cùng các bản clone tùy chỉnh. Âm thanh mặc định là audio/pcm ở 24000 Hz, cấu hình riêng cho đầu vào và đầu ra.

Đây là cấu hình phiên trợ lý dùng, lắp ráp trong assistant.py:

def build_session_config(voice="ara", instructions=SYSTEM_PROMPT, sample_rate=24000):
    # The model needs to know "today" or it guesses the year for a date like "July 6th".
    instructions = f"{instructions}\nToday's date is {date.today().isoformat()}."
    return {
        "voice": voice,
        "instructions": instructions,
        "turn_detection": None,  # manual turns for file-based input
        "audio": {
            "input": {"format": {"type": "audio/pcm", "rate": sample_rate}},
            "output": {"format": {"type": "audio/pcm", "rate": sample_rate}},
        },
        "tools": [CHECK_AVAILABILITY_TOOL],
    }

Trường instructions là system prompt. Prompt cho phòng khám được giữ ngắn gọn vì các câu trả lời dài bằng giọng nói khó theo dõi:

You are a voice appointment assistant for a small clinic. Help callers book,
reschedule, cancel, or ask questions about appointments, services, and hours.
Answer whatever the caller asks that relates to the clinic. Keep responses short
and natural for a phone conversation. Ask one question at a time. Confirm
important details before taking action. Use the availability tool before offering
a time slot. Escalate to a human for medical, urgent, sensitive, or unclear
requests. If a caller asks about something unrelated to the clinic, say briefly
that it is outside what you can help with, then steer back to booking. If you
cannot make out what the caller said, ask them to repeat it instead of repeating
your last message.

Câu về leo thang giúp tác tử của phòng khám tránh đưa lời khuyên y khoa. Hai câu cuối giúp giữ đúng phạm vi và tránh lặp vòng khi người gọi không rõ ràng. Cấu hình cũng thêm ngày hôm nay vì trong thử nghiệm thực tế, mô hình có thể đoán sai năm cho các ngày như "ngày 6 tháng 7".

Tinh chỉnh phát hiện lượt nói

Phát hiện lượt nói là cách tác tử quyết định bạn đã ngừng nói. Đặt turn_detection.type thành server_vad thì máy chủ kết thúc lượt khi im lặng. Để null thì bạn tự kiểm soát lượt bằng cách commit bộ đệm âm thanh, đây là cách tôi dùng cho luồng theo tệp.

Server VAD có ba thiết lập đáng chú ý: threshold ấn định âm lượng cần đạt để tính là lời nói, silence_duration_ms ấn định độ dài khoảng ngắt để kết thúc lượt, và prefix_padding_ms giữ lại một chút âm thanh trước khi lời nói bắt đầu. Nếu tác tử hay cắt lời người dùng, hãy tăng silence_duration_ms trước tiên.

Gửi âm thanh tới tác tử

Giờ chúng ta gửi giọng nói của người gọi. Âm thanh phải khớp định dạng phiên: mono PCM 16-bit ở 24000 Hz, mã hóa base64 và gửi theo từng khối.

Client stream tệp theo lát cắt, rồi commit bộ đệm để đánh dấu kết thúc lượt:

async def send_audio(self, pcm_bytes, chunk_ms=100, commit=True):
    bytes_per_chunk = int(self._sample_rate * 2 * chunk_ms / 1000)
    for start in range(0, len(pcm_bytes), bytes_per_chunk):
        chunk = pcm_bytes[start:start + bytes_per_chunk]
        await self._t.send({
            "type": "input_audio_buffer.append",
            "audio": base64.b64encode(chunk).decode(),
        })
    if commit:
        await self._t.send({"type": "input_audio_buffer.commit"})
    self.cost.audio_seconds += pcm_seconds(pcm_bytes, self._sample_rate)

Nếu tần số lấy mẫu hoặc mã hóa không khớp session.update, bạn có thể nghe rè hoặc im lặng thay vì lỗi rõ ràng. Âm thanh đi qua input_audio_buffer.append, vì vậy tính phí theo thời lượng thay vì theo mỗi thông điệp.

Nhận phản hồi bằng giọng

Sau khi bạn yêu cầu phản hồi, âm thanh đến dưới dạng response.output_audio.delta, bản chép đến dưới dạng response.output_audio_transcript.delta, và response.done đóng lượt.

Client gom tất cả vào một vòng lặp async duy nhất:

async def _collect_response(self):
    audio = bytearray()
    transcript, calls = [], []
    while True:
        event = await self._recv()
        etype = event["type"]
        if etype == "response.output_audio.delta":
            audio += base64.b64decode(event["delta"])
        elif etype == "response.output_audio_transcript.delta":
            transcript.append(event.get("delta", ""))
        elif etype == "response.function_call_arguments.done":
            calls.append(event)
        elif etype == "response.done":
            break
    return bytes(audio), "".join(transcript), calls

Giải mã các delta âm thanh, nối theo thứ tự và ghi kết quả ra tệp response.wav. Để ghi lại chính lời người gọi, đặt audio.input.transcription và đọc conversation.item.input_audio_transcription.completed.

Xây dựng workflow Trợ lý Đặt lịch

Giờ các phần ghép thành hội thoại: yêu cầu đặt lịch, câu hỏi làm rõ, kiểm tra khả dụng, đề xuất khung giờ, xác nhận. Để mang bối cảnh qua các lượt, mỗi lượt mới kết nối lại với conversation id và bật nối lại phiên.

Thêm gọi công cụ vào tác tử giọng nói

Với phòng khám, tác tử phải kiểm tra khả dụng trước khi hứa khung giờ. Công cụ tùy chỉnh là cách mô hình chạm tới mã của bạn: nó phát ra yêu cầu, ứng dụng của bạn chạy hàm, và bạn gửi kết quả về.

Công cụ là một hàm thuần cùng schema JSON đưa vào cấu hình phiên. Đây là schema từ tools.py:

CHECK_AVAILABILITY_TOOL = {
    "type": "function",
    "name": "check_availability",
    "description": "Look up open appointment slots for a service on a given date. "
                   "Always call this before offering the caller a time.",
    "parameters": {
        "type": "object",
        "properties": {
            "service": {"type": "string", "description": "Service requested."},
            "date": {"type": "string", "description": "Requested date as YYYY-MM-DD."},
        },
        "required": ["service", "date"],
    },
}

Vòng lặp có hình dạng cố định. Khi mô hình muốn dùng công cụ, nó gửi response.function_call_arguments.done kèm đối số. Bạn chạy hàm, trả về function_call_output, rồi gửi response.create để tác tử tiếp tục. Thiếu response.create cuối cùng, tác tử sẽ im lặng.

Giải thích vòng khứ hồi gọi công cụ. Ảnh: Tác giả.

Các hàm tùy chỉnh như vậy chạy trong mã của bạn. Bản demo Streamlit đăng ký thêm ba hàm từ cùng tệp: book_appointment, transfer_to_human, và end_call. Các công cụ dựng sẵn như tìm kiếm web, tìm kiếm trên X, tìm kiếm bộ sưu tập và công cụ MCP từ xa, chạy trên máy chủ của xAI.

Xử lý lỗi công cụ

Công cụ sẽ lỗi, và tác tử giọng nói mà giả định thành công có thể hứa một khung giờ không tồn tại. ToolRegistry.execute của tôi không bao giờ raise: một tra cứu thất bại trả về dict {"error": ...}.

def execute(self, name, arguments):
    handler = self._handlers.get(name)
    if handler is None:
        return {"error": f"unknown tool: {name}"}
    try:
        return handler(**arguments)
    except ToolError as exc:
        return {"error": str(exc)}

Một trạng thái lỗi tường minh ngăn tác tử coi lệnh gọi công cụ thất bại là thành công.

Thêm theo dõi chi phí

Trước khi phục vụ cho bất kỳ ai, hãy biết chi phí một cuộc gọi. Âm thanh tính $0,05 mỗi phút, tính cả phần bạn gửi và nhận. Sự kiện đầu vào văn bản tính $0,004 mỗi cái. Kết quả function_call_output và sự kiện response.create không bị tính phí.

Client theo dõi trong quá trình, nên chi phí là thuộc tính bạn có thể đọc bất cứ lúc nào:

@property
def audio_usd(self):
    rate = 0.05 + (0.01 if self.telephony else 0.0)
    return self.audio_seconds / 60 * rate

@property
def total_usd(self):
    return self.audio_usd + self.text_usd + self.tool_usd

Số điện thoại do xAI cấp thêm phụ phí thoại $0,01 mỗi phút, và tiện ích sẽ áp dụng khi bạn đặt telephony=True. Các công cụ do xAI lưu trữ tính riêng: tìm kiếm web và X khoảng $5 mỗi nghìn lệnh gọi, tìm kiếm tệp khoảng $2,50.

Xử lý lỗi và tình huống biên

Đa số lỗi rơi vào danh sách ngắn sau:

• Thiếu hoặc API key không hợp lệ trả 401 ở bắt tay kết nối, nên kiểm tra key trước

• Đội bị chặn trả 403, còn giới hạn tốc độ trả 429, hãy thử lại với backoff

• Cấu hình phiên sai trả 400, thường là gõ nhầm tên trường

• Định dạng âm thanh không hỗ trợ sẽ cho tiếng rè thay vì lỗi, nên khớp tần số phiên

• Thiếu response.create sau khi trả kết quả công cụ sẽ khiến tác tử treo

• Cố gắng đặt trùng có thể gây vấn đề thực sự, nên đừng thử lại một cách mù quáng

Thử lại thao tác đọc thất bại như check_availability là an toàn, nhưng thử lại thao tác ghi thất bại như đặt lịch thực sự có thể đặt trùng. Mọi hành động thay đổi dữ liệu cần kiểm tra tính idempotent trước.

Dùng token tạm thời cho ứng dụng phía client

Mọi thứ đến giờ giả định mã chạy trên máy chủ của bạn, nơi API key thuộc về. Nếu trình duyệt hoặc ứng dụng di động kết nối trực tiếp, hãy dùng ephemeral token.

Máy chủ của bạn gọi POST https://api.x.ai/v1/realtime/client_secrets với key của bạn, nhận về phản hồi token, và chuyển giá trị token cho client. Khi tôi chạy, phản hồi gồm value và expires_at:

@app.post("/session")
async def create_session():
    async with httpx.AsyncClient() as client:
        response = await client.post(
            CLIENT_SECRETS_URL,
            headers={"Authorization": f"Bearer {os.environ['XAI_API_KEY']}"},
            json={"expires_after": {"seconds": 300}},
        )
    return response.json()

Trình duyệt không thể đặt header WebSocket tùy chỉnh, nên token đi kèm trong header sec-websocket-protocol với tiền tố xai-client-secret..

Biến workflow thành một endpoint FastAPI

Một endpoint cho phép frontend hoặc dịch vụ khác gọi workflow. Route xác thực thân yêu cầu bằng mô hình Pydantic, nhận thông điệp dạng văn bản đã định kiểu hoặc đường dẫn âm thanh, và trả về bản chép, âm thanh phản hồi, nhật ký công cụ, độ trễ và chi phí ước tính.

@app.post("/appointments/voice")
async def appointments_voice(body: VoiceRequest):
    fail = {"check_availability"} if body.simulate_tool_failure else None
    assistant = AppointmentAssistant(voice=body.voice, telephony=body.telephony, fail_tools=fail)
    if body.text:
        result = await assistant.run_live(text=body.text, conversation_id=body.conversation_id)
    else:
        pcm = load_wav_as_pcm(body.audio_path, 24000)
        result = await assistant.run_live(pcm, conversation_id=body.conversation_id)
    return {
        "transcript": result.transcript,
        "audio_wav_base64": base64.b64encode(encode_wav_bytes(result.audio, 24000)).decode(),
        "tool_calls": result.tool_calls,
        "latency_seconds": round(result.latency_s, 3),
        "estimated_cost_usd": round(result.cost.total_usd, 6),
        "audio_seconds": round(result.cost.audio_seconds, 2),
        "conversation_id": result.conversation_id,
    }

Chạy với uvicorn app:app --reload và mở http://localhost:8000/docs. Đọc XAI_API_KEY từ môi trường máy chủ và không bao giờ nhận nó từ thân yêu cầu.