Realtime Agent 实时语音代理

2026-04-16 AI Agent 编程 4.1k 字 15 分钟阅读
📑 目录
  1. 概述
  2. WebSocket 低延迟优化与音频编解码的实时系统
  3. 正文
  4. 相关阅读
  5. 参考文档
  6. 完整实战示例：低延迟语音客服系统
  7. 实时语音系统的数据流架构
  8. 常见问题与调试
  9. 与其他方案对比
  10. 实时语音会话的状态机设计与双缓冲音频队列
  11. 生产环境部署与性能优化
    1. WebSocket 集群的实践要点
    2. 端到端延迟的关键指标
    3. 全球边缘节点部署的架构考量
    4. 运维团队的协作建议

概述

WebSocket 低延迟优化与音频编解码的实时系统

Realtime Agent 基于 gpt-realtime-1.5 模型和 WebSocket 协议，实现了端到端延迟低至数百毫秒的语音对话。这个延迟水平对于自然的语音交互至关重要——研究表明，当对话延迟超过 700ms 时，用户会明显感到"不自然"，交互体验显著下降。

实现低延迟的关键技术栈包括：

  1. 流式音频输入：客户端通过 WebSocket 持续发送音频流，而不是等待用户说完再发送完整录音。这消除了录音等待的延迟。
  2. 增量式处理：模型在收到音频流的同时就开始处理，而不是等到音频结束。当检测到用户停顿时，模型可以立即开始生成响应。
  3. 语音活动检测（VAD）：通过算法检测用户何时开始说话和停止说话，实现自动的"按键通话"替代。VAD 的准确性和延迟直接影响交互的流畅度——过于敏感的 VAD 会导致模型在用户停顿（如思考时）过早介入，而过于迟钝的 VAD 则会导致响应延迟。
  4. 流式 TTS 输出：模型的文本响应被实时转换为语音，并分段发送给客户端，用户可以在完整响应生成完成前就开始听到开头部分。

音频编解码是另一个影响延迟和带宽的关键因素。常见的选择包括：

  • PCM：无压缩，音质最好，但带宽消耗大（16kHz 单声道约 256kbps）。
  • Opus：高效的语音编解码器，在同等音质下带宽仅为 PCM 的 10%-20%，且支持可变码率。
  • AAC：广泛用于流媒体，但延迟略高于 Opus。

Opus 通常是实时语音交互的首选，因为它在低延迟和带宽效率之间取得了最佳平衡。

从系统设计来看，Realtime Agent 的架构是一个典型的管道（Pipeline）：音频输入 -> VAD -> ASR（自动语音识别）-> LLM -> TTS -> 音频输出。每个阶段的延迟都会累积，因此优化需要从全链路角度考虑，而非单独优化某个环节。

RealtimeAgent + RealtimeRunner：基于 WebSocket 的低延迟语音交互 Agent。

正文

相关阅读

参考文档

完整实战示例：低延迟语音客服系统

以下示例展示了如何构建一个支持 VAD 控制、噪音抑制和优雅中断的实时语音客服系统：

import asyncio
import json
from dataclasses import dataclass
from typing import AsyncIterator
from agents import Agent, Runner
from agents.realtime import RealtimeAgent, RealtimeConfig


@dataclass
class VoiceSessionConfig:
    language: str = "zh-CN"
    voice: str = "alloy"
    vad_threshold: float = 0.5
    interruption_mode: str = "polite"  # polite, immediate


class VoiceCustomerService:
    """实时语音客服系统。"""

    def __init__(self, config: VoiceSessionConfig):
        self.config = config
        self.session_stats = {
            "total_turns": 0,
            "avg_latency_ms": 0.0,
            "interruptions": 0,
        }

    async def handle_session(self, websocket) -> None:
        """处理一个完整的语音会话。"""
        agent = RealtimeAgent(
            name="VoiceAgent",
            instructions="""
你是智能语音客服助手。规则：
1. 用中文回答  #语气亲切自然
2. 回答简洁  #不超过3句话
3. 如果用户问题复杂  #建议转文字渠道
4. 遇到情绪激动的用户  #先安抚再解决问题
""".strip(),
            model="gpt-realtime-1.5",
            realtime_config=RealtimeConfig(
                voice=self.config.voice,
                language=self.config.language,
                vad_enabled=True,
                vad_threshold=self.config.vad_threshold,
            ))

实时语音系统的数据流架构

下图展示了从用户语音输入到语音输出的端到端数据流：
mermaid
flowchart LR
U[用户麦克风] --> VAD[VAD 检测]
VAD --> STT[语音转文字]
STT --> A[Agent 处理]
A --> LLM[LLM 推理]
LLM --> TTS[文字转语音]
TTS --> MIX[音频混合]
MIX --> SP[扬声器输出]
A --> T[工具调用]
T --> A
style VAD fill:#c5e0b4,stroke:#5a4a3a
style STT fill:#e8d5b5,stroke:#5a4a3a
style LLM fill:#e8d5b5,stroke:#5a4a3a
style TTS fill:#e8d5b5,stroke:#5a4a3a


端到端延迟是实时语音系统的生命线。每个环节的延迟都会累积，总延迟超过 800ms 就会让用户明显感知到卡顿。

## 实时语音的关键技术挑战

实时语音 Agent 面临的工程挑战远比文本 Agent 复杂。

**打断处理（Barge-in）**：

用户在 Agent 说话期间可能插话打断。系统需要实时检测用户开始说话的事件，并立即停止当前语音输出，处理用户的新输入。这要求 VAD（语音活动检测）具有极低的延迟（< 100ms），并且音频 pipeline 支持随时中断。

```python
class BargeInHandler:
    def __init__(self):
        self.speaking = False
        self.current_tts_task = None

    async def on_user_speech_start(self):
        self.speaking = True
        if self.current_tts_task and not self.current_tts_task.done():
            self.current_tts_task.cancel()
            await self.audio_player.stop()

    async def on_user_speech_end(self, transcript: str):
        self.speaking = False
        result = await self.agent.process(transcript)
        self.current_tts_task = asyncio.create_task(
            self.tts.speak(result)
        )

回声消除（AEC）：

如果用户使用扬声器而非耳机，Agent 输出的语音会被麦克风重新采集，形成回声。如果不消除回声，STT 会将 Agent 的话误识别为用户的输入，导致对话循环。WebRTC 提供了成熟的 AEC 算法，建议在客户端集成。

网络抖动缓冲：

语音数据通过 WebSocket 传输时，网络抖动可能导致音频不连贯。客户端应实现 jitter buffer，缓存 100-200ms 的音频数据，以平滑网络波动带来的断续。

多轮对话的上下文管理：

语音对话的上下文管理与文本对话不同。用户通常不会重复已提及的信息（如还是刚才那个订单），Agent 需要正确理解这种指代关系。建议维护一个对话实体表，记录对话中提到的关键实体（人名、订单号、日期），在指代消解时作为参考。

实时语音 Agent 的成功不仅取决于技术实现，还高度依赖语音交互设计。Agent 的语音回复应该自然、有节奏感，避免长时间沉默或过于急促。适当的填充词（如让我想想、好的）可以给用户反馈，减少等待焦虑。语音交互的延迟优化也很重要：预连接在用户打开应用时预先建立 WebSocket 连接；本地 VAD 在客户端实现语音活动检测；模型预热在服务启动时发送虚拟请求预热缓存。对于多轮对话，利用对话历史的局部性原理，只保留最近 N 轮上下文，避免历史过长导致处理延迟增加。

常见问题与调试

问题一：音频延迟过高导致对话不自然

如果端到端延迟超过 1 秒，用户体验会显著下降。优化方法：

  1. 使用有线网络或 5GHz WiFi，避免 2.4GHz 频段的干扰和拥塞。
  2. 选择地理位置最近的 OpenAI 接入点，减少网络往返时间。
  3. 在客户端实现音频预缓冲，平滑网络抖动带来的影响。
  4. 对于 TTS 输出，使用更低质量的语音模型（如降低采样率）来换取更低的延迟。

问题二：VAD 误触发导致频繁打断

在嘈杂环境中，VAD 可能将背景噪音误判为用户说话，导致模型频繁中断自己的回答。调优建议：

  1. 在客户端实现噪音抑制（NS）和回声消除（AEC）预处理。
  2. 调高 VAD 阈值，要求更高的音频能量才判定为说话。
  3. 增加"最小说话时长"参数，过滤掉短暂的噪音 burst。
  4. 实现"礼貌打断"模式：即使用户开始说话，也允许模型完成当前句子。

问题三：中文语音识别的准确性问题

实时语音识别对中文的准确性可能低于英文，特别是在涉及专业术语、人名或地名时。改进措施：

  1. 在 RealtimeConfig 中提供自定义词汇表（hotwords），提升特定词汇的识别率。
  2. 结合上下文进行纠错：如果 ASR 结果在语义上不合理，用 LLM 进行后处理修正。
  3. 对于关键信息（如电话号码、地址），要求用户通过 DTMF（按键输入）或文字确认。

与其他方案对比

维度Agents SDK RealtimeLiveKit + OpenAITwilio + Whisper
延迟低（端到端优化）中（需组装）高（串行管道）
集成复杂度低（SDK 原生）中（需配置 LiveKit）高（需自建管道）
语音质量高（OpenAI TTS）高（可自定义 TTS）中（依赖外部 TTS）
适用场景快速原型、标准客服复杂多媒体交互传统电话系统集成

LiveKit 是目前实时音视频领域最专业的开源框架，它提供了完善的房间管理、媒体路由和跨平台 SDK，适合构建复杂的实时交互应用（如多人视频会议中的 AI 助手）。Twilio + Whisper 的组合则更适合需要与传统电话网络（PSTN）集成的场景，如自动外呼系统。Agents SDK 的 Realtime Agent 提供了最简单的接入路径，适合已经使用 Agents SDK 构建文本 Agent、希望快速扩展到语音场景的团队。

实时语音会话的状态机设计与双缓冲音频队列

语音交互的复杂性不仅来自音频编解码，更来自会话状态的动态变化。一个完整的语音对话会经历"空闲 -> 监听中 -> 处理中 -> 播放中 -> 空闲"的循环，而用户的中断、网络抖动、识别错误都会使状态迁移变得复杂。使用**状态机模式（State Machine Pattern）**来管理会话生命周期，是保障系统稳定性的关键设计。

stateDiagram-v2
    [*] --> Idle: 会话开始
    Idle --> Listening: 检测到语音活动
    Listening --> Processing: VAD 判定说话结束
    Processing --> Playing: LLM + TTS 生成完成
    Playing --> Idle: 播放完成
    Playing --> Listening: 用户打断（中断事件）
    Processing --> Listening: 识别为空/取消
    Idle --> Listening: 唤醒词触发
    Playing --> Processing: 网络重连后恢复

状态机的核心优势在于将复杂的条件判断转化为明确的状态迁移表。例如，"用户打断"这一事件在"Playing"状态下会触发切换到"Listening"，但在"Processing"状态下可能被忽略（等待当前处理完成）。这种显式化的设计消除了大量隐式的 if-else 嵌套，使得代码的可测试性和可维护性大幅提升。

另一个关键技术点是双缓冲音频队列（Double-Buffered Audio Queue）。在实时语音系统中，音频数据的生产（麦克风采集）和消费（网络发送）速度不匹配是常态。使用单队列会导致频繁的锁竞争和内存拷贝；双缓冲策略则通过交替使用两个缓冲区，实现无锁化的数据交换：

import asyncio
from collections import deque
from typing import Optional


class DoubleBufferAudioQueue:
    """双缓冲音频队列：生产/消费解耦，减少锁竞争。"""

    def __init__(self, buffer_size: int = 3200):  # 200ms @ 16kHz mono
        self.front = bytearray(buffer_size)   # 消费者读取
        self.back = bytearray(buffer_size)    # 生产者写入
        self.back_pos = 0
        self.front_ready = asyncio.Event()
        self.lock = asyncio.Lock()

    async def write(self, chunk: bytes) -> Optional[bytes]:
        """写入音频块，当 back 满时交换缓冲区并返回 front 数据。"""
        async with self.lock:
            remaining = len(self.back) - self.back_pos
            if len(chunk) <= remaining:
                self.back[self.back_pos:self.back_pos + len(chunk)] = chunk
                self.back_pos += len(chunk)
                return None
            else:
                # back 已满，交换 front/back
                self.front, self.back = self.back, self.front
                self.back[:len(chunk)] = chunk
                self.back_pos = len(chunk)
                self.front_ready.set()
                return bytes(self.front)

    async def read(self) -> bytes:
        """阻塞等待 front 缓冲区就绪。"""
        await self.front_ready.wait()
        self.front_ready.clear()
        return bytes(self.front)


class AudioPipeline:
    """基于双缓冲的音频处理管道，应用观察者模式解耦处理阶段。"""

    def __init__(self):
        self.queue = DoubleBufferAudioQueue()
        self.observers = []

    def add_observer(self, callback):
        self.observers.append(callback)

    async def produce(self, mic_stream):
        """生产者：从麦克风读取并写入队列。"""
        async for chunk in mic_stream:
            packet = await self.queue.write(chunk)
            if packet:
                for obs in self.observers:
                    await obs(packet)

    async def consume(self):
        """消费者：读取完整缓冲区并分发。"""
        while True:
            packet = await self.queue.read()
            for obs in self.observers:
                await obs(packet)

上述代码展示了两个经典设计模式的结合：双缓冲模式解决了生产-消费速率不匹配问题，观察者模式则将音频采集、VAD 检测、网络发送等阶段解耦。每个处理阶段只需注册为观察者，无需关心其他阶段的实现细节。这种架构使得新增处理阶段（如添加噪音抑制模块）对现有代码的侵入性降到最低。

在生产环境中，还应考虑为状态机添加超时保护和异常恢复机制。例如，"Processing"状态如果持续超过 10 秒，应自动降级为文字回复或提示用户重试；"Playing"状态下如果检测到连续 3 次用户打断，应切换到"人工接管"状态。这些边界情况的处理，往往决定了系统在真实场景下的可用性。

生产环境部署与性能优化

WebSocket 集群的实践要点

将本章节的技术应用到生产环境时，首要考虑的是稳定性与可观测性。建议采用渐进式 rollout 策略：先在开发环境验证核心逻辑，再迁移到预发布环境进行压力测试，最后才全量上线。部署过程中应配置完善的日志收集和指标监控，确保任何问题都能被快速发现和定位。

具体来说，需要在基础设施层面做好以下准备：容器资源限制（CPU/内存）、网络策略配置（防火墙规则、服务网格）、持久化存储选型（SSD vs 标准盘）以及备份恢复方案。对于高可用要求严格的场景，建议部署多实例并配置负载均衡，避免单点故障导致服务中断。

端到端延迟的关键指标

监控是生产系统的生命线。针对本章节涉及的功能，建议重点跟踪以下指标：请求延迟（P50/P95/P99）、错误率（4xx/5xx/超时）、吞吐量（QPS/TPS）以及资源利用率（CPU/内存/磁盘/网络）。这些指标应接入统一的监控大盘，并设置合理的告警阈值。

除了基础指标，还应关注业务层面的指标。例如功能成功率、用户满意度、成本消耗趋势等。通过将技术指标与业务指标关联分析，可以更准确地评估系统改进的实际价值，避免陷入"为了优化而优化"的陷阱。

全球边缘节点部署的架构考量

随着业务规模增长，单实例部署很快会成为瓶颈。扩展性设计应在项目初期就纳入考量，而非事后补救。水平扩展通常比垂直扩展更具成本效益，但也引入了分布式系统的复杂性（数据一致性、服务发现、负载均衡等）。

在扩展过程中，建议遵循"无状态优先"原则：将状态外置到独立的存储层（如 Redis、PostgreSQL），使计算层可以随时水平扩容。对于无法避免的状态（如会话、缓存），采用分布式一致性协议或最终一致性模型来管理。定期进行容量规划和压力测试，确保系统在流量峰值时仍能稳定运行。

运维团队的协作建议

技术方案的落地离不开高效的团队协作。建议建立清晰的运维手册（Runbook），涵盖常见故障的诊断步骤、应急处理流程和升级路径。同时，通过定期的复盘会议，将线上事故转化为团队的学习素材，持续完善系统的健壮性。

在工具链方面，推荐将本章节的配置和脚本纳入版本控制（Git），并使用 Infrastructure as Code（IaC）工具（如 Terraform、Ansible）管理基础设施变更。这不仅能提高部署效率，还能确保环境一致性，减少"在我机器上能跑"的问题。