作者 | 崔皓

審校 | 重樓

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

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

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

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

開篇

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

為什么需要 Realtime API？

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

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

嘗鮮 Realtime API？

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

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

1.會話區域

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

2.語音設置

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

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

3.功能與模型配置

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

目前該界面已經給Plus 用戶開放使用權限，開發者可以配置和測試 Realtime API，了解其響應效果并調整設置。

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

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

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

通過官網宣傳可知，該 API 的音頻功能由全新的 GPT-4o 模型提供支持，具體版本為 gpt-4o-realtime-preview。而在幾周內，開發者還將迎來 gpt-4o-audio-preview，該版本支持通過文本或音頻輸入與 GPT-4o 模型交互，并生成文本或音頻輸出，或兩者兼備的響應形式。

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

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

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

為什么不直接調用API ？

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

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

Client 端的角色：Client（客戶端）是負責發起操作和接收用戶輸入的地方。在這個架構中，客戶端可以是與用戶交互的界面，比如一個聊天應用、網頁、移動端應用等。客戶端需要將用戶的請求傳遞給服務端，并且根據返回的結果展示給用戶。

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

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

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

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

多客戶端協作：在一些實時系統中，比如聊天系統、監控系統，多個客戶端可能同時與服務器交互，事件驅動模型允許多客戶端之間實時共享和同步數據，而不需要每個請求都單獨進行阻塞調用。

Realtime API的特點

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

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

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

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

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

我們通過一段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。在連接成功后，發送一個請求，讓模型使用 "text" 模態，并為用戶提供幫助。隨后，當服務器發送消息時，我們會通過 "message" 事件接收到響應，并將其打印出來。

import WebSocket from "ws";

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

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 構造函數來創建一個 WebSocket 連接。Authorization 是API 的認證信息，用于驗證開發者的身份。process.env.OPENAI_API_KEY 是從環境變量中讀取的 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", ...)：這是一個事件監聽器，當 WebSocket 連接成功打開時，會觸發 "open" 事件。我們使用匿名函數來處理這個事件。

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

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

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

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

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

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

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

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'}));

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

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

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

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

在這段代碼中，通過 WebSocket 向服務器發送一個 conversation.item.create 事件，其中 item 是一條用戶發送的文本消息。這一事件告知服務器用戶輸入了 "Hello!"，并請求模型生成適當的響應。

音頻交互

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

1.引入模塊

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

fs模塊用于讀取本地文件系統中的音頻文件。

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

2.floatTo16BitPCM 函數

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;
}

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

3.base64EncodeAudio 函數

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);
}

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

4.解碼音頻文件并獲取通道數據

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

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

5.發送音頻數據

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'}));

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

事件類型定義

前面幾段代碼我們發現無論是Client還是Server 在發送和接受event的時候都會定義event類型，比較常見的有conversation.item.create 創建對話，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 是一個事件類型，主要用于在對話中添加一個新項目（例如消息或函數調用）。此事件包含以下關鍵屬性：

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：項目狀態（"completed"、"in_progress"、"incomplete"）。
• role：消息發送者的角色（"user"、"assistant"、"system"）。
• content：消息內容。
• call_id、name、arguments、output：用于函數調用相關的詳細信息。

Realtime AI 最佳實踐

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

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

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

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

1.connect()(函數)  

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

2.VoiceToolExecutor (類)  

此類通過函數調用異步執行工具，當與用戶交互時，如果需要用到外部工具就會啟用該類。

包含屬性：

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

函數：

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

3.OpenAIVoiceReactAgent (類)  

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

屬性：

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

函數：

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

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

總結

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

作者介紹

崔皓，51CTO社區編輯，資深架構師，擁有18年的軟件開發和架構經驗，10年分布式架構經驗。