数字人客服真正的难点不在"嘴型对不对"，而在两件事：用户说到一半时能不能停下来听，以及说完之后能不能真的去查订单、改地址、发验证码。这篇教程从零搭一条可跑通的链路：音色与形象绑定 → 实时音视频流 → 语音打断 → 业务动作调用。
按下面的步骤做完，你会得到一个本地可运行的服务端 + 浏览器前端：对着麦克风说话，页面上出现一个会说话的数字人；你中途插话，它立刻闭嘴并转入倾听；你说"帮我查一下订单 12345"，它会调用你写的查询接口，然后用数字人形象把结果念出来。
文中涉及的模型 ID、音色 ID、形象 ID、WebSocket 地址、字段名，都请以官方文档当前版本为准。不同账号权限下能拿到的形象资源也不一样。
前置条件清单
- 一个已开通 Live / Live Avatar 能力的账号，以及可用的 API Key
- 控制台里已经创建好的形象 ID（avatar）和音色 ID（voice），二者是独立的，可以组合
- Python 3.10+ 与 Node.js 18+（前端只用浏览器原生 API 也可以，Node 主要用于工具函数侧）
- 本地麦克风与扬声器，建议戴耳机
- 一张能查看 WebSocket 流量的调试手段：浏览器 DevTools 或
wscat之类的工具 - 一个可被调用的业务接口，或者先用假数据函数顶上
装依赖：
```bash
python -m venv .venv
source .venv/bin/activate # Windows 用 .venv\Scripts\activate
pip install websockets python-dotenv sounddevice
```
包名与最小版本以官方文档为准，这里只列常用的几个。
第一步：想清楚数据链路，再动手写代码
数字人客服本质上是一条双向流：
```text
麦克风 → 16kHz PCM → WebSocket 上行
WebSocket 下行 → 24kHz PCM 音频 + 视频帧 → 浏览器播放/渲染
WebSocket 下行 → 事件（打断、工具调用、回合结束）→ 业务逻辑
```
三个关键约定先记住，后面所有坑都跟它们有关：
1. 上行音频是 16kHz、16 位、单声道、小端 PCM，不是 mp3，也不是 48kHz。
2. 下行音频是 24kHz PCM，浏览器播放前要自己转成 AudioBuffer。
3. 打断既发生在服务端，也发生在你本地。服务端会发一个"被打断"的信号，但你的播放队列必须自己清空，否则会出现"它已经停了，声音还在放"。
第二步：把形象和音色写成配置
不要把形象 ID 写死在代码里，用配置文件管起来，方便后面做 A/B。
```yaml
config.yaml
avatar:
avatar_id: "<控制台创建的形象 ID>"
voice_id: "<官方文档列出的预置音色之一>"
locale: "zh-CN"
session:
模型 ID 以官方文档当前版本为准
model: "<Live Avatar 对应的模型 ID>"
response_modalities: ["AUDIO", "VIDEO"]
persona: |
你是一名电商售后客服，名字叫小满。
规则：
1. 每次回答控制在三句话以内，口语化，不要念标点符号。
2. 需要查数据时，先说一句"我帮您查一下"，再调用工具。
3. 查不到结果时，明确说查不到，并给一个下一步建议，不要编造。
4. 用户情绪激动时先安抚，不要重复解释规则。
```
persona 这段是数字人好不好用的分水岭。写得越像"给真人客服的岗前培训"，效果越稳；写成"你是一个乐于助人的助手"，出来的就是一段段念稿。
第三步：建立会话并绑定形象
会话建立分三步：连接、发 setup、收到 setup 完成确认。在收到确认之前不要发音频，这是新手最常见的丢帧原因。
```python
live_session.py
import asyncio, base64, json, os
import websockets
from dotenv import load_dotenv
load_dotenv()
WS_URL = os.environ["LIVE_WS_URL"] # 以官方文档给出的地址为准
API_KEY = os.environ["GEMINI_API_KEY"]
class LiveAvatarSession:
def __init__(self, cfg):
self.cfg = cfg
self.ws = None
self.ready = asyncio.Event()
def _setup_payload(self):
return {
"setup": {
"model": self.cfg["session"]["model"],
"generationConfig": {
"responseModalities": self.cfg["session"]["response_modalities"],
"speechConfig": {
"voiceConfig": {
"prebuiltVoiceConfig": {
"voiceName": self.cfg["avatar"]["voice_id"]
}
}
},
},
形象相关字段以官方文档当前版本为准
"avatarConfig": {"avatarId": self.cfg["avatar"]["avatar_id"]},
"systemInstruction": {"parts": [{"text": self.cfg["persona"]}]},
"tools": [QUERY_ORDER_TOOL],
"realtimeInputConfig": {
"automaticActivityDetection": {
"disabled": False,
"prefixPaddingMs": 300,
"silenceDurationMs": 700,
}
},
}
}
async def connect(self):
self.ws = await websockets.connect(
f"{WS_URL}?key={API_KEY}",
max_size=16 * 1024 * 1024, # 视频帧很大，别用默认上限
)
await self.ws.send(json.dumps(self._setup_payload()))
while True:
msg = json.loads(await self.ws.recv())
if "setupComplete" in msg:
self.ready.set()
return
```
silenceDurationMs 是"用户停顿多久算说完"。设太小，用户换气就被判成说完；设太大，客服反应迟钝。中文对话从 600～900ms 之间开始调。
第四步：把麦克风音频推上去
```python
audio_io.py
import base64, json, sounddevice as sd
SAMPLE_RATE = 16000
BLOCK = 1024
async def pump_mic(session, stop: asyncio.Event):
loop = asyncio.get_running_loop()
with sd.InputStream(samplerate=SAMPLE_RATE, channels=1,
dtype="int16", blocksize=BLOCK) as stream:
while not stop.is_set():
data, _ = await loop.run_in_executor(None, stream.read, BLOCK)
await session.ws.send(json.dumps({
"realtimeInput": {
"audio": {
"data": base64.b64encode(data.tobytes()).decode(),
"mimeType": f"audio/pcm;rate={SAMPLE_RATE}",
}
}
}))
```
两个细节：stream.read 是阻塞的，必须丢到 executor 里，否则会卡住整个事件循环；mimeType 里的采样率要和实际采集率一致，写错了模型听到的是快放或慢放。
如果前端直接连，浏览器侧用 WebAudio 采集，注意 AudioContext 的 sampleRate 默认是 48kHz，需要自己重采样到 16kHz 再切片发送。
第五步：接收音视频并播放
服务端会把音频和视频帧混在同一个回合里下发。外层循环只做分发，播放队列单独管。
```python
async def pump_downstream(session, player, video_sink):
async for raw in session.ws:
msg = json.loads(raw)
if tc := msg.get("toolCall"):
await handle_tool_call(session, tc)
continue
sc = msg.get("serverContent")
if not sc:
continue
关键：用户插话，模型被中断
if sc.get("interrupted"):
player.flush() # 立刻清空本地待播队列
video_sink.pause_speaking()
continue
for part in sc.get("modelTurn", {}).get("parts", []):
inline = part.get("inlineData")
if not inline:
continue
payload = base64.b64decode(inline["data"])
mime = inline.get("mimeType", "")
if mime.startswith("audio/"):
player.push(payload)
elif mime.startswith("video/") or mime.startswith("image/"):
video_sink.push(payload)
if sc.get("turnComplete"):
video_sink.stop_speaking()
```
player.flush() 不能做成"淡出"。数字人说话是逐帧渲染的，淡出会让你看到嘴还在动、声音渐渐消失，观感很怪。正确做法是丢弃队列里所有未播的 buffer，并从当前时刻重新计时。
浏览器端的播放队列大致这样：
```javascript
// player.js —— 24kHz PCM16 队列播放
const SAMPLE_RATE = 24000;
class PcmPlayer {
constructor(ctx) {
this.ctx = ctx;
this.queue = [];
this.nextTime = 0;
this.sources = new Set();
}
push(int16Buffer) {
const view = new Int16Array(int16Buffer);
const f32 = new Float32Array(view.length);
for (let i = 0; i < view.length; i++) f32[i] = view[i] / 32768;
const buf = this.ctx.createBuffer(1, f32.length, SAMPLE_RATE);
buf.copyToChannel(f32, 0);
this.queue.push(buf);
this._schedule();
}
_schedule() {
while (this.queue.length) {
const buf = this.queue.shift();
const src = this.ctx.createBufferSource();
src.buffer = buf;
src.connect(this.ctx.destination);
const startAt = Math.max(this.ctx.currentTime, this.nextTime);
src.start(startAt);
this.nextTime = startAt + buf.duration;
this.sources.add(src);
src.onended = () => this.sources.delete(src);
}
}
flush() {
for (const s of this.sources) { try { s.stop(); } catch (_) {} }
this.sources.clear();
this.queue.length = 0;
this.nextTime = 0; // 从"现在"重新开始排
}
}
```
flush() 里的 nextTime = 0 这一行容易漏。漏了之后下一次 Math.max(currentTime, nextTime) 会算到一个过去的时刻，导致音频瞬间挤在一起播出去。
第六步：把打断做扎实
打断由三层组成，缺一层都会露馅。
第一层：服务端检测。 打开自动活动检测，让服务端判断用户在说话。它会下发 interrupted，这是主路径。
第二层：本地静音兜底。 网络抖动时 interrupted 可能晚到几百毫秒。本地做一个简易能量检测，音量连续超过阈值约 200ms，就先自行 player.flush()，等服务端信号到了再对齐。
```python
def is_speech(pcm_bytes, threshold=800):
import array
samples = array.array("h", pcm_bytes)
if not samples:
return False
rms = (sum(s * s for s in samples) / len(samples)) ** 0.5
return rms > threshold
```
阈值不要照抄，用 print(rms) 在你的实际环境下测一遍安静房间和说话时的数值，取中间。
第三层：回声消除。 这是打断失败的头号元凶。数字人的声音从扬声器出来又被麦克风收进去，被当成"用户在说话"，于是它一直自己打断自己。解决办法按性价比排序：戴耳机 > 开浏览器或系统的 AEC > 把麦克风增益调低 > 提高本地 VAD 阈值。
另外，打断时给一个即时的视觉反馈：嘴型停住、状态文案从"正在回答"变成"正在听"。用户看到反应了，才会继续说下去。
第七步：接业务动作
用工具调用（function calling）把业务能力挂上去。工具描述写得越像给新人看的接口文档，模型调用得越准。
```python
QUERY_ORDER_TOOL = {
"functionDeclarations": [{
"name": "query_order",
"description": "根据订单号查询订单状态与物流节点。仅在用户明确给出订单号时调用。",
"parameters": {
"type": "object",
"properties": {
"order_no": {"type": "string", "description": "纯数字订单号"},
"need_logistics": {"type": "boolean", "description": "是否需要物流轨迹"}
},
"required": ["order_no"]
}
}]
}
```
处理调用并把结果回填：
```python
async def handle_tool_call(session, tool_call):
responses = []
for call in tool_call.get("functionCalls", []):
name = call.get("name")
args = call.get("args", {}) or {}
try:
if name == "query_order":
result = await query_order(**args) # 你的真实接口
else:
result = {"error": f"unknown function: {name}"}
except Exception as e:
result = {"error": str(e)}
responses.append({
"id": call.get("id"),
"name": name,
"response": {"result": result},
})
await session.ws.send(json.dumps({"toolResponse": {"functionResponses": responses}}))
```
三个经验：
- 别让模型沉默地查数据。 在
persona里要求它先说"我帮您查一下"，再说结论。否则用户会面对 2～3 秒完全静止的数字人，很容易以为掉线了。 - 接口要加超时。 业务接口慢到 8 秒以上，体验还不如直接告诉用户"稍后回电"。
- 失败要能说出口。 返回
{"error": "订单不存在"}这种结构化结果，比抛异常让模型自己猜要好。
如果业务动作需要用户确认（比如改地址、退款），加一个二次确认的工具，让模型先复述再执行。这一步在真客服场景里省不掉。
第八步：本地联调与压测
上线前至少跑完这几项：
```bash
1. 打通链路：只开音频，确认有声音回来
python -m app.session --mode audio
2. 打断测试：让数字人念一段长文本，中途连续插话三次
python -m app.session --mode avatar --stress interrupt
3. 工具测试：构造 20 条不同口述方式的订单号（带空格、带停顿）
python -m app.session --mode avatar --stress tools
```
记录三个数字：首帧延迟（用户说完到数字人开口）、打断响应（用户插话到声音停止）、工具往返（调用到念出结果）。这三个数字决定了用户觉得"像不像真人"。
常见坑与排错
听不到声音。 九成是采样率。上行写 16k 实际采 48k，或者下行按 24k 采还当成 16k 播，都会变成怪声或静音。先打印实际采样率。
数字人自问自答。 回声进麦克风。先戴耳机复测，能解决就不用改代码。
打断后还在说话。 检查 player.flush() 是否真的清空了 sources；浏览器里 AudioBufferSourceNode.stop() 之后节点不会自动断开，但声音会停，如果你用了中间 GainNode 做淡出，记得把 gain 也复位。
工具调用循环。 模型反复调用同一个函数，通常是因为返回值里没有它需要的字段，或者返回了它读不懂的字符串。把返回结构改成扁平、字段名直白的 JSON。
长会话突然断开。 需要心跳与重连。如果服务端提供会话恢复的能力（resumption 之类的机制），把恢复句柄存下来，重连后带上，避免用户重讲一遍。具体字段以官方文档为准。
不要在前端放 API Key。 浏览器直连意味着密钥暴露。标准做法是自建一个薄代理：前端连你的服务，你的服务持有密钥并转发 WebSocket 帧。
注意计费口径。 实时音视频通常按连接时长计，不是按调用次数。用户挂断页面记得主动关闭会话，闲置超过一定时间也要服务端强制收尾。具体价格与计费规则以官方页面为准。
下一步建议
链路跑通之后，按这个顺序加固：
1. 接知识库。 把常见问题做成检索，作为工具的另一个分支，减少模型自由发挥。
2. 加对话记忆。 每个会话一份短期摘要，跨会话按用户 ID 存长期偏好，避免每次都问"您的订单号是多少"。
3. 做降级路径。 连续两次识别失败或工具超时，直接给出转人工入口。数字人客服的价值是分担，不是硬撑。
4. 埋点与质检。 记录每轮的首帧延迟、打断次数、工具调用成功率，抽样做人工听感评分。改 persona 时对比这些数字，别凭感觉调。
5. 音色与形象 A/B。 形象 ID 和音色 ID 是解耦的，同一套业务逻辑可以组合出多个版本，用真实转化数据决定保留哪个。