Grok Voice Agent Builder: คู่มือปฏิบัติในภาษา Python

xAI เปิดตัว Voice Agent Builder เป็นคอนโซลสำหรับสร้างเอเจนต์เสียง อธิบายผังการโทร แนบเอกสารและเครื่องมือ และเลือกเสียงได้

เวลาเปิดคอนโซลทดสอบเอเจนต์เสียง สิ่งที่สำคัญกว่าบันทึกเปิดตัวคือส่วนที่ต้องต่อเข้ากับโค้ด: จะตั้งค่าเซสชัน WebSocket อย่างไร เสียงเดินทางอย่างไร เรียกใช้เครื่องมือที่ไหน ค่าใช้จ่ายเท่าไร และแอปอื่นจะเรียกเวิร์กโฟลว์ได้อย่างไร

โค้ดด้านล่างจะสร้างโฟลว์นั้นใหม่โดยตรงกับ Voice Agent API โดยจะใช้ผู้ช่วยนัดหมายคลินิกที่ตรวจสอบความว่าง ตอบกลับด้วยเสียง ติดตามต้นทุน จัดการความล้มเหลวของเครื่องมือ และเปิดเผยปลายทาง FastAPI

Grok Voice Agent Builder คืออะไร

Voice Agent Builder คือคอนโซลของ xAI สำหรับสร้างและดีพลอยเอเจนต์เสียงบน Grok Voice เปิดตัวเวอร์ชันเบต้าเมื่อ 1 กรกฎาคม 2026 แทนที่จะใช้บริการแยกกันระหว่างสpeeชทูเท็กซ์ โมเดลภาษา และเท็กซ์ทูสpeeช จะใช้เส้นทางโมเดลเสียงเดียว

คอนโซลมีระบบโทรศัพท์ การดึงเอกสาร เครื่องมือและคอนเนคเตอร์ ราวกันตก เซิร์ฟเวอร์ MCP ระยะไกล และบันทึกการโทรพร้อมไฟล์เสียง ถอดความ และเทรซ

เสียงคิดค่าบริการเป็นรายนาที คอนโซลยังเป็นเบต้าอยู่ จึงใช้ API โดยตรง

กลไกการทำงานของ Grok Voice Agent API ใต้ Builder

ใต้คอนโซลคือ Voice Agent API เป็น WebSocket แบบเรียลไทม์ที่เปิดเผยรันไทม์เดียวกับที่ Builder ใช้

Builder อยู่เหนือ Voice API ภาพโดยผู้เขียน

โมเดลที่ใช้ที่นี่คือ grok-voice-think-fast-1.0 ส่วนแอลิแอส grok-voice-latest ชี้ไปที่โมเดลล่าสุด ฉันใช้ที่นี่ แต่สำหรับแอปที่ดีพลอยแล้วจะแนบท้ายด้วยเวอร์ชันที่ระบุ xAI รายงานคะแนน 67.3% สำหรับโมเดลนี้บนกระดานผู้นำ τ-voice Bench; ฉันมองว่าเป็นข้อมูลประกอบหนึ่งจุด ไม่ใช่การรับประกัน

บันทึกความเข้ากันได้: API เข้ากันได้กับ OpenAI Realtime API หากมีโค้ดที่คุยกับปลายทางเรียลไทม์ของ OpenAI อยู่แล้ว ส่วนใหญ่แค่เปลี่ยน base URL และคีย์

ภาพรวมโปรเจกต์: เราจะสร้างอะไร

ผู้ช่วยคลินิกรับอินพุตเสียง ตอบกลับด้วยเสียงที่สร้าง ถามคำถามติดตาม ตรวจสอบความว่างก่อนเสนอช่วงเวลา และโอนให้เจ้าหน้าที่เมื่อจำเป็น ตัวอย่างหลักใช้เครื่องมือเดียว; เดโม Streamlit เพิ่มการจอง การโอน และการวางสาย

บทเรียนหลักแยกเป็นสี่ไฟล์ แต่ละไฟล์ทำงานอย่างหนึ่ง:

• voice_client.py เก็บไคลเอนต์ WebSocket ตัวช่วยด้านเสียง และการติดตามต้นทุน

• tools.py เก็บ check_availability พร้อมเครื่องมือเดโมเพิ่มเติมที่ใช้โดย Streamlit

• assistant.py เก็บพรอมป์ระบบ การตั้งค่าเซสชัน และเวิร์กโฟลว์

• app.py ให้บริการทั้งหมดผ่าน FastAPI

สี่ไฟล์นี้คือเส้นทางผ่านบทความ รีโปยังมี app_streamlit.py สำหรับเดโมแบบภาพ และ run.py เป็นตัวเรียกใช้บน Windows แต่จะกลับมาดูสองไฟล์นี้ภายหลังเมื่อโฟลว์หลักทำงานแล้ว

ข้อกำหนดเบื้องต้น

ก่อนรันโค้ด ต้องมี Python 3.10 ขึ้นไป บัญชี xAI คีย์ API จาก console.x.ai เครดิตแบบเติมเงิน และความคุ้นเคยพื้นฐานกับตัวแปรสภาพแวดล้อม JSON และ WebSocket

ตั้งค่าโปรเจกต์

สร้างโฟลเดอร์และสภาพแวดล้อมเสมือน แล้วติดตั้งแพ็กเกจ:

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

ตรึงเวอร์ชันแพ็กเกจเหล่านี้ใน requirements.txt เพื่อให้เช็คเอาท์ใหม่ใช้สภาพแวดล้อมเดียวกัน

สร้างไฟล์ .env ไว้ข้างไฟล์ Python:

XAI_API_KEY=xai-your-key-here

เพิ่ม .env ลงใน .gitignore คีย์ API ควรอยู่บนเซิร์ฟเวอร์เท่านั้น

สร้าง Voice Agent

มาเริ่มลงมือกัน

เชื่อมต่อ Grok Voice Agent API ผ่าน WebSocket

ขั้นแรกคือตั้งการเชื่อมต่อ ส่งผ่านโมเดลเป็นคิวรีพารามิเตอร์ และคีย์ของคุณเป็น bearer token ในการจับมือเชื่อมต่อ:

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())

เมื่อใช้คีย์จริง อีเวนต์แรกที่เห็นคือ session.created ซึ่งหมายความว่าเปิดซ็อกเก็ตแล้วและพร้อมสำหรับการตั้งค่า

อีเวนต์ Session created ยืนยันการเชื่อมต่อ ภาพโดยผู้เขียน

ตั้งค่าเซสชันเสียง

ซ็อกเก็ตที่เชื่อมต่อยังไม่ใช่เอเจนต์ที่ตั้งค่าแล้ว กำหนดรูปร่างโดยส่งอีเวนต์ session.update พร้อมออบเจ็กต์ session

เสียง รูปแบบเสียง และคำสั่ง

สามการตั้งค่าที่ใช้บ่อยคือเสียง รูปแบบเสียง และพรอมป์ระบบ Realtime API มีเสียงให้เลือกห้าชื่อ eve, ara, rex, sal, และ leo รวมถึงโคลนแบบกำหนดเอง ค่าเริ่มต้นของเสียงคือ audio/pcm ที่ 24000 Hz โดยตั้งค่าอินพุตและเอาต์พุตแยกกัน

นี่คือการตั้งค่าเซสชันที่ผู้ช่วยใช้ ประกอบใน 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],
    }

ฟิลด์ instructions คือพรอมป์ระบบ พรอมป์ของคลินิกนี้ยังคงสั้นเพราะคำตอบเสียงยาวๆ ฟังยาก:

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.

บรรทัดเรื่องการส่งต่อช่วยกันไม่ให้เอเจนต์คลินิกให้คำแนะนำทางการแพทย์ สองบรรทัดสุดท้ายช่วยให้เกาะขอบเขตและหยุดลูปเมื่อผู้โทรไม่ชัด การตั้งค่ายังผนวกวันที่ของวันนี้ เพราะจากการทดสอบจริง โมเดลอาจเดาปีผิดสำหรับวันที่อย่าง "6 July"

ปรับแต่งการตรวจจับการผลัดกันพูด

การตรวจจับการผลัดกันพูดคือวิธีที่เอเจนต์ตัดสินว่าคุณพูดจบ ตั้งค่า turn_detection.type เป็น server_vad ให้เซิร์ฟเวอร์ตัดจบเมื่อเงียบ เว้นไว้เป็น null แล้วควบคุมการผลัดกันพูดด้วยการ commit บัฟเฟอร์เสียง ซึ่งฉันใช้สำหรับโฟลว์ไฟล์

Server VAD มีสามค่าที่ควรรู้: threshold กำหนดความดังขั้นต่ำที่นับเป็นคำพูด silence_duration_ms กำหนดความยาวช่วงเงียบที่จบการผลัดกันพูด และ prefix_padding_ms เก็บเสียงก่อนเริ่มพูดเล็กน้อย หากเอเจนต์ชอบขัดจังหวะ ให้ลองเพิ่ม silence_duration_ms ก่อน

ส่งเสียงไปยังเอเจนต์

ต่อไปคือส่งเสียงของผู้โทร รูปแบบเสียงต้องตรงกับการตั้งค่าเซสชัน: โมโน PCM 16 บิต ที่ 24000 Hz เข้ารหัสเป็น base64 และส่งเป็นชิ้นๆ

ไคลเอนต์จะสตรีมไฟล์เป็นส่วนๆ แล้ว commit บัฟเฟอร์เพื่อระบุว่าจบการผลัดกันพูด:

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)

หากอัตราสุ่มหรือการเข้ารหัสไม่ตรงกับ session.update อาจได้ยินซ่าๆ หรือเงียบแทนที่จะเป็นข้อผิดพลาดชัดเจน เสียงผ่าน input_audio_buffer.append จึงคิดค่าบริการตามระยะเวลา ไม่ใช่ต่อข้อความ

รับคำตอบเสียง

หลังจากขอคำตอบแล้ว เสียงจะมาถึงเป็น response.output_audio.delta ถอดความมาถึงเป็น response.output_audio_transcript.delta และ response.done จะปิดการผลัดกันพูด

ไคลเอนต์รวบรวมทั้งหมดในลูป async เดียว:

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

ถอดรหัสเดลต้าเสียง ต่อเรียงตามลำดับ แล้วเขียนผลลัพธ์เป็นไฟล์ response.wav หากต้องการจับคำพูดของผู้โทร ตั้งค่า audio.input.transcription และอ่าน conversation.item.input_audio_transcription.completed

สร้างเวิร์กโฟลว์ผู้ช่วยนัดหมาย

ตอนนี้ชิ้นส่วนต่างๆ กลายเป็นบทสนทนา: คำขอจอง คำถามชี้แจง ตรวจสอบความว่าง เสนอช่วงเวลา ยืนยัน เพื่อพกพาบริบทข้ามการผลัดกันพูด แต่ละเทิร์นใหม่จะเชื่อมต่ออีกครั้งด้วย conversation id และเลือกใช้การกลับมาใช้เซสชันเดิม

เพิ่มการเรียกใช้เครื่องมือให้เอเจนต์เสียง

สำหรับคลินิก เอเจนต์ต้องตรวจสอบความว่างก่อนสัญญาเวลานัด เครื่องมือกำหนดเองคือวิธีที่โมเดลเข้าถึงโค้ดของคุณ: โมเดลส่งคำขอ แอปของคุณรันฟังก์ชัน แล้วคุณส่งผลลัพธ์กลับ

เครื่องมือคือฟังก์ชันธรรมดาพร้อมสคีมา JSON ที่ใส่ในตั้งค่าเซสชัน นี่คือสคีมาจาก 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"],
    },
}

ลูปมีรูปทรงคงที่ เมื่อโมเดลต้องการเครื่องมือ มันจะส่ง response.function_call_arguments.done พร้อมอาร์กิวเมนต์ คุณรันฟังก์ชัน ส่งกลับ function_call_output แล้วส่ง response.create ต่อเพื่อให้เอเจนต์ทำงานต่อ หากพลาด response.create สุดท้าย เอเจนต์จะเงียบ

อธิบายรอบการเรียกใช้เครื่องมือ ภาพโดยผู้เขียน

ฟังก์ชันกำหนดเองแบบนี้รันในโค้ดของคุณ เดโม Streamlit ลงทะเบียนอีกสามฟังก์ชันจากไฟล์เดียวกัน: book_appointment, transfer_to_human, และ end_call เครื่องมือในตัว เช่น การค้นหาเว็บ การค้นหาใน X การค้นหาในคอลเลกชัน และเครื่องมือ MCP ระยะไกล จะรันบนเซิร์ฟเวอร์ของ xAI

จัดการความล้มเหลวของเครื่องมือ

เครื่องมือมีวันล้มเหลว และเอเจนต์เสียงที่สมมุติว่าทุกอย่างสำเร็จอาจเผลอสัญญาช่วงเวลาที่ไม่มีอยู่ ToolRegistry.execute ของฉันจะไม่ raise: การค้นหาที่ล้มเหลวจะส่งกลับเป็นดิกชันนารี {"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)}

สถานะผิดพลาดที่ชัดเจนช่วยกันไม่ให้เอเจนต์ตีความการเรียกใช้เครื่องมือที่ล้มเหลวว่าเป็นความสำเร็จ

เพิ่มการติดตามต้นทุน

ก่อนให้บริการใคร ควรรู้ว่าหนึ่งสายมีค่าใช้จ่ายเท่าไร เสียงคิดที่ $0.05 ต่อนาที โดยนับทั้งที่ส่งและรับ อีเวนต์ข้อความอินพุตคิดที่ $0.004 ต่อครั้ง ผลลัพธ์จาก function_call_output และอีเวนต์ response.create ไม่คิดค่าใช้จ่าย

ไคลเอนต์ติดตามระหว่างทำงาน จึงอ่านค่าใช้จ่ายได้ทุกจุด:

@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

หมายเลขที่ xAI จัดให้มีค่าบริการโทรศัพท์เพิ่ม $0.01 ต่อนาที ซึ่งตัวช่วยจะนำไปใช้เมื่อกำหนด telephony=True เครื่องมือที่โฮสต์โดย xAI คิดค่าใช้จ่ายแยก: ค้นหาเว็บและค้นหาใน X ประมาณ $5 ต่อการเรียกหนึ่งพันครั้ง และค้นหาไฟล์ประมาณ $2.50

จัดการข้อผิดพลาดและกรณีขอบ

ความล้มเหลวส่วนใหญ่อยู่ในรายการสั้นๆ:

• ขาดคีย์ API หรือคีย์ไม่ถูกต้องจะส่งคืน 401 ตอนจับมือเชื่อมต่อ จึงควรตรวจสอบคีย์ก่อน

• ทีมถูกบล็อกจะได้ 403 และการจำกัดอัตราจะได้ 429 ซึ่งควรลองซ้ำพร้อม backoff

• การตั้งค่าเซสชันผิดรูปจะได้ 400 มักเป็นสะกดชื่อฟิลด์ผิด

• รูปแบบเสียงที่ไม่รองรับจะให้เสียงซ่าๆ ไม่ใช่ข้อผิดพลาด จึงต้องให้อัตราสุ่มตรงกับเซสชัน

• ขาด response.create หลังผลลัพธ์เครื่องมือทำให้เอเจนต์ค้าง

• ความพยายามจองซ้ำอาจก่อปัญหาจริง จึงอย่าลองซ้ำแบบไม่ยั้งคิด

การลองซ้ำการอ่านที่ล้มเหลวอย่าง check_availability ปลอดภัย แต่การลองซ้ำการเขียนที่ล้มเหลวอย่างการจองจริงอาจทำให้จองซ้ำ สำหรับการกระทำที่เปลี่ยนข้อมูล ต้องตรวจสอบ idempotency ก่อน

ใช้โทเค็นอายุสั้นสำหรับแอปฝั่งไคลเอนต์

ทั้งหมดนี้สมมุติว่าโค้ดรันบนเซิร์ฟเวอร์ของคุณ ซึ่งคีย์ API ควรอยู่ หากเบราว์เซอร์หรือแอปมือถือเชื่อมต่อโดยตรง ให้ใช้ ephemeral tokens.

เซิร์ฟเวอร์ของคุณเรียก POST https://api.x.ai/v1/realtime/client_secrets ด้วยคีย์ของคุณ รับการตอบกลับเป็นโทเค็น แล้วส่งค่าโทเค็นให้ไคลเอนต์ ในการรันทดสอบของฉัน การตอบกลับมี value และ 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()

เบราว์เซอร์ตั้งค่า WebSocket headers กำหนดเองไม่ได้ จึงส่งโทเค็นในเฮดเดอร์ sec-websocket-protocol พร้อมคำนำหน้า xai-client-secret.

เปลี่ยนเวิร์กโฟลว์เป็นปลายทาง FastAPI

ปลายทางช่วยให้ส่วนหน้า หรือบริการอื่น เรียกเวิร์กโฟลว์ได้ เส้นทางจะตรวจสอบเนื้อหาแพ็กด้วยโมเดล Pydantic รับข้อความแบบไทป์หรือพาธไฟล์เสียง และส่งกลับถอดความ เสียงตอบกลับ บันทึกเครื่องมือ เวลาแฝง และประมาณการค่าใช้จ่าย

@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,
    }

รันด้วย uvicorn app:app --reload แล้วเปิด http://localhost:8000/docs อ่าน XAI_API_KEY จากสภาพแวดล้อมบนเซิร์ฟเวอร์ และอย่ายอมรับจากเนื้อหาแพ็กของคำขอ