模型适配、流式事件与最终提交：练习答案

参考实现

追加到正文文件：

import json

def collect(events):
    chunks, finished = [], False
    for event in events:
        if finished:
            raise ValueError("event after completion")
        if event["type"] == "delta": chunks.append(event["text"])
        elif event["type"] == "done": finished = True
        else: raise ValueError("unknown event")
    if not finished: raise RuntimeError("incomplete")
    value = json.loads("".join(chunks))
    if set(value) != {"answer", "source"} or not all(isinstance(v, str) for v in value.values()):
        raise ValueError("invalid output")
    return value

ok = [{"type":"delta", "text":'{"answer":"68 F",'}, {"type":"delta", "text":'"source":"calculator"}'}, {"type":"done"}]
bad = [ok + [{"type":"delta", "text":"x"}],
       [{"type":"delta", "text":'{"answer":68,"source":"c"}'}, {"type":"done"}]]
for events in bad:
    try: collect(events)
    except ValueError: pass
    else: raise AssertionError("invalid output committed")
print("post-completion and schema rejected")

前者是事件协议错误，后者是业务输出契约错误，日志中应分别分类。

返回正文

深化页面练习解答

运行时适配器

内部事件至少区分文本增量、工具调用、工具结果、用量、完成、失败和取消，并携带 run ID、序号和 request ID。供应商重放按事件序号去重，业务重试则创建新的 attempt 和审计记录。

可观测执行器

span 可按 run → model call → tool call → queue job 嵌套；日志增加成本、缓存命中和供应商 request ID。密钥、令牌、密码、完整敏感输入和未脱敏工具结果不能进入普通日志，使用哈希、分类和关联 ID 替代。

背压与取消

序号、确认序号和结果版本用于重连与去重。同步请求有连接和读取超时，流式请求还要有空闲超时，长任务使用整体 deadline 和队列租约。工具结果和完成事件不能按普通文本丢弃。