AI Agent Engineer

AI Agent Engineer

复现流式、取消与重试

4 min read

复现流式、取消与重试

实践目标

用确定性测试分别复现流式分段、取消后保留上下文和 Provider 重试，学会区分三种看似相似的“响应中断”。

前置知识

  • 已理解 asyncio cancellation；
  • 已读 Provider retry 和 Runtime Event；
  • 当前实践不需要 DeepSeek key。

需要阅读的代码

  • nanobot/providers/base.py::chat_stream_with_retry
  • nanobot/providers/base.py::_run_with_retry
  • nanobot/agent/loop.py::_cancel_active_tasks
  • nanobot/agent/loop.py::_dispatch
  • nanobot/agent/runner.py::run
  • nanobot/api/server.py::handle_chat_completions
  • tests/providers/test_provider_retry.py
  • tests/agent/test_stop_preserves_context.py
  • tests/test_api_stream.py

先写下预测

为三种情况分别写：

情况会不会重试会不会保留部分文本会不会保存 checkpoint
Provider 在首个 delta 前 429
stream 已输出后普通错误
用户 /stop

实践 A：流式重试边界

运行：

Set-Location "example\nanobot\code-EN"
python -m pytest tests\providers\test_provider_retry.py `
  -k "does_not_retry_after_emitting_content or retries_timeout_in_new_stream_segment" `
  -vv

给这些位置打断点：

  • chat_stream_with_retry() 设置 has_streamed_content；
  • _run_with_retry() 的 should_retry_guard；
  • 新 stream segment 分支。

预期观察

  • 已输出内容后，普通失败默认不能无痕重试；
  • stream stall timeout 是明确处理的恢复例外；
  • 若入口支持 segment，会先结束旧 segment，再开始新 segment；
  • retry policy 既依赖错误类型，也依赖已经发生的外部可见输出。

实践 B：取消后上下文

运行：

python -m pytest tests\agent\test_stop_preserves_context.py -vv

观察：

  • active task 如何被 cancel；
  • CancelledError 在哪一层捕获；
  • runtime checkpoint 如何恢复；
  • pending user turn 如何清理；
  • Session 保存了哪些中间消息。

预期观察

取消不是抹掉整个 Turn。已经完成的 tool/assistant 中间状态会尽力保留，以便下一轮理解发生过什么。

实践 C：API SSE

运行：

python -m pytest tests\test_api_stream.py -vv

观察：

  • process_direct() 接收的 on_stream / on_stream_end；
  • delta 如何变成 SSE chunk；
  • final response 和 [DONE]；
  • exception/timeout 如何关闭流。

一张诊断表

现象第一观察点不要先做
一个字都没输出就等待Provider retry metadata、on_retry_wait重启整个 Gateway
输出一半后重头开始stream id/segment、stall timeout假设是前端重复渲染
/stop 后下轮仍看到工具结果checkpoint restore认定 Session 损坏
SSE 已断但后台仍运行request cancellation 是否传到 task直接删除 Session

完成条件

  • 三组测试均能运行或明确记录环境阻塞；
  • 能分别画出 retry、cancel、SSE 三条调用链；
  • 修正实践前预测表；
  • 至少指出一个“外部可见后不可安全重放”的原因。

验证方法

保持 code-EN 只读，用 IDE 单步或 pytest --pdb 观察现有 fake response 顺序和断言。若确实要改变响应顺序，把最小 fake case 复制到 $nanobotLab 或本学习库的示例区域再实验；不要编辑源仓库测试，也不要为了让实验通过修改生产 retry policy。

复盘问题

  1. 为什么 stream stall 可以作为普通错误后的特殊恢复例外？
  2. CancelledError 为什么通常不应被改成普通错误字符串？
  3. SSE 连接断开是否一定意味着 Provider task 已取消？
  4. checkpoint 保存和外部副作用回滚有什么本质区别？

下一篇：编写一个 Workspace Skill

Backlinks