从 0 搭一个 AI Agent(四):SSE 跑通之后,三处“契约错配”与它们的修复边界
英文标题:从 0 搭一个 AI Agent(四):SSE 跑通之后,三处“契约错配”与它们的修复边界
SSE 跑通后,核心是修复三处契约错配:取消依赖 send 抛错、aclose 竞态掩盖关闭、客户端 204/clean close 重试错位;并补 done 续传、重试预算与优雅关闭。
8.3 KB
中文
上一版文章把断连、续传、重试、优雅关闭的修复清单列了出来。这一版不增加新条目,只做一件事:把每条修复背后的“契约假设”说清楚 。因为这些问题之所以反复出现,不是因为某一行代码写错了,而是因为应用层、框架层、运行时层、客户端库,对同一件事持有不一致的假设 ,而没有任何一层负责仲裁。
三处错配贯穿全文:
- 服务端取消语义 :Starlette 0.49+ 假设
send()在客户端断开时抛OSError,uvicorn 的send()在断开后直接return。 - 生成器关闭语义 :应用层假设
aclose()是同步的、可以随时调用,Python 异步生成器在“running”状态下并发调用aclose()会抛RuntimeError,而这个RuntimeError会被except Exception捕获,掩盖关闭信号。 - 客户端终止语义 :SSE 规范用 204 表示“不要重连”,
fetch-event-source的默认策略在 204 和 clean close 两个点上都与标准错位。
下面逐个展开。
一、服务端取消:Starlette 的快速路径假设了 uvicorn 不履行的契约
Starlette 0.41.2 的 StreamingResponse.__call__ 无条件运行 stream_response(send) 和 listen_for_disconnect(receive) 的 anyio task group,谁先结束谁取消对方。客户端断开时,listen_for_disconnect 读到 http.disconnect,取消 body task,CancelledError 一路传播进生成器,finally 正常执行。
Starlette 0.49.1+ 为 spec_version >= (2, 4) 加了一条快速路径:跳过 listen_for_disconnect,直接 await stream_response(send),依赖 send() 在客户端断开时抛 OSError 来终止。这条路径的设计依据来自 ASGI 2.4 规范:“If send() is called on a closed connection the server should raise a server-specific subclass of OSError”。-1
uvicorn 通告 spec_version = "2.4",所以快速路径总是被选中。但 uvicorn 的 send() 在 self.disconnected 为 True 时直接return,不抛任何异常;connection_lost() 也不 .cancel() 对应的 ASGI 任务。结果:except OSError 分支是死代码,生成器继续迭代,每个 chunk 被静默丢弃。Kiln 在 /run_comparison 上复现了完整现象——关闭浏览器标签页后,Running eval job for ... 日志继续输出,LLM 调用持续触发,直到进程重启。
这里容易产生一个追问:uvicorn 不是曾经加过 ClientDisconnect(IOError) 吗?确实,uvicorn 的 changelog 显示 0.27.0 引入过“在 send() 上抛 ClientDisconnect(IOError)”的行为。但这条行为的稳定性 才是关键。Starlette 的快速路径假设 send() 在每一次 向死连接写入时都抛,而 uvicorn 的抛错行为并未在版本间形成这样的稳定契约。PR #3629 的讨论本身就暴露了分歧:一方主张让原始 OSError 传播,另一方认为 StreamingResponse 原来的 ClientDisconnect 转换“其实没问题”。这种分歧恰恰说明,把取消语义完全委托给“send() 一定抛错”本身就是脆弱的。-1
修复的方向不是“捕获 OSError”,而是不依赖这条契约 。CancellableStreamingResponse 无条件走旧的 task-group + listen_for_disconnect 路径,忽略 spec_version:
class CancellableStreamingResponse(StreamingResponse):
async def __call__(self, scope, receive, send):
...
Kiln 的实现刻意让 __call__ 的方法体“与 Starlette 0.52.1 的 else 分支完全相同,不做 spec_version 检查,不做 try/except OSError”,并明确注明“不要以任何理由重新引入 spec_version 分支”。is_disconnected() 仍然是应用层最可靠的兜底,因为它读的是 receive() 的 disconnect 消息,不依赖 send() 的行为。但框架层修复和兜底是两件事:兜底让生成器最终 停下来,CancellableStreamingResponse 让取消即时 传播。
二、生成器关闭:aclose() 在“running”状态下是异步竞态,不是同步清理
当框架取消生成器任务时,CancelledError 是 BaseException,不会被 except Exception 误捕获。但 aclose() 注入的是 GeneratorExit,同样绕过 except Exception——问题出在 aclose() 本身可能抛RuntimeError。
Python 的异步生成器有一个“running”标志。当生成器正在执行一次 asend() / athrow() / __anext__() 的调用栈中时,它是 running 的;此时并发调用 aclose(),会抛 RuntimeError: aclose(): asynchronous generator is already running(或 async generator ignored GeneratorExit)。这个 RuntimeError 是 Exception 的子类,会被 except Exception 捕获。于是你向一个正在被关闭的连接 yield 一个 error 事件,同时掩盖真正的关闭信号。
关键是要把触发条件说准。普通await 时生成器是 suspended 的,aclose() 会正常注入 GeneratorExit,不会报 RuntimeError。触发 RuntimeError 的场景是并发关闭的竞态 :生成器尚在一次 asend() / athrow() 的执行中,外部 aclose() 进来了。vLLM 的一个修复 PR 把这个机制写得很直白:“Task cancellation is asynchronous, so the generator may still be running when aclose() is called. In that case, aclose() raises: RuntimeError: asynchronous generator is already running”。该 PR 的修法是先取消并 await 所有 pending 的 anext() 任务,等对应迭代器真正停下来之后再调用 aclose()。这从侧面印证了:不 await 被取消的任务就直接aclose(),是在和生成器的 running 状态赛跑。-20
应用层的修法不需要那么复杂,但原则是一样的:清理放finally,不在 finally 里 yield,except Exception 只做重新抛出。
async def event_stream(prompt, request):
try:
async for token in llm.stream(prompt):
if await request.is_disconnected():
return
yield sse_frame(str(token), event="token")
yield sse_frame("[DONE]", event="done")
except Exception:
raise
finally:
await llm.aclose()
finally 对 GeneratorExit 和 CancelledError 都会执行,且不会把它们转换成普通 Exception。在 finally 里 yield 会破坏关闭语义,所以 except Exception 分支只重新抛出。如果生成器可能处于 running 状态被并发关闭,更稳妥的做法是在 finally 里捕获 RuntimeError 后等待一个事件循环 tick 再重试aclose() ,而不是在 except Exception 里做任何向连接写入的事。
三、客户端终止:204 和 clean close,默认策略在两个方向上与标准相反
SSE 规范用 HTTP 204 告诉客户端“没有更多内容,不要重连”。原生 EventSource 遵循这个语义。fetch-event-source 的默认错误策略则不同:它会对 204 持续重试,直到你显式覆盖 。有资料在对照 @microsoft/fetch-event-source 2.0.1 时把这个行为写得很明确:“a default error policy that retries a 204 forever until you override it”。与此同时,它把 clean close 视为流的结束 ——一次部署导致的干净关闭可能让读取提前终止,而 EventSource 在 clean close 后是会重连的。-9
也就是说,“不保证”不是模糊的风险,而是两个具体的错位方向:该停的 204 可能不停,该继续读的 clean close 可能提前结束 。修复是在 onopen 里显式识别 204 并抛 FatalError,同时在 onclose 里对未收到 done 的关闭抛 RetriableError:
async onopen(response) {
if (response.status === 204) {
throw new FatalError('流已结束(204)')
}
}
退避也必须自己算,fetch-event-source 的默认重试间隔不递增:
onerror(err) {
if (err instanceof FatalError) throw err
if (retryCount >= MAX_RETRIES) throw new FatalError('重连超限')
retryCount++
const delay = Math.min(30_000, 1000 * 2 ** retryCount)
* (0.8 + Math.random() * 0.4)
return delay
}
而 retryCount 只在流明确推进后重置 ——收到 meta 或 token,不在 onopen 重置。否则“连接成功但服务端始终不发 done 就关闭”这条路径会让重试预算被反复清零,MAX_RETRIES 永远不触发。
四、续传边界:done 的 id 和“补发 done”是两件事
SSE 规范中,“没有 id 字段”和“id: 空值”不同:前者保留上一个 Last-Event-ID,后者重置。如果 done 不携带可解析的 id,客户端在收到 done 前断连,重连时的 Last-Event-ID 仍然是最后一个 token 的 {request_id}:N,服务端会从 N 继续回放,而 done 本身不会被重新发送。
if resume_from >= len(tokens):
yield sse_frame("[DONE]", event="done",
event_id=f"{request_id}:done")
return
for i, token in enumerate(tokens[resume_from:], start=resume_from + 1):
yield sse_frame(token, event="token",
event_id=f"{request_id}:{i}")
yield sse_frame("[DONE]", event="done",
event_id=f"{request_id}:done")
回放分支有两条出口 :缓存已耗尽(resume_from >= len(tokens))和回放完毕(for 循环结束后)。两条出口都必须发送带可解析id 的 done。只给 done 加 id 但不补发,客户端仍然等不到结束信号;只补发但不带 id,下次重连仍然无法把 done 识别为终止态。parse_last_event_id 中 seq == "done" 返回 -1 的哨兵分支保留,作为“明确收到 done 后重连”的快速路径,但缓存回放末尾的补发不能省。
五、优雅关闭与幂等键:两个被低估的“尽早”和“一个分隔符”
timeout_graceful_shutdown 默认 None 意味着不设等待上限 。SSE 生成器在没有外部终止信号时不会自行返回,两者叠加就是无限等待。修复分两步:在 lifespan 的 shutdown 阶段尽早广播 sentinel ,并把 timeout_graceful_shutdown 调小作为兜底。
“尽早”需要锚定具体位置:sentinel 广播应该发生在 lifespan 上下文中 yield 之后、任何可能被 uvicorn 强制取消的 await 清理之前,不要放在finally 里。Omnigent 的 PR 讨论中已经指出这个陷阱:finally 可能在 graceful timeout 超时、ASGI 任务已被强制取消之后才执行,sentinel 被塞进一个消费者已经不在的队列,等于 no-op。
幂等键的修复只有一行,但方向不能错:
cache_key = f"{user_id}:{request_id}"
user_id 从认证信息获取,不由客户端单独提供。requestId 的 UUID 碰撞概率可以忽略,但恶意客户端可以构造已知的 requestId 去读另一个用户的缓存——这不是碰撞,是访问控制缺失。
六、优先级与验证方法
如果只能改三处,按“不改的后果是否不可见地持续发生”排序:
retryCount不在onopen重置,只在meta/token后重置。 否则重试上限是摆设,实际是无限循环。- 回放分支的两条出口都补发带可解析
id的done。 否则续传在边界上永远等不到结束信号,客户端会陷入“回放完 token、但没有结束信号”的循环。 - 优雅关闭尽早广播 sentinel,并把
timeout_graceful_shutdown调小。 否则 Ctrl-C 卡住的不仅是进程,还有部署流水线。
CancellableStreamingResponse 的升级、GeneratorExit 的 finally 兜底、幂等键加用户维度、204 显式终止,属于第二梯队:不改不一定立刻出事,但每一条的修复成本都远低于排查成本。
最后,给一个可操作的验证方法 ,比读任何清单都直接。在 SSE 生成器里加一行日志,记录每次 yield 时的 id(request) 或一个递增计数器;然后在长任务运行中途关闭浏览器标签页,观察服务端日志。如果日志继续输出 ,说明 send() 路径没有终止生成器——这时 is_disconnected() 是否生效、CancellableStreamingResponse 是否必要,一测便知。同样的方法适用于 aclose():在 finally 里打一行日志,如果客户端断开后 finally 没有在预期时间内执行,说明取消信号没有传播进来。
上一篇让 demo 看起来跑起来了;这一篇把“为什么它会在你看不见的地方继续跑”拆成了三处契约错配。对 AI Agent 来说,看不见的消耗比看得见的报错更贵。
注 :文中版本基准为 Starlette 0.49.1+(含 0.52.x)、uvicorn(spec_version = "2.4")、@microsoft/fetch-event-source 2.0.1。uvicorn 的 send() 行为在 0.27.0 附近有过 ClientDisconnect(IOError) 的短暂窗口,但 Starlette 快速路径所依赖的“每次写入死连接都稳定抛 OSError”这一契约,在当前主流组合下并未成立。如果你运行在不同小版本上,建议用第六节的日志方法确认自己的实际行为,而不是直接套用结论。
在做前端 + AI / 全栈 AI 的朋友,欢迎交流。我目前在看相关方向的机会,玄镜是手上真实跑了一年的项目。
English / 原文
上一版文章把断连、续传、重试、优雅关闭的修复清单列了出来。这一版不增加新条目,只做一件事:把每条修复背后的“契约假设”说清楚 。因为这些问题之所以反复出现,不是因为某一行代码写错了,而是因为应用层、框架层、运行时层、客户端库,对同一件事持有不一致的假设 ,而没有任何一层负责仲裁。
三处错配贯穿全文:
- 服务端取消语义 :Starlette 0.49+ 假设
send()在客户端断开时抛OSError,uvicorn 的send()在断开后直接return。 - 生成器关闭语义 :应用层假设
aclose()是同步的、可以随时调用,Python 异步生成器在“running”状态下并发调用aclose()会抛RuntimeError,而这个RuntimeError会被except Exception捕获,掩盖关闭信号。 - 客户端终止语义 :SSE 规范用 204 表示“不要重连”,
fetch-event-source的默认策略在 204 和 clean close 两个点上都与标准错位。
下面逐个展开。
一、服务端取消:Starlette 的快速路径假设了 uvicorn 不履行的契约
Starlette 0.41.2 的 StreamingResponse.__call__ 无条件运行 stream_response(send) 和 listen_for_disconnect(receive) 的 anyio task group,谁先结束谁取消对方。客户端断开时,listen_for_disconnect 读到 http.disconnect,取消 body task,CancelledError 一路传播进生成器,finally 正常执行。
Starlette 0.49.1+ 为 spec_version >= (2, 4) 加了一条快速路径:跳过 listen_for_disconnect,直接 await stream_response(send),依赖 send() 在客户端断开时抛 OSError 来终止。这条路径的设计依据来自 ASGI 2.4 规范:“If send() is called on a closed connection the server should raise a server-specific subclass of OSError”。-1
uvicorn 通告 spec_version = "2.4",所以快速路径总是被选中。但 uvicorn 的 send() 在 self.disconnected 为 True 时直接return,不抛任何异常;connection_lost() 也不 .cancel() 对应的 ASGI 任务。结果:except OSError 分支是死代码,生成器继续迭代,每个 chunk 被静默丢弃。Kiln 在 /run_comparison 上复现了完整现象——关闭浏览器标签页后,Running eval job for ... 日志继续输出,LLM 调用持续触发,直到进程重启。
这里容易产生一个追问:uvicorn 不是曾经加过 ClientDisconnect(IOError) 吗?确实,uvicorn 的 changelog 显示 0.27.0 引入过“在 send() 上抛 ClientDisconnect(IOError)”的行为。但这条行为的稳定性 才是关键。Starlette 的快速路径假设 send() 在每一次 向死连接写入时都抛,而 uvicorn 的抛错行为并未在版本间形成这样的稳定契约。PR #3629 的讨论本身就暴露了分歧:一方主张让原始 OSError 传播,另一方认为 StreamingResponse 原来的 ClientDisconnect 转换“其实没问题”。这种分歧恰恰说明,把取消语义完全委托给“send() 一定抛错”本身就是脆弱的。-1
修复的方向不是“捕获 OSError”,而是不依赖这条契约 。CancellableStreamingResponse 无条件走旧的 task-group + listen_for_disconnect 路径,忽略 spec_version:
class CancellableStreamingResponse(StreamingResponse):
async def __call__(self, scope, receive, send):
...
Kiln 的实现刻意让 __call__ 的方法体“与 Starlette 0.52.1 的 else 分支完全相同,不做 spec_version 检查,不做 try/except OSError”,并明确注明“不要以任何理由重新引入 spec_version 分支”。is_disconnected() 仍然是应用层最可靠的兜底,因为它读的是 receive() 的 disconnect 消息,不依赖 send() 的行为。但框架层修复和兜底是两件事:兜底让生成器最终 停下来,CancellableStreamingResponse 让取消即时 传播。
二、生成器关闭:aclose() 在“running”状态下是异步竞态,不是同步清理
当框架取消生成器任务时,CancelledError 是 BaseException,不会被 except Exception 误捕获。但 aclose() 注入的是 GeneratorExit,同样绕过 except Exception——问题出在 aclose() 本身可能抛RuntimeError。
Python 的异步生成器有一个“running”标志。当生成器正在执行一次 asend() / athrow() / __anext__() 的调用栈中时,它是 running 的;此时并发调用 aclose(),会抛 RuntimeError: aclose(): asynchronous generator is already running(或 async generator ignored GeneratorExit)。这个 RuntimeError 是 Exception 的子类,会被 except Exception 捕获。于是你向一个正在被关闭的连接 yield 一个 error 事件,同时掩盖真正的关闭信号。
关键是要把触发条件说准。普通await 时生成器是 suspended 的,aclose() 会正常注入 GeneratorExit,不会报 RuntimeError。触发 RuntimeError 的场景是并发关闭的竞态 :生成器尚在一次 asend() / athrow() 的执行中,外部 aclose() 进来了。vLLM 的一个修复 PR 把这个机制写得很直白:“Task cancellation is asynchronous, so the generator may still be running when aclose() is called. In that case, aclose() raises: RuntimeError: asynchronous generator is already running”。该 PR 的修法是先取消并 await 所有 pending 的 anext() 任务,等对应迭代器真正停下来之后再调用 aclose()。这从侧面印证了:不 await 被取消的任务就直接aclose(),是在和生成器的 running 状态赛跑。-20
应用层的修法不需要那么复杂,但原则是一样的:清理放finally,不在 finally 里 yield,except Exception 只做重新抛出。
async def event_stream(prompt, request):
try:
async for token in llm.stream(prompt):
if await request.is_disconnected():
return
yield sse_frame(str(token), event="token")
yield sse_frame("[DONE]", event="done")
except Exception:
raise
finally:
await llm.aclose()
finally 对 GeneratorExit 和 CancelledError 都会执行,且不会把它们转换成普通 Exception。在 finally 里 yield 会破坏关闭语义,所以 except Exception 分支只重新抛出。如果生成器可能处于 running 状态被并发关闭,更稳妥的做法是在 finally 里捕获 RuntimeError 后等待一个事件循环 tick 再重试aclose() ,而不是在 except Exception 里做任何向连接写入的事。
三、客户端终止:204 和 clean close,默认策略在两个方向上与标准相反
SSE 规范用 HTTP 204 告诉客户端“没有更多内容,不要重连”。原生 EventSource 遵循这个语义。fetch-event-source 的默认错误策略则不同:它会对 204 持续重试,直到你显式覆盖 。有资料在对照 @microsoft/fetch-event-source 2.0.1 时把这个行为写得很明确:“a default error policy that retries a 204 forever until you override it”。与此同时,它把 clean close 视为流的结束 ——一次部署导致的干净关闭可能让读取提前终止,而 EventSource 在 clean close 后是会重连的。-9
也就是说,“不保证”不是模糊的风险,而是两个具体的错位方向:该停的 204 可能不停,该继续读的 clean close 可能提前结束 。修复是在 onopen 里显式识别 204 并抛 FatalError,同时在 onclose 里对未收到 done 的关闭抛 RetriableError:
async onopen(response) {
if (response.status === 204) {
throw new FatalError('流已结束(204)')
}
}
退避也必须自己算,fetch-event-source 的默认重试间隔不递增:
onerror(err) {
if (err instanceof FatalError) throw err
if (retryCount >= MAX_RETRIES) throw new FatalError('重连超限')
retryCount++
const delay = Math.min(30_000, 1000 * 2 ** retryCount)
* (0.8 + Math.random() * 0.4)
return delay
}
而 retryCount 只在流明确推进后重置 ——收到 meta 或 token,不在 onopen 重置。否则“连接成功但服务端始终不发 done 就关闭”这条路径会让重试预算被反复清零,MAX_RETRIES 永远不触发。
四、续传边界:done 的 id 和“补发 done”是两件事
SSE 规范中,“没有 id 字段”和“id: 空值”不同:前者保留上一个 Last-Event-ID,后者重置。如果 done 不携带可解析的 id,客户端在收到 done 前断连,重连时的 Last-Event-ID 仍然是最后一个 token 的 {request_id}:N,服务端会从 N 继续回放,而 done 本身不会被重新发送。
if resume_from >= len(tokens):
yield sse_frame("[DONE]", event="done",
event_id=f"{request_id}:done")
return
for i, token in enumerate(tokens[resume_from:], start=resume_from + 1):
yield sse_frame(token, event="token",
event_id=f"{request_id}:{i}")
yield sse_frame("[DONE]", event="done",
event_id=f"{request_id}:done")
回放分支有两条出口 :缓存已耗尽(resume_from >= len(tokens))和回放完毕(for 循环结束后)。两条出口都必须发送带可解析id 的 done。只给 done 加 id 但不补发,客户端仍然等不到结束信号;只补发但不带 id,下次重连仍然无法把 done 识别为终止态。parse_last_event_id 中 seq == "done" 返回 -1 的哨兵分支保留,作为“明确收到 done 后重连”的快速路径,但缓存回放末尾的补发不能省。
五、优雅关闭与幂等键:两个被低估的“尽早”和“一个分隔符”
timeout_graceful_shutdown 默认 None 意味着不设等待上限 。SSE 生成器在没有外部终止信号时不会自行返回,两者叠加就是无限等待。修复分两步:在 lifespan 的 shutdown 阶段尽早广播 sentinel ,并把 timeout_graceful_shutdown 调小作为兜底。
“尽早”需要锚定具体位置:sentinel 广播应该发生在 lifespan 上下文中 yield 之后、任何可能被 uvicorn 强制取消的 await 清理之前,不要放在finally 里。Omnigent 的 PR 讨论中已经指出这个陷阱:finally 可能在 graceful timeout 超时、ASGI 任务已被强制取消之后才执行,sentinel 被塞进一个消费者已经不在的队列,等于 no-op。
幂等键的修复只有一行,但方向不能错:
cache_key = f"{user_id}:{request_id}"
user_id 从认证信息获取,不由客户端单独提供。requestId 的 UUID 碰撞概率可以忽略,但恶意客户端可以构造已知的 requestId 去读另一个用户的缓存——这不是碰撞,是访问控制缺失。
六、优先级与验证方法
如果只能改三处,按“不改的后果是否不可见地持续发生”排序:
retryCount不在onopen重置,只在meta/token后重置。 否则重试上限是摆设,实际是无限循环。- 回放分支的两条出口都补发带可解析
id的done。 否则续传在边界上永远等不到结束信号,客户端会陷入“回放完 token、但没有结束信号”的循环。 - 优雅关闭尽早广播 sentinel,并把
timeout_graceful_shutdown调小。 否则 Ctrl-C 卡住的不仅是进程,还有部署流水线。
CancellableStreamingResponse 的升级、GeneratorExit 的 finally 兜底、幂等键加用户维度、204 显式终止,属于第二梯队:不改不一定立刻出事,但每一条的修复成本都远低于排查成本。
最后,给一个可操作的验证方法 ,比读任何清单都直接。在 SSE 生成器里加一行日志,记录每次 yield 时的 id(request) 或一个递增计数器;然后在长任务运行中途关闭浏览器标签页,观察服务端日志。如果日志继续输出 ,说明 send() 路径没有终止生成器——这时 is_disconnected() 是否生效、CancellableStreamingResponse 是否必要,一测便知。同样的方法适用于 aclose():在 finally 里打一行日志,如果客户端断开后 finally 没有在预期时间内执行,说明取消信号没有传播进来。
上一篇让 demo 看起来跑起来了;这一篇把“为什么它会在你看不见的地方继续跑”拆成了三处契约错配。对 AI Agent 来说,看不见的消耗比看得见的报错更贵。
注 :文中版本基准为 Starlette 0.49.1+(含 0.52.x)、uvicorn(spec_version = "2.4")、@microsoft/fetch-event-source 2.0.1。uvicorn 的 send() 行为在 0.27.0 附近有过 ClientDisconnect(IOError) 的短暂窗口,但 Starlette 快速路径所依赖的“每次写入死连接都稳定抛 OSError”这一契约,在当前主流组合下并未成立。如果你运行在不同小版本上,建议用第六节的日志方法确认自己的实际行为,而不是直接套用结论。
在做前端 + AI / 全栈 AI 的朋友,欢迎交流。我目前在看相关方向的机会,玄镜是手上真实跑了一年的项目。