函数调用分发器:超时、重试、去重

函数调用分发器:超时、重试、去重

本节摘要:分发器是外壳为 schema 许下的每个承诺买单的地方——超时、重试、去重、错误映射,全在一条缝上。它坐在循环(第 20 节)与注册表(第 21 节)之间:传输(第 22 节)喂数给循环,循环把工具调用交给分发器,分发器调注册表、跑 handler、返回结果或 JSON-RPC 形状的错误信封。本节实现每调用超时(返回类型化错误而非挂死循环)、指数退避抖动重试、幂等键去重(慢原调用与重试竞态不跑两次)、handler 异常映射到统一错误信封,以及并发扇出的信号量限流。

对应原课程:Phase 19 · Lesson 23 · function-call-dispatcher(原英文 phases/19-capstone-projects/23-function-call-dispatcher/docs/en.md)。本节属「Agent Harness 深度构建赛道」第四节。

学习目标

阅读完本节,你应当能够:

1. 给工具 handler 包每调用超时,返回类型化错误而非挂死循环。
2. 应用带抖动、带上限尝试次数的指数退避重试。
3. 在幂等键上去重重试,使与慢原调用竞态的重试不跑两次。
4. 把 handler 异常与传输故障映射到循环已理解的单一错误信封。
5. 用并发限制约束并行分发,使四十个工具调用的扇出不耗尽事件循环。

一、问题与直觉

分发器是唯一知道定时器、重试、幂等的层。循环不知道,注册表不知道,handler 不知道。这个隔离就是要点。一个非幂等的 db.write 超时了,可能提交了也可能没提交,重试会重复写——分发器必须尊重注册表记录的 idempotent 标志。

一个真实的生产 bug:重试在原调用还在途时触发。第一次卡在 4.9 秒(刚好低于超时),重试在 5 秒触发,两个请求对同一后端竞态。如果工具是 payments.charge,你扣了两次款。

二、从零实现

超时:每个工具有默认超时(注册表记录带 timeout_ms),分发器可被每调用覆盖。用 asyncio.wait_for,超时时取消 handler 任务返回 DispatchError(kind="timeout")。对非幂等工具,超时默认不重试——db.write 超时可能已提交,重试重复写。幂等工具才重试。

重试:最多三次,指数退避带抖动:

attempt 1  -> delay 0
attempt 2  -> delay 0.1s * (1 + random[0..0.5])
attempt 3  -> delay 0.4s * (1 + random[0..0.5])
​

只有 timeout 与 transient 错误重试。schema 错误是确定性的,重试不改变结果只烧预算,不重试。重试循环尊重外壳预算:调用方预算剩余工具调用为零时,首次尝试即失败返回 kind="budget_exceeded"。

幂等键去重:分发器接受可选 idempotency_key。若同键在途,分发器等在途 future 返回其结果。缓存完成后再留 60 秒,吸收迟到的重试。键是调用方责任——外壳从规划器派生 f"{step_id}:{tool_name}:{hash(args)}"。分发器不发明键,因为仅从参数派生键会让两个语义不同的调用长得一样。

错误信封(统一形状):

DispatchError
  kind        : "timeout" | "transient" | "schema" | "not_found" | "internal" | "budget_exceeded"
  message     : str
  attempts    : int
  jsonrpc_code: int   (-32601/-32602/-32603 之一)
​

循环把 kind 映射到下一状态:schema 与 not_found 进 on_error 并触发重规划;timeout 与 transient 进 on_error 且视尝试次数决定是否重规划;budget_exceeded 触发 on_budget_exceeded。

三、并发扇出限流

gather(*calls) 同时跑所有协程。四十个工具调用就是四十个开 socket 或四十个子进程管道,多数后端不喜欢一个客户端四十个并行连接。分发器用信号量包 gather,默认并发限八。每个调用派发前获取信号量,完成时释放。调用方看到 gather 形状的输出,实际调度是有界的。

sem = asyncio.Semaphore(8)
async def gather_bounded(calls):
    async def bounded(c):
        async with sem:
            return await dispatch(*c)
    return await asyncio.gather(*[bounded(c) for c in calls])
​

四、可复用产物

code/main.py 定义 Dispatcher、DispatchError、TransientError。分发器构造时取一个注册表,异步 dispatch(name, args, ...) 是唯一入口。每尝试超时在 _run_with_retries 内用 asyncio.wait_for 内联施加。gather_bounded(calls) 跑多个带并发限的分发。code/tests/test_dispatcher.py 覆盖超时触发、transient 重试、schema 错误不重试、幂等去重(两个同键并发调用塌成一次 handler 调用)、并发限流(信号量在起作用)。测试用 asyncio.sleep(0) 与确定性 Counter handler,毫秒级完成、不依赖墙钟。

五、框架对比

分发器是把「schema 承诺」兑现的地方。注册表加形状(第 21 节),传输加线格式(第 22 节),分发器加时间与韧性。三者的隔离让它们各自可测、可换。生产分发器会再加两样:一是每个转移的结构化日志(循环的事件流已给,但分发器也应发 dispatch.attempt 与 dispatch.retry 事件);二是断路器——窗口内 N 次失败后,工具进冷却期,派发立即返回 kind="circuit_open" 而非尝试 handler。两者都叠在本分发器上而不改契约。

六、练习

1. 非幂等不重试:注册一个 db.write 非幂等工具,让它超时,确认分发器返回 kind="timeout" 而非重试。
2. 幂等去重:用同一 idempotency_key 并发派发两次,确认 handler 只跑一次,两个调用方拿到同结果。
3. schema 不重试:让一个工具因参数类型错失败,确认 kind="schema" 不触发重试。
4. 并发限流:派发 40 个调用,默认并发 8,确认信号量把在途数压在 8。
5. 预算耗尽:传一个剩余工具调用为零的预算,确认首次尝试即 kind="budget_exceeded"。

本节要点回顾

1. 分发器是韧性的缝:超时/重试/去重/错误映射,循环/注册表/handler 都不知道。
2. 超时按幂等性:幂等重试,非幂等不重试(避免重复写)。
3. 指数退避带抖动:三尝试,只有 timeout/transient 重试,schema 不重试。
4. 幂等键去重:同键在途则等 future,键是调用方责任(含 step_id)。
5. 统一错误信封:六种 kind 映射到循环的下一状态。
6. 信号量限流:默认并发 8,40 扇出不耗尽事件循环。

下一节,我们建「计划-执行控制流」——把分发器接到一个能规划、执行、反思的 Agent 上。