作者 | 崔皓

審校 | 重樓

OpenAI 推出的 Realtime API 標志著語音交互技術(shù)的一次重大突破。它允許開發(fā)者構(gòu)建低延遲、高效率的多模態(tài)對話體驗，支持文本和音頻輸入輸出，為語音助手、在線教育、游戲等場景帶來了新的可能性。

傳統(tǒng)的語音交互模式存在明顯延遲，需要經(jīng)過“聲音->文字->文字推理->聲音”的轉(zhuǎn)換過程，導致情感、重點和口音的喪失，影響用戶體驗。Realtime API 通過直接流式傳輸音頻輸入輸出，優(yōu)化了這一過程，實現(xiàn)了更加自然、流暢的對話體驗。它還能夠自動處理中斷，并支持函數(shù)調(diào)用，使得語音助手能夠更加智能地響應用戶請求。

Realtime API 使用 WebSocket 協(xié)議進行雙向通信，并通過事件機制實現(xiàn)消息的發(fā)送和接收。開發(fā)者可以通過監(jiān)聽不同的事件來完成消息的發(fā)送和接受，而且事件驅(qū)動機制非常適合處理異步通信。

本文詳細介紹了 Realtime API 的功能、原理和應用場景，并通過代碼示例展示了如何使用該 API 創(chuàng)建語音交互應用。

開篇

OpenAI最近推出的Realtime API，標志著人工智能交互技術(shù)的一個重大突破。這個API目前處于公開測試階段，允許開發(fā)者在其應用中構(gòu)建低延遲的多模態(tài)對話體驗，支持文本和音頻作為輸入和輸出。這意味著，通過Realtime API，開發(fā)者可以創(chuàng)建出自然、實時的語音對語音交互，無需經(jīng)過中間的文本轉(zhuǎn)換，從而實現(xiàn)更加流暢和互動的用戶體驗。此API的發(fā)布，不僅降低了開發(fā)門檻，還推動了AI技術(shù)的應用創(chuàng)新，為語音助手、在線教育、游戲等場景帶來了新的可能性。

為什么需要 Realtime API？

在追求極致用戶體驗的今天，傳統(tǒng)的語音助手技術(shù)面臨著重大挑戰(zhàn)。在過去，為了構(gòu)建類似的語音交互體驗，開發(fā)人員必須依賴一系列復雜的流程：首先使用類似Whisper語音識別模型來轉(zhuǎn)錄音頻，也就是將人類輸入的聲音轉(zhuǎn)化為文字的形式。接著，轉(zhuǎn)化之后的文字傳遞給LLM（大預言模型，例如：GPT-4o）進行推理，最后將LLM推理出來的文字通過文本轉(zhuǎn)語音的模型來播放輸出。相應過程經(jīng)歷了“聲音->文字->文字推理->聲音”，方法不僅會引入明顯的延遲，還常常導致情感、重點和口音的喪失，從而影響用戶體驗。

然而，隨著OpenAI Realtime API的推出，這一切都得到了顯著改善。Realtime API通過直接流式傳輸音頻輸入和輸出優(yōu)化了這一過程，實現(xiàn)了更加自然和實時的對話體驗。它還能夠自動處理中斷，類似于ChatGPT中的高級語音模式。此外，實時API允許創(chuàng)建持久的WebSocket連接，以便與GPT-4o交換消息，支持函數(shù)調(diào)用，這使得語音助手能夠更加智能地響應用戶請求。例如，助手可以代表用戶下訂單或檢索相關(guān)客戶信息，以個性化其響應。簡而言之，Realtime API不僅解決了“實時”交互的主要問題，還通過函數(shù)調(diào)用功能，滿足了客戶對個性化服務的需求，為語音交互應用帶來了革命性的進步。

嘗鮮 Realtime API？

 既然Realtime API 可以讓用戶與大模型通過語音方式進行實時交互，那么就讓我們走近Realtime API 感受一下它的魅力吧！通過OpenAI的Playground的頁面就可以訪問Realtime API，如下圖所示。這是 OpenAI Playground 的界面，用于訪問 Realtime API 功能。該界面為開發(fā)者提供了一個簡單易用的平臺，用來測試實時的語音交互應用。

在這個界面中，開發(fā)者可以進行以下操作：

1.會話區(qū)域

中心區(qū)域顯示的是實時會話窗口，任何語音或文本交互都會在這里出現(xiàn)。在靠下方的區(qū)域中，用戶可以通過點擊“Start session”按鈕啟動一個會話，與 API 進行交互。

2.語音設(shè)置

右側(cè)提供了一些基本設(shè)置，如選擇語音模型（例如 Alloy）和配置語音活動檢測。

Threshold（閾值）、Prefix padding（前綴填充）、Silence duration（靜默時長）：這些參數(shù)允許用戶微調(diào)語音檢測的靈敏度，確保系統(tǒng)能夠準確地捕捉和處理語音輸入。

3.功能與模型配置

開發(fā)者可以添加自定義的功能（Functions），這為構(gòu)建復雜的語音交互場景提供了可能。還有模型的溫度（Temperature）和最大生成字數(shù)（Max tokens）的配置，用來控制模型生成內(nèi)容的創(chuàng)意程度和響應長度。

目前該界面已經(jīng)給Plus 用戶開放使用權(quán)限，開發(fā)者可以配置和測試 Realtime API，了解其響應效果并調(diào)整設(shè)置。

我們可以點擊“Start session”開啟一次兌換，此時需要獲取麥克風的權(quán)限，在對話之前需要在右上方的“System instructions”中告訴模型要執(zhí)行的指令。

當出現(xiàn)下圖“Start talking”的時候就可以與LLM進行對話了。

如下圖所示，我們輸入簡單的“Who are you？”（00:03）時，LLM 在很短時間（00:04）立即給出了回復：“I’m an AI here to help out!” 。目前來說是Realtime API的Beta 版本，對話的token 有一定限制。

通過官網(wǎng)宣傳可知，該 API 的音頻功能由全新的 GPT-4o 模型提供支持，具體版本為 gpt-4o-realtime-preview。而在幾周內(nèi)，開發(fā)者還將迎來 gpt-4o-audio-preview，該版本支持通過文本或音頻輸入與 GPT-4o 模型交互，并生成文本或音頻輸出，或兩者兼?zhèn)涞捻憫问健?/p>

不過從使用價格方面來看，是有一點小貴，如下：

文本Token：每 100 萬個輸入Token 5 美元，每 100 萬個輸出Token 20 美元。

音頻Token：每 100 萬個輸入Token 100 美元，輸出Token 200 美元。這意味著，開發(fā)者大致會為每分鐘音頻輸入支付 0.06 美元，音頻輸出支付 0.24 美元。

為什么不直接調(diào)用API ？

從上面的Realtime API 演示可以看出，僅僅通過輸入語音就可以和GPT 模型進行對話，這個和ChatGPT的功能類似，僅僅把輸入的文字改成了語音，看上去好像就是直接調(diào)用API接口就可以完成功能，為什么硬生生搞出一個Realtime API的概念。細心的你可能注意到了，在語音對話的場景中有一個特點，用戶與大模型之間需要創(chuàng)建一個會話，基于這個會話會有多次的，實時的交流，這里的“實時”就是Realtime API與普通API存在不同的地方。

由此我們可以推斷出Realtime API執(zhí)行的基本邏輯，也就是模型、外部系統(tǒng)、用戶輸入之間的協(xié)同工作：

Client 端的角色：Client（客戶端）是負責發(fā)起操作和接收用戶輸入的地方。在這個架構(gòu)中，客戶端可以是與用戶交互的界面，比如一個聊天應用、網(wǎng)頁、移動端應用等?？蛻舳诵枰獙⒂脩舻恼埱髠鬟f給服務端，并且根據(jù)返回的結(jié)果展示給用戶。

Server 端的角色：Server（服務器端）是主要的處理中心，它負責：接收從客戶端發(fā)來的請求。與 OpenAI 的模型（API）交互，獲取初步的結(jié)果。根據(jù)情況，決定是否需要進一步調(diào)用外部系統(tǒng)或函數(shù)（如數(shù)據(jù)庫查詢、外部服務請求）。處理外部請求的結(jié)果，并通過事件機制將結(jié)果推送回客戶端。需要注意的是這里的Client 和Server 需要保持一個實時的，“長”連接保證信息溝通的順暢。

為了實現(xiàn)Client 與Server 之間的信息溝通，于是Realtime 引入了事件機制：

異步操作處理：Server 在與外部系統(tǒng)通信時，有可能需要等待數(shù)據(jù)返回，而不希望在這期間阻塞整個流程。事件驅(qū)動機制使得 Server 可以發(fā)起請求后立即返回，而在外部操作完成時，通過事件通知系統(tǒng)完成操作并返回結(jié)果。

事件驅(qū)動的更新：比如在某些場景下，外部系統(tǒng)可能隨時推送新的狀態(tài)更新或結(jié)果，這時候事件機制可以使得 Server 自動處理并通過事件將結(jié)果通知客戶端。

多客戶端協(xié)作：在一些實時系統(tǒng)中，比如聊天系統(tǒng)、監(jiān)控系統(tǒng)，多個客戶端可能同時與服務器交互，事件驅(qū)動模型允許多客戶端之間實時共享和同步數(shù)據(jù)，而不需要每個請求都單獨進行阻塞調(diào)用。

Realtime API的特點

我們通過使用OpenAI 提供的Playground 了解了Realtime API的基本功能，雖然目前測試的Token量有限，但讓我們對它的了解更近了一步。然后，分析了Realtime API 執(zhí)行的基本邏輯，通過Client 與Server 創(chuàng)建長連接，利用事件機制在他們之間發(fā)送實時消息。

在接下來的部分，我們將詳細介紹 Realtime API 的功能和應用，在Realtime API中特別實現(xiàn)了一個實時 API。這個實時API是一種有狀態(tài)、基于事件的API，通過 WebSocket 進行通信。這個如何理解呢？這里就需要理解Realtime API 機制的三個特點：

WebSocket 通信：該 API 使用 WebSocket 協(xié)議進行雙向通信。與傳統(tǒng)的 HTTP 請求-響應模式不同，WebSocket 允許客戶端和服務器之間建立持久連接，降低了延遲，提高了數(shù)據(jù)傳輸?shù)男?，特別適合實時語音、文本或其他模態(tài)數(shù)據(jù)的交互。

設(shè)置狀態(tài)：由于需要保持實時通信，因此需要通過WebSocket實現(xiàn)長連接。那么就需要針對 WebSocket 連接保持其會話的上下文，從而在整個會話中跟蹤用戶的請求和響應，這樣才能支撐持續(xù)、多輪次的對話場景。

事件交互：在創(chuàng)建會話通道之后，就需要定義溝通方式，Realtime API 是基于事件來收發(fā)消息的，開發(fā)者可以通過監(jiān)聽不同的事件完成消息的發(fā)送和接受，而且事件驅(qū)動機制非常適合處理異步通信。

我們通過一段Realtime API的代碼讓大家更近一步了解，代碼如下：

import WebSocket from "ws";
const url = "wss://api.openai.com/v1/realtime?model=gpt-4o-realtime-preview-2024-10-01";
const ws = new WebSocket(url, {
    headers: {
        "Authorization": "Bearer " + process.env.OPENAI_API_KEY,
        "OpenAI-Beta": "realtime=v1",
    },
});
ws.on("open", function open() {
    console.log("Connected to server.");
    ws.send(JSON.stringify({
        type: "response.create",
        response: {
            modalities: ["text"],
            instructions: "Please assist the user.",
        }
    }));
});
ws.on("message", function incoming(message) {
    console.log(JSON.parse(message.toString()));
});

代碼中，通過 wss://api.openai.com/v1/realtime URL 建立一個 WebSocket 連接，并指定使用模型 gpt-4o-realtime-preview-2024-10-01。在連接成功后，發(fā)送一個請求，讓模型使用 "text" 模態(tài)，并為用戶提供幫助。隨后，當服務器發(fā)送消息時，我們會通過 "message" 事件接收到響應，并將其打印出來。

import WebSocket from "ws";

使用 WebSocket 模塊，它是基于 WebSocket 協(xié)議的一個庫，用于在客戶端和服務器之間創(chuàng)建持久的、全雙工通信通道。ws 模塊是 Node.js 環(huán)境中常用的 WebSocket 實現(xiàn)。

const url = "wss://api.openai.com/v1/realtime?model=gpt-4o-realtime-preview-2024-10-01";

url 是 WebSocket 的服務器地址，使用的是 OpenAI Realtime API 的 WebSocket 連接地址。wss:// 表示這是一個加密的 WebSocket 連接，類似于 https。model=gpt-4o-realtime-preview-2024-10-01 代表我們要使用的模型版本。

const ws = new WebSocket(url, {
    headers: {
        "Authorization": "Bearer " + process.env.OPENAI_API_KEY,
        "OpenAI-Beta": "realtime=v1",
    },
});

使用 WebSocket 構(gòu)造函數(shù)來創(chuàng)建一個 WebSocket 連接。Authorization 是API 的認證信息，用于驗證開發(fā)者的身份。process.env.OPENAI_API_KEY 是從環(huán)境變量中讀取的 OpenAI API 密鑰，確保安全地訪問 API。OpenAI-Beta 是額外的請求頭，指定使用的是 Realtime API 的測試版本。"realtime=v1" 表示這是 Realtime API 的第一個版本。

ws.on("open", function open() {
    console.log("Connected to server.");
    ws.send(JSON.stringify({
        type: "response.create",
        response: {
            modalities: ["text"],
            instructions: "Please assist the user.",
        }
    }));
});

ws.on("open", ...)：這是一個事件監(jiān)聽器，當 WebSocket 連接成功打開時，會觸發(fā) "open" 事件。我們使用匿名函數(shù)來處理這個事件。

console.log("Connected to server.") 用于輸出成功連接的消息到控制臺，確認連接已經(jīng)建立。

ws.send(...) 通過 WebSocket 發(fā)送一條 JSON 格式的消息給服務器，表示希望創(chuàng)建一個響應。

"type": "response.create" 定義了請求的類型，這里表示要創(chuàng)建一個新的響應。"modalities": ["text"] 指定了使用的模態(tài)（即交互形式），此處是文本模態(tài)。"instructions": "Please assist the user." 是我們發(fā)給 API 的指令，讓它協(xié)助用戶進行對話。

ws.on("message", function incoming(message) {
    console.log(JSON.parse(message.toString()));
});

ws.on("message", ...)：當從服務器接收到一條消息時，會觸發(fā) "message" 事件。message.toString() 將接收到的二進制消息轉(zhuǎn)換為字符串形式。JSON.parse() 用于將字符串解析成 JSON 對象。console.log() 將解析后的消息對象打印到控制臺，便于調(diào)試和查看返回結(jié)果。

通過上面的操作，我們就與Realtime API建立了連接，接著就可以通過發(fā)送事件的方式，讓Realtime API幫我們完成各種不同的功能。

接下來，試圖通過如下代碼讓Realtime API針對我們輸入的文字產(chǎn)生響應。

const event = {
  type: 'conversation.item.create',
  item: {
    type: 'message',
    role: 'user',
    content: [
      {
        type: 'input_text',
        text: 'Hello!'
      }
    ]
  }
};
ws.send(JSON.stringify(event));
ws.send(JSON.stringify({type: 'response.create'}));

上述代碼首先創(chuàng)建event對象，通過 WebSocket 發(fā)送用戶的文本輸入（Hello!）到服務器。Event對象中定義了type: 'conversation.item.create' ，從字面理解為對話。如果說每次客戶端與服務器的WebSocket的連接為一個session（會話），一旦客戶端創(chuàng)建會話，它就會發(fā)送包含文本和音頻塊的 JSON 格式的event（事件）。服務器將包含語音輸出的音頻、該語音輸出的文本記錄和函數(shù)調(diào)用進行響應。

一個session（會話）中有可能包含多個conversation（對話）。每次對話的內(nèi)容可以不同，可以聊天氣，可以聊商品訂單，當時他們都保存在一個session（會話）中，也就是通過一個長連接session支持多次對話。

event 對象中的 item 屬性用于向 Realtime API 發(fā)送不同類型的消息或事件。在這個例子中，item 具有一個名為 message 的屬性，它表示一條消息，用戶發(fā)送的文本或音頻都可以封裝在這里。item 可以是三種類型之一：message、function_call 或 function_call_output。

• message: 表示一條消息，包含文本或音頻的輸入。
• function_call: 表示模型希望調(diào)用某個工具或函數(shù)。
• function_call_output: 表示函數(shù)調(diào)用的返回結(jié)果。

在這段代碼中，通過 WebSocket 向服務器發(fā)送一個 conversation.item.create 事件，其中 item 是一條用戶發(fā)送的文本消息。這一事件告知服務器用戶輸入了 "Hello!"，并請求模型生成適當?shù)捻憫?/p>

音頻交互

前面通過兩段代碼描述了如何通過Client 與Server 端進行WebSocket連接，利用event發(fā)送消息得到響應。接下來利用相同的手法，通過event 傳送音頻信息獲得響應。我們需要將音頻文件轉(zhuǎn)換為 PCM16 格式的 base64 編碼數(shù)據(jù)，并通過 WebSocket 發(fā)送到服務端作為用戶輸入的音頻消息。接下來是代碼的片段：

1.引入模塊

import fs from 'fs';
import decodeAudio from 'audio-decode';

fs模塊用于讀取本地文件系統(tǒng)中的音頻文件。

audio-decode 庫用來將音頻文件解碼為原始的音頻字節(jié)（`AudioBuffer` 對象），便于后續(xù)處理。

2.floatTo16BitPCM 函數(shù)

function floatTo16BitPCM(float32Array) {
  const buffer = new ArrayBuffer(float32Array.length * 2);
  const view = new DataView(buffer);
  let offset = 0;
  for (let i = 0; i < float32Array.length; i++, offset += 2) {
    let s = Math.max(-1, Math.min(1, float32Array[i]));
    view.setInt16(offset, s < 0 ? s * 0x8000 : s * 0x7fff, true);
  }
  return buffer;
}

函數(shù)用于將 Float32Array（浮點數(shù)組格式的音頻數(shù)據(jù)）轉(zhuǎn)換為 PCM16 格式的 ArrayBuffer。PCM16 是常見的音頻編碼格式，將浮點值（-1 到 1 之間）轉(zhuǎn)換為 16 位整數(shù)。DataView對象允許你對二進制 ArrayBuffer 進行逐字節(jié)操作。setInt16 方法用于寫入 16 位整數(shù)。

3.base64EncodeAudio 函數(shù)

function base64EncodeAudio(float32Array) {
  const arrayBuffer = floatTo16BitPCM(float32Array);
  let binary = '';
  let bytes = new Uint8Array(arrayBuffer);
  const chunkSize = 0x8000; // 32KB chunk size
  for (let i = 0; i < bytes.length; i += chunkSize) {
    let chunk = bytes.subarray(i, i + chunkSize);
    binary += String.fromCharCode.apply(null, chunk);
  }
  return btoa(binary);
}

該函數(shù)首先調(diào)用 `floatTo16BitPCM` 將音頻數(shù)據(jù)轉(zhuǎn)換為 PCM16 格式。接下來，它將 `ArrayBuffer` 轉(zhuǎn)換為 `Uint8Array`，方便逐字節(jié)處理。然后，它將音頻數(shù)據(jù)按 32KB 塊（`chunk`）讀取，并通過 `String.fromCharCode` 將字節(jié)轉(zhuǎn)換為字符串形式。最后，為了便于通過網(wǎng)絡傳輸音頻數(shù)據(jù)，使用 `btoa` 將該二進制字符串編碼為 base64。

4.解碼音頻文件并獲取通道數(shù)據(jù)

const myAudio = fs.readFileSync('./path/to/audio.wav');
const audioBuffer = await decodeAudio(myAudio);
const channelData = audioBuffer.getChannelData(0); // 僅處理單聲道音頻

fs.readFileSync 同步讀取音頻文件的內(nèi)容。decodeAudio 將音頻文件解碼為 AudioBuffer，并通過 getChannelData(0)獲取音頻的通道數(shù)據(jù)。這里假設(shè)處理的是單聲道音頻（如果是多聲道，需要處理多個通道）。

5.發(fā)送音頻數(shù)據(jù)

const base64AudioData = base64EncodeAudio(channelData);
const event = {
  type: 'conversation.item.create',
  item: {
    type: 'message',
    role: 'user',
    content: [
      {
        type: 'input_audio',
        audio: base64AudioData
      }
    ]
  }
};
ws.send(JSON.stringify(event));
ws.send(JSON.stringify({type: 'response.create'}));

通過調(diào)用 base64EncodeAudio 函數(shù)，將音頻通道數(shù)據(jù)編碼為 base64 格式。構(gòu)建 WebSocket 事件，其中 item是類型為 message 的用戶輸入，且內(nèi)容是 base64 編碼的音頻數(shù)據(jù)（input_audio）。ws.send(JSON.stringify(event)) 通過 WebSocket 向服務器發(fā)送這條消息。隨后，通過發(fā)送 response.create 請求，要求服務器生成對應的響應。

事件類型定義

前面幾段代碼我們發(fā)現(xiàn)無論是Client還是Server 在發(fā)送和接受event的時候都會定義event類型，比較常見的有conversation.item.create 創(chuàng)建對話，respnse.create響應。如下圖所示，實際上在Realtime API中定義了9個Client 事件類型和28個Server事件類型。

以我們常用的conversation.item.create類型為例， 其完全體的定義如下：

{
    "event_id": "event_345",
    "type": "conversation.item.create",
    "previous_item_id": null,
    "item": {
        "id": "msg_001",
        "type": "message",
        "status": "completed",
        "role": "user",
        "content": [
            {
                "type": "input_text",
                "text": "Hello, how are you?"
            }
        ]
    }
}

conversation.item.create 是一個事件類型，主要用于在對話中添加一個新項目（例如消息或函數(shù)調(diào)用）。此事件包含以下關(guān)鍵屬性：

1. event_id：可選的客戶端生成的 ID，用于標識事件。
2. type：事件類型，必須為 "conversation.item.create"。
3. previous_item_id：新項目插入后，前一個項目的 ID。
4. item：要添加的項目，包含：

• id：項目唯一 ID。
• type：項目類型（"message"、"function_call"、"function_call_output"）。
• status：項目狀態(tài)（"completed"、"in_progress"、"incomplete"）。
• role：消息發(fā)送者的角色（"user"、"assistant"、"system"）。
• content：消息內(nèi)容。
• call_id、name、arguments、output：用于函數(shù)調(diào)用相關(guān)的詳細信息。

Realtime AI 最佳實踐

在OpenAI推出Realtime API的第一時間，LangChain也推出了與之相關(guān)的開源項目： Voice ReAct Agent，它是一個基于OpenAI實時API的ReAct風格代理的最佳實現(xiàn)。它通過LangChain工具列表調(diào)用工具，并且用戶可以編寫和傳遞自定義工具給模型。目前該項目包含Python 和TypeScript兩個版本。具體的安裝、使用過程可以參考官方鏈接，整個過程非常簡單，這里不做贅述。

我們把關(guān)注點放到代碼實現(xiàn)上， 如下圖所示，這里是Python Server端的代碼，__init__.py。 它主要完成了下面三個方面的任務，

• WebSocket 連接：connect()負責管理與 OpenAI API 的 WebSocket 連接，發(fā)送和接收數(shù)據(jù)。
• 工具執(zhí)行：VoiceToolExecutor 負責工具調(diào)用的異步執(zhí)行，確保并發(fā)操作的安全性。
• 實時 API 代理：OpenAIVoiceReactAgent 管理與 OpenAI 實時 API 的交互，執(zhí)行工具并根據(jù)輸入流式傳輸響應。

由于篇幅有限，這里不對所有代碼進行講解，只針對主要的類和函數(shù)進行描述如下：

1.connect()(函數(shù))  

這是一個異步上下文管理器，用于建立到 OpenAI API 的 WebSocket 連接。它接收 API 密鑰、模型和 URL，并返回兩個組件：用于發(fā)送數(shù)據(jù)的函數(shù) send_event 和用于接收事件的迭代器 event_stream。該函數(shù)保證在使用后正確關(guān)閉 WebSocket。

2.VoiceToolExecutor (類)  

此類通過函數(shù)調(diào)用異步執(zhí)行工具，當與用戶交互時，如果需要用到外部工具就會啟用該類。

包含屬性：

• tools_by_name: 工具名稱與 BaseTool 對象的字典。
• _trigger_future: 管理函數(shù)調(diào)用觸發(fā)的 asyncio.Future 對象。
• _lock: 用于安全處理并發(fā)操作的asyncio.Lock。

函數(shù)：

•   _trigger_func()：等待 future 對象返回工具調(diào)用數(shù)據(jù)。
•   add_tool_call(tool_call)：添加工具調(diào)用，確保不會被其他并發(fā)調(diào)用覆蓋。
•   _create_tool_call_task(tool_call)：創(chuàng)建并運行處理工具調(diào)用的任務，使用工具的 `ainvoke()` 方法解析 JSON 參數(shù)并處理錯誤。
•   output_iterator()：持續(xù)返回任務結(jié)果的主循環(huán)，管理并發(fā)任務并處理錯誤。

3.OpenAIVoiceReactAgent (類)  

這是核心代理類，負責管理用戶輸入和 Realtime API 實時交互，并處理工具的執(zhí)行。

屬性：

• model: 使用的 OpenAI 模型。
• api_key: 以 SecretStr 格式安全存儲的 API 密鑰。
• instructions: 可選的模型指令。
• tools: 可用的工具列表。
• url: OpenAI API 的 WebSocket URL。

函數(shù)：

• aconnect(input_stream, send_output_chunk)：連接 OpenAI API 并管理實時的輸入輸出通訊。它發(fā)送工具信息和指令，監(jiān)聽響應，處理工具輸出，并流式傳輸對話內(nèi)容，使用 amerge 合并多個輸入輸出流。該方法還處理特定的響應類型并觸發(fā)必要的工具調(diào)用。
• audio-playback-worklet.js：實現(xiàn)了一個 AudioPlaybackWorklet，負責將接收到的 PCM 數(shù)據(jù)解碼并播放。它包含了handleMessage 方法，將傳入的音頻數(shù)據(jù)存入緩沖區(qū)；process 方法負責將緩沖區(qū)的數(shù)據(jù)輸出到揚聲器，按每次的緩沖量來處理數(shù)據(jù)。
• audio-processor-worklet.js：實現(xiàn)了 PCMAudioProcessor，將麥克風捕獲的 Float32 音頻數(shù)據(jù)轉(zhuǎn)換為 Int16 格式，然后通過 postMessage 發(fā)送到主線程，供后續(xù)處理。
• Index.html：通過WebSocket("ws://localhost:3000/ws")與服務器建立連接后，即可實現(xiàn)音頻的實時傳輸和處理。為此，我們創(chuàng)建了一個Player類來初始化音頻上下文，并利用AudioWorkletNode（引用audio-playback-worklet.js）播放服務器傳來的音頻數(shù)據(jù)。同時，設(shè)計Recorder類，用于獲取用戶麥克風輸入，通過audio-processor-worklet.js提供的方法處理音頻數(shù)據(jù)，將其分片編碼為base64格式，然后通過WebSocket發(fā)送到服務器。在接收到服務器返回的音頻流后，客戶端會對其進行解碼，并傳遞給播放器，從而實現(xiàn)音頻的播放功能。整個流程形成了一個閉環(huán)，確保了音頻從錄入到播放的順暢進行。

看完服務端代碼，再看看客戶端代碼，如下圖所示：

總結(jié)

Realtime API 改變了傳統(tǒng)語音交互模式，通過流式傳輸音頻輸入輸出，實現(xiàn)低延遲、高效率的語音對話。它支持文本和音頻兩種模態(tài)，并提供事件驅(qū)動機制，方便開發(fā)者構(gòu)建復雜的語音交互場景。通過 WebSocket 連接和事件交互，Realtime API 為開發(fā)者提供了強大的工具，推動語音交互應用的創(chuàng)新和發(fā)展。本文介紹了 Realtime API 的功能、原理和應用，并通過代碼示例展示了其使用方法，為開發(fā)者提供實用的參考和指導。

作者介紹

崔皓，51CTO社區(qū)編輯，資深架構(gòu)師，擁有18年的軟件開發(fā)和架構(gòu)經(jīng)驗，10年分布式架構(gòu)經(jīng)驗。