用 GPT-Realtime 构建全天候零售客服：avatarin 的两周实战启示

python -m pip install websocket-client
export OPENAI_API_KEY="replace-with-your-key"
export OPENAI_REALTIME_MODEL="replace-with-current-realtime-model"
python realtime_retail_demo.py question.pcm answer.pcm

import base64
import json
import os
import sys
import websocket

if len(sys.argv) != 3:
    raise SystemExit("usage: python realtime_retail_demo.py input.pcm output.pcm")

api_key = os.environ["OPENAI_API_KEY"]
model = os.environ["OPENAI_REALTIME_MODEL"]
url = f"wss://api.openai.com/v1/realtime?model={model}"

ws = websocket.create_connection(
    url,
    header=[f"Authorization: Bearer {api_key}"],
)

ws.send(json.dumps({
    "type": "session.update",
    "session": {
        "modalities": ["text", "audio"],
        "instructions": (
            "You are a multilingual retail assistant. "
            "Answer briefly in the customer's language. "
            "Never invent prices, inventory, warranty terms, or order status. "
            "When verified data is unavailable, offer a human handoff."
        ),
        "input_audio_format": "pcm16",
        "output_audio_format": "pcm16"
    }
}))

with open(sys.argv[1], "rb") as source:
    while chunk := source.read(12_000):
        ws.send(json.dumps({
            "type": "input_audio_buffer.append",
            "audio": base64.b64encode(chunk).decode("ascii")
        }))

ws.send(json.dumps({"type": "input_audio_buffer.commit"}))
ws.send(json.dumps({"type": "response.create"}))

with open(sys.argv[2], "wb") as output:
    while True:
        event = json.loads(ws.recv())
        event_type = event.get("type")

        if event_type == "response.audio.delta":
            output.write(base64.b64decode(event["delta"]))
        elif event_type == "response.done":
            break
        elif event_type == "error":
            raise RuntimeError(json.dumps(event, ensure_ascii=False))

ws.close()
print(f"audio response written to {sys.argv[2]}")