บทเรียน 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 เทียบกับ 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.08 ต่อนาทีของออดิโอ เทียบกับ $0.05 ของ 1.0 ดังนั้นการใช้อะไลอัสที่ไม่ตรึงเวอร์ชันจะแพงขึ้นโดยที่โค้ดคุณไม่ได้เปลี่ยนแม้แต่บรรทัดเดียว ตรึงสตริงแบบมีเวอร์ชันในทุกสิ่งที่คุณดีพลอย

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

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

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

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

• voice_client.py ห่อ WebSocket ติดตามบิลลิง และให้ตัวช่วยส่ง/รับ

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

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

• token_server.py เป็นเอ็นด์พอยต์ FastAPI เล็กๆ สำหรับออกโทเค็นชั่วคราว

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

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

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

ต้องมี บัญชี SpaceXAI พร้อมคีย์ API การจัดการบิลลิงที่มีเงิน (ไม่มีระดับฟรีถาวร และเครดิตโปรโมชันสำหรับบัญชีใหม่ไม่พอ) และความคุ้นเคยกับ asyncio และ WebSockets พอที่จะตามได้โดยไม่ต้องอธิบาย 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 ไม่กี่วินาทีหลังเชื่อมต่อส่วนใหญ่ พูดถึงไว้เพื่อไม่ให้คุณเข้าใจผิดว่าเป็นเออร์เรอร์

รูปแบบออดิโอและการขนส่ง

Codec และการขนส่งเป็นคนละเรื่อง Codec ซึ่งตั้งใต้ 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 และพารามิเตอร์โมเดล และข้อความแรกที่คุณส่งจะกำหนดทุกอย่างเกี่ยวกับพฤติกรรมของเอเจนต์:

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 ให้เรียบง่ายขึ้น แทนการพอร์ตคำต่อคำ ของฉันบอกเอเจนต์ให้ตอบสั้น ถามทีละคำถาม และอ่านสิ่งที่กำลังจะเขียนกลับก่อนลงมือยืนยันด้วยเสียงเป็นเรื่องประสบการณ์ผู้ใช้ ไม่ใช่มาตรการความปลอดภัย แอปของคุณยังต้องบังคับใช้การอนุญาตบนการเขียนเองอยู่ดี

สิ่งหนึ่งที่ทำให้ฉันแปลกใจ: หากสตริงโมเดลไม่รู้จัก จะไม่ยกเออร์เรอร์ตอนเชื่อมต่อ แต่ fallback เงียบๆ ไปยัง 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(),
    }))

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 บวกกับฟังก์ชัน 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 ปลอดภัยสำหรับการรีทรายหากมีการหมดเวลา การเขียนไม่ใช่: รีทราย update_delivery_instructions หลังหมดเวลาแบบคลุมเครืออาจทำให้เปลี่ยนซ้ำสองครั้ง บรรทัดยืนยันในพรอมป์ไม่หยุดสิ่งนี้ ให้ใช้ idempotency key หรือการตรวจจับซ้ำแทนสำหรับงานเขียน

ใส่กติกาการปฏิเสธในฟังก์ชันด้วย 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 เพื่อหั่นรายการผู้ช่วยให้เหลือเท่าที่ผู้ใช้ได้ยิน เอกสารยืนยันว่ามีอยู่แต่ไม่ได้ระบุเวลาที่ควรยิงระหว่าง barge-in สดๆ ต้องทดสอบจังหวะเอง

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

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

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

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 จะทำให้ประวัติการสนทนาที่เอเจนต์เสียงบันทึกไว้หายไปและจึงใช้กับ resumption ไม่ได้

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

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

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

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

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

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

streamlit run app_streamlit.py