บทช่วยสอน Grok Voice Think Fast 2.0 API: สร้างเอเจนต์เสียงแบบเรียลไทม์ด้วย Python

Grok Voice Think Fast 2.0 ของ SpaceXAI เป็นโมเดลแปลงเสียงเป็นเสียง ส่งเสียงเข้าไปผ่าน WebSocket และรับเสียงกลับมา ระหว่างนั้นตัวโมเดลสามารถให้เหตุผลและพูดต่อได้ขณะฟังก์ชันที่มันตัดสินใจเรียกกำลังทำงานอยู่ ไม่มีขั้นตอนแยกเป็น speech-to-text และไม่มีขั้นตอนแยกเป็น text-to-speech

SpaceXAI ประกาศ Think Fast 2.0 เมื่อ 29 กรกฎาคม 2026: เสียงแรกมาไวขึ้น พฤติกรรม full-duplex เสถียรขึ้น (ฟังไปพูดไปแทนที่จะผลัดกันอย่างเคร่งครัด) และเรียกใช้เครื่องมือได้เร็วตั้งแต่ช่วงต้นของเทิร์น จะไม่ยืดเยื้อเรื่องเบนช์มาร์ก เพราะสำหรับบทช่วยสอน สิ่งที่สำคัญคือโค้ดต้องเปลี่ยนตรงไหน

เราจะสร้างเอเจนต์เสียงฝ่ายบริการลูกค้าสำหรับร้านค้าออนไลน์ ผู้โทรสามารถสอบถามคำสั่งซื้อ เปลี่ยนคำแนะนำการจัดส่ง ยกเลิก ขัดจังหวะเอเจนต์กลางประโยค และกลับมาสนทนาต่อหลังการเชื่อมต่อหลุด นี่คือเส้นทางผ่าน API ไม่ใช่เครื่องมือสร้าง Voice Agent แบบ no-code ที่บทช่วยสอน Grok Voice Agent Builder ของเราได้อธิบายไว้ เริ่มจากตรงนั้นหากเน้นทำผ่านคอนโซลก่อน

Grok Voice Think Fast 2.0 คืออะไร?

Grok Voice Think Fast 2.0 คือโมเดลใหม่ล่าสุดของ SpaceXAI สำหรับ Speech to Speech API ซึ่งเป็นชื่อผลิตภัณฑ์ที่คนส่วนใหญ่เรียกรวม ๆ ว่า Grok Voice หากยังคุ้นกับชื่อบริษัทว่า xAI ก็เจ้าเดียวกัน: ถูกควบรวมเข้า SpaceX และรีแบรนด์เป็น SpaceXAI เมื่อ 6 กรกฎาคม 2026 ฝั่ง API ไม่ได้เปลี่ยนตามแบรนด์ ดังนั้นตัวระบุทั้งหมดด้านล่างยังคงเป็น xai ตั้งแต่ตัวแปร XAI_API_KEY ไปจนถึงโฮสต์ api.x.ai.

สแตกเสียงแบบดั้งเดิมจะต่อเรียงสามบริการ: speech-to-text, โมเดลภาษา, แล้ว text-to-speech และทุกการส่งต่อเพิ่มทั้งความหน่วงและโอกาสที่บริบทจะสูญหาย Think Fast 2.0 รวมทั้งหมดไว้ในโมเดลเดียว รับเข้าเป็นเสียงหรือข้อความ และส่งออกเป็นเสียงหรือข้อความผ่านการเชื่อมต่อเดียวกัน

WebSocket แบบเสียงต่อเสียง เทียบกับสถาปัตยกรรมแบบสามบริการ ภาพโดยผู้เขียน

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

บนเบนช์มาร์กที่ SpaceXAI อ้างถึงจาก Artificial Analysis Think Fast 2.0 ทำคะแนน 82.9% บน Speech to Speech Index เทียบกับ 75.7% ของรุ่น 1.0 และลดเวลาเสียงแรกจาก 1.25 วินาที เหลือ 0.70 วินาที ตัวเลขจากผู้ขายบนเบนช์มาร์กทั่วไปเป็นสมมติฐานเกี่ยวกับโฟลว์ของผู้โทรของคุณ ไม่ใช่แผนการทดสอบ

จะเจอสตริงชื่อโมเดลสามแบบ: grok-voice-latest, grok-voice-think-fast-2.0 และ grok-voice-think-fast-1.0 อะลิแอสสะดวกตอนทำต้นแบบ แต่ไม่เสถียรพอสำหรับงานอื่น

เมื่อทดสอบวันที่ 4 สิงหาคม 2026 grok-voice-latest ยังชี้ไปที่ grok-voice-think-fast-1.0 โดย บันทึกประจำรุ่นของ SpaceXAI กำหนดตารางย้ายไป Think Fast 2.0 ในวันถัดไป การสลับครั้งนี้คือการเปลี่ยนราคาเช่นกัน จาก $0.05 เป็น $0.08 ต่อนาทีของเสียง ดังนั้นการใช้อะลิแอสที่ไม่ปักหมุดเวอร์ชันจะทำให้แพงขึ้นโดยที่โค้ดคุณไม่เปลี่ยนสักบรรทัด ปักหมุดสตริงเวอร์ชันในทุกสิ่งที่นำขึ้นใช้งาน

เราจะสร้างอะไร

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

โปรเจ็กต์แบ่งเป็นไฟล์เล็กหลายไฟล์แทนการเขียนสคริปต์ไฟล์เดียว เพราะแต่ละส่วนทำหน้าที่ต่างกัน และจะได้ทดสอบแยกกัน นี่คือโครงสร้าง:

• config.py โหลดคีย์ API และเก็บสตริงโมเดล อัตราสุ่มตัวอย่าง และ URL ของเอนด์พอยต์

• voice_client.py หุ้ม WebSocket ติดตามการคิดเงิน และเปิดให้ใช้ตัวช่วยส่ง/รับ

• tools.py กำหนดฟังก์ชันเกี่ยวกับออเดอร์ และสโตร์ออเดอร์ขนาดเล็กในหน่วยความจำแทนฐานข้อมูลจริง

• assistant.py เก็บพรอมป์ระบบ การตั้งค่าเซสชัน และลูปเหตุการณ์ที่เชื่อมทุกอย่างเข้าด้วยกัน

• token_server.py เป็นเอนด์พอยต์ FastAPI ขนาดเล็กที่สร้างโทเค็นชั่วคราว

• app_streamlit.py ใช้ไคลเอนต์เดียวกันบนการโทรผ่านเบราว์เซอร์สด เดี๋ยวจะกลับมาพูดถึงหลังส่วนการทดสอบ

เส้นทางการสอนเริ่มจากเทอร์มินัล เดโมจะเพิ่มไมโครโฟน

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

ต้องมี บัญชี SpaceXAIพร้อมคีย์ API การจัดการเรียกเก็บเงินที่เติมเงินไว้ (ไม่มีฟรีเทียร์ถาวร และเครดิตโปรโมชันสำหรับบัญชีใหม่ไม่เพียงพอ) และความคุ้นเคยกับ asyncio และ WebSocket พอที่จะตามได้โดยไม่ต้องอธิบาย await แบบบรรทัดต่อบรรทัด

ตัวอย่างเริ่มต้นอย่างรวดเร็วของ SpaceXAI ใช้แพ็กเกจ websockets ตรง ๆ แทน SDK เฉพาะ และเราก็เช่นกัน เอกสารไม่ได้ระบุเวอร์ชัน Python ที่ต้องการ ฉันทดสอบบน 3.11

เก็บคีย์ API ไว้บนเซิร์ฟเวอร์ หากเบราว์เซอร์หรือแอปมือถือคุยกับ Voice API โดยตรง ให้ใช้ โทเค็นชั่วคราว แทนคีย์จริง ซึ่งจะกล่าวถึงในส่วนความปลอดภัยด้านล่าง

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

ทุกไฟล์ด้านล่างอยู่ใน repo ของโปรเจ็กต์ จึงโคลนได้แทนการก็อปปี้สแน็กเก็ต:

git clone https://github.com/KhalidAbdelaty/grok-voice-think-fast-2.0.git
cd grok-voice-think-fast-2.0
pip install -r requirements.txt

websockets จัดการการเชื่อมต่อเรียลไทม์ และ python-dotenv อ่านคีย์ของคุณ ที่เหลือครอบคลุมเอนด์พอยต์โทเค็นและเดโมบนเบราว์เซอร์ ใส่คีย์ของคุณใน .env:

XAI_API_KEY=xai-your-key-here

นั่นคือส่วนใหญ่ของการตั้งค่า ส่วนที่น่าสนใจคือการเชื่อมต่อ

ทำความเข้าใจกับ Grok Voice Realtime API

Grok Voice คือชื่อผลิตภัณฑ์ สิ่งที่คุณเขียนโค้ดคุยด้วยจริง ๆ คือเอนด์พอยต์ WebSocket ที่ wss://api.x.ai/v1/realtime และทั้งบทสนทนาเกิดขึ้นเป็นสตรีมของเหตุการณ์ JSON ผ่านซ็อกเก็ตเดียว

วงจรชีวิตของเหตุการณ์

การเชื่อมต่อมีรูปแบบตายตัว: เซิร์ฟเวอร์ส่ง session.created และ conversation.created ทันทีที่คุณเชื่อมต่อ คุณส่ง session.update เพื่อกำหนดค่าเสียงและเครื่องมือ เซิร์ฟเวอร์ยืนยันด้วย session.updated และจากนั้นคุณสร้างรายการบทสนทนาและร้องขอคำตอบ ฉันรันทดสอบด้วยคีย์จริงและลำดับตรงตามเอกสารทุกอย่าง

• session.update (ไคลเอนต์) กำหนดค่าเสียง คำสั่ง เครื่องมือ และรูปแบบเสียง

• conversation.item.create (ไคลเอนต์) เพิ่มข้อความผู้ใช้ ข้อความผู้ช่วย หรือผลลัพธ์เครื่องมือ

• response.create (ไคลเอนต์) ขอให้โมเดลพูด; ระบบ VAD ของเซิร์ฟเวอร์ส่งให้อัตโนมัติ

• response.output_audio.delta และ response.output_audio_transcript.delta (เซิร์ฟเวอร์) สตรีมคำตอบขณะกำลังสร้าง

• response.done (เซิร์ฟเวอร์) ปิดเทิร์น

มีสองอย่างที่มักทำให้สะดุด หน้าเอกสาร Speech to Speech ที่ลิงก์ไว้ด้านบนพูดถึงเหตุการณ์ conversation.item.created ระหว่างการกู้คืนเซสชัน แต่ รายการอ้างอิงเหตุการณ์ที่เป็นมาตรฐานมีแค่ conversation.item.added และนั่นคือสิ่งที่มาถึงในทุกการทดสอบที่ฉันรัน ดังนั้นให้โค้ดอิงกับอันนี้ คุณยังจะเห็นเหตุการณ์ ping ที่ไม่มีเอกสารหลังเชื่อมต่อไปไม่กี่วินาที กล่าวถึงไว้เพื่อจะได้ไม่อ่านว่าเป็นข้อผิดพลาด

รูปแบบเสียงและการขนส่ง

โค้เดกและการขนส่งเป็นตัวเลือกแยกกัน โค้เดกกำหนดภายใต้ audio.input.format และ audio.output.format ได้แก่ audio/pcm (Linear16, ค่าเริ่มต้น 24000 Hz), audio/pcmu หรือ audio/pcma (G.711 ที่ 8 kHz สำหรับโทรศัพท์) หรือ audio/opus (24 kHz) ส่วนการขนส่งคือวิธีส่งไบต์เหล่านั้นบนสาย:

• json (ค่าเริ่มต้น) ส่งเสียงเป็นข้อความ base64 ภายใน input_audio_buffer.append และ response.output_audio.delta ตรวจสอบและดีบักง่าย

• binary ส่งไบต์โค้เดกดิบเป็นเฟรมไบนารีของ WebSocket ข้ามโอเวอร์เฮด base64 โดยต้องให้ลูปรับแยกสาขาตามชนิดข้อความ

เริ่มด้วย JSON ตัวอย่างในเอกสารทั้งหมดใช้แบบนี้ ตรวจสอบได้ง่าย และโอเวอร์เฮด base64 ไม่ใช่คอขวดของงานเอเจนต์สนับสนุน ย้ายไป binary เมื่อมีเหตุผลจากการวัดจริง

ความเข้ากันได้กับ OpenAI Realtime API

ข้ามส่วนนี้ได้หากไม่เคยใช้ OpenAI Realtime API สำหรับคนอื่น ๆ Speech to Speech API ตาม OpenAI Realtime API ได้ใกล้เคียงพอที่โค้ดไคลเอนต์ส่วนใหญ่จะพอร์ตโดยเปลี่ยน base URL และคีย์ แต่ไม่ถึงกับแทนที่ได้ทันที

ทรานสคริปต์มาถึงเป็น conversation.item.input_audio_transcription.updated ที่นี่แทน delta ของ OpenAI เหตุการณ์บางอย่างของ OpenAI ไม่รองรับ และ SpaceXAI เพิ่มส่วนขยายของตัวเอง: force_message สำหรับบรรทัดเปิดเผยแบบสคริปต์, resumption สำหรับการเชื่อมต่อใหม่ และ replace สำหรับแก้การออกเสียงชื่อแบรนด์ก่อน text-to-speech

สร้างเอเจนต์เสียงแบบเรียลไทม์

พอเรื่องโปรโตคอลแล้ว นี่คือไคลเอนต์ที่คุยกับมัน

เชื่อมต่อและกำหนดค่าเซสชัน

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

import asyncio
import json
import os
import websockets

MODEL = "grok-voice-think-fast-2.0"  # pin the version, not grok-voice-latest

async def connect():
    url = f"wss://api.x.ai/v1/realtime?model={MODEL}"
    ws = await websockets.connect(
        url, additional_headers={"Authorization": f"Bearer {os.environ['XAI_API_KEY']}"}
    )
    await ws.send(json.dumps({
        "type": "session.update",
        "session": {
            "voice": "eve",
            "instructions": SYSTEM_PROMPT,
            "turn_detection": {"type": "server_vad"},
            "tools": ORDER_TOOLS,
            "resumption": {"enabled": True},
        }
    }))

return ws

instructions คือพรอมป์ระบบ และโมเดลนี้ต้องการพรอมป์สั้น ๆ บันทึกการย้ายเวอร์ชันของ SpaceXAI แนะนำให้ทำพรอมป์ที่เขียนสำหรับโมเดลเสียงยุค GPT ให้เรียบง่ายขึ้น แทนการพอร์ตแบบคำต่อคำ ของฉันสั่งให้เอเจนต์ตอบสั้น ถามทีละคำถาม และอ่านสิ่งที่จะเขียนกลับก่อนลงมือ ยืนยันแบบพูดเป็นเรื่องประสบการณ์ใช้งานที่ดี ไม่ใช่มาตรการความปลอดภัย แอปพลิเคชันของคุณยังต้องบังคับใช้อำนาจอนุญาตบนการเขียนนั้นเอง

สิ่งหนึ่งที่ทำให้ประหลาดใจ: หากสตริงชื่อโมเดลไม่รู้จัก จะไม่ยกข้อผิดพลาดตอนเชื่อมต่อ แต่ถอยกลับไป grok-voice-think-fast-1.0 เงียบ ๆ ลดระดับคำขอที่เสียเงินเพราะพิมพ์ผิดโดยไม่บอก เป็นค่าเริ่มต้นที่แปลก บันทึกฟิลด์ session.model ของ session.created ครั้งหนึ่งตอนเริ่มระบบและตรวจว่าตรงกับที่ขอ

เอาต์พุตเทอร์มินัลแสดง session.created หลังการเชื่อมต่อ ภาพโดยผู้เขียน

สตรีมเสียงของผู้ใช้

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

async def send_audio_chunk(ws, pcm_bytes: bytes):
    await ws.send(json.dumps({
        "type": "input_audio_buffer.append",
        "audio": base64.b64encode(pcm_bytes).decode(),
    }))

Server VAD มีปุ่มปรับสามตัว และการตั้งค่าผิดเป็นสาเหตุที่พบบ่อยที่สุดที่ทำให้เอเจนต์เสียง “ดูเสีย” ขณะที่ล็อกไม่แสดงข้อผิดพลาด ค่าเหล่านี้ไม่ปรากฏในเอคโค่ของ session.updated โดยค่าเริ่มต้น จึงควรตรวจตามเอกสารแทนการเดา

• threshold (0.1 ถึง 0.9 ค่าเริ่มต้น 0.85) ระดับความดังที่นับว่าเป็นคำพูด; เพิ่มสำหรับห้องที่มีเสียงรบกวน ลดหากคนพูดเบาถูกมองข้าม

• silence_duration_ms ระยะเวลาที่ผู้โทรเงียบก่อนที่เซิร์ฟเวอร์จะปิดเทิร์น; สั้นไปจะตัดบทกลางความคิด ยาวไปจะรู้สึกเชื่องช้า

• prefix_padding_ms (ค่าเริ่มต้น 333) ชิ้นเสียงที่เก็บไว้ก่อนตรวจจับคำพูดเล็กน้อย เพื่อไม่ให้พยางค์แรกถูกตัด

ปรับ silence_duration_ms ก่อนหากผู้โทรถูกตัดบทตอนหยุดคิดบ่อย ๆ นี่คือค่าที่ฉันจับก่อนอีกสองตัว

รับและเล่นคำตอบ

เสียงมาถึงเป็นชิ้นเล็ก ๆ ในรูปแบบ response.output_audio.delta และจุดประสงค์ของการสตรีมคือเล่นทีละชิ้นทันทีที่มาถึง แทนการรอ response.done.

async def play_response(ws):
    async for message in ws:
        event = json.loads(message)
        if event["type"] == "response.output_audio.delta":
            chunk = base64.b64decode(event["delta"])
            speaker.write(chunk)  # your playback call goes here
        elif event["type"] == "response.output_audio_transcript.delta":
            print(event["delta"], end="", flush=True)

เก็บทรานสคริปต์ไว้แม้ในโปรดักชัน มันคือเครื่องมือดีบักที่ถูกที่สุดเมื่อผู้โทรบอกว่าเอเจนต์ “พูดอะไรแปลก ๆ”

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

เอเจนต์เสียงที่ทำได้แค่พูด คือแชตบอตที่มีไมโครโฟน

สร้างเครื่องมือเกี่ยวกับออเดอร์

แต่ละเครื่องมือคือ JSON schema บวกฟังก์ชัน Python ธรรมดาฝั่งเรา โมเดลไม่แตะฐานข้อมูลโดยตรง มันเห็นเฉพาะสิ่งที่ฟังก์ชันของเราส่งกลับ

ORDER_TOOLS = [
    {
        "type": "function",
        "name": "check_order_status",
        "description": "Look up the status, ETA, and delivery instructions for an order.",
        "parameters": {
            "type": "object",
            "properties": {
                "order_number": {"type": "string", "description": "e.g. ORD-1042"},
            },
            "required": ["order_number"],
        },
    },
    # find_orders, update_delivery_instructions, cancel_order,
    # create_support_ticket, check_ticket_status and transfer_to_human
    # all follow the same shape
]

งานอ่านอย่าง check_order_status ปลอดภัยสำหรับการลองใหม่หากเกิด timeout แต่งานเขียนไม่ใช่: ลองใหม่ที่ update_delivery_instructions หลัง timeout ที่คลุมเครืออาจทำให้การเปลี่ยนแปลงเดิมถูกใช้ซ้ำ บรรทัดยืนยันในพรอมป์ไม่หยุดสิ่งนี้ ให้คีย์ idempotency หรือการตรวจจับซ้ำกับงานเขียนแทน

ใส่เงื่อนไขปฏิเสธในฟังก์ชันด้วย cancel_order ส่งเหตุผลและทางเลือกกลับแทนการยกเลิกออเดอร์ที่จัดส่งแล้ว เพราะพรอมป์ที่ว่า “อย่ายกเลิกออเดอร์ที่จัดส่งแล้ว” เป็นเพียงคำแนะนำ แต่ฟังก์ชันที่ปฏิเสธคือข้อบังคับ

จัดการลูปการเรียกใช้เครื่องมือ

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

async def handle_tool_call(ws, event):
    args = json.loads(event["arguments"])
    result = execute(event["name"], args)  # never raises; errors come back as {"error": ...}
    await ws.send(json.dumps({
        "type": "conversation.item.create",
        "item": {
            "type": "function_call_output",
            "call_id": event["call_id"],
            "output": json.dumps(result),
        },
    }))

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

มีข้อควรระวังที่ SpaceXAI ระบุไว้และฉันก็เจอในครั้งแรก: การส่ง response.create ทันทีที่ผลลัพธ์เครื่องมือออก อาจทับซ้อนกับประโยคเกริ่นนำที่เอเจนต์ยังพูดอยู่ ครั้งหนึ่งมันเริ่มด้วย “ฉันจะตรวจสอบสถานะออเดอร์ ORD-1042 เดี๋ยวนี้” แล้วเรียกเครื่องมือกลางประโยค ดังนั้นการตอบกลับทันทีจะพูดทับตัวเอง

รอให้เสียงของเทิร์นปัจจุบันจบ และแสดงสถานะ “กำลังคิด” สั้น ๆ คั่นกลาง

โฟลว์การเรียกใช้เครื่องมือก่อนจะดำเนินคำตอบต่อ ภาพโดยผู้เขียน

จัดการการขัดจังหวะและสถานะบทสนทนา

มีสองปัญหาแยกกัน ผู้โทรพูดทับเอเจนต์กลางคำตอบ และ WebSocket หลุดต้องกลับมาต่อ

รองรับการขัดจังหวะตามธรรมชาติ

เมื่อเปิด server_vad การพูดแทรก (barge-in) จะเกิดขึ้นอัตโนมัติฝั่งเซิร์ฟเวอร์: ทันทีที่ตรวจว่าผู้โทรเริ่มพูดอีกครั้ง มันจะส่งสัญญาณ input_audio_buffer.speech_started และหยุดสร้างคำตอบเดิม หน้าที่ของคุณคือฝั่งไคลเอนต์ ล้างเสียงที่เข้าคิวไว้เพื่อให้เอเจนต์เงียบแทนที่จะพูดจบประโยคที่ไม่มีใครอยากฟัง

if event["type"] == "input_audio_buffer.speech_started":
    playback_queue.clear()

สำหรับเซสชันแบบแมนนวลที่ไม่ใช้ VAD response.cancel ทำงานเดียวกันนี้ตามคำขอ นอกจากนี้ยังมี conversation.item.truncate เพื่อหั่นรายการของผู้ช่วยให้เหลือเท่าที่ได้ยินจริง เอกสารยืนยันว่ามี แต่ไม่ได้บอกว่าจะยิงเมื่อใดระหว่างการพูดแทรกแบบสด ดังนั้นต้องทดสอบจังหวะด้วยตนเอง

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

กลับเข้าเซสชันที่หลุด

การกู้คืนเซสชันต้องเปิดใช้งานเองและไม่ใช่ความจำ ตั้งค่า resumption.enabled: true บน session.update ดึง ID จากเหตุการณ์ conversation.created และถ้าซ็อกเก็ตหลุด ให้เชื่อมต่อใหม่ด้วย ?conversation_id=<id> ใน URL และเปิดใช้งานอีกครั้งบนการเชื่อมต่อใหม่

async def reconnect(conversation_id):
    url = f"wss://api.x.ai/v1/realtime?model={MODEL}&conversation_id={conversation_id}"
    ws = await websockets.connect(url, additional_headers=auth_header)
    await ws.send(json.dumps({"type": "session.update", "session": {"resumption": {"enabled": True}}}))
    return ws

เทิร์นที่แคชไว้ ทรานสคริปต์ การเรียกเครื่องมือ และผลลัพธ์เครื่องมือจะเล่นซ้ำก่อนคำถามถัดไป และแคชจะหายไปหลังไม่มีการใช้งาน 30 นาที ฉันทดสอบโดยถามเรื่องคำสั่งซื้อ ทำการตัดการเชื่อมต่อ แล้วเชื่อมใหม่เพื่อถามต่อโดยไม่ต้องพูดซ้ำ เอเจนต์หยิบยก ETA ต่อได้ถูกต้อง

มีข้อควรระวังที่ไม่มีเอกสาร: การเล่นซ้ำไม่ได้มาทันที ดังนั้นคำถามที่ยิงทันทีที่ซ็อกเก็ตเปิดอาจแซงหน้าและตอบกลับโดยไม่มีความทรงจำของเทิร์นก่อนหน้า รอสักครู่ก่อนจะโทษ resumption

บันทึกเทอร์มินัลของเซสชันที่กู้คืน ภาพโดยผู้เขียน

อย่าใช้สิ่งนี้แทนการบันทึกสถานะออเดอร์ในฐานข้อมูลของคุณเอง หากแคชหมดอายุหรือผู้โทรโทรกลับมาวันพรุ่งนี้ คุณจะเริ่มจากศูนย์ตามที่ออกแบบไว้

รักษาความปลอดภัยและติดตามเอเจนต์

อย่าใส่คีย์ API ถาวรในโค้ดเบราว์เซอร์หรือมือถือ หากไคลเอนต์เชื่อมต่อโดยตรงแทนผ่านเซิร์ฟเวอร์ของคุณ ให้สร้างโทเค็นอายุสั้น:

from fastapi import FastAPI
import httpx, os

app = FastAPI()

@app.post("/session")
async def create_session():
    async with httpx.AsyncClient() as client:
        response = await client.post(
            "https://api.x.ai/v1/realtime/client_secrets",
            headers={"Authorization": f"Bearer {os.environ['XAI_API_KEY']}"},
            json={"expires_after": {"seconds": 300}},
        )
    return response.json()  # {"value": "xai-realtime-client-secret-...", "expires_at": ...}

เบราว์เซอร์ไม่สามารถตั้งค่าเฮดเดอร์ Authorization ที่กำหนดเองบนการจับมือ WebSocket ได้ จึงส่งโทเค็นผ่านเฮดเดอร์ sec-websocket-protocol แทน โดยมีคำนำหน้า xai-client-secret..

เซิร์ฟเวอร์สร้างโทเค็น เบราว์เซอร์เข้าร่วมการโทร ภาพโดยผู้เขียน

การคิดเงินมีสองมิเตอร์ เสียงที่ส่งหรือรับ คิดที่ $0.08 ต่อนาทีดังที่กล่าวไปก่อนหน้า ซึ่งเท่ากับ $4.80 ต่อชั่วโมง และทุก conversation.item.create ที่ไม่ใช่เสียงและไม่ใช่ function_call_output มีค่าใช้จ่ายคงที่ $0.004 response.create ไม่คิดเงินเลย แต่ละ response.done มีออบเจกต์ usage ซึ่งในการทดสอบของฉันมีรายงาน output_audio_seconds ควบคู่กับ billable_audio_seconds แยกต่างหาก ให้คิดเงินจากสิ่งเหล่านั้น ไม่ใช่จากการประมาณ

ข้อจำกัดที่มีเอกสารของ Speech to Speech API คือ 10 เซสชันพร้อมกันต่อทีม และขีดจำกัดเซสชัน 120 นาที ทั้งหมดใน us-east-1 อย่าวางแผนความจุจากตัวเลขของ Voice Agent API ซึ่งต่างกัน

ด้านความเป็นส่วนตัว ควรระบุให้ชัดเจน คำถาม-คำตอบด้านความปลอดภัยของ SpaceXAI ระบุว่าคำขอและคำตอบของ API ถูกเก็บไว้แบบเข้ารหัส 30 วันเพื่อมอนิเตอร์การละเมิด และไม่ถูกใช้เพื่อฝึกโดยไม่ได้รับอนุญาต และทีมสามารถเปิดใช้ Zero Data Retention ได้ แม้ ZDR จะทำให้ประวัติบทสนทนาของเอเจนต์เสียงที่ถูกเก็บหายไปและจึงใช้ร่วมกับการกู้คืนเซสชันไม่ได้

หากต้องเปิดเผยว่ามีการบันทึกสายหรือจัดการด้วย AI นั่นคือหน้าที่ของส่วนขยาย force_message ที่กล่าวถึงก่อนหน้า บรรทัดจะเล่นตามที่เขียน ไม่ใช่ตามที่โมเดลถอดความ

ทดสอบเอเจนต์เสียง

สถานะ 200 บนการจับมือ WebSocket ไม่ได้บอกว่าเอเจนต์ทำถูกต้องหรือไม่ ทดสอบผลลัพธ์ ไม่ใช่แค่การเชื่อมต่อ

• การค้นหาออเดอร์ปกติ ตรวจคำตอบที่พูดกับระเบียน ไม่ใช่แค่ว่ามีคำตอบมา
• การตอบที่ถูกขัดจังหวะ ยืนยันว่าเสียงหยุดและเอเจนต์ตอบรับคำขอใหม่
• การอัปเดตการจัดส่งที่ต้องยืนยัน ตรวจเทียบกับระเบียนออเดอร์
• การปฏิเสธ เช่นการยกเลิกออเดอร์ที่จัดส่งแล้ว ซึ่งเอเจนต์ต้องอธิบายกฎแทนการขอโทษ
• หมายเลขออเดอร์ที่ไม่รู้จัก ตรวจให้แน่ว่าเอเจนต์บอกตามนั้นแทนการแต่งสถานะขึ้นมา
• เครื่องมือที่ส่งข้อผิดพลาดกลับมา ตรวจว่าเอเจนต์พูดข้อผิดพลาดแทนการนิ่งไป
• การเชื่อมต่อใหม่และการกู้คืน รวมถึงหน้าต่างการเล่นซ้ำที่เจอข้างต้น
• เสียงรบกวนมาก การพูดเร็ว และผู้โทรที่สะกดตัวเลขและที่อยู่ทีละตัว

ฉันรันทดสอบส่วนใหญ่กับคีย์จริงระหว่างเขียน ความล้มเหลวที่น่าสนใจเป็นเชิงพฤติกรรม ไม่ใช่ข้อผิดพลาด: เวลาเล่นซ้ำใน resumption ข้างต้น และค่า threshold ของ VAD ที่เกินช่วงถูกยอมรับแทนการปฏิเสธ เรื่องแบบนี้จะหลุดไปแบบเสีย ๆ หากทดสอบแค่เส้นทางปกติ เพิ่มการทดสอบหลายภาษา และดูส่วนคำถามที่พบบ่อยสำหรับลูกเล่นในการระบุชื่อภาษา

สองในนั้นคุณพิมพ์ทดสอบไม่ได้ app_streamlit.py เป็นหน้า Streamlit ที่ใส่การโทรสดในเบราว์เซอร์: ไมโครโฟนสตรีมเข้า WebSocket เดียวกันผ่าน WebRTC เสียงของเอเจนต์สตรีมกลับ และซ็อกเก็ตเปิดอยู่ตลอด

streamlit run app_streamlit.py