从临时字幕到最终输入：轮次归并与重复控制
前置是真实SDK会话和设备生命周期。一次语音输入同时有媒体、会话item和转录事件身份。网络连接换了不代表业务意图换了，文字暂时显示出来也不代表可以执行订单操作。本节先让真实SDK消费可控转录事件，再给出有条件的在线录音入口。
三种身份与两类文本
event_id去重一次事件，item_id + content_index定位哪一段音频，应用分配的connection身份隔离重连前后的事件。同一item的delta用于构造临时字幕；completed包含最终文本，应替换临时显示。delta本身是增量，本例没有虚构“revision”协议事件。故障夹具故意让最终文本与临时文本不同，检查应用不会把二者都提交。
不同轮次的completed可能乱序。input_audio_buffer.committed.previous_item_id保存对话顺序，不能用识别完成顺序推断谁先说话。识别失败、输入取消和已经final是不同状态：失败不能被晚到成功悄悄复活；已经final后的业务撤销也不能靠清空字幕实现。
实时音频模型和独立ASR可以对同一录音产生不同理解，转录只是证据之一。关键金额/时间仍需用户确认，最终转录并不自动变成授权工具参数。协议依据为官方转录指南，实验固定openai2.20.0的事件类型，未混入新版Live TranscriptGrouper接口。
真实SDK、本机通道与最终文本账本
沿上节Python3.12环境，安装openai[realtime]==2.20.0 websockets==15.0.1。保存为transcript_lab.py，运行python transcript_lab.py。代码只监听本机临时端口；音频为静音夹具，服务端返回规定文本，不能证明语音识别质量。服务端先接收两轮append/commit，再故意以第二轮优先的顺序完成转录。
python
import asyncio,base64,json,sqlite3,tempfile
from pathlib import Path
from openai import AsyncOpenAI
from openai.types.realtime import RealtimeServerEvent
from pydantic import TypeAdapter
from websockets.asyncio.server import serve
PREFIX='conversation.item.input_audio_transcription.'
schema=TypeAdapter(RealtimeServerEvent)
class Transcripts:
def __init__(self,path,session):
self.db=sqlite3.connect(path)
self.db.execute('CREATE TABLE IF NOT EXISTS finals(session TEXT,item TEXT,part INTEGER,text TEXT,PRIMARY KEY(session,item,part))')
self.session=session;self.items={};self.seen={};self.ready=[]
def cancel(self,item):
# 已完成文本不能当成尚未提交的草稿撤销；撤销业务另有合同。
row=self.items[item]
if row['state']=='final':raise ValueError('already final')
row['state']='cancelled';row['text']=''
def accept(self,session,event):
if session!=self.session:return 'old-session'
e=event.model_dump(exclude_none=True);eid=e['event_id']
raw=json.dumps(e,sort_keys=True)
if eid in self.seen:
if self.seen[eid]!=raw:raise ValueError('event ID collision')
return 'duplicate-event'
if len(self.seen)>=1000:raise ValueError('session event budget')
self.seen[eid]=raw
kind=e['type'];item=e.get('item_id')
if kind=='input_audio_buffer.committed':
if item in self.items:raise ValueError('item committed twice')
if len(self.items)>=20:raise ValueError('turn budget')
self.items[item]={'previous':e.get('previous_item_id'),'text':'','state':'partial'}
return 'committed'
if not kind.startswith(PREFIX):return 'unhandled'
if item not in self.items:raise ValueError('missing commit; resync instead of guessing')
row=self.items[item]
if e['content_index']!=0:raise ValueError('this adapter permits one audio part')
if row['state']=='cancelled':return 'cancelled'
if kind==PREFIX+'failed':
if row['state']!='final':row['state']='failed';row['text']=''
return 'failed'
if row['state']=='failed':return 'failed'
if kind==PREFIX+'delta':
if row['state']=='final':return 'late-delta'
if len(row['text'])+len(e['delta'])>4000:raise ValueError('caption budget')
row['text']+=e['delta'];return 'partial'
if kind==PREFIX+'completed':
text=e['transcript']
if not text.strip() or len(text)>4000:raise ValueError('empty or long final')
key=(session,item,0)
old=self.db.execute('SELECT text FROM finals WHERE session=? AND item=? AND part=?',key).fetchone()
if old and old[0]!=text:raise ValueError('conflicting final')
with self.db:self.db.execute('INSERT OR IGNORE INTO finals VALUES(?,?,?,?)',(*key,text))
row.update(text=text,state='final') # 最终文本替换临时显示，不是再append。
if not old:self.ready.append((key,text))
return 'final' if not old else 'duplicate-final'
return 'unhandled'
def ordered(self):
# 不能按completed到达顺序展示，也不推断缺失previous。
result=[];previous=None
while len(result)<len(self.items):
children=[key for key,v in self.items.items() if v['previous']==previous and key not in result]
if len(children)!=1:raise ValueError('missing predecessor, fork or cycle')
previous=children[0];result.append(previous)
return result
def close(self):self.db.close()
def event(kind,eid,item,**fields):
return {'type':kind,'event_id':eid,'item_id':item,**fields}
def final(eid,item,text):
return event(PREFIX+'completed',eid,item,content_index=0,transcript=text,usage={'type':'duration','seconds':.1})
async def main():
with tempfile.TemporaryDirectory() as directory:
state=Transcripts(Path(directory)/'captions.sqlite','connection-1')
events=[event('input_audio_buffer.committed','c1','u1',previous_item_id=None),
event('input_audio_buffer.committed','c2','u2',previous_item_id='u1'),
event(PREFIX+'delta','d1','u1',content_index=0,delta='明天九点'),
final('f2','u2','查询订单'),final('f1','u1','明天下午九点'),
final('f1-duplicate','u1','明天下午九点'),
event(PREFIX+'delta','late','u1',content_index=0,delta='错误追加')]
async def server(ws):
requests=[json.loads(await ws.recv()) for _ in range(5)]
assert [r['type'] for r in requests]==['session.update','input_audio_buffer.append','input_audio_buffer.commit','input_audio_buffer.append','input_audio_buffer.commit']
assert requests[0]['session']['audio']['input']['transcription']['model']=='gpt-4o-transcribe'
assert len(base64.b64decode(requests[1]['audio']))==4800
for e in events:
schema.validate_python(e);await ws.send(json.dumps(e))
async with serve(server,'127.0.0.1',0) as fixture:
port=fixture.sockets[0].getsockname()[1]
async with AsyncOpenAI(api_key='fixture',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 conn:
await conn.session.update(session={'type':'transcription','audio':{'input':{
'format':{'type':'audio/pcm','rate':24000},'turn_detection':None,
'transcription':{'model':'gpt-4o-transcribe'}}}})
await conn.input_audio_buffer.append(audio=base64.b64encode(bytes(4800)).decode())
await conn.input_audio_buffer.commit()
await conn.input_audio_buffer.append(audio=base64.b64encode(bytes(4800)).decode())
await conn.input_audio_buffer.commit()
async with asyncio.timeout(5):
async for e in conn:
state.accept('connection-1',e)
if e.event_id=='d1':assert not state.ready and state.items['u1']['text']=='明天九点'
if e.event_id=='late':break
assert state.ordered()==['u1','u2']
assert state.items['u1']['text']=='明天下午九点' and len(state.ready)==2
# 完成顺序u2/u1不同于说话顺序u1/u2；调度业务时另选顺序策略。
assert [key[1] for key,text in state.ready]==['u2','u1']
assert state.accept('old-connection',schema.validate_python(final('old','u1','旧字幕')))=='old-session'
state.close()
print('SDK transcript: partial replaced, turns ordered, final deduplicated')
if __name__=='__main__':asyncio.run(main())输出SDK transcript: partial replaced, turns ordered, final deduplicated。观察第一轮只有delta时ready为空，界面可以显示“明天九点”，业务消费者还没有最终输入；completed把它替换为“明天下午九点”。第二轮先完成也不会覆写第一轮字幕，ordered仍是u1、u2。重复completed即使换了event_id，也被SQLite主键挡住；相同item却给不同最终文本会显式报冲突。
SQLite这里保存的是最终转录收件箱，ready是当前进程的新到达通知，不是可靠消息队列。消费者重启后要从finals按消费游标取未处理项，或在同一数据库事务写outbox；不能只依赖内存ready保证交付。真正工具提交要复用业务intent ID和审批/幂等账本，不能拿连接item ID当付款键。例子限制20轮、1000事件、4000字；生产按会话轮转和保留期清理，不能无限保留seen。
ordered遇到缺前驱、分叉或循环就拒绝。网络有序只保证同一连接发送顺序，不保证不同转录任务完成顺序；重连要重新建立连接身份。这个程序不支持把半个旧会话内存偷偷恢复到新连接，未final的旧字幕应标为中断，由产品决定重新录入或明确重试。保存的旧final可用于审计和业务去重，但不能证明远端支持会话续传。
同一归并器连接有权使用的录音
保存下段为online_transcribe.py，与transcript_lab.py同目录。准备本人有权使用的0.1–30秒、24kHz单声道PCM16的speech.wav；设置OPENAI_API_KEY及账户支持的OPENAI_TRANSCRIBE_MODEL，例如本基线支持的gpt-4o-transcribe。运行会计费，本次未执行云端分支。新转录模型的额外参数和SDK版本需单独核对，不能照搬模型名推断字段。
python
import asyncio,base64,os,wave,uuid
from openai import AsyncOpenAI
from transcript_lab import Transcripts,PREFIX
async def main():
with wave.open('speech.wav','rb') as wav:
if (wav.getnchannels(),wav.getsampwidth(),wav.getframerate())!=(1,2,24000):
raise ValueError('mono PCM16 24kHz required')
if not .1<=wav.getnframes()/24000<=30:raise ValueError('0.1–30 seconds required')
raw=wav.readframes(wav.getnframes())
identity=str(uuid.uuid4());state=Transcripts('captions.sqlite',identity)
try:
async with AsyncOpenAI() as client:
async with client.realtime.connect(extra_query={'intent':'transcription'}) as conn:
await conn.session.update(session={'type':'transcription','audio':{'input':{
'format':{'type':'audio/pcm','rate':24000},'turn_detection':None,
'transcription':{'model':os.environ['OPENAI_TRANSCRIBE_MODEL']}}}})
async def send():
for i in range(0,len(raw),4800):
await conn.input_audio_buffer.append(audio=base64.b64encode(raw[i:i+4800]).decode())
await conn.input_audio_buffer.commit()
async def receive():
async for e in conn:
if e.type=='error':raise RuntimeError(e.error.message)
state.accept(identity,e)
if e.type==PREFIX+'failed':raise RuntimeError('ASR failed')
if e.type==PREFIX+'delta':print('临时:',state.items[e.item_id]['text'])
if e.type==PREFIX+'completed':print('最终:',state.items[e.item_id]['text']);return
raise RuntimeError('connection closed without final')
async with asyncio.timeout(90):
async with asyncio.TaskGroup() as group:
group.create_task(send());group.create_task(receive())
finally:state.close()
if __name__=='__main__':asyncio.run(main())发送与接收在TaskGroup中并发，错误会取消同组任务；总90秒截止避免永久等final。这里按录音文件批送，不模拟麦克风采集速度。界面只打印临时/最终文字，最终文本仍未触发业务工具。实时录音接入时，使用上一节已经拥有会话身份的采集流，同时保留时长与字节预算。
练习
- 第一轮先取消，再收到completed；第二轮收到failed后又收到completed。两轮都不得进入ready。
- 同一item出现不同最终文本应报冲突。关闭重开SQLite后重放相同final，不得再次发新到达通知。
- 新连接s2收到s1的旧final，不更新当前字幕；解释为何业务幂等仍不能仅用
s2:item。