OpenAI Realtime API 的「說明書」。

很多研究 ChatGPT 的人，在使用后不久就會開始搗鼓 ChatGPT API。它是 OpenAI 提供的開放程序接口，讓開發者可以把業界最先進的大模型引入到自己的產品中，構建聊天機器人、虛擬助手等等。近一年來，依靠這套工具打造的熱門 App 已有不少。

Realme API 是 OpenAI 最新發布的 API 能力，它在今年 10 月 1 日推出，可幫助開發人員構建快速語音轉語音的智能化體驗。

近日，在 OpenAI DevDay 新加坡站，來自 Daily.co 的工程師受邀演講，介紹了基于 OpenAI 的能力構建語音 AI 智能體的方法。

一直以來，該公司的工程師們一直在使用實時 API 搭建產品。在活動之后，演講人發布了本篇博客，談了談構建 Pipecat 時的經驗教訓。

博客地址：https://www.latent.space/p/realtime-api

本篇博客的內容已通過 OpenAI 員工審核。Pipecat 是由 Daily 發起的開源項目，現在是一個實時 API 框架。

鏈接：https://pipecat.ai

對于想要構建大模型語音產品的人們來說，這篇內容可能包含你所需要的一切，從 24khz/G.711 音頻、RTMP、HLS、WebRTC 的技術細節到中斷 / VAD，再到成本、延遲、工具調用和上下文管理。

讓我們看看他們是怎么說的。

（以下為博客原文，機器之心做了不改變原意的編譯、整理。）

首先，我們愿意分享一些在使用原始實時 API（無框架、無外部依賴）時積累的心得，特別是在準備新加坡 DevDay 演講的過程中。標準的 OpenAI 參考應用包含了許多內置功能，我們盡量精簡這些功能，專注于語音活動檢測（VAD）和函數調用，從而打造了一個「簡易實時控制臺」演示。

實際上，語音活動檢測（VAD）并不總是完美無缺，而且演示語音應用時很少能在完全安靜的環境下進行。因此，我們建議在演示中始終配備「靜音」和「強制回復」按鈕，正如我們的演示所示。此外，該演示還展示了如何簡單添加和插入記憶，以及如何展示對話雙方的記錄。

從 Pipeline 到端到端模型

在我的大部分職業生涯中，我都在研究人與人之間對話的網絡基礎設施 —— 用于構建低延遲媒體流、視頻通話和大數據協作環境等的工具。2023 年 3 月，GPT-4 還是一個純文本模型。但為 GPT-4 開發語音模式相對容易。我整合了一個語音轉文本系統，將語音輸入轉換成文本提示，然后將 GPT-4 的文本輸出送入一個文本轉語音的音頻生成器中。

[語音輸入] ? [ ASR ] ? [ GPT4 ] ? [ TTS ] ? [語音輸出] —— 內容來自 DevDay Realtime API Talk：https://www.youtube.com/watch?v=mVR90WmA34U

這種多模型 pipeline 方法并不新鮮。我們撥打電話客戶支持熱線時使用的「自然語言處理」系統就是這樣工作的。新方法是 pipeline 核心的 GPT-4 大型語言模型。你可以用 GPT-4 進行真正的對話。

很明顯，那些較舊的 NLP 系統已經過時了。但顯然，新的挑戰也已出現。

這個系統的延遲很長。GPT-4 需要一秒鐘左右才能開始生成響應。STT 和 TTS 模型又增加了一兩秒鐘。

有時 GPT-4 會偏離軌道。弄清楚如何檢測這個問題，并把問題出現的幾率降到最低，似乎是一件相當困難的事情。

一些經典的 NLP 問題仍然存在，例如句尾端點（弄清楚 LLM 應該何時響應）和中斷處理。

GPT-4 擅長對話，但沒有很好的方法與現有的后端系統集成。

現有 TTS 模型的語音輸出質量，一聽就知道是機器人聲。

自 GPT-4 發布以來的 20 個月里，人工智能提升的速度令人驚嘆。當前版本的 GPT-4 擅長遵循提示、專注于任務并減少了幻覺。函數調用和結構化數據輸出是可靠的。模型響應迅速，我們擁有快速、價格合理且質量相當高的 TTS 模型。

最新的 GPT-4 功能是原生音頻輸入和輸出。GPT-4——升級版的 GPT-4o——現在有了自己的聲音和耳朵！

實時 API

10 月 1 日，OpenAI 發布了一款低延遲、多模態 API，該 API 利用了 GPT-4o 出色的「語音到語音」功能。這個新的「實時 API」能夠管理對話狀態、實現短語端點（輪流檢測）、提供雙向音頻流，并支持用戶中斷 LLM 的輸出。

使用此 API，最簡單的處理 pipeline 如下所示：

[ 語音輸入 ] ? [ GPT-4o ] ? [ 語音輸出 ] 

我一直在幫助客戶、朋友和與我一起在開源項目上工作的人們使用 OpenAI Realtime API。

使用這個新 API 與使用 OpenAI HTTP 推理 API 完全不同。新的 Realtime API 是有狀態的。它在 WebSocket 連接之上定義了一個雙向事件協議。

最近，我寫了關于 Realtime API 的優點、我認為還需要一篇博客來講如何有效使用 Realtime API 。無論你是工程師還是開發者，都希望能從這篇博客找到一些有用的背景信息、代碼片段，和可以節省你時間的具體細節。

在適當的地方，我會提供鏈接到 Pipecat 的源代碼和示例。Pipecat 是一個開源的、供應商中立的 Python 框架，專為實時、多模式 AI 代理和應用程序設計。它支持 GPT-4o 及其實時版本，以及超過 40 種其他 AI API 和服務，還涵蓋了多種網絡傳輸選項，包括 WebSockets、WebRTC、HTTP、SIP 以及 PSTN / 撥入 / 撥出等。

Pipecat 還附帶一個大型核心功能庫，用于上下文管理、內容審核、用戶狀態管理、事件處理、腳本跟蹤以及語音（和視頻）代理的其他重要構建塊。

• 完整的 Pipecat OpenAI Realtime API 集成在這里：https://github.com/pipecat-ai/pipecat/tree/main/src/pipecat/services/openai_realtime_beta
• 這是一個使用 Realtime API 的單文件示例語音機器人：https://github.com/pipecat-ai/pipecat/blob/main/examples/foundational/19-openai-realtime-beta.py

OpenAI 實時 API 的架構

對話語音是 OpenAI 實時 API 支持的核心用例。對話語音 API 需要：

• 管理多個用戶和 LLM 輪次的對話狀態；
• 確定用戶何時結束對話（并期待 LLM 的響應）；
• 處理用戶中斷 LLM 輸出；
• 用戶語音的文本轉錄、函數調用和 LLM 上下文的操作對于許多用例也很重要。

OpenAI 的實時 API 通過定義一系列通過 WebSocket 連接發送和接收的事件來實現這些功能。該 API 包括 9 種客戶端事件（從客戶端發送到服務器的事件）和 28 種服務器事件（從服務器發送到客戶端的事件）。以下是所有 37 個事件的 Pydantic 事件定義：

https://github.com/pipecat-ai/pipecat/blob/main/src/pipecat/services/openai_realtime_beta/events.py

這個事件結構有很多優點。Python 中包括所有導入和 asyncio 樣板的最小命令行客戶端大約只有 75 行代碼。

https://x.com/kwindla/status/1843118281911308355

音頻以 base64 編碼塊的形式發送和接收，嵌入在 input_audio_buffer.append 和 audio.delta 事件中。

API 當前支持未壓縮的 16 位、24khz 音頻和壓縮的 G.711 音頻。

• G.711 僅用于電話用例；與其他更現代的編解碼器選項相比，音頻質量相對較差。
• 未壓縮的 16 位、24khz 音頻的比特率為 384 千比特 / 秒。base64 編碼開銷將標稱比特率推高至約 500 kbs。但 permessage-deflate 標準壓縮將使比特率降至 300-400 kbs。
• 如果你關心的是實現實時延遲，那么 300kbs 的媒體流比人們通常希望通過 WebSocket 連接發送的媒體流要大。我們在后文也會談談延遲和 WebSockets。

延遲

人類希望在正常對話中得到快速響應，對話的響應時間為 500 毫秒是正常的，AI 長時間的停頓會讓人感覺不自然。

所以如果你正在構建對話式 AI 應用程序，語音到語音的延遲大概是 800 毫秒。盡管當今的 LLM 很難始終如一地實現這一點。

OpenAI Realtime API 提供了非常好的推理延遲效果。對于位于美國的客戶，API 的第一個字節時間約為 500 毫秒。如果我們的目標是總語音到語音延遲為 800 毫秒，那么音頻處理和短語端點大約需要 300 毫秒。在理想條件下，這幾乎是一項不可能完成的任務。

測量語音到語音延遲的過程十分簡單。你只需錄制對話，將錄音導入音頻編輯軟件，觀察音頻波形，并測量從用戶語音結束到 LLM 語音輸出開始之間的時間。如果你正在開發打算實際投產的對話式語音應用，定期監控延遲數據是非常重要的。在這些測試中加入模擬網絡數據包丟失和抖動的功能，將會加分。

以編程方式測量真正的語音到語音延遲具有挑戰性，因為部分延遲發生在操作系統內部深處。因此，大多數可觀察性工具僅測量推理到第一個字節的時間。這是總語音到語音延遲的合理代理，但請注意，此測量不包括短語端點時間（請參閱下一節）。

許多「小」事情可能會引發延遲的「蝴蝶效應」。例如，藍牙音頻設備可能會增加幾百毫秒的延遲。有關導致在 Web 瀏覽器中運行的語音到語音應用程序中延遲的所有因素的更多詳細信息，請參閱此推文、博客以及 AI.Engineer 的演講：

• 推文鏈接：https://x.com/kwindla/status/1806129490411900940
• 博客鏈接：https://www.daily.co/blog/the-worlds-fastest-voice-bot/

句尾檢測與打斷

在對話中，人們會輪流說話。在語音 AI 應用中，對話雙方絲滑地切換回合需要做到兩部分：

• 應用判斷人類是否已經說完，開始期待系統的響應。這被稱為短語終點檢測或回合檢測（turn detection）。大多數應用都會嘗試自動檢測一個回合結束了，但有些應用會在界面上設置按鈕，用戶按住按鈕才能說話，松開按鈕表示說話結束。
• 如果用戶在 LLM 說話時打斷了它，一般來說，系統應該立即停止播放 LLM 的語音輸出。然而，對于一些應用場景，這種行為需要是可配置的：有時候，即使用戶打斷了 LLM，也希望 LLM 能把話說完。

OpenAI 實時 API 內置了自動句尾檢測和處理打斷的功能。這些功能由 VAD（Voice Activity Detection）實現。自動輪次檢測默認是開啟的，但可以隨時關閉。

以下是 OpenAI 關于自動輪次檢測的文檔鏈接：

https://platform.openai.com/docs/guides/realtime/realtime-api-beta#input-audio-buffer

有多種 VAD 參數可配置，其中最重要的是 silence_duration_ms，即用戶停止說話后，VAD 等待的時間長度（以毫秒為單位）。

在這段時間后，會觸發 input_audio_buffer.speech_stopped 事件并開始推理。

OpenAI 在服務器端維護了一個音頻緩沖區，應用程序可以通過發送 input_audio_buffer.append 事件持續地添加音頻幀。在自動輪次檢測模式下，應用程序只需持續發送音頻數據，依靠 OpenAI 服務器端的 VAD 來識別用戶何時開始和停止說話。

當用戶停止說話時，會觸發多個 API 事件，LLM 隨即開始生成響應。若用戶再次開始說話，任何正在進行的響應將被中斷，音頻輸出也會被清除。

這種方法非常簡單高效（無需編寫任何客戶端代碼），并且效果顯著。然而，在以下三種情況下，應用可能會選擇關閉 OpenAI 的自動輪次檢測功能：

• 不希望允許應用被打斷時
• 像微信一樣「按鍵說話」樣式的用戶界面
• 開發者使用其他句尾檢測方法

禁用 OpenAI 的自動輪次檢測功能后，客戶端需在用戶語音回合結束時發送兩個 Realtime API 事件：input_audio_buffer.commit 和 response.create。

以下是在用戶開始說話時，如何調用這兩個事件的 Pipecat 代碼示例：

https://github.com/pipecat-ai/pipecat/blob/main/src/pipecat/services/openai_realtime_beta/openai.py#L145

OpenAI VAD 似乎比 Pipecat 中的默認短語端點實現對背景噪音更敏感。Pipecat 使用輸入音頻能量的平滑運行平均值來相對于背景噪音自動調平。它還會忽略音頻中的短尖峰，即使它們具有相當高的語音置信度，并支持高級音頻處理。

OpenAI 的 silence_duration_ms 參數默認為 500ms，Pipecat 將這個參數稱為 stop_secs。

這是一個很好的折衷方案，介于 LLM 響應時間過長和 LLM 響應過快之間。對于某些用例，最好選擇較長的靜音持續時間。例如，在工作面試環境中，讓人們在談話時有更多時間思考他們的答案通常會提供更好的體驗。在這種情況下，800ms 甚至 1s 都是理想的。

使用標準 VAD 時，除了語音 AI 演示之外，我們通常不建議將設置設置為低于 500 毫秒！有一些技術可以通過使用上下文感知短語端點來補充標準 VAD，或進行推測（貪婪）推理，或兩者兼而有之，從而獲得更快的響應時間。這些技術超出了本文的范圍，但如果您對它們感興趣，Pipecat Discord 是一個閑逛的好地方。

Pipecat 的基本 VAD 實現在這里：

https://github.com/pipecat-ai/pipecat/blob/1d4be0139aeff2ee8cc214c81ae00e5948e35977/src/pipecat/audio/vad/vad_analyzer.py#L86

管理上下文

多輪對話是一系列用戶輸入和 LLM 響應。

LLM 本身是無狀態的，因此每次有用戶輸入時，都需要將所有相關對話歷史記錄發送到 LLM。如果你之前構建過對話式 LLM 應用程序（文本或語音），你會熟悉跟蹤對話歷史記錄并使用該歷史記錄創建不斷增加的「上下文」。

OpenAI Realtime API 為你進行對話管理，有兩個巨大的好處：更簡單的代碼和更低的延遲。

代碼更簡單，因為你不必跟蹤應用程序中的對話歷史記錄。

延遲較低有幾個原因。每次你希望 LLM 生成響應時，不必重新發送大型上下文。這節省了一些網絡開銷。此外，當前的音頻輸入可以流式傳輸到 OpenAI 服務器，以便在請求推理時立即使用。最后，OpenAI 可以實現上下文緩存等內部優化。這一切帶來一次巨大的勝利！

有兩個限制需要注意：最大上下文長度為 128000 tokens，單個對話的最大時間為 15 分鐘。

在音頻對話中，你不太可能遇到 token 限制。音頻每分鐘使用大約 800 tokens。

然而，15 分鐘時長對于某些應用程序可能是一個限制。

目前無法通過 OpenAI Realtime API 檢索對話上下文、將「助手」音頻消息加載到上下文中或可靠地加載多消息歷史記錄。

參閱此存儲庫以獲取測試用例：https://github.com/kwindla/openai-realtime-test-cases

不過，實現持久對話和長時間對話是可能的。你需要將對話歷史記錄保存為文本。然后，在重新啟動對話時，發送完整的對話歷史記錄（和適當的提示）作為新對話中的第一條消息。

以下是 Pipecat 代碼，它使用 OpenAI HTTP API 支持的相同消息列表格式來初始化對話：

https://github.com/pipecat-ai/pipecat/blob/main/src/pipecat/services/openai_realtime_beta/context.py#L76

關于上下文管理的其他一些事情也值得討論。

LLM 的音頻生成速度比語音輸出速度更快。OpenAI 將服務器端 LLM 響應添加到對話上下文中，速度與生成的速度一樣快。但講話的播放速度較慢。如果用戶中斷 LLM，則用戶將只能聽到 LLM 響應的一部分。在大多數情況下，您希望對話歷史記錄僅包含用戶實際聽到的 LLM 響應部分。

您需要發送對話.item.truncate 事件以強制服務器端上下文匹配用戶聽到的音頻范圍。請注意，無論您是否使用自動轉彎檢測 (server_vad)，您都需要執行此操作。

以下是 Pipecat 代碼，用于計算用戶聽到的音頻的持續時間并調用對話.item.truncate：

https://github.com/pipecat-ai/pipecat/blob/main/src/pipecat/services/openai_realtime_beta/context.py#L49

對于許多用例來說，用戶輸入和 LLM 輸出的轉錄都很重要。保存對話以便用戶稍后可以返回就是一個例子。許多企業用例需要對話記錄來滿足內容審核、后處理需求或合規性原因。

OpenAI Realtime API 始終提供 LLM 輸出的轉錄。輸入轉錄默認關閉，但可以通過在配置會話時設置 input_audio_transcription 字段來啟用。

輸出轉錄由 LLM 本地生成，與音頻輸出緊密匹配。輸入轉錄由單獨的模型生成，并不總是與模型 “聽到” 的內容匹配。對于某些用例來說，這可能是一個問題。如果轉錄數據包含語言字段，這也會很有用。（許多語音人工智能用例都是多語言的。）

目前還沒有辦法將輸出轉錄與語音定時對齊。這使得當用戶中斷時很難截斷文本輸出，并且很難構建諸如單詞精確的流文本字幕之類的東西。

輸入音頻轉錄也可能落后于模型輸出幾秒鐘。如果您需要使用轉錄進行內容審核，您可能需要使用您自己的轉錄模型和門短語終結于轉錄完成或內容審核檢查本身之后。

最后，請注意 Realtime API 消息的內容格式與 OpenAI HTTP API 的格式不同。以下是從 HTTP API 格式轉換為 Realtime API 格式的 Pipecat 代碼：

函數調用

函數調用在 OpenAI Realtime API 中運行得非常好（所有 GPT-4 系列模型都是如此）。

• 與對話消息的格式一樣，工具格式與 OpenAI HTTP API 略有不同。
• 此 HTTP API 工具列表（只有一個條目）：

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_current_weather",
            "description": "Get the current weather in a given location",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "The city and state, e.g. San Francisco, CA",
                    },
                },
                "required": ["location"],
            }
        }
    }
]

成為實時 API 中的工具列表：

tools = [
    {
        "type": "function",
        "name": "get_current_weather",
        "description": "Get the current weather",
        "parameters": {
            "type": "object",
            "properties": {
                "location": {
                    "type": "string",
                    "description": "The city and state, e.g. San Francisco, CA",
                },
                "format": {
                    "type": "string",
                    "enum": ["celsius", "fahrenheit"],
                    "description": "The temperature unit to use. Infer this from the users location.",
                },
            },
            "required": ["location", "format"],
        },
    }
]

可以通過兩種方式從 API 獲取函數調用事件：

• 通過流事件 response.function_call_arguments.delta 和 function_call_arguments.done
• 作為 response.done 事件的一部分

如果您從 HTTP API 進行移植并希望保留盡可能多的現有代碼結構，則流事件可能會很有用。但令人高興的是，Realtime API 使得從 response.done 事件中提取函數調用結構變得非常簡單。流對于函數調用來說并不是很有用 —— 在調用函數之前，您需要完整的函數調用結構 —— 并且在使用 HTTP API 時，從流式響應塊中組裝函數調用數據一直是一個小麻煩。

下面是從 response.done 事件的輸出字段中提取函數調用信息的 Pipecat 代碼：https://github.com/pipecat-ai/pipecat/blob/main/src/pipecat/services/openai_realtime_beta/openai.py#L469

成本

對話式 AI 的使用場景，成本通常會隨著會話長度呈指數增長。

大多數對話式 AI 應用和 API 都會在每次推理請求中使用完整的會話歷史，OpenAI 實時 API 也不例外。

不過，OpenAI 實時 API 能夠自動緩存并重復利用已發送的輸入 tokens。緩存的音頻 tokens 成本比非緩存的低 80%，在長對話中這可以大幅降低成本。

一般的對話成本：

-  1 minute  — $0.11
  -  2 minutes - $0.26
  -  5 minutes - $0.92 
  - 10 minutes - $2.68
  - 15 minutes - $5.28

OpenAI 不會為接近靜音的音頻輸入生成 tokens。成本估算基于假設對話中 70% 的時間是用戶實際說話的時間。如果實際說話時間比例更低，成本會相應降低。

預計 OpenAI 會持續降低實時 API 的成本。但如果你的預算緊張，可以考慮每隔幾輪對話重置上下文，用文本替換音頻消息，也可以使用摘要功能進一步減少輸入 tokens 的數量。

以下是一個成本計算器表格，你可以調整其中的參數算一算輸入 token 的成本：https://docs.google.com/spreadsheets/d/1EL-mjqlmj4ehug8BjmgmAFm9uFZtZXY9N9EvqLm8Ebc/edit?usp=sharing

WebSockets 還是 WebRTC？

OpenAI 實時 API 使用 WebSockets 進行網絡傳輸。

WebSockets 非常適合用于服務器之間的通信，尤其是在對延遲要求不高的場景中，以及在原型開發和一般性開發測試中。

如果你正在開發一個基于 OpenAI 實時 API 的瀏覽器或原生移動應用，并且對會話延遲有嚴格要求，建議使用 WebRTC 連接。

具體來說，用 WebRTC 將音頻從你的應用發送到服務器，接收音頻，然后在服務器端直接調用 OpenAI 實時 API。

Pipecat 支持 WebRTC、WebSockets 和 HTTP 傳輸，可以輕松構建「客戶端」——「服務器」—— [推理服務器] 的架構。

但對于生產環境中的客戶端 —— 服務器實時媒體連接，不推薦使用 WebSockets。理由如下：

WebSockets 基于 TCP 協議，因此音頻流會遇到「首阻塞問題」。如果某個數據包延遲，后續的數據包也會被阻塞。此外，TCP 協議會嘗試重新發送丟失的數據包，但在實時場景中，如果這些數據包到達時已經過期，就毫無意義，還可能進一步增加延遲。

用 WebRTC 就可以解決這些問題。首先，WebRTC 使用的 Opus 音頻編解碼器與 WebRTC 的帶寬估算和數據包調度（擁塞控制）邏輯緊密結合。這使得 WebRTC 音頻流能夠很好地適應各種現實網絡環境。

Opus 音頻編解碼器具有非常優秀的前向糾錯能力，能有效處理音頻流中的丟包問題，保持音頻穩定，（不過，這需要網絡能及時丟棄延遲的數據包且沒有首阻塞問題才行）。

WebRTC 發送和接收的音頻會自動添加時間戳，因此播放和中斷邏輯的實現都變得非常簡單。而在 WebSockets 中，處理起來則要困難得多。

此外， WebRTC 自帶出色的回聲消除、噪聲消減和自動增益控制功能。如果使用 WebSockets，則需要自己想辦法將這些音頻處理功能集成到應用中。

最后，在長距離網絡傳輸中，延遲和不穩定性是不可避免的。為了減少這些問題，可以通過「靠近用戶的中轉節點」（路由器）優化連接，從而提升實時媒體的性能，而 WebRTC 平臺會自動幫你完成這部分優化。

• 如果你對媒體傳輸網絡協議的設計感興趣，可以參考這篇關于 RTMP、HLS 和 WebRTC 的技術概覽：
  https://www.daily.co/blog/video-live-streaming/
• 如果想深入了解 WebRTC 的路由機制，可以參考 Daily 詳解其全球 WebRTC 基礎設施的文章：
  https://www.daily.co/blog/global-mesh-network/

回聲消除和音頻處理

幾乎所有支持對話的語音應用都需要處理音頻的功能，特別是回聲消除。

Media Capture and Streams API 為 Chrome、Safari 和 Edge 瀏覽器提供了相對成熟可靠的回聲消除功能，開發者可以放心使用。

我們強烈建議不要將 Firefox 作為開發和測試的瀏覽器。Firefox 的回聲消除和音頻流管理功能較差且不穩定。你可能得花費大量時間修復 Firefox 的特有 bug，而這些問題在其他瀏覽器中完全不存在。

因此，建議優先針對 Chrome 和 Safari 開發功能，后續再考慮是否要適配 Firefox。

瀏覽器的 WebRTC 和原生 SDK 通常會默認開啟高質量的回聲消除、消減底噪和自動增益控制功能。

需要注意的是，回聲消除必須在客戶端設備上完成，而其他類型的音頻處理可以在服務器端實現。

例如，Pipecat 集成了 Krisp 的降噪和分離說話人模型，可以在服務器端處理音頻。

API 設計的題外話

每個 API 都是工程的產物，要權衡軟件設計和開發中的各種限制。

優秀的 API 力求需求明確，找到功能細分的「黃金點」，可以讓簡單的事情更簡單，也能讓復雜的事情成為可能。

把 OpenAI 實時 API 集成到 Pipecat 中是一件很有趣的事情。這兩種支持對話式語音 AI 的方法有很多相同之處。

OpenAI 的事件（event）可以直接映射成 Pipecat 的幀類型。OpenAI 實時 API 解決的問題類型或關注的領域，與 Pipecat 用戶已經習慣的問題非常相似，對 OpenAI 的設計會很熟悉，不需要花太多時間去適應。

但它們在設計的底層架構上有很大的差異，這種差異可能是因為兩者的定位和使用場景不同。

OpenAI 實時 API 的事件架構可以輕松集成到任何編程語言或框架中。發送事件時，只需傳輸一些 JSON（或類似格式）數據；接收事件時，通過讀取循環將數據分發到相應函數即可。

相比之下，Pipecat 是一種數據流架構，受多年來多媒體數據處理框架（如 GStreamer）的影響較大，在設計上強調模塊化和流水線化。

在 OpenAI 實時 API 中，核心構建塊是「事件（event）」；在 Pipecat 中，核心構建塊是「幀處理器（frame processor）」。

一個 Pipecat 中的語音到語音循環可能看起來是這樣的：

pipeline = Pipeline(
    [
        transport.input(), 
        context_aggregator.user(),
        openai_realtime_llm, 
        context_aggregator.assistant(),
        transport.output()
    ]
)

在此基礎上添加語音轉文本和文本轉語音的處理器，這種架構可以兼容任何大型語言模型（LLM）。

pipeline = Pipeline(
    [
        transport.input(), 
        context_aggregator.user(),
        stt,
        llm,
        tts, 
        context_aggregator.assistant(),
        transport.output()
    ]
)

以下是一個更復雜的 Pipecat 流程，整合了多種功能模塊，能夠高效地處理客戶端命令和事件。通過 Webhook 實現函數調用， 內置了處理計費、監控流程，錯誤管理功能。

pipeline = Pipeline(
    [
        el,
        transport.input(),
        rtvi,
        user_aggregator,
        openai_realtime_llm,
        rtvi_speaking,
        rtvi_user_transcription,
        rtvi_bot_llm,
        rtvi_bot_transcription,
        webhooks_processor,
        ml,
        rtvi_metrics,
        transport.output(),
        rtvi_bot_tts,
        assistant_aggregator,
    ]
)

實時語音 API 需要判斷你的話在哪里結束，好接上對話。以下一個是減少延遲的實驗性方案。

其中，語音活動檢測（VAD）負責聽聲音有沒有停下來，LLM 來判斷剛才說的是不是完整的一句話，是不是有話沒說完。這兩個判斷將放在并行的子流程中同時運行。

pipeline = Pipeline(
    [
        transport.input(),
        stt,
        context_aggregator.user(),
        ParallelPipeline(
            [
                # Pass everything except UserStoppedSpeaking to the elements after
                # this ParallelPipeline
                FunctionFilter(filter=block_user_stopped_speaking),
            ],
            [
                # Ignore everything except an OpenAILLMContextFrame. Pass a specially constructed
                # LLMMessagesFrame to the statement classifier LLM. The only frame this
                # sub-pipeline will output is a UserStoppedSpeakingFrame.
                statement_judge_context_filter,
                statement_llm,
                completeness_check,
            ],
            [
                # Block everything except OpenAILLMContextFrame and LLMMessagesFrame
                FunctionFilter(filter=pass_only_llm_trigger_frames),
                llm,
                bot_output_gate,  # Buffer all llm output until notified.
            ],
        ),
        tts,
        user_idle,
        transport.output(),
        context_aggregator.assistant(),
    ]
)

更多資源

OpenAI 的實時 API 文檔：

• https://platform.openai.com/docs/guides/realtime/realtime-api-beta#quickstart

API 參考文檔：

• https://platform.openai.com/docs/api-reference/realtime

優秀的實時 API 控制臺應用：

• https://github.com/openai/openai-realtime-console
• @swyx 的簡化版：https://github.com/swyxio/simple-realtime-console

Pipecat 代碼，可以通過 Pipecat 來調用 OpenAI 的實時功能，尤其是其中的 Pydantic 模版可能對其他項目特別有用：

• https://github.com/pipecat-ai/pipecat/tree/main/src/pipecat/services/openai_realtime_beta