qwen-audio-3.0-realtime-plus

即時語音對話模型，接受語音與文字，並以語音或文字回應。設計重點是自然的交談節奏，以及與工具互動。

Ailoot · 即時語音 GET /v1/realtime?model=qwen-audio-3.0-realtime-plus qwen-audio-3.0-realtime-plus

關於模型

此說明介紹模型的能力。Ailoot 可用的輸入、輸出與選項，依所選方法列於下方。

Qwen-Audio-3.0-Realtime-Plus 是 Alibaba 的即時語音對話模型。它接受語音與文字，以音訊與文字回應，並支援自然交換話語的雙向對話。聲音可分段傳送與接收，讓互動隨對話進行。

判斷對方是否說完

可透過聲音活動、語意完整度或明確控制，判斷一次發話的邊界。在模型開發商提供的模式中，須在傳送第一段音訊前選定模式；開始對話後若要變更，必須建立新的連線。這種控制是否可用，取決於選定的方法。語意模式會考量停頓與填充詞：短暫猶豫不一定表示使用者已說完。使用者的新發話可中斷正在形成的回應，讓對話依下一段發話繼續。語意模式下，可用指定對話者的語音樣本引導辨識，以減少其他人聲及背景雜音的干擾。

指令設定助理角色及交談風格。語音表達可考量語調、速度與情緒；3.0 版本提供五種系統聲音，以及依語音樣本設定的聲音。模型支援 11 種語言，包括俄語，也支援模型開發商列出的漢語方言變體。在模型開發商的設定方式中，聲音於第一次更新工作階段設定時指定；同一段對話之後傳入的新聲音會被忽略。這些設定須由選定的呼叫方式支援。

工具與對話歷史

支援函式呼叫。模型提出所需動作，連接的應用程式執行動作，再將結果傳回對話。這讓對話能圍繞實際資料展開，並在操作完成後繼續。語音助理可利用工具結果，同時保留對話脈絡。模型開發商的功能也包括內建網路搜尋，前提是選定的方法有提供。網路搜尋與函式呼叫不能在同一工作階段同時啟用。

預設歷史保留 20 輪對話，可設定的範圍為一至 50 輪。累積音訊的上限為 300 秒；超過限制時，較舊的歷史片段會被刪除。這與單次發話的長度不同：歷史設定決定繼續對話時，哪些先前素材仍可使用。

文字脈絡與聲音

正式脈絡長度為 40 960 個 Token。輸入長度上限為 16 384 個 Token，輸出上限為 8192；推理模式也採用相同的輸入與輸出上限，另外限制推理鏈最多為 1024 個 Token。

串流聲音使用單聲道、16 位元 PCM：輸入為 16 kHz，輸出為 24 kHz。文字與音訊資料屬於同一段對話。此特定模型未宣稱支援視覺輸入或輸出圖片；專長是語音、文字及語音互動。

Ailoot · 即時語音 GET /v1/realtime?model=qwen-audio-3.0-realtime-plus qwen-audio-3.0-realtime-plus

qwen-audio-3.0-realtime-plus

Alibaba /qwen-audio-3.0-realtime-plus

入門

即時語音對話模型，接受語音與文字，並以語音或文字回應。設計重點是自然的交談節奏，以及與工具互動。

接受的輸入: 文字 · 音訊
結果: 文字 · 音訊

Ailoot 可用 · 輸入與方法

Ailoot · 即時語音 GET /v1/realtime?model=qwen-audio-3.0-realtime-plus

輸入: text, audio
輸出: text, audio
輸出格式: 工作階段事件

欄位	類型	要求	值與限制	預設值
欄位`query.model`	類型`string`	要求必填	值與限制 qwen-audio-3.0-realtime-plus	預設值—
欄位`session.input_audio_format`	類型`string`	要求選填	值與限制 pcm16, pcm	預設值—
欄位`session.output_audio_format`	類型`string`	要求選填	值與限制 pcm16, pcm	預設值—
欄位`session.instructions`	類型`string`	要求選填	值與限制限制適用對象: 所選的工作階段政策	預設值—
欄位`session.max_history_turns`	類型`integer`	要求選填	值與限制最小值: 1 限制適用對象: 所選的工作階段政策	預設值—
欄位`input_audio_buffer.append.audio`	類型`base64_string`	要求必填	值與限制輸入內容: PCM16 單聲道；輸入 16 kHz，輸出 24 kHz；位元組長度須非零且為偶數	預設值—

輸入素材組合

realtime_text_audio

方法參數與生命週期

參數	規格
參數`material_types`	規格text, audio
參數`required`	規格null
參數`optional_controls`	規格null
參數`content_type`	規格WebSocket

生命週期： exact retained published method

方法結構與 ID 均為精確值；供應商能力與准入狀態分開。

方案

Ailoot 方法費率

Ailoot 即時語音GET /v1/realtime?model=qwen-audio-3.0-realtime-plus

輸入: US$0.0008 / 1K
輸出: US$0.0064 / 1K
輸入: US$0.0048 每分鐘音訊
輸出: US$0.018 每分鐘輸出音訊

Ailoot 會在請求前預留估算金額。取得結果後，依實測用量結算最終費用。串流結算採用最終用量紀錄。預留金額與結算的運作方式 →

請求範例

Ailoot · 即時語音

Python WebSocket · 文字Python WebSocket · 音訊

# Install the documented client once: python3 -m pip install websockets
import base64, json, os, time, wave
from pathlib import Path
from websockets.sync.client import connect

key = os.environ.get("AILOOT_API_KEY", "")
if not key.strip():
    raise SystemExit("Set your AILOOT_API_KEY before connecting")
file = Path(os.environ.get("AILOOT_TEXT_FILE", ""))
if not file.is_file(): raise SystemExit("Set AILOOT_TEXT_FILE to your actual UTF-8 material")
material = file.read_text(encoding="utf-8")
if not material.strip(): raise SystemExit("Text material must not be empty")
out = Path("realtime-results")
out.mkdir(exist_ok=True)
audio, texts, complete = {}, {}, []
current_response = None
pending = False
deadline = time.monotonic() + 570  # Local client bound; lower server/session limits still apply.

def receive(ws, log):
    global current_response, pending
    left = deadline - time.monotonic()
    if left <= 0:
        raise TimeoutError("Client session limit; preserve IDs and partial results, do not replay")
    event = json.loads(ws.recv(timeout=min(60, left)))
    log.write(json.dumps(event, ensure_ascii=False) + "\n")
    log.flush()
    kind = event.get("type")
    if kind == "error":
        raise RuntimeError(json.dumps(event, ensure_ascii=False))
    if kind == "response.created":
        current_response = event["response"]["id"]
    if kind in {"response.audio.delta", "response.output_audio.delta", "response.text.delta", "response.output_text.delta", "response.audio_transcript.delta", "response.output_audio_transcript.delta", "response.text.done", "response.audio_transcript.done"}:
        identity = (event["response_id"], event["item_id"], event["output_index"], event["content_index"])
        if kind in {"response.audio.delta", "response.output_audio.delta"}:
            raw = base64.b64decode(event["delta"], validate=True)
            if len(raw) % 2: raise ValueError("Output PCM16 chunk must have an even byte length")
            audio.setdefault(identity, bytearray()).extend(raw)
        elif kind.endswith(".done"):
            texts[identity] = event.get("text", event.get("transcript", ""))
        else:
            texts[identity] = texts.get(identity, "") + event["delta"]
            print(event["delta"], end="", flush=True)
    if kind == "response.done":
        response = event["response"]
        complete.append(response)
        if current_response is None: current_response = response["id"]
        if response["id"] == current_response: pending = False
    return event

def wait_for(ws, log, kind):
    while True:
        event = receive(ws, log)
        if event.get("type") == kind: return event

def send(ws, event):
    ws.send(json.dumps(event, ensure_ascii=False))

try:
    # One exact model query; no reconnect/retry loop and no provider credentials.
    with connect("wss://api.ailoot.ai/v1/realtime?model=qwen-audio-3.0-realtime-plus", additional_headers={"Authorization": "Bearer " + key}) as ws:
        with (out / "events.jsonl").open("w", encoding="utf-8") as log:
            wait_for(ws, log, "session.created")
            send(ws, {"type":"session.update", "session":{"modalities":["audio","text"], "input_audio_format":"pcm", "output_audio_format":"pcm", "turn_detection":None}})
            wait_for(ws, log, "session.updated")
            send(ws, {"type":"conversation.item.create", "item":{"type":"message", "role":"user", "content":[{"type":"input_text", "text":material}]}})
            send(ws, {"type":"response.create"})
            pending = True
            try:
                while pending: receive(ws, log)
            except KeyboardInterrupt:
                # Cancel the actual current inference, then preserve its returned status.
                send(ws, {"type":"response.cancel"})
                while pending: receive(ws, log)
    if not complete or complete[-1].get("status") != "completed":
        raise RuntimeError("Response did not complete normally; inspect saved events and status")
finally:
    manifest = {"responses":complete, "parts":[]}
    for number, identity in enumerate(dict.fromkeys(list(audio) + list(texts))):
        record = {"response_id":identity[0], "item_id":identity[1], "output_index":identity[2], "content_index":identity[3]}
        if identity in audio:
            raw = bytes(audio[identity])
            pcm_file = out / ("audio-" + str(number) + ".pcm")
            wav_file = out / ("audio-" + str(number) + ".wav")
            pcm_file.write_bytes(raw)
            with wave.open(str(wav_file), "wb") as wav:
                wav.setnchannels(1); wav.setsampwidth(2); wav.setframerate(24000); wav.writeframes(raw)
            record.update({"pcm":str(pcm_file), "wav":str(wav_file), "sample_rate_hz":24000})
        if identity in texts:
            text_file = out / ("text-" + str(number) + ".txt")
            text_file.write_text(texts[identity], encoding="utf-8")
            record["text"] = str(text_file)
        manifest["parts"].append(record)
    (out / "results.json").write_text(json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8")
    print("\nSaved every received event and result in", out)

# Install the documented client once: python3 -m pip install websockets
import base64, json, os, time, wave
from pathlib import Path
from websockets.sync.client import connect

key = os.environ.get("AILOOT_API_KEY", "")
if not key.strip():
    raise SystemExit("Set your AILOOT_API_KEY before connecting")
file = Path(os.environ.get("AILOOT_AUDIO_FILE", ""))
if not file.is_file(): raise SystemExit("Set AILOOT_AUDIO_FILE to your raw PCM16 mono 16-kHz recording")
material = file.read_bytes()
if not material or len(material) % 2 or material.startswith(b"RIFF"):
    raise SystemExit("Use nonempty signed little-endian PCM16 mono 16-kHz bytes, not a WAV header")
out = Path("realtime-results")
out.mkdir(exist_ok=True)
audio, texts, complete = {}, {}, []
current_response = None
pending = False
deadline = time.monotonic() + 570  # Local client bound; lower server/session limits still apply.

def receive(ws, log):
    global current_response, pending
    left = deadline - time.monotonic()
    if left <= 0:
        raise TimeoutError("Client session limit; preserve IDs and partial results, do not replay")
    event = json.loads(ws.recv(timeout=min(60, left)))
    log.write(json.dumps(event, ensure_ascii=False) + "\n")
    log.flush()
    kind = event.get("type")
    if kind == "error":
        raise RuntimeError(json.dumps(event, ensure_ascii=False))
    if kind == "response.created":
        current_response = event["response"]["id"]
    if kind in {"response.audio.delta", "response.output_audio.delta", "response.text.delta", "response.output_text.delta", "response.audio_transcript.delta", "response.output_audio_transcript.delta", "response.text.done", "response.audio_transcript.done"}:
        identity = (event["response_id"], event["item_id"], event["output_index"], event["content_index"])
        if kind in {"response.audio.delta", "response.output_audio.delta"}:
            raw = base64.b64decode(event["delta"], validate=True)
            if len(raw) % 2: raise ValueError("Output PCM16 chunk must have an even byte length")
            audio.setdefault(identity, bytearray()).extend(raw)
        elif kind.endswith(".done"):
            texts[identity] = event.get("text", event.get("transcript", ""))
        else:
            texts[identity] = texts.get(identity, "") + event["delta"]
            print(event["delta"], end="", flush=True)
    if kind == "response.done":
        response = event["response"]
        complete.append(response)
        if current_response is None: current_response = response["id"]
        if response["id"] == current_response: pending = False
    return event

def wait_for(ws, log, kind):
    while True:
        event = receive(ws, log)
        if event.get("type") == kind: return event

def send(ws, event):
    ws.send(json.dumps(event, ensure_ascii=False))

try:
    # One exact model query; no reconnect/retry loop and no provider credentials.
    with connect("wss://api.ailoot.ai/v1/realtime?model=qwen-audio-3.0-realtime-plus", additional_headers={"Authorization": "Bearer " + key}) as ws:
        with (out / "events.jsonl").open("w", encoding="utf-8") as log:
            wait_for(ws, log, "session.created")
            send(ws, {"type":"session.update", "session":{"modalities":["audio","text"], "input_audio_format":"pcm", "output_audio_format":"pcm", "turn_detection":None}})
            wait_for(ws, log, "session.updated")
            for offset in range(0, len(material), 640):
                send(ws, {"type":"input_audio_buffer.append", "audio":base64.b64encode(material[offset:offset+640]).decode("ascii")})
                time.sleep(0.02)
            send(ws, {"type":"input_audio_buffer.commit"})
            wait_for(ws, log, "input_audio_buffer.committed")
            send(ws, {"type":"response.create"})
            pending = True
            try:
                while pending: receive(ws, log)
            except KeyboardInterrupt:
                # Cancel the actual current inference, then preserve its returned status.
                send(ws, {"type":"response.cancel"})
                while pending: receive(ws, log)
    if not complete or complete[-1].get("status") != "completed":
        raise RuntimeError("Response did not complete normally; inspect saved events and status")
finally:
    manifest = {"responses":complete, "parts":[]}
    for number, identity in enumerate(dict.fromkeys(list(audio) + list(texts))):
        record = {"response_id":identity[0], "item_id":identity[1], "output_index":identity[2], "content_index":identity[3]}
        if identity in audio:
            raw = bytes(audio[identity])
            pcm_file = out / ("audio-" + str(number) + ".pcm")
            wav_file = out / ("audio-" + str(number) + ".wav")
            pcm_file.write_bytes(raw)
            with wave.open(str(wav_file), "wb") as wav:
                wav.setnchannels(1); wav.setsampwidth(2); wav.setframerate(24000); wav.writeframes(raw)
            record.update({"pcm":str(pcm_file), "wav":str(wav_file), "sample_rate_hz":24000})
        if identity in texts:
            text_file = out / ("text-" + str(number) + ".txt")
            text_file.write_text(texts[identity], encoding="utf-8")
            record["text"] = str(text_file)
        manifest["parts"].append(record)
    (out / "results.json").write_text(json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8")
    print("\nSaved every received event and result in", out)

接收並儲存結果

Ailoot · 即時語音

工作階段事件

保留所有已完成的結果

請保留所有完整傳回的回應、檔案與內容區塊。

請依每個實際回應 ID，組合所有回傳的音訊增量片段與文字／事件結果。

需要檢查原始回應時

請保留實際的工作階段／回應／事件 ID 與紀錄憑證。

錯誤與後續步驟

回應	條件	下一步
回應426 · `invalid_request_error`	條件未進行 WebSocket 升級	下一步請使用 WebSocket 用戶端，並帶入 Upgrade: websocket。

別名

模型別名

Ailoot qwen-audio-3.0-realtime-plus

方法路徑別名

請使用上方列出的 API 方法路徑。

qwen-audio-3.0-realtime-plus