Eleven Realtime 的接入成本

要判断 Eleven Realtime 这类实时语音接口要投入多少人力，先不要列功能清单。更稳妥的路径是先做一个只做回声的最小通话：把音频按固定分片推上去、把回来的音频落盘、确认一轮双向流式能自然结束。跑通这一轮之后，链路、鉴权、异常处理和联调各自大概要几天，才有依据可谈；否则报价基本靠拍脑袋，联调和弱网验证往往被漏算。

适用场景：准备接入实时语音、但手上没有可对照实现的团队。操作动作：先实现只做回声的通话骨架，一轮跑通后再补业务。验证方式：能接通、能听见、能打断三条都能在日志或落盘文件里找到证据。风险边界：回声跑通只说明链路和凭据可用，不代表多并发、弱网和打断时序已经清楚，这几项仍需单独排期确认。

先把“最小可用”写成验收条目

“最小可用”如果留在口头，做的过程中一定会不断加需求，工期也就跟着失控。建议在动手前把三条验收标准写成可判定的条目，每条都要有“怎么算通过”的观察方式，而不是主观感觉。

能接通：服务端收到会话建立事件、客户端拿到可用的临时凭据、连接在日志里出现明确的建立时间点。判定方式是看日志里 session_start 与握手成功各出现一次，且返回码正常。
能听见：把回传音频写入本地文件，用播放器能正常播放，且内容与推上去的音频可对应。判定方式是落盘文件时长与输入时长在同一量级，不是零字节、也不是满屏静音。
能打断：在回传播放过程中发送一次中断动作，日志里能看到回传提前结束、状态回到可再次输入的档位。判定方式是中断前后各有一轮事件记录，且中断后仍能再发起一轮，不需要重连。

这三条只要有一条靠“听起来还行”来判，后面就一定会返工。把判定方式定成日志字段和落盘文件，验收就变成了查表。

写一个只做回声的通话骨架

骨架的目标只有一个：把音频收发链路跑通，业务逻辑一律不接。下面是一份通用结构的示例，事件名和字段名按实际接口替换，不要照抄当作官方接口使用。

# echo_call.py —— 只验证链路，不接业务逻辑 import os, json, base64, asyncio, wave, websockets CFG = { 'endpoint': 'wss://REPLACE_ME', # 服务商实时端点，按实际环境替换 'model': 'REPLACE_ME', # 模型或音色标识，按控制台取值替换 'in_rate': 16000, # 输入采样率，须与采集侧一致 'out_rate': 16000, # 输出采样率，须与回传格式一致 'chunk_ms': 20, # 分片时长，决定单帧字节数 } def b64e(b): return base64.b64encode(b).decode() def b64d(s): return base64.b64decode(s) async def send_loop(ws, wav_path): with wave.open(wav_path, 'rb') as w: n = int(CFG['in_rate'] * CFG['chunk_ms'] / 1000) while True: frames = w.readframes(n) if not frames: break await ws.send(json.dumps({'type': 'input_audio.append', 'audio': b64e(frames)})) await asyncio.sleep(CFG['chunk_ms'] / 1000) # 按实时节奏压入，不要一次性灌完 async def recv_loop(ws, out_path): with wave.open(out_path, 'wb') as w: w.setnchannels(1); w.setsampwidth(2); w.setframerate(CFG['out_rate']) async for raw in ws: evt = json.loads(raw) if evt.get('type') == 'response.audio.delta': w.writeframes(b64d(evt['delta'])) async def main(): token = os.environ['REALTIME_EPHEMERAL_TOKEN'] # 由本地服务端换取，不写进代码 headers = {'Authorization': 'Bearer ' + token} async with websockets.connect(CFG['endpoint'], extra_headers=headers) as ws: await asyncio.gather(send_loop(ws, 'in.wav'), recv_loop(ws, 'out.wav')) asyncio.run(main())

需要按实际环境替换的位置有四处：端点地址、模型或音色标识、采样率与分片时长、以及鉴权头的字段名（有的库版本用 additional_headers，按所用库的实际参数调整）。分片读取用 sleep 控制节奏是关键，直接循环把整个文件灌进去，服务端通常来不及处理，还会掩盖真实的时序问题。接收侧先落盘而不是边收边播，是为了让“能听见”这条验收有文件可查。

把鉴权与密钥放到服务端

长期密钥一旦出现在客户端代码或前端包里，就等于公开。建议的做法是：客户端不直接持有长期密钥，而是先向自己的服务端换一个短期凭据，再拿这个凭据去连实时接口。

读取方式：服务端进程从环境变量读取长期密钥，例如 os.environ['REALTIME_API_KEY']，不要写进配置文件仓库，也不要打印到启动日志。
转发调用位置：在服务端提供一个换取临时凭据的接口，这个方法就是唯一调用长期密钥的地方；客户端骨架里的 token 从这个接口拿。调用位置集中在一处，后续换密钥、加限流都只改这里。
日志脱敏处理点：在写日志之前统一过一层脱敏函数，把 Authorization 头、token 字段、密钥片段替换为固定占位符；换取凭据的请求与响应各记一条，但只记结果状态和过期时间，不记凭据本身。

验证方式是：把客户端产物搜一遍，确认搜不到长期密钥；再查看一次日志输出，确认凭据和密钥都是占位符而非原文。

记录一次完整通话的资源消耗

这一节的目的只是攒原始数据，为后面谈成本留依据，不给金额，也不做推算结论。三项必须采集：

通话时长：客户端在发送结束和接收结束各打一个时间戳，取跨度；服务端同时记录会话建立与关闭时间。两个值对不上时，差额本身也是排查线索。
上下行音频字节数：发送侧累计分片解码后的原始字节数，接收侧累计音频增量事件的字节数。统计的是音频载荷，不含协议头和编码开销，这一点要在记录里写清楚。
连接数：服务端在会话建立和关闭时对进程内计数器做增减，日志里输出当时的并发会话数。只靠客户端自报并不可靠，服务端计数才有意义。

建议统一成一行结构化日志，便于后面按通话聚合：

call_id=... evt=call_end dur_ms=... tx_audio_bytes=... rx_audio_bytes=... frames_sent=... frames_recv=... concurrent_sessions=... close_reason=...

把一次正常结束、一次主动打断、一次中途断开这三种情况各记一轮，比只记成功路径更有参考价值。

把工作量拆成可估算的条目

有了通话记录和验收条目，就可以按维度拆任务，并给每项写清验证产出。下面这份清单可以直接拿去排期，具体天数按各自环境判断。

链路：分片发送、接收落盘、节奏控制。验证产出是一个可播放的回传文件，以及发送与接收帧数基本对齐的日志。
鉴权：服务端换取短期凭据、客户端注入、过期与重取。验证产出是客户端产物中无长期密钥，且凭据过期后能自动重取并继续通话。
异常处理：连接中断重试策略、超时判定、错误码归类、半途失败时的资源释放。验证产出是构造至少三类异常（网络断开、凭据失效、服务端报错）后，每种都能在日志中定位到原因并回到可重试状态。
联调：把回调、状态事件与业务侧对接，确认打断时序与并发下的行为。验证产出是一份联调记录，包含正常、打断、并发三种场景各自的观察结果与遗留问题清单。

按这四项拆开，人力估算就不再依赖“感觉这个接口复杂不复杂”。跑通一轮回声通话的实际耗时，通常比继续开会讨论更能说明问题。