Chapter 05

综合：Agent 工具调度器

前四章分别给了概念、机制、实操、陷阱。这一章把它们拧成一个真实项目——一个 AI Agent 的工具调度层。重点不是"照步骤实现"，而是判别：每个设计点，你该选第几章的哪个方案，为什么。

本章你将建立的 schema

  • 把 01–04 的概念整合进一个端到端可运行的系统
  • 在真实约束下做判别：catchTag vs catchAll、exit vs fail-fast、Layer vs 写死、timeout vs 手动取消
  • 验收一个并发、超时、失败隔离、可中断、可测试的调度器

5.1项目背景

为一个 AI Agent 构建工具调度层。Agent 在一轮里决定调用若干工具（search、weather、calculator……），每个工具是一次外部调用。调度层的要求：

  • 有界并发：同时最多跑 N 个，别打爆下游 API。
  • 单项超时：每个工具最多等 2 秒，超时即取消那一个。
  • 失败隔离：一个工具失败或超时，不影响其他工具的结果汇总。
  • 共享依赖：所有工具共用一个 HttpClient，测试时可换假实现。
  • 可被上游中断：用户取消整轮请求时，所有在飞的工具干净停止、释放资源。
Agent 上游 工具列表 调度层 runTools Effect.all {concurrency:3} 用户取消 interrupt tool: search timeout · exit tool: weather timeout · exit tool: calc timeout · exit HttpClient 服务（Layer 注入）
图 5.1每个工具各自 timeout + exit，共享一个 HttpClient。注意：上游的 interrupt（红）打到调度层，会沿结构化并发自动传到所有工具 fiber——这是 §2.5 的保证在系统层面的兑现。

5.2设计任务：判别决策（不是步骤实现）

动手前先做四个判别。每个都在前几章见过两个方案，这个场景下哪个对？

表 5.1 · 四个判别点
决策点备选这个场景选哪个？为什么
错误处理01.6 catchTag / 04.5 catchAllcatchTag/exit。要区分工具失败种类、不能静默吞错（P5）
结果收集02.4 Effect.all fail-fast / 02.2 exit 隔离每项 exit。一个工具失败不该让整轮汇总失败（P10）
依赖写死 new HttpClient() / 01.7 Tag+LayerTag + Layer。测试换假实现、生产共享单例（§2.3）
超时取消02.5 表 AbortController / Effect.timeoutEffect.timeout。自动中断且 finalizer 必跑（§2.5）
调度器设计决策 复杂并发 + 强错误保证? 否 Promise.allSettled 是 失败要隔离? 否 Effect.all 直接 是 本项目这条路 exit + timeout + Layer
图 5.2判别走到底，落在最下面那条红色路径。注意：每个分叉点对应表 5.1 的一行——决策树不是"实现步骤"，是"为什么这个场景排除了另一个方案"。

5.3自己实现

按下面的验收标准，自己写一版 runTools + 一个测试用 HttpClient Layer。先别看参考实现。

验收 checklist（可观测行为）

  • 4 个工具、并发设 3：能观察到不是一拥而上（用日志打时间戳验证最多 3 个同时在飞）。
  • 一个工具 Effect.never（永不返回）：它在 2 秒被取消，其余工具正常返回结果。
  • 一个工具 Effect.fail：该工具 exit 是 Failure，其余是 Success，整轮不抛。
  • 测试用 Layer 与生产 Layer 可互换，runTools 一行不改。

5.4亲手画一张图

亲手画一张图

合上教程，在纸上画出你的调度器在运行那一刻的 fiber 树：父 fiber（runTools）下挂几个子 fiber（每个工具一个）。只画 4 个节点。画完回到图 5.1 对照——你画的图里，"上游中断"那根箭头指向父还是子？finalizer 在哪一层运行？（答案：指向父，靠结构化并发自动向下传到子；每个子的资源 finalizer 在各自被中断时运行。）

5.5参考实现

参考实现（写完自己版本再展开）
dispatcher.tsTypeScript
import { Effect, Context, Layer, Data, Exit, Console } from "effect"

// ── 错误（01.6）──
class ToolError extends Data.TaggedError("ToolError")<{
  readonly tool: string
  readonly reason: string
}> {}

// ── 共享服务（01.7 + 02.3）──
class HttpClient extends Context.Tag("app/HttpClient")<
  HttpClient,
  { readonly get: (url: string) => Effect.Effect<string, ToolError> }
>() {}

interface Tool {
  readonly name: string
  readonly call: Effect.Effect<string, ToolError, HttpClient>
}

const mkTool = (name: string, url: string): Tool => ({
  name,
  call: Effect.gen(function* () {
    const http = yield* HttpClient
    return yield* http.get(url)
  }),
})

// ── 调度：有界并发(02.4) + 单项超时(02.5) + 失败隔离(02.2)──
const runTools = (tools: ReadonlyArray<Tool>) =>
  Effect.all(
    tools.map((t) =>
      t.call.pipe(
        Effect.timeout("2 seconds"),
        Effect.exit,
        Effect.map((exit) => ({ name: t.name, exit })),
      ),
    ),
    { concurrency: 3 },
  )

// ── 测试用实现：换成生产实现只改这一个 Layer ──
const HttpClientTest = Layer.succeed(HttpClient, {
  get: (url) =>
    url.includes("slow")
      ? Effect.never                                  // 模拟超时
      : url.includes("boom")
        ? Effect.fail(new ToolError({ tool: url, reason: "boom" }))
        : Effect.succeed(`result of ${url}`),
})

const program = runTools([
  mkTool("search", "search?q=effect"),
  mkTool("weather", "weather/tokyo"),
  mkTool("slow", "slow-endpoint"),
  mkTool("boom", "boom-endpoint"),
]).pipe(
  Effect.tap((rows) =>
    Effect.forEach(rows, (r) =>
      Console.log(
        `${r.name}: ${Exit.isSuccess(r.exit) ? r.exit.value : "FAILED/TIMEOUT"}`,
      ),
    ),
  ),
  Effect.provide(HttpClientTest), // R: HttpClient → never
)

Effect.runPromise(program)
// search:  result of search?q=effect
// weather: result of weather/tokyo
// slow:    FAILED/TIMEOUT      （2 秒被中断，没卡死）
// boom:    FAILED/TIMEOUT      （fail 被隔离在自己的 Exit）

关键决策说明：① Effect.exit 是失败隔离的核心——没有它，boom 的失败会让整个 Effect.all 短路。② timeout 让 slow（Effect.never）在 2 秒被中断，而不是永远挂起。③ 依赖只在最外层 provide 一次，业务代码（mkTool/runTools）只认 HttpClient 这个 Tag。④ 整个 program 仍是个值——上游想取消，对它 Fiber.interrupt 即可，中断沿结构化并发传到每个工具。

5.6反思问题

  1. 你的实现里哪个决策最难做？回去看是哪一章帮你定的？
  2. 如果需求变成"任一工具失败就整轮失败"（强一致），你要改哪一行？（提示：去掉 Effect.exit，让 Effect.all 恢复 fail-fast。）
  3. 如果要给整轮加 5 秒总预算（不只是单项 2 秒），加在哪里？这和 03 章进阶挑战是同一个问题吗？

§本章 self-check

  1. 为什么 runTools 的返回类型 R 是 HttpClient 而不是 never？什么时候才变 never？
  2. 把并发从 3 改成 "unbounded"，对"打爆下游"这条要求意味着什么？
答案
  1. 因为工具 call 里 yield* HttpClient，依赖被记进 R（02.3）。直到最外层 Effect.provide(HttpClientTest) 才把 R 消成 never、可运行。
  2. "unbounded" 会让所有工具同时在飞，违背"有界并发别打爆下游"。具体数字（如 3）才是限流；这正是 P8 的教训。