同步、背压、延迟与质量评估:练习答案
参考实现
独立运行:
python
def latest_by_media_time(frames):
return max(frames,key=lambda x:x['pts']) if frames else None
assert latest_by_media_time([{'pts':99},{'pts':33},{'pts':66}]) == {'pts':99}
assert latest_by_media_time([]) is None还应拒绝其他会话的帧,并明确是否允许未来时间戳;PTS只有在同一时钟基准下可比较。
返回题目。
质量与恢复
独立保存 answer_quality.py,Python3.11+运行;正文完整SQLite重开实验直接保留,再添加以下时延变体。
python
import sqlite3,tempfile
from pathlib import Path
def edits(a,b):
row=list(range(len(b)+1))
for i,x in enumerate(a,1):
new=[i]
for j,y in enumerate(b,1):
new.append(min(new[-1]+1,row[j]+1,row[j-1]+(x!=y)))
row=new
return row[-1]
reference='book room for June six'.split()
hypothesis='book room for June seven'.split()
assert edits(reference,hypothesis)/len(reference)==.2
# 大部分文字正确仍可订错日期,任务断言是独立门槛。
assert hypothesis[-1]!='six'
assert edits(['yes'],['no','no','no'])/1==3
trace={'speech_end':1.,'vad_end':1.2,'first_audio':1.6,'play_start':1.8,
'interruption':2.,'last_old_sample':2.04}
assert abs(trace['play_start']-trace['speech_end']-.8)<1e-9
assert abs(trace['last_old_sample']-trace['interruption']-.04)<1e-9
# 真实SQLite重开:新连接不能把同一业务意图当作新动作。
with tempfile.TemporaryDirectory() as directory:
path=Path(directory)/'operations.db'
def open_db():
db=sqlite3.connect(path)
db.execute('CREATE TABLE IF NOT EXISTS operation (business_id TEXT PRIMARY KEY, argument TEXT, result TEXT)')
return db
def apply(db,business_id,argument):
with db:
db.execute('INSERT OR IGNORE INTO operation VALUES (?,?,?)',(business_id,argument,'confirmed:'+argument))
old=db.execute('SELECT argument,result FROM operation WHERE business_id=?',(business_id,)).fetchone()
if old[0]!=argument:raise ValueError('conflicting replay')
return old[1]
db=open_db();first=apply(db,'intent1','June six');db.close()
db=open_db();assert apply(db,'intent1','June six')==first
try:apply(db,'intent1','June seven')
except ValueError:pass
else:raise AssertionError('changed intent reused old confirmation')
assert db.execute('SELECT COUNT(*) FROM operation').fetchone()[0]==1
db.close()
print('word errors, heard latency and durable operation identity passed')
model_delay=trace['first_audio']-trace['speech_end']
trace['play_start']+=.4
assert abs(trace['play_start']-trace['speech_end']-1.2)<1e-9
assert abs(trace['first_audio']-trace['speech_end']-model_delay)<1e-9
print('playback delay isolated from model delay')一次替换和两次插入使yes→no no no的编辑距离为3,除以一个参考词得到3。不同日期重放在同一业务键下必须失败,原账本记录保持不变。