把 Eleven Realtime 接进客服外呼，第一件要跑通的不是话术，而是两件能被观测到的事：用户说完到听见第一个字，这段耗时究竟落在哪一段；以及机器正在播报时用户插话，服务端是否真的停止了音频下发。这两件事都能用一条最小双向流式链路加日志打点测出来，不必先搭完整业务系统。拿到可复算的日志后，再拿数字去对业务能容忍的阈值，决定是否扩量。下面所有步骤都按可复现、可核验写，阈值和结论由你用自己环境的记录填。

先用一条最小双向流式链路把首字延迟拆成四个时间戳：采集结束、请求发出、首个音频包、首个可听字。再用相邻差值判断耗时压在采集、上行、模型还是下行播放。打断逻辑则用固定时机回灌用户语音，只认服务端是否停止下发音频帧这个可观测事实。两项都拿到反复可算的日志后再谈扩量，单次通话的偶发结果不作为依据。

搭一条最小双向流式通话链路

先确认基础通路：能连上、能把本地音频按固定分片发出去、能把返回的音频写进文件并回放。这条链路不通，后面所有延迟数字都不成立。骨架用通用 WebSocket 形态写，地址、鉴权、会话初始化消息类型全部用占位符，接入时逐项替换；分片长度按采集侧实际帧长定，20ms 一帧是常见做法，具体要结合你的采样率和编码确认。

import asyncio, json, time, websockets

URL = '{REALTIME_WS_URL}?model={MODEL}'        # 占位符，按实际服务替换
HEADERS = {'Authorization': 'Bearer {API_KEY}'}
CALL_ID = '{CALL_ID}'
OUT = 'recv_' + CALL_ID + '.pcm'

def log(event, **kw):
    print(json.dumps({'event': event, 'call_id': CALL_ID,
                      'ts_mono': time.monotonic(),
                      'ts_ms': int(time.time() * 1000), **kw},
                     ensure_ascii=False))

async def run(chunks, chunk_ms=20):
    log('conn_start')
    async with websockets.connect(URL, extra_headers=HEADERS, max_size=None) as ws:
        log('conn_open')
        await ws.send(json.dumps({'type': '{SESSION_INIT}',
                                  'audio': {'format': 'pcm16', 'rate': 16000}}))
        log('session_init_sent')

        async def sender():
            for i, c in enumerate(chunks):
                if i == 0:
                    log('chunk_first_sent')
                await ws.send(c)
                if i % 25 == 0:
                    log('chunk_progress', seq=i)
                await asyncio.sleep(chunk_ms / 1000)
            log('chunk_last_sent', seq=len(chunks) - 1)

        async def receiver():
            n = 0
            with open(OUT, 'wb') as f:
                while True:
                    msg = await ws.recv()
                    if isinstance(msg, bytes):
                        if n == 0:
                            log('first_audio_pkt', size=len(msg))
                        n += 1
                        f.write(msg)
                    else:
                        evt = json.loads(msg)
                        log('srv_event', type=evt.get('type'))
                        if evt.get('type') in ('{DONE}', '{ERROR}'):
                            break
            log('recv_end', frames=n)

        await asyncio.gather(sender(), receiver())

asyncio.run(run([b'\x00\x00' * 320] * 50))   # 占位音频，换成真实采集分片

每步都留了日志点：conn_start 与 conn_open 看握手和鉴权耗时，session_init_sent 看会话初始化是否被接受，chunk_first_sent 与 chunk_last_sent 圈定上行窗口，first_audio_pkt 是下行起点，recv_end 记总帧数。验证方式：OUT 落盘文件能被播放器回放出可辨识语音、且 recv_end.frames 大于 0；否则先查鉴权、会话初始化消息格式和音频格式声明，不要继续往后测延迟。

把首字延迟拆成四段打点

四个时间戳必须在同一进程里用单调时钟取，墙钟仅作展示，避免系统对时跳变把差值算错。字段建议统一为 ts_capture_end_ms、ts_request_sent_ms、ts_first_audio_pkt_ms、ts_first_audible_ms，用 call_id 加 turn_id 归到同一回合。

• capture_end：上行最后一帧进入发送队列的时刻，在采集回调里打，不要在发送循环里打。
• request_sent：首个上行音频帧真正写出连接的时刻，用来吃掉排队和序列化开销。
• first_audio_pkt：首个返回二进制音频包到达客户端的时刻。
• first_audible：播放侧首个超过能量阈值的帧，或 ASR 首个带文本的 partial——二选一，全程保持一致，换判据前先标注。

import json, sys

rows = [json.loads(l) for l in open(sys.argv[1], encoding='utf-8')]
turn = {}
for r in rows:
    if 'turn_id' in r and 'event' in r:
        turn.setdefault(r['turn_id'], {})[r['event']] = r['ts_mono']

def pct(xs, p):
    xs = sorted(xs)
    return xs[min(len(xs) - 1, int(round(p / 100 * (len(xs) - 1))))]

up, model, down, total = [], [], [], []
for _, m in turn.items():
    try:
        a, b = m['capture_end'], m['request_sent']
        c, d = m['first_audio_pkt'], m['first_audible']
    except KeyError:
        continue                      # 缺点位的回合不参与统计，先补齐打点
    up.append(b - a); model.append(c - b)
    down.append(d - c); total.append(d - a)

for name, xs in (('up', up), ('model', model), ('down', down), ('total', total)):
    if xs:
        print(name, 'n=%d p50=%d p90=%d max=%d' % (len(xs), pct(xs, 50), pct(xs, 90), max(xs)))

看结果再定归因方向：up 偏大先查采集分片长度和上行带宽，model 偏大查会话参数与会话初始化是否每回合重做，down 偏大查客户端缓冲和播放起播策略。若上行与下行不在同一台机器上，四段差值不能直接相减，需要先各自夹住本机时钟再对齐。

在播报中注入用户语音测打断

注入时点不要随机：等客户端收满固定数量的下行音频包后再回灌，例如第 N 包（N 放进配置），回灌用一段固定的 PCM 文件，保证每轮输入一致、失败可复现。判定标准只看一件事：注入之后服务端是否停止下发音频帧，而不是客户端把音量调小或本地丢弃。

INJECT_AFTER_PKT = 25          # 播报收满这么多包后注入，按环境调
injected = False

# 接在上面 receiver 的 bytes 分支里
if n == INJECT_AFTER_PKT and not injected:
    injected = True
    log('inject_start', frames=n)
    await ws.send(open('user_talkback.pcm', 'rb').read())
    # 之后继续记录：frames、last_audio_ts、是否收到中断类事件

观测点固定三个：注入时刻 inject_start、注入后继续收到的音频帧数与最后一帧时间戳 last_audio_ts、服务端事件里中断或截断类消息的到达时刻。失败时的典型日志特征通常是：收到中断事件后仍有成批 audio 帧到达；帧数一路涨到播报文本自然结束；last_audio_ts 明显晚于注入时刻；或者完全没收到中断类事件，但客户端侧已经把播报播完。把这些日志特征记下来，比记“打断失败”四个字有用得多。打断能力要按场景区分——用户短促应答和长句插话可能走不同路径，两类都要各测一轮。

失败态覆盖：断网、长静音、超长输入

这三类场景的作用是摸清链路行为边界，不是验证稳定性，因此只记录观察到的现象，不写推测性结论。构造方式：断网用本机防火墙丢包或直接杀掉网络命名空间；长静音发送全零 PCM 若干秒；超长输入连续发送超出业务常见长度的音频。每种场景跑一次，把字段填进下表。

表格留空是有意的：同一份代码在不同网络和账号配额下表现可能不同，结论应以你自己跑出来的日志为准。填完后重点看两件事——异常发生后链路是“明确报错”还是“静默卡住”，以及日志里能否凭 call_id 还原出完整时间线。

用实测数字决定是否扩量

阈值从业务侧来：外呼场景里用户对沉默的容忍、话术节奏允许的响应窗口，应由产品与客服运营确认，工程侧不要自己定。工程侧负责的是把指标测准，并如实填进下面这张判定表；任一项没通过就先修该项，不要用客户端静音、队列丢弃这类止血手段掩盖问题。

指标                 采集方式                         实测值   业务阈值   判定
首字延迟 p50        由四段日志复算 total
首字延迟 p90        由四段日志复算 total
上行段 up            request_sent - capture_end
模型段 model         first_audio_pkt - request_sent
下行段 down          first_audible - first_audio_pkt
打断停播延迟         注入时刻到最后一帧音频
打断成功率           停止下发次数 / 注入次数
断网行为是否明确      失败态表格第 1 行
长静音行为是否明确    失败态表格第 2 行
超长输入行为是否明确  失败态表格第 3 行

建议的扩量前提是：分段耗时能定位到单一瓶颈，且不是采集端抖动造成；打断在后端可观测地停止下发，而不只是客户端听不见；三类失败态都有明确日志表现，能凭日志判断发生了什么。三条同时满足再谈扩量，只满足一两条时，扩量会把偶发问题放大成客服现场的口碑问题。