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

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

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

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

開篇

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

為什么需要 Realtime API？

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

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

嘗鮮 Realtime API？

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

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

1.會(huì)話區(qū)域

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

2.語音設(shè)置

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

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

3.功能與模型配置

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

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

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

當(dāng)出現(xiàn)下圖“Start talking”的時(shí)候就可以與LLM進(jìn)行對(duì)話了。

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

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

不過從使用價(jià)格方面來看，是有一點(diǎn)小貴，如下：

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

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

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

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

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

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

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

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

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

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

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

Realtime API的特點(diǎn)

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

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

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

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

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

我們通過一段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 建立一個(gè) WebSocket 連接，并指定使用模型 gpt-4o-realtime-preview-2024-10-01。在連接成功后，發(fā)送一個(gè)請(qǐng)求，讓模型使用 "text" 模態(tài)，并為用戶提供幫助。隨后，當(dāng)服務(wù)器發(fā)送消息時(shí)，我們會(huì)通過 "message" 事件接收到響應(yīng)，并將其打印出來。

import WebSocket from "ws";

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

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

url 是 WebSocket 的服務(wù)器地址，使用的是 OpenAI Realtime API 的 WebSocket 連接地址。wss:// 表示這是一個(gè)加密的 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)建一個(gè) WebSocket 連接。Authorization 是API 的認(rèn)證信息，用于驗(yàn)證開發(fā)者的身份。process.env.OPENAI_API_KEY 是從環(huán)境變量中讀取的 OpenAI API 密鑰，確保安全地訪問 API。OpenAI-Beta 是額外的請(qǐng)求頭，指定使用的是 Realtime API 的測試版本。"realtime=v1" 表示這是 Realtime API 的第一個(gè)版本。

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", ...)：這是一個(gè)事件監(jiān)聽器，當(dāng) WebSocket 連接成功打開時(shí)，會(huì)觸發(fā) "open" 事件。我們使用匿名函數(shù)來處理這個(gè)事件。

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

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

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

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

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

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

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

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

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

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

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

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

音頻交互

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

1.引入模塊

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

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

audio-decode 庫用來將音頻文件解碼為原始的音頻字節(jié)（`AudioBuffer` 對(duì)象），便于后續(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（浮點(diǎn)數(shù)組格式的音頻數(shù)據(jù)）轉(zhuǎn)換為 PCM16 格式的 ArrayBuffer。PCM16 是常見的音頻編碼格式，將浮點(diǎn)值（-1 到 1 之間）轉(zhuǎn)換為 16 位整數(shù)。DataView對(duì)象允許你對(duì)二進(jìn)制 ArrayBuffer 進(jìn)行逐字節(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)絡(luò)傳輸音頻數(shù)據(jù)，使用 `btoa` 將該二進(jìn)制字符串編碼為 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è)處理的是單聲道音頻（如果是多聲道，需要處理多個(gè)通道）。

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 向服務(wù)器發(fā)送這條消息。隨后，通過發(fā)送 response.create 請(qǐng)求，要求服務(wù)器生成對(duì)應(yīng)的響應(yīng)。

事件類型定義

前面幾段代碼我們發(fā)現(xiàn)無論是Client還是Server 在發(fā)送和接受event的時(shí)候都會(huì)定義event類型，比較常見的有conversation.item.create 創(chuàng)建對(duì)話，respnse.create響應(yīng)。如下圖所示，實(shí)際上在Realtime API中定義了9個(gè)Client 事件類型和28個(gè)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 是一個(gè)事件類型，主要用于在對(duì)話中添加一個(gè)新項(xiàng)目（例如消息或函數(shù)調(diào)用）。此事件包含以下關(guān)鍵屬性：

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

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

Realtime AI 最佳實(shí)踐

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

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

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

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

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

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

2.VoiceToolExecutor (類)  

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

包含屬性：

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

函數(shù)：

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

3.OpenAIVoiceReactAgent (類)  

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

屬性：

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

函數(shù)：

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

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

總結(jié)

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

作者介紹

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