摘要 — 实时 API

1. 概述

构建一个端到端的语音机器人，它可以监听你的麦克风，实时进行语音反馈，并总结长对话，从而保证质量永不下降。

你将学到

1. 实时麦克风流 → OpenAI 实时（语音到语音）端点。
2. 即时字幕和语音回放，在每一次交互中。
3. 对话状态容器，存储每一条用户/助手消息。
4. 自动“上下文修剪” – 当 token 窗口变得非常大时（可配置），旧的对话轮次会被压缩成一个摘要。
5. 可扩展的设计，你可以适应以支持客服机器人、自助服务终端或多语言助手。

先决条件

| 要求 | 详情 |

需要帮助设置密钥？

遵循官方快速入门指南。

注意：

1. GPT-4o-Realtime 支持 128k token 的上下文窗口，但在某些用例中，随着你将更多 token 塞入上下文窗口，你可能会注意到性能下降。
2. Token 窗口 = 模型当前在会话中保留的所有 token（单词和音频 token）。

一行安装（在新的单元格中运行）

# 运行一次以安装或升级依赖项（如果已安装，请注释掉）
# !pip install --upgrade openai websockets sounddevice simpleaudio

# 标准库导入
import os
import sys
import io
import json
import base64
import pathlib
import wave
from dataclasses import dataclass, field
from typing import List, Literal

# 第三方库导入
import asyncio
import numpy as np
import sounddevice as sd         # 麦克风捕获
import simpleaudio               # 扬声器播放
import websockets                # WebSocket 客户端
import openai                    # OpenAI Python SDK >= 1.14.0

# 安全地设置你的 API 密钥
openai.api_key = os.getenv("OPENAI_API_KEY", "")
if not openai.api_key:
    raise ValueError("未找到 OPENAI_API_KEY – 请设置环境变量或编辑此单元格。")

2. Token 使用 — 文本 vs 语音

大 token 窗口非常宝贵，你使用的每个额外 token 都会增加延迟和成本。 对于音频，输入 token 窗口的增加速度远快于纯文本，因为必须表示幅度、时序和其他声学细节。

实际上，对于相同的句子，音频的 token 数通常是文本的 ≈ 10 倍。

• GPT-4o 实时支持高达 128k token，并且随着 token 数量的增加，指令遵循性可能会发生漂移。
• 每个用户/助手轮次都会消耗 token → 窗口只会增长。
• 策略：将旧的轮次总结成一条助手消息，保留最后几条原始轮次，然后继续。

3. 辅助函数

以下辅助函数将使我们能够运行完整的脚本。

3.1 对话状态

与基于 HTTP 的聊天补全不同，实时 API 维护一个开放的、有状态的会话，包含两个关键组件：

| 组件 | 目的 |

此笔记本将这些组件封装在一个简单的 ConversationState 对象中，以保持你的逻辑清晰，跟踪历史记录，并在上下文窗口填满时管理摘要。

@dataclass
class Turn:
    """对话中的一次发声（用户**或**助手）。"""
    role: Literal["user", "assistant"]
    item_id: str                    # 服务器分配的标识符
    text: str | None = None         # 转录就绪后填充

@dataclass
class ConversationState:
    """会话所需的所有可变数据 — 仅此而已。"""
    history: List[Turn] = field(default_factory=list)         # 有序日志
    waiting: dict[str, asyncio.Future] = field(default_factory=dict)  # 待处理的转录获取
    summary_count: int = 0

    latest_tokens: int = 0          # 上一次回复后的窗口大小
    summarising: bool = False       # 保护，防止同时运行两次摘要

一个快速查看转录的辅助函数：

def print_history(state) -> None:
    """美观地打印到目前为止的对话记录。"""
    print("—— 对话记录 ———————————————————————————————")
    for turn in state.history:
        text_preview = (turn.text or "").strip().replace("\n", " ")
        print(f"[{turn.role:<9}] {text_preview}  ({turn.item_id})")
    print("——————————————————————————————————————————")

3.2 · 流式音频

我们将原始 PCM-16 麦克风数据直接流式传输到实时 API。

管道是：麦克风 ─► 异步队列 ─► WebSocket ─► 实时 API

3.2.1 捕获麦克风输入

我们将从一个协程开始，该协程：

• 以 24 kHz、单声道、PCM-16（实时 API 接受的格式之一）打开默认麦克风。
• 将流切片成约 40 ms 的块。
• 将每个块转储到一个 asyncio.Queue 中，以便另一个任务（下一节）将其转发给 OpenAI。

async def mic_to_queue(pcm_queue: asyncio.Queue[bytes]) -> None:
    """
    捕获原始 PCM-16 麦克风音频，并将约 CHUNK_DURATION_MS 的块推送到 *pcm_queue*，
    直到周围的任务被取消。

    参数
    ----------
    pcm_queue : asyncio.Queue[bytes]
        PCM-16 帧（小端序 int16）的目标队列。
    """
    blocksize = int(SAMPLE_RATE_HZ * CHUNK_DURATION_MS / 1000)

    def _callback(indata, _frames, _time, status):
        if status:                               # XRuns、设备更改等。
            print("⚠️", status, file=sys.stderr)
        try:
            pcm_queue.put_nowait(bytes(indata))  # 1 次性入队
        except asyncio.QueueFull:
            # 如果上游（WebSocket）跟不上，则丢弃帧。
            pass

    # RawInputStream 是同步的；使用上下文管理器进行包装以自动关闭。
    with sd.RawInputStream(
        samplerate=SAMPLE_RATE_HZ,
        blocksize=blocksize,
        dtype="int16",
        channels=1,
        callback=_callback,
    ):
        try:
            # 保持协程活动，直到被调用者取消。
            await asyncio.Event().wait()
        finally:
            print("⏹️  麦克风流已关闭。")

3.2.2 将音频块发送到 API

我们的麦克风任务现在正在用原始 PCM-16 块填充 asyncio.Queue。 下一步：从该队列中提取块，对它们进行base64 编码（协议要求 JSON 安全文本），并将每个块作为 input_audio_buffer.append 事件发送到实时 WebSocket。

# 用于对音频块进行 base64 编码的辅助函数
b64 = lambda blob: base64.b64encode(blob).decode()

async def queue_to_websocket(pcm_queue: asyncio.Queue[bytes], ws):
    """从队列读取音频块并将其作为 JSON 事件发送。"""
    try:
        while (chunk := await pcm_queue.get()) is not None:
            await ws.send(json.dumps({
                "type": "input_audio_buffer.append",
                "audio": b64(chunk),
            }))
    except websockets.ConnectionClosed:
        print("WebSocket 已关闭 – 正在停止上传器")

3.2.3 处理传入事件

一旦音频到达服务器，实时 API 就会通过相同的 WebSocket 推送一系列 JSON 事件。 理解这些事件对于以下几点至关重要：

• 打印实时字幕
• 将增量音频播放给用户
• 维护准确的 对话状态，以便稍后进行上下文修剪

| 事件类型 | 到达时间 | 重要性 | 典型处理逻辑 |

3.3 检测何时进行摘要

实时模型维护一个大型 128k token 窗口，但随着上下文的增加，质量可能在达到限制之前就已下降。

我们的目标：一旦运行窗口接近安全阈值（笔记本默认2000 token），就自动摘要，然后本地和服务器端删除被替换的轮次。

我们监控 response.done 中返回的 latest_tokens。当它超过 SUMMARY_TRIGGER 并且我们拥有的轮次多于 KEEP_LAST_TURNS 时，我们启动一个后台摘要协程。

我们将除最后 2 个轮次之外的所有内容压缩成一个法语段落，然后：

1. 将该段落作为新的助手消息插入到对话的顶部。

2. 删除用于摘要的消息项。

我们稍后将询问 Voice Agent 摘要的语言是什么，以测试摘要是否成功插入到实时 API 对话上下文中。

async def run_summary_llm(text: str) -> str:
    """调用一个轻量级模型来总结 `text`。"""
    resp = await asyncio.to_thread(lambda: openai.chat.completions.create(
        model=SUMMARY_MODEL,
        temperature=0,
        messages=[
            {"role": "system", "content": "用一个简洁的段落以法语总结以下对话，以便将其用作未来对话的上下文。"},
            {"role": "user", "content": text},
        ],
    ))
    return resp.choices[0].message.content.strip()

重要的实现细节：

• 摘要被追加为 SYSTEM 消息，而不是 ASSISTANT 消息。测试表明，在长时间的对话中，使用 ASSISTANT 消息进行摘要可能会导致模型错误地从音频响应切换到文本响应。通过使用 SYSTEM 消息进行摘要（其中还可以包含其他自定义指令），我们向模型明确表示这些是设置上下文的指令，从而防止它错误地采用正在进行的对话的用户-助手交互的模式。

async def summarise_and_prune(ws, state):
    """总结旧轮次，在服务器端删除它们，并在本地+远程预先插入一个摘要轮次。"""
    state.summarising = True
    print(
        f"⚠️  Token 窗口 ≈{state.latest_tokens} ≥ {SUMMARY_TRIGGER}。正在总结…",
    )
    old_turns, recent_turns = state.history[:-KEEP_LAST_TURNS], state.history[-KEEP_LAST_TURNS:]
    convo_text = "\n".join(f"{t.role}: {t.text}" for t in old_turns if t.text)

    if not convo_text:
        print("没有可总结的内容（转录仍在等待中）。")
        state.summarising = False

    summary_text = await run_summary_llm(convo_text) if convo_text else ""
    state.summary_count += 1
    summary_id = f"sum_{state.summary_count:03d}"
    state.history[:] = [Turn("assistant", summary_id, summary_text)] + recent_turns

    print_history(state)    

    # 在服务器上创建摘要
    await ws.send(json.dumps({
        "type": "conversation.item.create",
        "previous_item_id": "root",
        "item": {
            "id": summary_id,
            "type": "message",
            "role": "system",
            "content": [{"type": "input_text", "text": summary_text}],
        },
    }))

    # 删除旧项目
    for turn in old_turns:
        await ws.send(json.dumps({
            "type": "conversation.item.delete",
            "item_id": turn.item_id,
        }))

    print(f"✅ 已插入摘要 ({summary_id})")

    state.summarising = False

以下函数允许我们随时间轮询转录。这对于用户音频尚未立即转录的情况很有用，因此我们可以稍后检索最终结果。

async def fetch_full_item(
    ws, item_id: str, state: ConversationState, attempts: int = 1
):
    """
    请求服务器获取完整的对话项目；如果转录字段仍然为 null，则重试最多 5 次。
    完成后解析等待的 future。
    """
    # 如果已经有待处理的获取请求，只需等待它
    if item_id in state.waiting:
        return await state.waiting[item_id]

    fut = asyncio.get_running_loop().create_future()
    state.waiting[item_id] = fut

    await ws.send(json.dumps({
        "type": "conversation.item.retrieve",
        "item_id": item_id,
    }))
    item = await fut

    # 如果转录仍然缺失，则重试（最多 5 次）
    if attempts < 5 and not item.get("content", [{}])[0].get("transcript"):
        await asyncio.sleep(0.4 * attempts)
        return await fetch_full_item(ws, item_id, state, attempts + 1)

    # 完成 – 删除标记
    state.waiting.pop(item_id, None)
    return item

4. 端到端工作流程演示

运行以下两个单元格以启动交互式会话。按 Ctrl-C 停止录制。

注意： 此笔记本使用 SUMMARY_TRIGGER = 2000 和 KEEP_LAST_TURNS = 2 以便快速演示摘要。 在生产环境中，您应根据应用程序的需求调整这些值。

• 典型的 SUMMARY_TRIGGER 值在 20,000–32,000 token 之间，具体取决于性能在您的用例中随上下文增大而下降的程度。

# 音频/配置旋钮
SAMPLE_RATE_HZ    = 24_000   # pcm16 所需
CHUNK_DURATION_MS = 40       # 音频捕获的块大小
BYTES_PER_SAMPLE  = 2        # pcm16 = 2 字节/样本
SUMMARY_TRIGGER   = 2_000    # 当上下文 ≥ 此值时进行摘要
KEEP_LAST_TURNS   = 2       # 保留这些轮次不变
SUMMARY_MODEL     = "gpt-4o-mini"  # 更便宜、快速的摘要器

# --------------------------------------------------------------------------- #
# 🎤 实时会话                                                          #
# --------------------------------------------------------------------------- #
async def realtime_session(model="gpt-4o-realtime-preview", voice="shimmer", enable_playback=True):
    """
    主协程：连接到实时端点，启动辅助任务，
    并在一个大的异步 for 循环中处理传入事件。
    """
    state = ConversationState()  # 每次运行时重置状态

    pcm_queue: asyncio.Queue[bytes] = asyncio.Queue()
    assistant_audio: List[bytes] = []

    # ----------------------------------------------------------------------- #
    # 打开到实时 API 的 WebSocket 连接                                        #
    # ----------------------------------------------------------------------- #
    url = f"wss://api.openai.com/v1/realtime?model={model}"
    headers = {"Authorization": f"Bearer {openai.api_key}", "OpenAI-Beta": "realtime=v1"}

    async with websockets.connect(url, extra_headers=headers, max_size=1 << 24) as ws:
        # ------------------------------------------------------------------- #
        # 等待服务器发送 session.created                                     #
        # ------------------------------------------------------------------- #
        while json.loads(await ws.recv())["type"] != "session.created":
            pass
        print("session.created ✅")

        # ------------------------------------------------------------------- #
        # 配置会话：语音、模态、音频格式、转录                                 #
        # ------------------------------------------------------------------- #
        await ws.send(json.dumps({
            "type": "session.update",
            "session": {
                "voice": voice,
                "modalities": ["audio", "text"],
                "input_audio_format": "pcm16",
                "output_audio_format": "pcm16",
                "input_audio_transcription": {"model": "gpt-4o-transcribe"},
            },
        }))

        # ------------------------------------------------------------------- #
        # 启动后台任务：麦克风捕获 → 队列 → websocket                         #
        # ------------------------------------------------------------------- #
        mic_task = asyncio.create_task(mic_to_queue(pcm_queue))
        upl_task = asyncio.create_task(queue_to_websocket(pcm_queue, ws))

        print("🎙️ 现在说话（按 Ctrl-C 退出）…")

        try:
            # ------------------------------------------------------------------- #
            # 主事件循环：处理来自 websocket 的传入事件                           #
            # ------------------------------------------------------------------- #
            async for event_raw in ws:
                event = json.loads(event_raw)
                etype = event["type"]

                # --------------------------------------------------------------- #
                # 用户刚说话 ⇢ conversation.item.created (role = user)           #
                # --------------------------------------------------------------- #
                if etype == "conversation.item.created" and event["item"]["role"] == "user":
                    item = event["item"]
                    text = None
                    if item["content"]:
                        text = item["content"][0].get("transcript")

                    state.history.append(Turn("user", event["item"]["id"], text))

                    # 如果转录尚未可用，稍后获取它
                    if text is None:
                        asyncio.create_task(fetch_full_item(ws, item["id"], state))

                # --------------------------------------------------------------- #
                # 转录已获取 ⇢ conversation.item.retrieved                         #
                # --------------------------------------------------------------- #
                elif etype == "conversation.item.retrieved":
                    content = event["item"]["content"][0]
                    # 填充历史中缺失的转录
                    for t in state.history:
                        if t.item_id == event["item"]["id"]:
                            t.text = content.get("transcript")
                            break

                # --------------------------------------------------------------- #
                # 助手的音频以增量方式到达                                        #
                # --------------------------------------------------------------- #
                elif etype == "response.audio.delta":
                    assistant_audio.append(base64.b64decode(event["delta"]))

                # --------------------------------------------------------------- #
                # 助手回复完成 ⇢ response.done                                    #
                # --------------------------------------------------------------- #
                elif etype == "response.done":
                    for item in event["response"]["output"]:
                        if item["role"] == "assistant":
                            txt = item["content"][0]["transcript"]
                            state.history.append(Turn("assistant", item["id"], txt))
                            # print(f"\n🤖 {txt}\n")
                    state.latest_tokens = event["response"]["usage"]["total_tokens"]
                    print(f"—— response.done  (窗口 ≈{state.latest_tokens} token) ——")
                    print_history(state)

                    # 获取任何仍然缺失的用户转录
                    for turn in state.history:
                        if (turn.role == "user"
                            and turn.text is None
                            and turn.item_id not in state.waiting):
                            asyncio.create_task(
                                fetch_full_item(ws, turn.item_id, state)
                            )

                    # 回放收集到的音频，一旦回复完成
                    if enable_playback and assistant_audio:
                        simpleaudio.play_buffer(b"".join(assistant_audio), 1, BYTES_PER_SAMPLE, SAMPLE_RATE_HZ)
                        assistant_audio.clear()

                    # 如果上下文过大则进行摘要 – 在后台运行，以免阻塞对话
                    if state.latest_tokens >= SUMMARY_TRIGGER and len(state.history) > KEEP_LAST_TURNS and not state.summarising:
                        asyncio.create_task(summarise_and_prune(ws, state))

        except KeyboardInterrupt:
            print("\n正在停止…")
        finally:
            mic_task.cancel()
            await pcm_queue.put(None)
            await upl_task

# 运行实时会话（此单元格将阻塞直到您停止它）
await realtime_session()

session.created ✅
🎙️ 现在说话（按 Ctrl-C 退出）…
—— response.done  (窗口 ≈979 token) ——
—— 对话记录 ———————————————————————————————
[user     ] Can you tell me a quick story?  (item_BTuMOcpUqp8qknKhLzlkA)
[assistant] Once upon a time, in a cozy little village, there was a cat named Whiskers who was always getting into trouble. One sunny day, Whiskers found a mysterious glowing stone in the garden. Curious, he pawed at it, and poof! The stone granted him the ability to talk to birds. Whiskers and his new bird friends had grand adventures, solving mysteries and exploring the village. And from that day on, Whiskers was known as the most adventurous cat in the village. The end.  (item_BTuMPRWxqpv0ph6QM46DK)
——————————————————————————————————————————
—— response.done  (窗口 ≈2755 token) ——
—— 对话记录 ———————————————————————————————
[user     ] Can you tell me a quick story?  (item_BTuMOcpUqp8qknKhLzlkA)
[assistant] Once upon a time, in a cozy little village, there was a cat named Whiskers who was always getting into trouble. One sunny day, Whiskers found a mysterious glowing stone in the garden. Curious, he pawed at it, and poof! The stone granted him the ability to talk to birds. Whiskers and his new bird friends had grand adventures, solving mysteries and exploring the village. And from that day on, Whiskers was known as the most adventurous cat in the village. The end.  (item_BTuMPRWxqpv0ph6QM46DK)
[user     ] Can you tell me three extremely funny stories?  (item_BTuNN64LdULM21OyC4vzN)
[assistant] Sure, let's dive into some giggle-worthy tales:  **Story One:** There was a forgetful baker named Benny who baked a hundred cakes for a big wedding. But on the big day, he forgot where he put them! The entire town joined in to find the missing cakes, only to discover Benny had stored them in his neighbor's garage, thinking it was his pantry. The wedding turned into a town-wide cake feast!  **Story Two:** A mischievous dog named Sparky loved to play pranks. One day, he swapped his owner's phone with a squeaky toy, causing a hilarious mix-up of barks, squeaks, and confused calls. Sparky's owner ended up having a full conversation with the mailman, all in squeaks!  **Story Three:** In a small town, a parrot named Polly became a local celebrity for reciting tongue twisters. One day, Polly challenged the mayor to a tongue twister duel. The mayor, tongue-tied and laughing, declared Polly the official town jester. Polly squawked with pride, and the town rang with laughter for days.  (item_BTuNNpNxki5ynSQ5c3Xsa)
——————————————————————————————————————————
⚠️  Token 窗口 ≈2755 ≥ 2000。正在总结…
—— 对话记录 ———————————————————————————————
[assistant] L'utilisateur a demandé une histoire rapide, et l'assistant a raconté celle d'un chat nommé Whiskers qui, après avoir trouvé une pierre mystérieuse dans son jardin, a obtenu le pouvoir de parler aux oiseaux. Avec ses nouveaux amis oiseaux, Whiskers a vécu de grandes aventures, résolvant des mystères et explorant le village, devenant ainsi le chat le plus aventurier du village.  (sum_001)
[user     ] Can you tell me three extremely funny stories?  (item_BTuNN64LdULM21OyC4vzN)
[assistant] Sure, let's dive into some giggle-worthy tales:  **Story One:** There was a forgetful baker named Benny who baked a hundred cakes for a big wedding. But on the big day, he forgot where he put them! The entire town joined in to find the missing cakes, only to discover Benny had stored them in his neighbor's garage, thinking it was his pantry. The wedding turned into a town-wide cake feast!  **Story Two:** A mischievous dog named Sparky loved to play pranks. One day, he swapped his owner's phone with a squeaky toy, causing a hilarious mix-up of barks, squeaks, and confused calls. Sparky's owner ended up having a full conversation with the mailman, all in squeaks!  **Story Three:** In a small town, a parrot named Polly became a local celebrity for reciting tongue twisters. One day, Polly challenged the mayor to a tongue twister duel. The mayor, tongue-tied and laughing, declared Polly the official town jester. Polly squawked with pride, and the town rang with laughter for days.  (item_BTuNNpNxki5ynSQ5c3Xsa)
——————————————————————————————————————————
✅ 已插入摘要 (sum_001)
—— response.done  (窗口 ≈2147 token) ——
—— 对话记录 ———————————————————————————————
[assistant] L'utilisateur a demandé une histoire rapide, et l'assistant a raconté celle d'un chat nommé Whiskers qui, après avoir trouvé une pierre mystérieuse dans son jardin, a obtenu le pouvoir de parler aux oiseaux. Avec ses nouveaux amis oiseaux, Whiskers a vécu de grandes aventures, résolvant des mystères et explorant le village, devenant ainsi le chat le plus aventurier du village.  (sum_001)
[user     ] Can you tell me three extremely funny stories?  (item_BTuNN64LdULM21OyC4vzN)
[assistant] Sure, let's dive into some giggle-worthy tales:  **Story One:** There was a forgetful baker named Benny who baked a hundred cakes for a big wedding. But on the big day, he forgot where he put them! The entire town joined in to find the missing cakes, only to discover Benny had stored them in his neighbor's garage, thinking it was his pantry. The wedding turned into a town-wide cake feast!  **Story Two:** A mischievous dog named Sparky loved to play pranks. One day, he swapped his owner's phone with a squeaky toy, causing a hilarious mix-up of barks, squeaks, and confused calls. Sparky's owner ended up having a full conversation with the mailman, all in squeaks!  **Story Three:** In a small town, a parrot named Polly became a local celebrity for reciting tongue twisters. One day, Polly challenged the mayor to a tongue twister duel. The mayor, tongue-tied and laughing, declared Polly the official town jester. Polly squawked with pride, and the town rang with laughter for days.  (item_BTuNNpNxki5ynSQ5c3Xsa)
[user     ]   (item_BTuPLaCv8ATdIwAQ2rLgO)
[assistant] Sure! The first summary I provided between us was in French.  (item_BTuPLa7BaSQToGCVOmfBK)

我们与我们的语音 AI 进行了一次对话。经过几轮后，总 token 数达到了 SUMMARY_MAX，这触发了对话摘要步骤。这生成了早期消息的摘要。

由于总共有 N = 4 条消息，我们总结了前 N - 2 = 2 条消息：

—— 对话记录 ———————————————————————————————
[user     ] Can you tell me a quick story?  (item_BTuMOcpUqp8qknKhLzlkA)
[assistant] Once upon a time, in a cozy little village, there was a cat named Whiskers who was always getting into trouble. One sunny day, Whiskers found a mysterious glowing stone in the garden. Curious, he pawed at it, and poof! The stone granted him the ability to talk to birds. Whiskers and his new bird friends had grand adventures, solving mysteries and exploring the village. And from that day on, Whiskers was known as the most adventurous cat in the village. The end.  (item_BTuMPRWxqpv0ph6QM46DK)

然后我们使用 root: true 标志创建了一个法语摘要并将其插入到对话历史中。这确保了摘要出现在对话的第一条消息中。之后，我们使用 "type": "conversation.item.delete" 删除被摘要的原始项目。

为了验证摘要插入，我们询问了语音 AI 摘要的语言是什么。它正确地回答：

[assistant] Sure! The first summary I provided between us was in French.  (item_BTuPLa7BaSQToGCVOmfBK)

5 · 实际应用

上下文摘要对于长时间运行的语音体验非常有用。 以下是一些用例创意：

| 用例 | 附加价值 | 有何用处 |

6 · 后续步骤和深入阅读

尝试使用此笔记本，并尝试将上下文摘要集成到您的应用程序中。

您可以尝试的几件事： | 尝试这个… | 你将学到什么 |

资源：

• OpenAI 实时指南
• OpenAI 实时对话
• OpenAI 实时 API 参考
• 语音 AI 和语音代理