扣子WebSocket大模型语音交互原理与实现
·
扣子 WebSocket 大模型语音交互：从原理到落地的全链路深度解析
本文系统性地拆解了字节跳动扣子（Coze）平台基于 WebSocket 的实时语音对话技术，涵盖全双工通信原理、ASR-LLM-TTS 流水线架构、VAD 转弯检测机制、音频编解码管道，并提供 Web SDK、Python SDK、ESP32 嵌入式设备三种场景的完整实现方案。
目录
- 一、引言：为什么选择 WebSocket 做语音交互
- 二、总体架构：全双工实时对话系统
- 三、核心技术链路深度剖析
- 四、WebSocket 全双工通信原理
- 五、Turn Detection：对话轮次管理
- 六、音频管道全链路
- 七、具体实现：三大场景实战
- 八、完整事件流序列图
- 九、常见问题与调优建议
- 十、总结与展望
- 参考资料
一、引言：为什么选择 WebSocket 做语音交互
在 AI 大模型时代，语音交互正在从传统的"一问一答"进化为实时、双向、可打断的自然对话模式。想象你和朋友聊天——你可以随时插话、打断、补充，对方也能立刻感知并调整回应。这种《Her》电影中的 AI 对话体验，正是扣子 WebSocket 语音交互技术追求的目标。
传统方案通常采用 HTTP 短连接 + 轮询 的方式：
用户录音 → HTTP上传 → 等待识别 → HTTP返回文本 → HTTP请求LLM → 等待生成 → HTTP请求TTS → 返回音频
这种模式的痛点很明显：
- ❌ 延迟高：每个环节都是请求-响应，端到端延迟 3~8 秒
- ❌ 不支持打断：必须等 AI 说完才能说下一句
- ❌ 连接开销大：每次请求都要 TCP 三次握手 + TLS 握手
- ❌ 流式体验差：无法边说话边识别，必须录完整句再发送
而 WebSocket 全双工方案 彻底改变了这一局面：
- ✅ 全双工通信：音频可以边采集边发送，AI 回复可以边生成边播放
- ✅ 端到端延迟 < 800ms：流式处理，首字延迟极低
- ✅ 支持打断：用户可以随时说话，AI 立即停止并切换为聆听模式
- ✅ 长连接复用：一次握手，持续通信，省去重复握手开销
二、总体架构：全双工实时对话系统
扣子 WebSocket 语音交互采用经典的 “ASR → LLM → TTS” 三层流水线，架设在 WebSocket 全双工通道之上：
┌─────────────────────────────────────────────────────────────────────┐
│ 客户端 (Client) │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────────────┐ │
│ │ 麦克风 │──▶│ AI 降噪 │──▶│ PCM/Opus │──▶│ │ │
│ │ 采集 │ │ (可选) │ │ 编码 │ │ WebSocket │ │
│ └──────────┘ └──────────┘ └──────────┘ │ 全双工通道 │ │
│ │ │ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ wss://ws.coze.cn│ │
│ │ 扬声器 │◀──│ 音量控制 │◀──│ Opus/PCM │◀──│ │ │
│ │ 播放 │ │ │ │ 解码 │ │ │ │
│ └──────────┘ └──────────┘ └──────────┘ └────────┬─────────┘ │
└─────────────────────────────────────────────────────────┼───────────┘
│
═══════════════════════════════╪═══════
│
┌─────────────────────────────────────────────────────────┼───────────┐
│ 扣子服务端 (Server) │ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌────────▼─────────┐ │
│ │ 扬声器 │◀──│ TTS │◀──│ LLM │◀──│ ASR 语音识别 │ │
│ │ 播放 │ │ 语音合成│ │ 大模型 │ │ │ │
│ └──────────┘ └──────────┘ └──────────┘ └──────────────────┘ │
│ ┌──────────────────┐ │
│ │ VAD 语音活动 │ │
│ │ 检测 (可选) │ │
│ └──────────────────┘ │
└─────────────────────────────────────────────────────────────────────┘
核心分层：
| 层次 | 职责 | 关键组件 |
|---|---|---|
| 公共 API 层 | 音频输入控制、对话管理、状态查询 | WsChatClient / RealtimeClient |
| 基础连接层 | WebSocket 生命周期、心跳、事件分发 | BaseWsChatClient |
| 音频采集层 | 麦克风拾音、AI 降噪、PCM 编码 | PcmRecorder |
| 音频播放层 | 播放队列调度、音量控制、本地回听 | WavStreamPlayer |
| 编解码层 | Opus ↔ PCM 双向转换 | OpusDecoder / OpusEncoder |
| 同步层 | 音频帧与文本字幕对齐 | SentenceSynchronizer |
| 网络传输层 | WebSocket 协议栈 | WebSocket API |
三、核心技术链路深度剖析
3.1 ASR —— 语音转文本
扣子的 ASR 并非传统意义上的独立语音识别服务，而是深度集成了大语言模型的上下文理解能力。
技术特点：
- 上下文增强识别：LLM 会根据对话历史来纠正常见的同音词错误。例如对话在聊"扣子平台"，ASR 就不会把 “kòu zi” 识别成"扣子"以外的词
- 中英混合识别：支持"帮我查一下 GPT-4 的 API 文档"这种中英夹杂的表达
- 热词定制：可为特定场景预定义专有名词词表
- 中文准确率 ≥ 92%：在正常噪声环境下
- 端到端延迟约 800ms：从用户说完到识别结果返回
与传统 ASR 的对比：
传统 ASR Pipeline：
音频 → VAD切分 → 声学模型 → 语言模型 → 解码 → 文本
扣子 LLM-ASR Pipeline：
音频 → VAD切分 → 声学特征提取 → LLM端到端识别(融合上下文) → 文本
↑
对话历史记忆
3.2 LLM —— 大模型大脑
扣子的 LLM 层是整个对话系统的"大脑"，它不仅负责理解用户意图和生成回复，还管理着：
- 记忆系统：多轮对话上下文（默认保留最近 5 轮），支持 Redis 持久化
- 知识库集成：可挂载企业文档、FAQ、产品手册等私有知识
- 插件/技能系统：天气查询、日程管理、智能家居控制等外部能力
- 工作流编排：多步骤复杂任务的自动拆解与执行
LLM 生成的文本会并行做两件事：
- 流式推送给客户端作为字幕展示（
CONVERSATION_MESSAGE_DELTA） - 送入 TTS 引擎进行语音合成
3.3 TTS —— 文本转语音
扣子 TTS 采用端到端神经网络架构：
文本 → 文本前端(分词/韵律预测) → Transformer声学模型 → HiFiGAN声码器 → 音频波形
关键指标：
| 指标 | 数值 |
|---|---|
| 自然度 MOS 分 | 4.5 ~ 4.7 |
| 合成速度 | 3.2x 实时（即 1 秒音频仅需 0.3 秒合成） |
| 支持语言 | 28+ 种 |
| 情感表达 | 8 种基础情感 |
| 输出编码 | Opus / PCM / G.711A / G.711U |
SSML 标记语言 可用于精细控制：
<speak>
你好，<break time="500ms"/>
我是扣子<prosody rate="slow" pitch="high">AI 助手</prosody>
</speak>
四、WebSocket 全双工通信原理
4.1 为什么是 WebSocket 而不是 HTTP？
用快递站来比喻就很容易理解：
| HTTP | WebSocket | |
|---|---|---|
| 比喻 | 寄信——写一封信，寄出去，等回信 | 打电话——建立连接后实时双向通话 |
| 连接 | 每个请求建立一个连接 | 一次握手，持久复用 |
| 方向 | 客户端请求 → 服务端响应 | 双向任意时刻发送 |
| 开销 | 每次 TCP + TLS 握手 | 仅一次握手 |
| 流式 | 需 SSE 或 chunked transfer | 原生支持帧传输 |
| 适用 | REST API、文件下载 | 实时通信、游戏、语音 |
对于语音交互场景，WebSocket 是唯一合理的选择——你不可能等用户说完一整句话、HTTP 上传、等返回、再播放，这种体验完全不可接受。
4.2 协议栈与连接生命周期
┌─────────────────────────────────────────────────┐
│ 应用层：Coze 语音协议 │
│ (CHAT_UPDATE / INPUT_AUDIO_BUFFER_APPEND / │
│ CONVERSATION_AUDIO_DELTA ...) │
├─────────────────────────────────────────────────┤
│ 帧层：WebSocket 帧 (RFC 6455) │
│ 支持文本帧(JSON控制消息) + 二进制帧(音频数据) │
├─────────────────────────────────────────────────┤
│ 传输层：TLS 1.2+ (WSS) │
│ 传输层：TCP │
└─────────────────────────────────────────────────┘
连接全过程：
1. HTTP Upgrade 握手 (客户端 → 服务端)
GET wss://ws.coze.cn/v1/chat HTTP/1.1
Upgrade: websocket
Connection: Upgrade
2. 服务端确认升级 (101 Switching Protocols)
3. 配置阶段
客户端发送 CHAT_UPDATE 事件，协商音频参数：
- 输入格式: PCM 16bit / 48kHz / 单声道
- 输出格式: Opus / 24kHz
- 转弯检测: server_vad / client_interrupt
- 降噪参数: NSNG / SOFT
4. 对话阶段（全双工）
- 客户端持续发送 INPUT_AUDIO_BUFFER_APPEND (音频帧)
- 服务端流式返回 CONVERSATION_AUDIO_DELTA (AI语音)
- 服务端流式返回 CONVERSATION_MESSAGE_DELTA (字幕文本)
- 任意时刻可发送 interrupt 打断
5. 断开 (任一端发送 Close 帧)
五、Turn Detection：对话轮次管理
Turn Detection（轮次检测）是实时语音对话中最关键的机制——它决定了 “谁在说话” 和 “什么时候该换对方说”。
5.1 Server VAD 模式
Server VAD（Voice Activity Detection，语音活动检测）模式完全由服务端自动判断用户是否在说话。
工作流程：
用户开始说话
│
├── 客户端持续采集音频并发送 INPUT_AUDIO_BUFFER_APPEND
│
├── 服务端 VAD 检测到语音能量 → 触发 INPUT_AUDIO_BUFFER_SPEECH_STARTED
│ └── 客户端自动停止当前 AI 播放 (打断效果)
│
├── 用户继续说...
│
├── 服务端 VAD 检测到静音持续 ≥ silence_duration_ms → 触发 SPEECH_STOPPED
│ └── 服务端开始 ASR → LLM → TTS
│
├── 客户端收到 CONVERSATION_AUDIO_DELTA → 播放 AI 回复
│
└── AI 回复结束 → CONVERSATION_CHAT_COMPLETED → 等待下一轮
关键配置参数：
| 参数 | 默认值 | 说明 |
|---|---|---|
prefix_padding_ms |
600ms | 语音开始前的音频缓冲（防止切掉开头） |
silence_duration_ms |
500ms | 判定用户说完的静音阈值 |
turn_detection.type |
server_vad |
使用服务端 VAD |
适用场景：免提对话、智能音箱、车载助手
5.2 Client Interrupt（PTT）模式
Client Interrupt 模式由客户端手动控制录音的开始和结束，类似对讲机的"按键说话"（Push-to-Talk）。
工作流程：
用户按下按钮
│
├── 客户端调用 startRecord()
├── 发送 INPUT_AUDIO_BUFFER_APPEND (含音频帧)
│
├── 用户继续说话...
│
├── 用户松开按钮
│
├── 客户端调用 stopRecord()
├── 发送 INPUT_AUDIO_BUFFER_COMPLETE
│
├── 服务端 ASR → LLM → TTS
│
├── 客户端播放 AI 回复
│
└── 如用户中途按 interrupt() → 立即打断 AI
关键 API：
// 开始录音
client.startRecord();
// 停止录音并提交
client.stopRecord();
// 打断 AI 回复
client.interrupt();
适用场景：移动端 App、微信小程序、硬件按钮设备
5.3 两种模式对比与选型建议
| 维度 | Server VAD | Client Interrupt |
|---|---|---|
| 语音检测 | 服务端自动 | 客户端手动 |
| 录音控制 | 自动 | startRecord() / stopRecord() |
| 本地回听 | ✅ 支持 | ❌ 强制禁用 |
| 打断方式 | 自动（用户开口即打断） | 手动 interrupt() |
| 带宽消耗 | 较高（持续上传音频） | 较低（仅按键时上传） |
| 实现复杂度 | 低（客户端无需状态管理） | 中（需管理按钮状态） |
| 误触发风险 | 有（环境噪声可能误触发） | 无 |
| 适用场景 | 免提对话、智能音箱 | 移动 App、对讲机、硬件设备 |
选型建议：音视频通话类应用选 Server VAD；按键对讲类应用选 Client Interrupt。ESP32 等嵌入式设备推荐 Client Interrupt 以节省带宽和功耗。
六、音频管道全链路
6.1 采集管道（Recording Pipeline）
从物理麦克风到 WebSocket 发送的完整链路：
┌──────────────────────────────────────────────────────────────┐
│ 采集管道 (Client → Server) │
│ │
│ 物理麦克风 │
│ │ │
│ ▼ │
│ ┌──────────────┐ │
│ │ MediaStream │ getUserMedia({ audio: true }) │
│ │ API (浏览器) │ │
│ └──────┬───────┘ │
│ │ │
│ ▼ │
│ ┌──────────────┐ 可选：Agora AI 降噪引擎 │
│ │ AI 降噪 │ - NSNG (非稳态噪声：人声、交通) │
│ │ (可选) │ - STATIONARY_NS (稳态噪声：风扇、空调) │
│ └──────┬───────┘ │
│ │ │
│ ▼ │
│ ┌──────────────┐ │
│ │ PCM 编码 │ 格式: PCM 16-bit little-endian │
│ │ 16bit/48kHz │ 采样率: 16000 / 24000 / 48000 Hz │
│ │ 单声道 │ 声道: Mono │
│ └──────┬───────┘ │
│ │ │
│ ▼ │
│ ┌──────────────┐ 可选：带宽受限时启用 │
│ │ Opus 压缩 │ - 16kbps @ 16kHz │
│ │ (可选) │ - 帧长 20ms ~ 60ms │
│ └──────┬───────┘ │
│ │ │
│ ▼ │
│ ┌──────────────┐ │
│ │ Base64 编码 │ 转为文本传输（也可二进制帧） │
│ └──────┬───────┘ │
│ │ │
│ ▼ │
│ ┌──────────────┐ │
│ │ WebSocket │ INPUT_AUDIO_BUFFER_APPEND 事件 │
│ │ 发送 │ │
│ └──────────────┘ │
└──────────────────────────────────────────────────────────────┘
编码格式对比：
| 编码 | 码率 | 延迟 | 音质 | 适用场景 |
|---|---|---|---|---|
| PCM 16bit 48kHz | ~768 kbps | 零 | 无损 | 宽带网络、Web 端 |
| PCM 16bit 16kHz | ~256 kbps | 零 | 良好 | 中等带宽 |
| Opus 16kbps | 16 kbps | ~20ms/帧 | 良好 | 低带宽、4G、嵌入 |
6.2 播放管道（Playback Pipeline）
从服务端接收 AI 语音到扬声器播放：
┌──────────────────────────────────────────────────────────────┐
│ 播放管道 (Server → Client) │
│ │
│ WebSocket 接收 │
│ CONVERSATION_AUDIO_DELTA 事件 │
│ │ │
│ ▼ │
│ ┌──────────────┐ │
│ │ 音频队列 │ audioDeltaList (FIFO) │
│ │ 缓冲 │ 保证帧按序播放 │
│ └──────┬───────┘ │
│ │ │
│ ▼ │
│ ┌──────────────┐ │
│ │ 格式解码 │ Opus → PCM / G.711 → PCM │
│ │ (Opus/PCM/ │ OpusDecoder │
│ │ G.711) │ │
│ └──────┬───────┘ │
│ │ │
│ ▼ │
│ ┌──────────────┐ │
│ │ WavStream │ 基于 AudioWorklet 实现 │
│ │ Player │ 低延迟播放，支持动态音量控制 │
│ └──────┬───────┘ │
│ │ │
│ ▼ │
│ ┌──────────────┐ │
│ │ 物理扬声器 │ setPlaybackVolume(0.0 ~ 1.0) │
│ └──────────────┘ │
│ │
│ 并行： │
│ ┌──────────────┐ │
│ │ 字幕同步 │ SentenceSynchronizer │
│ │ (可选) │ 音频帧 ↔ 文字精确对齐 │
│ └──────────────┘ │
└──────────────────────────────────────────────────────────────┘
6.3 AI 降噪引擎
扣子集成了 Agora AI 降噪引擎，可在客户端对采集的音频进行实时降噪：
| 模式 (mode) | 噪声类型 | 典型场景 |
|---|---|---|
NSNG |
非稳态噪声 | 交通声、旁人说话、键盘敲击 |
STATIONARY_NS |
稳态噪声 | 空调声、风扇声、电流声 |
| 等级 (level) | 效果 | 代价 |
|---|---|---|
SOFT |
轻度降噪，保留语音细节 | 语音失真极小（推荐） |
AGGRESSIVE |
强力降噪 | 可能轻微影响语音自然度 |
⚠️ 注意：AI 降噪模块依赖 Web Worker，需要在构建工具中正确配置 Worker 加载路径。
七、具体实现：三大场景实战
7.1 Web 前端实现
环境准备：
npm install @coze/api
# 或者使用实时通信专用包
npm install @coze/realtime-api
完整实现代码：
import {
RealtimeClient,
EventNames,
RealtimeUtils
} from "@coze/realtime-api";
// ========================================
// 第一步：检查设备权限
// ========================================
async function checkPermissions() {
const result = await RealtimeUtils.checkDevicePermission();
if (!result.audio) {
throw new Error("请授予麦克风访问权限");
}
}
// ========================================
// 第二步：初始化客户端
// ========================================
const client = new RealtimeClient({
// 基础配置
baseURL: "https://api.coze.cn", // 国内站
// baseURL: "https://api.coze.com", // 国际站
accessToken: "pat_xxxxxxxxxxxxx", // 个人访问令牌
botId: "your_bot_id", // 扣子 Bot ID
// 语音配置
voiceId: "your_voice_id", // TTS 音色
debug: true, // 开启调试日志
// 降噪配置
suppressStationaryNoise: true, // 抑制稳态噪声
suppressNonStationaryNoise: false, // 不抑制非稳态噪声
// RTC 房间信息获取（使用火山引擎 RTC）
getRoomInfo: async () => {
// 通过你的后端 API 获取 RTC Token
const response = await fetch("/api/coze/room-token");
return response.json();
// 返回格式: { token, uid, room_id, app_id }
},
});
// ========================================
// 第三步：注册事件监听
// ========================================
// 连接成功
client.on(EventNames.CONNECTED, () => {
console.log("✅ 语音对话已建立");
});
// 断开连接
client.on(EventNames.DISCONNECTED, () => {
console.log("❌ 连接已断开");
});
// 对话状态变化
client.on(EventNames.CONVERSATION_CHAT_IN_PROGRESS, () => {
console.log("🤔 AI 正在思考...");
});
// 文本字幕流（逐字/逐句推送）
client.on(EventNames.CONVERSATION_MESSAGE_DELTA, (_, event) => {
console.log("📝 AI 说:", event.data.content);
});
// 语音转写结果
client.on(EventNames.CONVERSATION_AUDIO_TRANSCRIPT_UPDATE, (_, event) => {
console.log("🎤 识别到:", event.data.content);
});
// AI 语音开始播放
client.on(EventNames.CONVERSATION_AUDIO_STARTED, () => {
console.log("🔊 AI 开始说话");
});
// AI 语音播放完毕
client.on(EventNames.CONVERSATION_AUDIO_COMPLETED, () => {
console.log("🔇 AI 说完");
});
// 对话回合完成
client.on(EventNames.CONVERSATION_CHAT_COMPLETED, () => {
console.log("✅ 对话回合结束");
});
// 错误处理
client.on(EventNames.SERVER_ERROR, (_, event) => {
console.error("❌ 错误:", event.data.msg);
});
// ========================================
// 第四步：连接并开始对话
// ========================================
async function startConversation() {
try {
await checkPermissions();
await client.connect();
console.log("开始对话——直接说话即可打断 AI");
// Server VAD 模式下，会自动检测语音和打断
// 无需额外操作，直接说话即可！
} catch (error) {
console.error("连接失败:", error);
}
}
// ========================================
// 交互控制
// ========================================
// 打断 AI（Client Interrupt 模式下使用）
function interruptBot() {
client.interrupt();
console.log("⏹️ 已打断 AI");
}
// 静音/取消静音
function toggleMute() {
const muted = client.toggleAudioEnable();
console.log(muted ? "🔇 已静音" : "🎤 已开启麦克风");
}
// 调节 AI 音量
function setVolume(volume: number) {
client.setPlaybackVolume(volume); // 0.0 ~ 1.0
}
// 切换输入设备
async function switchMicrophone() {
const devices = await RealtimeUtils.getAudioInputDevices();
await client.setAudioInputDevice(devices[0].deviceId);
}
// ========================================
// 第五步：断开连接
// ========================================
async function endConversation() {
await client.disconnect();
console.log("对话结束");
}
// 启动
startConversation();
7.2 Python SDK 实现
环境准备：
pip install cozepy
完整异步实现：
import asyncio
import pyaudio
import wave
from cozepy import AsyncCoze, AsyncTokenAuth, COZE_CN_BASE_URL
from cozepy.websockets.chat import (
AsyncWebsocketsChatEventHandler,
AsyncWebsocketsChatClient,
AsyncWebsocketsChatCreatedEvent,
AsyncWebsocketsConversationMessageDeltaEvent,
AsyncWebsocketsConversationAudioDeltaEvent,
AsyncWebsocketsConversationChatCompletedEvent,
)
# ========================================
# 第一步：定义事件处理器
# ========================================
class VoiceChatHandler(AsyncWebsocketsChatEventHandler):
def __init__(self):
self.transcript = "" # 用户语音识别结果
self.ai_response = "" # AI 文字回复
self.audio_chunks = [] # AI 语音数据
self.is_ai_speaking = False # AI 是否正在说话
async def on_chat_created(self, cli: AsyncWebsocketsChatClient,
event: AsyncWebsocketsChatCreatedEvent):
"""会话创建成功"""
print(f"✅ 会话建立: {event.data.chat_id}")
async def on_conversation_message_delta(
self, cli: AsyncWebsocketsChatClient,
event: AsyncWebsocketsConversationMessageDeltaEvent):
"""AI 回复文本流（Delta）"""
self.ai_response += event.data.content
print(event.data.content, end="", flush=True)
async def on_conversation_audio_delta(
self, cli: AsyncWebsocketsChatClient,
event: AsyncWebsocketsConversationAudioDeltaEvent):
"""AI 语音音频帧"""
self.audio_chunks.append(event.data.audio)
async def on_conversation_chat_completed(
self, cli: AsyncWebsocketsChatClient,
event: AsyncWebsocketsConversationChatCompletedEvent):
"""对话回合完成"""
print(f"\n✅ 对话完成")
self.is_ai_speaking = False
# ========================================
# 第二步：主流程
# ========================================
async def main():
# 初始化 Coze 客户端
coze = AsyncCoze(
auth=AsyncTokenAuth("pat_xxxxxxxxxxxxxxxxx"),
base_url=COZE_CN_BASE_URL, # 国内站
)
handler = VoiceChatHandler()
# 创建 WebSocket 语音聊天客户端
chat = coze.websockets.chat.create(
bot_id="your_bot_id",
on_event=handler,
)
async with chat() as client:
# ---- 配置音频参数 ----
await client.chat_update({
"event_type": "chat.update",
"data": {
"input_audio": {
"format": "pcm",
"codec": "pcm",
"sample_rate": 16000,
"channel": 1,
},
"output_audio": {
"codec": "opus",
"opus_config": {
"sample_rate": 24000,
"bitrate": 48000,
},
},
"turn_detection": {
"type": "server_vad", # 使用服务端 VAD
"prefix_padding_ms": 600,
"silence_duration_ms": 500,
},
},
})
print("🎤 开始对话... (按 Ctrl+C 退出)")
# ---- 音频采集与发送 ----
CHUNK = 3200 # 200ms @ 16kHz
FORMAT = pyaudio.paInt16
CHANNELS = 1
RATE = 16000
p = pyaudio.PyAudio()
stream = p.open(
format=FORMAT,
channels=CHANNELS,
rate=RATE,
input=True,
frames_per_buffer=CHUNK,
)
try:
while True:
# 从麦克风读取音频
audio_data = stream.read(CHUNK, exception_on_overflow=False)
# 发送给扣子服务端
await client.input_audio_buffer_append(audio_data)
# 短暂休眠，避免过度占用 CPU
await asyncio.sleep(0.01)
except KeyboardInterrupt:
print("\n👋 对话结束")
finally:
stream.stop_stream()
stream.close()
p.terminate()
# ========================================
# 启动
# ========================================
if __name__ == "__main__":
asyncio.run(main())
7.3 ESP32 嵌入式实现
扣子官方提供了 ESP-COZE 组件，可直接在 ESP-IDF 环境中集成，支持 ESP32 / ESP32-S3 / ESP32-P4 等芯片。
环境准备：
# 在 ESP-IDF 项目中添加依赖
idf.py add-dependency "espressif/esp_coze^1.0.0"
CMakeLists.txt：
cmake_minimum_required(VERSION 3.5)
include($ENV{IDF_PATH}/tools/cmake/project.cmake)
# 主组件需声明对 esp_coze 的依赖
idf_component_register(
SRCS "main.c"
REQUIRES esp_coze
)
project(coze_voice_device)
main.c 核心代码：
#include <stdio.h>
#include "esp_log.h"
#include "esp_coze_api.h"
#include "esp_coze_audio.h"
static const char *TAG = "coze_voice";
// Coze 配置
#define COZE_TOKEN "pat_xxxxxxxxxxxxxxxxx" // 访问令牌
#define COZE_BOT_ID "your_bot_id" // Bot ID
#define COZE_VOICE_ID "voice_id" // 音色ID
// 音频配置 (适配 ESP32-S3)
#define AUDIO_SAMPLE_RATE 16000
#define AUDIO_FRAME_MS 60 // 60ms帧
#define AUDIO_CHUNK_SIZE (AUDIO_SAMPLE_RATE * 2 * AUDIO_FRAME_MS / 1000)
// 回调：接收到 AI 语音数据
static void on_audio_delta(uint8_t *data, size_t len, void *user_data) {
ESP_LOGI(TAG, "收到 AI 语音: %d 字节", len);
// 将 Opus 数据解码并通过 I2S 播放
esp_coze_audio_play(data, len);
}
// 回调：对话状态变化
static void on_chat_status(esp_coze_chat_status_t status, void *user_data) {
switch (status) {
case COZE_CHAT_STATUS_IDLE:
ESP_LOGI(TAG, "💤 空闲等待");
break;
case COZE_CHAT_STATUS_LISTENING:
ESP_LOGI(TAG, "🎤 聆听中...");
break;
case COZE_CHAT_STATUS_THINKING:
ESP_LOGI(TAG, "🤔 思考中...");
break;
case COZE_CHAT_STATUS_SPEAKING:
ESP_LOGI(TAG, "🔊 AI 说话中...");
break;
}
}
void app_main(void) {
ESP_LOGI(TAG, "=== 扣子语音设备启动 ===");
// 1. 初始化 Wi-Fi
// wifi_init_sta(); // 略，根据实际情况实现
// 2. 初始化音频硬件 (I2S 麦克风 + 扬声器)
esp_coze_audio_init_t audio_cfg = {
.sample_rate = AUDIO_SAMPLE_RATE,
.chunk_size = AUDIO_CHUNK_SIZE,
.input_gain = 1.0f, // 麦克风增益
.output_volume = 0.7f, // 扬声器音量
};
esp_coze_audio_init(&audio_cfg);
// 3. 初始化 Coze 客户端
esp_coze_config_t coze_cfg = {
.token = COZE_TOKEN,
.bot_id = COZE_BOT_ID,
.voice_id = COZE_VOICE_ID,
.ws_url = "wss://ws.coze.cn", // WebSocket 地址
.audio_format = COZE_AUDIO_PCM, // 输入格式 PCM
.output_codec = COZE_OUTPUT_OPUS, // 输出格式 Opus
.interaction_mode = COZE_MODE_CONTINUOUS, // 连续对话模式
.vad_config = {
.min_speech_ms = 64, // 最小语音片段 64ms
.min_silence_ms = 1000, // 最小静音 1000ms
},
.callback = {
.on_audio_delta = on_audio_delta,
.on_status_change = on_chat_status,
},
};
esp_err_t ret = esp_coze_init(&coze_cfg);
if (ret != ESP_OK) {
ESP_LOGE(TAG, "Coze 初始化失败: %d", ret);
return;
}
// 4. 建立 WebSocket 连接
ret = esp_coze_connect();
if (ret != ESP_OK) {
ESP_LOGE(TAG, "WebSocket 连接失败: %d", ret);
return;
}
ESP_LOGI(TAG, "✅ 已连接，开始对话！");
// 5. 主循环：持续采集麦克风数据并发送
uint8_t audio_buffer[AUDIO_CHUNK_SIZE];
while (true) {
// 从 I2S 麦克风读取
size_t bytes_read = esp_coze_audio_read(audio_buffer, AUDIO_CHUNK_SIZE);
if (bytes_read > 0) {
// 发送到扣子服务端
esp_coze_send_audio(audio_buffer, bytes_read);
}
vTaskDelay(pdMS_TO_TICKS(AUDIO_FRAME_MS));
}
}
三种交互模式选择：
typedef enum {
COZE_MODE_CONTINUOUS, // 连续对话：始终聆听，VAD 自动检测
COZE_MODE_WAKE_WORD, // 唤醒词模式：说唤醒词后开始对话
COZE_MODE_PUSH_BUTTON, // 按键模式：按下按钮开始对话
} esp_coze_interaction_mode_t;
💡 提示：ESP32 方案内置了完整的 3A 音频前端处理（AEC 回声消除 + AGC 自动增益 + NS 噪声抑制），无需额外实现。
八、完整事件流序列图
以下是 Server VAD 模式下，一次完整语音对话的 WebSocket 事件流：
时间轴 客户端 服务端
│
│──────── CHAT_UPDATE ────────────────────────→│ 配置音频参数
│ { input_audio: { format: pcm, │
│ sample_rate: 48000 }, │
│ output_audio: { codec: opus }, │
│ turn_detection: { │
│ type: server_vad, │
│ silence_duration_ms: 500 } } │ ← 参数协商
│ │
│←─────── CHAT_CREATED ──────────────────────│ 会话创建成功
│ { chat_id: "chat_xxx" } │
│ │
│──────── INPUT_AUDIO_BUFFER_APPEND ─────────→│ 持续发送音频帧
│──────── INPUT_AUDIO_BUFFER_APPEND ─────────→│ (每 20-60ms 一帧)
│──────── INPUT_AUDIO_BUFFER_APPEND ─────────→│
│ │
│←─────── INPUT_AUDIO_BUFFER_SPEECH_STARTED ─│ VAD 检测到语音开始
│ │ (自动清空播放队列)
│ │
│──────── INPUT_AUDIO_BUFFER_APPEND ─────────→│ 继续发送音频
│ ... (用户持续说话) ... │
│ │
│←─────── INPUT_AUDIO_BUFFER_SPEECH_STOPPED ─│ 静音 > 500ms，判定结束
│ │
│ ╔══════════════════╗ │
│ ║ ASR: 语音 → 文本║ │ 开始处理
│ ╚══════════════════╝ │
│ ╔══════════════════╗ │
│ ║ LLM: 理解 + 生成║ │
│ ╚══════════════════╝ │
│ ╔══════════════════╗ │
│ ║ TTS: 文本 → 语音║ │
│ ╚══════════════════╝ │
│ │
│←─────── CONVERSATION_CHAT_IN_PROGRESS ──────│ 开始推送回复
│←─────── CONVERSATION_MESSAGE_DELTA ─────────│ 文本字幕: "你好"
│←─────── CONVERSATION_MESSAGE_DELTA ─────────│ 文本字幕: "我是"
│←─────── CONVERSATION_AUDIO_DELTA ───────────│ 音频帧 1
│←─────── CONVERSATION_MESSAGE_DELTA ─────────│ 文本字幕: "扣子"
│←─────── CONVERSATION_AUDIO_DELTA ───────────│ 音频帧 2
│ ... (音频 + 字幕持续推送) ... │
│←─────── CONVERSATION_AUDIO_COMPLETED ───────│ 语音播放完毕
│←─────── CONVERSATION_CHAT_COMPLETED ────────│ 对话回合结束
│ │
│──────── INPUT_AUDIO_BUFFER_APPEND ─────────→│ (下一轮开始...)
│ (用户直接说话即可打断) │
Client Interrupt 模式下的差异：
│──────── startRecord() ────────────────────→│
│──────── INPUT_AUDIO_BUFFER_APPEND ─────────→│ 开始录音
│──────── INPUT_AUDIO_BUFFER_APPEND ─────────→│
│-------- stopRecord() ───────────────────→│ 结束录音
│──────── INPUT_AUDIO_BUFFER_COMPLETE ───────→│
│ │ 开始 ASR → LLM → TTS
│ │
│ (如果用户在 AI 说话时按下打断按钮) │
│──────── interrupt() ──────────────────────→│ ← 打断 AI
│←─────── CONVERSATION_AUDIO_COMPLETED ───────│ 立即停止播放
九、常见问题与调优建议
9.1 延迟优化
| 优化方向 | 方法 |
|---|---|
| 首字延迟 | 启用 LLM 流式输出，TTS 边接收边合成 |
| 采集延迟 | 减小音频帧长（20ms 代替 60ms），但需平衡带宽 |
| 网络延迟 | 就近接入（国内用 ws.coze.cn，国际用 ws.coze.com） |
| 播放延迟 | 使用 AudioWorklet 而非 ScriptProcessorNode |
9.2 打断体验调优
// 精细控制 VAD 参数
{
turn_detection: {
type: "server_vad",
prefix_padding_ms: 600, // 开头缓冲：防止切断首字
silence_duration_ms: 500, // 静音阈值：越小越灵敏，但也容易误判
}
}
silence_duration_ms太大 → 用户停下后等很久 AI 才回应silence_duration_ms太小 → 用户喘口气就把话切断了- 推荐值：中文对话 500~800ms，英文对话 300~600ms
9.3 带宽与音质平衡
场景 推荐编码 采样率 码率
──────────────────────────────────────────────────────
WiFi + Web 浏览器 PCM 48kHz ~768 kbps
4G 移动网络 Opus 24kHz ~48 kbps
4G 模组 + ESP32 Opus 16kHz ~16 kbps
弱网 / 卫星通信 Opus 8kHz ~8 kbps
9.4 常见报错排查
| 错误信息 | 原因 | 解决 |
|---|---|---|
MSys/Mingw is no longer supported |
在 Git Bash 中运行 ESP-IDF | 换用 PowerShell 或 CMD |
Could not open COM3 |
ESP32 未连接或驱动问题 | 按住 BOOT 键后按 RST 进入下载模式 |
no serial data received |
USB 线仅充电无数据 | 更换数据线 |
token expired |
访问令牌过期 | 在扣子控制台重新生成 PAT |
bot not found |
Bot ID 错误 | 检查 Bot 发布状态和 ID |
CM_PROB_PHANTOM |
CH340 幽灵设备（曾经连接现在断开） | 重新插拔 USB |
9.5 中断打断状态机
┌──────────────┐
startRecord │ IDLE │ interrupt()
┌──────────▶│ (空闲) │◀──────────┐
│ └──────┬───────┘ │
│ │ stopRecord() │
│ ▼ │
│ ┌──────────────┐ │
│ │ LISTENING │ │
│ │ (聆听) │ │
│ └──────┬───────┘ │
│ │ ASR→LLM→TTS │
│ ▼ │
│ ┌──────────────┐ │
└───────────│ SPEAKING │───────────┘
interrupt() │ (AI说话) │ interrupt()
└──────────────┘
十、总结与展望
核心要点回顾
-
WebSocket 全双工是基础：不同于 HTTP 的请求-响应模型，WebSocket 让语音交互真正实现了《Her》式的自然对话体验
-
ASR-LLM-TTS 三层流水线：每层都可以独立优化，流式处理是降低端到端延迟的关键
-
Turn Detection 是灵魂：Server VAD 适合免提，Client Interrupt 适合移动端和嵌入式设备
-
三种实现路径：
- Web SDK（
@coze/realtime-api）：功能最完整，适合浏览器场景 - Python SDK（
cozepy）：适合服务端集成和原型验证 - ESP-COZE（
espressif/esp_coze）：适合 IoT 硬件设备
- Web SDK（
-
音频管道要重视：降噪、编解码、设备管理直接影响用户体验
未来趋势
- 语音克隆：少量录音即可生成高相似度的个性化声音（>95%）
- 多模态融合：语音 + 视觉 + 触觉的多通道并行交互
- 边缘推理：ASR/TTS 模型下沉到端侧，进一步降低延迟
- 情感计算：识别用户情绪并在语音合成中动态调整情感表达
语音交互正在从"能用"走向"好用"，扣子 WebSocket 方案为我们打开了一扇通往《Her》世界的大门。无论是 Web 应用、移动 App 还是 ESP32 智能硬件，都能通过同一套协议接入大模型语音能力。
参考资料
- 扣子官方文档 - WebSocket Chat SDK
- coze-js GitHub 仓库
- coze-py GitHub 仓库
- ESP-COZE 组件
- 扣子 RTC 实时语音 OpenAPI
- @coze/realtime-api npm 包
作者注：本文基于扣子 WebSocket SDK v1.0+ 和 ESP-COZE v1.0.0 版本编写。代码示例仅供参考，实际使用时请替换为您自己的 Token 和 Bot ID。如有问题欢迎在评论区留言交流。
发布于 2026年6月 · 转载请联系作者
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
9
0- 0
已为社区贡献1条内容
所有评论(0)