真实SDK会话、打断与服务端历史截断
前置是第一章的PCM格式和本章轮次身份。现在把同样的状态边界放进真实SDK与本地WebSocket连接：客户端序列化事件，服务端收到JSON，返回经SDK模型校验的协议事件。服务端是故障夹具，不生成语音，不代表云端模型质量。
python -m pip install 'openai[realtime]==2.20.0' 'websockets==15.0.1'一个打断包含三个动作
先使旧response失效并清空播放队列，再请求服务端取消生成，最后用实际播放样本数截断assistant item。取消只影响继续生成；已经发到客户端但没播放的内容，必须从服务端可见历史中截掉，否则后续模型会以为用户已经听到它。response_id与item_id不是同一个身份，截断还指向content_index。
保存 realtime_socket_lab.py，运行 python realtime_socket_lab.py。端口由系统分配；不会连接OpenAI。播放设备回执是明确的20ms夹具，网络和SDK是真实运行。
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']==20
await emit(ws,dict(delta,event_id='late'))
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
# 设备回执夹具：只播出480样本(20ms)，其余仍在缓冲。
played_samples=480
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())收到100ms音频只确认播出20ms；因此truncate发送20，而非收到的100。打断后服务端故意再发旧response的音频，客户端因active已经清空而拒收，最后cancelled终态也不会把旧缓冲重新激活。状态图可以写成连接→监听→生成/播放→打断→监听；工具操作的业务状态独立于这个播放状态。
本例使用手动commit，同时注入speech_started测试打断分派器；这不是声学VAD实测。开启服务端VAD时，commit/自动response策略按该会话配置管理，不能既让服务端自动创建回应又每次额外response.create。静音阈值过短会抢话，过长会增加端点等待；背景音乐、回声和重叠说话需要真实录音的分组评估。
用自己的录音连接真实服务
以下独立 online_audio.py 需要 OPENAI_API_KEY、账户可用的 OPENAI_REALTIME_MODEL 和一段本人有权使用的24kHz单声道16bit WAV speech.wav，会消耗服务额度。本次未执行云端调用。它是手动提交的一轮录音，输出写成WAV，不冒充麦克风全双工播放器。
import asyncio,base64,os,wave
from openai import AsyncOpenAI
async def main():
with wave.open('speech.wav','rb') as wav:
if (wav.getnchannels(),wav.getsampwidth(),wav.getframerate())!=(1,2,24000):
raise ValueError('need mono PCM16 WAV at 24 kHz')
raw=wav.readframes(wav.getnframes())
if not 4800<=len(raw)<=24000*2*30:
raise ValueError('use 0.1 to 30 seconds of audio')
output=bytearray();done=False
async with AsyncOpenAI() as client:
async with client.realtime.connect(model=os.environ['OPENAI_REALTIME_MODEL']) as conn:
await conn.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}}}})
async with asyncio.timeout(90):
for start in range(0,len(raw),4800):
await conn.input_audio_buffer.append(audio=base64.b64encode(raw[start:start+4800]).decode())
await conn.input_audio_buffer.commit();await conn.response.create()
async for event in conn:
if event.type=='response.output_audio.delta':
output.extend(base64.b64decode(event.delta))
if len(output)>24000*2*60:raise ValueError('output audio budget')
elif event.type=='error':raise RuntimeError(event.error.message)
elif event.type=='response.done':
if event.response.status!='completed':raise RuntimeError(event.response.status)
done=True;break
if not done or not output:raise RuntimeError('no completed audio')
with wave.open('reply.wav','wb') as wav:
wav.setparams((1,2,24000,len(output)//2,'NONE','not compressed'))
wav.writeframes(output)
asyncio.run(main())文件批送不要求伪装成实时采集速度；实际设备流则按设备时钟送入并同时消费服务器事件，防止缓冲无限堆积。超时退出上下文关闭连接，不代表外部工具副作用回滚。增量转录在final前仍可修订，界面应区分临时字幕与最终文字；不要把每段转录都追加成独立用户命令。
维护旧会话形状
历史beta示例可能使用顶层modalities、input_audio_format和response.audio.delta；本例固定版本使用output_modalities、嵌套audio配置和response.output_audio.delta。先录下旧环境真实fixture，再做边界映射并回归；仅重命名事件不能证明取消、usage和历史语义兼容。翻译专用或新全双工接口也可能没有相同turn生命周期，不能把其事件混入本例分派器。
练习
将设备回执改为960样本，服务端应看到40ms。再让夹具发送两段迟到旧音频，仍只接受第一段。最后把done改为failed，应用必须报告失败，不能保留“已完成”状态；保留一个正常completed分支与一个cancelled分支的不同产品含义。