VAD、打断与陈旧事件：练习答案
参考实现
复用Player：
python
class Player:
def __init__(self):
self.generation = 0
self.queue = []
self.played_ms = 0
def start(self):
self.generation += 1
self.queue.clear()
self.played_ms = 0
return self.generation
def receive(self, generation, duration):
if generation == self.generation:
self.queue.append(duration)
def play_one(self):
if self.queue:
self.played_ms += self.queue.pop(0)
p = Player()
old = p.start(); p.receive(old,100); p.play_one()
heard = p.played_ms
new = p.start()
p.receive(old,200)
p.receive(new,50)
assert heard == 100
assert p.queue == [50]
p.play_one()
assert p.played_ms == 50
print('stale audio rejected')
p = Player()
a = p.start(); b = p.start(); c = p.start()
p.receive(a,100); p.receive(b,100)
assert p.queue == []
p.receive(c,80); p.play_one()
assert p.played_ms == 80单纯调用取消API不能保证旧事件不再到达；身份守卫是播放器自己的责任。
返回题目。
真实会话变体
下面完整文件使用本章相同SDK环境，保存 answer_session.py 运行；实际回环WebSocket验证40ms与两段迟到音频。
python
import asyncio,base64,json,struct
from openai import AsyncOpenAI
from openai.types.realtime import RealtimeServerEvent
from pydantic import TypeAdapter
from websockets.asyncio.server import serve
schema=TypeAdapter(RealtimeServerEvent)
audio=struct.pack('<h',1000)*2400 # 100ms mono PCM16 @24kHz
async def emit(ws,event):
schema.validate_python(event)
await ws.send(json.dumps(event))
async def server(ws):
requests=[json.loads(await ws.recv()) for _ in range(4)]
assert [r['type'] for r in requests]==[
'session.update','input_audio_buffer.append','input_audio_buffer.commit','response.create']
assert base64.b64decode(requests[1]['audio'])==audio
await emit(ws,{'type':'response.created','event_id':'e1',
'response':{'id':'r1','status':'in_progress'}})
delta={'type':'response.output_audio.delta','event_id':'e2','response_id':'r1',
'item_id':'assistant1','output_index':0,'content_index':0,
'delta':base64.b64encode(audio).decode()}
await emit(ws,delta)
await emit(ws,{'type':'input_audio_buffer.speech_started','event_id':'e3',
'audio_start_ms':120,'item_id':'user2'})
cancel=json.loads(await ws.recv());truncate=json.loads(await ws.recv())
assert cancel['type']=='response.cancel' and cancel['response_id']=='r1'
assert truncate['type']=='conversation.item.truncate'
assert truncate['item_id']=='assistant1' and truncate['audio_end_ms']==40
await emit(ws,dict(delta,event_id='late'))
await emit(ws,dict(delta,event_id='late2'))
await emit(ws,{'type':'response.done','event_id':'e4',
'response':{'id':'r1','status':'cancelled'}})
async def main():
async with serve(server,'127.0.0.1',0) as fixture:
port=fixture.sockets[0].getsockname()[1]
async with AsyncOpenAI(api_key='local-fixture-only',
websocket_base_url=f'ws://127.0.0.1:{port}') as client:
async with client.realtime.connect(model='local-fixture',
websocket_connection_options={'proxy':None}) as connection:
await connection.session.update(session={'type':'realtime',
'output_modalities':['audio'],'audio':{'input':{'format':{'type':'audio/pcm','rate':24000},
'turn_detection':None},'output':{'format':{'type':'audio/pcm','rate':24000}}}})
await connection.input_audio_buffer.append(audio=base64.b64encode(audio).decode())
await connection.input_audio_buffer.commit()
await connection.response.create()
accepted=0;active=None;item=None;played_samples=0;queue=[];done=False
async with asyncio.timeout(5):
async for event in connection:
if event.type=='response.created':active=event.response.id
elif event.type=='response.output_audio.delta' and event.response_id==active:
chunk=base64.b64decode(event.delta);queue.append(chunk)
item=event.item_id;accepted+=1
# 设备回执夹具：只播出960样本(40ms)，其余仍在缓冲。
played_samples=960
elif event.type=='input_audio_buffer.speech_started' and active:
old=active;active=None;queue.clear()
await connection.response.cancel(response_id=old)
if item:
await connection.conversation.item.truncate(item_id=item,content_index=0,
audio_end_ms=played_samples*1000//24000)
elif event.type=='error':raise RuntimeError(event.error.message)
elif event.type=='response.done':
assert event.response.status=='cancelled'
done=True;break
assert done and accepted==1 and queue==[]
print('real SDK socket, cancellation, truncation and late audio passed')
asyncio.run(main())失败终态与取消不是同一个产品结果。下面独立片段演示三个终态都显式处理，不把失败转成完成：
python
def terminal(status):
if status=='completed':return '完成'
if status=='cancelled':return '已中断'
raise RuntimeError('response failed: '+status)
assert terminal('completed')=='完成'
assert terminal('cancelled')=='已中断'
try:terminal('failed')
except RuntimeError:pass
else:raise AssertionError('failure accepted')设备与端点
静音块索引3、4分别结束于80、100ms，因此hangover=2在100ms发stop。纯音RMS约幅度/√2，0.01幅度低于0.05阈值；增益改变幅度，阈值不适合脱离设备固定。
保存 answer_gate.py，使用正文numpy环境独立运行：
python
import numpy as np
rate=16000;size=320 # 每块20ms
silence=np.zeros(size)
tone=.2*np.sin(2*np.pi*440*np.arange(size)/rate)
def boundaries(blocks,threshold=.05,hangover=3):
speaking=False;quiet=0;events=[]
for i,block in enumerate(blocks):
if len(block)!=size:raise ValueError('partial frame')
active=np.sqrt(np.mean(np.asarray(block,dtype=float)**2))>=threshold
if active:
quiet=0
if not speaking:events.append(('start',i*20));speaking=True
elif speaking:
quiet+=1
if quiet>=hangover:
events.append(('stop',(i+1)*20));speaking=False;quiet=0
return events
assert boundaries([silence,tone,tone,silence,silence,silence])==[('start',20),('stop',120)]
assert boundaries([silence]*6)==[]
assert boundaries([tone]*6)==[('start',0)]
print('energy gate and hangover passed')
assert boundaries([silence,tone,tone,silence,silence],hangover=2)[-1]==('stop',100)
assert boundaries([tone*.05]*6)==[]
print('endpoint variants passed')设备验收应观察track.readyState和浏览器采集指示，检查拒绝、迟到授权、停止及离开页面。纯函数测试不能代替这一手动检查；不要自动授予麦克风权限来让测试变绿。
转录归并变体
先将正文完整保存为transcript_lab.py，再把答案保存为asr_answers.py，同目录运行；导入正文不会自动打开socket。
python
from transcript_lab import Transcripts,schema,event,final,PREFIX
from pathlib import Path
import tempfile
with tempfile.TemporaryDirectory() as directory:
path=Path(directory)/'state.sqlite';s=Transcripts(path,'s1')
def send(e):return s.accept('s1',schema.validate_python(e))
send(event('input_audio_buffer.committed','c1','u1',previous_item_id=None))
send(event('input_audio_buffer.committed','c2','u2',previous_item_id='u1'))
s.cancel('u1')
assert send(final('late','u1','不要执行'))=='cancelled' and not s.ready
send(event(PREFIX+'failed','error','u2',content_index=0,
error={'type':'server_error','code':'asr_failed','message':'no speech','param':None}))
assert send(final('late2','u2','也不要执行'))=='failed'
send(event('input_audio_buffer.committed','c3','u3',previous_item_id='u2'))
send(final('f3','u3','订单42'))
assert len(s.ready)==1
try:send(final('conflict','u3','订单43'))
except ValueError as e:assert str(e)=='conflicting final'
else:raise AssertionError('conflict accepted')
s.close();s=Transcripts(path,'s1')
send(event('input_audio_buffer.committed','replay-commit','u3',previous_item_id=None))
assert send(final('replay-final','u3','订单42'))=='duplicate-final' and s.ready==[]
s.close();s=Transcripts(path,'s2')
assert s.accept('s1',schema.validate_python(final('old','u3','旧连接')))=='old-session'
s.close()
print('cancel, ASR failure, conflicting final and durable duplicate checks passed')数据库final去重只防同一连接item再次通知。新连接重复同一订票意图仍有新item，必须使用业务intent ID和参数摘要识别重复；取消字幕也不回滚已提交订单。
设备回调归属
将设备正文HTML存为device_lab.html，下面保存device_test.cjs，在同目录用Node.js20+执行node device_test.cjs。测试主动控制回调先后，验证程序资源归属；没有调用真实麦克风。
javascript
const fs=require('node:fs'),vm=require('node:vm'),assert=require('node:assert/strict');
const code=fs.readFileSync('device_lab.html','utf8').match(/<script>([\s\S]*?)<\/script>/)[1];
const nodes=Object.fromEntries(['start','stop','status','audio'].map(x=>[x,{}]));
const recordings=[],streams=[],handlers={};let pending=null,urlCount=0,revoked=0;
function makeStream(){const track={stopped:false,stop(){this.stopped=true}};const s={getTracks:()=>[track]};streams.push(s);return s}
class Recorder{constructor(s){this.stream=s;this.state='inactive';this.mimeType='audio/webm';recordings.push(this)}start(){this.state='recording'}stop(){this.state='inactive'}}
vm.runInNewContext(code,{document:{querySelector:x=>nodes[x.slice(1)]},navigator:{mediaDevices:{getUserMedia:()=>pending?pending:Promise.resolve(makeStream())}},MediaRecorder:Recorder,Blob,URL:{createObjectURL:()=>`blob:${++urlCount}`,revokeObjectURL:()=>revoked++},addEventListener:(n,f)=>handlers[n]=f});
(async()=>{
await nodes.start.onclick();nodes.stop.onclick();await nodes.start.onclick();
const newer=recordings[1];recordings[0].onstop();recordings[0].onerror();
assert.equal(newer.state,'recording');assert.equal(nodes.status.textContent,'本地录音中');
assert.equal(newer.stream.getTracks()[0].stopped,false);
nodes.stop.onclick();newer.onstop();assert.equal(urlCount,1);
let grant;pending=new Promise(resolve=>grant=resolve);const startPending=nodes.start.onclick();
nodes.stop.onclick();const late=makeStream();grant(late);await startPending;
assert(late.getTracks()[0].stopped);assert.equal(recordings.length,2);pending=null;
await nodes.start.onclick();const last=recordings[2];handlers.pagehide();last.onstop();
assert(last.stream.getTracks()[0].stopped);assert.equal(urlCount,1);assert.equal(revoked,1);
console.log('old stop/error, late permission and pagehide do not own new resources');
})();第一组让旧stop/error迟到，断言新录音仍在进行；第二组让许可Promise晚于停止，断言迟到track被释放；第三组pagehide后回调不得创建新URL。真实设备仍需手动检查权限拒绝和指示灯释放。