---
title: "分级 tool 注册表"
url: "https://turingblog.org/notes/patterns/graded-tools"
date: 2026-09-22
tags: ["LLM","Agent","TypeScript"]
description: "按副作用分 read / write / send 的 tool 注册表 —— 三道闸 + 两层防护，send 永远到不了 LLM 手上"
insight: "按副作用分 read / write / send 的 tool 注册表 —— 三道闸 + 两层防护，send 永远到不了 LLM 手上"
problem: "LLM 能调用的工具里混着会对外发消息、改状态的能力，仅凭「危险程度」的模糊判断管不住，一次幻觉就可能把消息真的发出去。"
---

把已有的能力函数包成统一、可枚举、可编排的 tool，并给每个 tool 标一个**副作用等级**：

| 等级 | 含义 | 能自主调吗 |
|---|---|---|
| `read` | 纯读：查数据、检索、抓取 | ✅ |
| `write` | 改本地状态：填表单、提交 | 固定流程里可以，**LLM 自主调不行** |
| `send` | 对外发出去：发消息、发邮件 | ❌ **永远人确认** |

分级依据是两个具体问题——**做完之后能撤销吗？外部的人会立刻看到吗？**——而不是「危险程度」这种模糊感觉。

- **零依赖**，三个文件，TypeScript
- 注册表 `invoke` 过**三道闸**：能力未授权 / 场景不匹配 / `send` + 自主编排
- 给 LLM 的 function 清单里**根本不含 `send` 类**（模型不会用它不知道存在的东西）
- 只读出口执行前**再查一次副作用等级**，防模型幻觉出 `ticket.submit` 这种名字
- 自带 wire 名映射：OpenAI 规范的 function name **不允许点号**，严格的网关会直接 400
- 附一个最小编排器：固定计划、结果串联、`stopIf` 提前结束

> 📖 设计取舍、以及「固定流程可以自主提交」和「LLM 可以自主调用提交」的区别 → [给 agent 的 tool 分级](/blog/agent-tool-side-effects)

## 怎么用

**1. 注册一个 tool**（薄 wrapper，不改被包的函数）：

```ts
import { toolRegistry } from './tools/registry';

toolRegistry.register({
  name: 'kb.search',              // <域>.<能力>
  domain: 'kb',
  title: '检索知识库',
  description: '按关键词检索知识库，返回最相关的若干条',   // ← 这句是给 LLM 看的
  inputSchema: {
    type: 'object',
    properties: { query: { type: 'string', description: '检索关键词' } },
    required: ['query'],
  },
  sideEffect: 'read',
  featureFlag: 'knowledgeBase',   // 可选：未授权则注册表不暴露它
  run: async ({ query }) => searchKnowledge(query),   // ← 你已有的能力函数
});
```

**2. 人触发（快捷键、按钮）：**

```ts
const r = await toolRegistry.invoke('kb.search', { query: '发票' }, { flags, scope });
```

**3. 交给 LLM（只读）：**

```ts
// 喂给模型的清单：已过滤未授权 / 不在当前场景 / 所有 send 类
const tools = toolRegistry.toFunctionSchemas({ flags, scope, autonomous: true });

// 模型回调时走只读出口，它会再查一次副作用等级
const r = await toolRegistry.invokeReadOnly(call.name, JSON.parse(call.arguments), { flags, scope });
```

**4. 跑一条固定计划：**

```ts
const plan = {
  name: '看这个会话在问什么',
  steps: [
    { tool: 'session.fetch', buildInput: (c) => (c.seed.id ? { id: c.seed.id } : undefined) },
    { tool: 'kb.search', buildInput: (c) => {
        const turns = c.byTool['session.fetch']?.data;
        const last = [...(turns ?? [])].reverse().find((t) => t.role === 'customer');
        return last ? { query: last.text.slice(0, 200) } : undefined;   // undefined = 跳过本步
      } },
  ],
};
await runPlan(plan, { id: 42 }, { flags, scope });
```

> ⚠ **`send` 类 tool 的 `run()` 里不该有发送代码。** 它只准备 payload、返回 `requiresConfirm`，真发在上层人确认之后。这不是约定而是结构——就算绕过所有闸门调用它，也发不出去。

## 验证

三个文件都过了 `tsc --strict`，另有 18 项行为测试覆盖：三道闸各自的放行与拒绝、`send` 在自主语境被拒且回 `requiresConfirm`、自主清单里不含 `send`、wire 名往返、只读出口拒绝 `write`、幻觉名返回结构化 error 而非抛异常、`run` 抛错被转成结果、编排器结果串联 / 缺 seed 整链跳过 / `stopIf` 提前结束 / 编排器里的 `send` 被拦。

## 代码

```ts
// src/tools/tool.ts
/**
 * Tool 层核心接口 —— 把已有的能力函数包装成统一、可枚举、可编排的 tool。
 *
 * 只加一层薄 wrapper + 元数据（schema / 描述 / 副作用标注），**不改被包的函数**。
 */

/**
 * 副作用等级 —— 决定 agent / LLM 能否自主调用。
 *
 * 分级依据是两个具体问题，不是「危险程度」这种模糊感觉：
 *   做完之后能撤销吗？外部的人会立刻看到吗？
 *
 * - `read`  纯读：查数据、检索、抓取 → 可自主
 * - `write` 改本地状态：填表单、提交 → 固定流程里可自主；**LLM 自主调不行**
 * - `send`  对外发出去：给客户发消息、发邮件 → **永远人确认**，LLM 绝不可自主
 */
export type SideEffect = 'read' | 'write' | 'send';

/** 极简 JSON Schema 子集：够描述 function-call 参数即可，不引入依赖。 */
export interface JsonSchema {
  type: 'object';
  properties: Record<
    string,
    {
      type: 'string' | 'number' | 'boolean' | 'array' | 'object';
      description?: string;
      enum?: readonly (string | number)[];
      default?: unknown;
    }
  >;
  required?: string[];
}

export interface ToolResult<O = unknown> {
  ok: boolean;
  data?: O;
  error?: string;
  /**
   * ★ `send` 类 tool 的 run() **永不真发** —— 它只准备 payload、返回 requiresConfirm，
   *   由上层界面拿给人确认之后再走真发分支。
   *   这不是约定而是结构：就算绕过所有闸门调用它，也发不出去。
   */
  requiresConfirm?: boolean;
  confirmPayload?: unknown;
}

export interface Tool<I = any, O = any> {
  /** 唯一名，建议 `<域>.<能力>`。也是给 LLM 的 function name（出口会做 wire 映射） */
  name: string;
  /** 功能域，用来分组和过滤 */
  domain: string;
  /** 给人看的短名（面板 / 日志） */
  title: string;
  /** 给 LLM 看的自然语言描述：说清楚做什么、什么时候该用 */
  description: string;
  inputSchema: JsonSchema;
  sideEffect: SideEffect;
  /** 仅在这些页面 / 场景可用。省略 = 任意 */
  scopes?: string[];
  /** 该能力未授权则注册表不暴露此 tool。省略 = 不 gate */
  featureFlag?: string;
  run(input: I): Promise<ToolResult<O>>;
}

/** 给 LLM 的 function schema 形态（OpenAI 风格）。 */
export interface FunctionSchema {
  name: string;
  description: string;
  parameters: JsonSchema;
}

/** 调用上下文 —— 注册表据此做 gating 与安全闸。 */
export interface InvokeCtx {
  /** 当前场景标识，与 Tool.scopes 比对。null = 未知 */
  scope?: string | null;
  /** 当前身份的能力开关，与 Tool.featureFlag 比对 */
  flags?: Record<string, boolean>;
  /**
   * ★ 是否 LLM 自主编排（true = 模型自己决定调什么，非人主动触发）。
   *   autonomous 时注册表拒绝 send 类，强制回到人确认路径。
   */
  autonomous?: boolean;
}
```

```ts
// src/tools/registry.ts
/**
 * ToolRegistry —— tool 的注册 / 枚举 / 统一调用入口。
 *
 * 统一 invoke 在这里过三道闸：
 *   1. featureFlag 未授权        → 拒绝
 *   2. scope 不匹配              → 拒绝
 *   3. sideEffect==='send' 且 autonomous → 拒绝（强制回人确认路径）
 */

import type {
  Tool,
  ToolResult,
  FunctionSchema,
  InvokeCtx,
  SideEffect,
} from './tool';

/* ------------------------------------------------------------------ *
 * wire 名映射
 *
 * OpenAI 规范里 function name 只允许 ^[a-zA-Z0-9_-]{1,64}$ —— **点号不合法**。
 * 严格校验的网关会直接 400，而且不告诉你是哪个字段错。
 *
 * 所以在出口做一层映射，**注册表内部的命名一个字不动**（那套名字通常已经被
 * 快捷键、编排器、调试钩子广泛引用，全局重命名的代价远大于加两行映射）。
 * ------------------------------------------------------------------ */
export const toWireName = (name: string): string => name.replace(/\./g, '__');
export const fromWireName = (name: string): string => name.replace(/__/g, '.');

export class ToolRegistry {
  private tools = new Map<string, Tool>();

  register(tool: Tool): this {
    if (this.tools.has(tool.name)) {
      console.warn(`[tools] 重复注册，覆盖：${tool.name}`);
    }
    this.tools.set(tool.name, tool);
    return this;
  }

  get(name: string): Tool | undefined {
    // 两种名字都认：内部名和 wire 名
    return this.tools.get(name) ?? this.tools.get(fromWireName(name));
  }

  /** 列出 tool，可按域 / 副作用 / 场景过滤。 */
  list(opts?: { domain?: string; sideEffect?: SideEffect; scope?: string | null }): Tool[] {
    let out = [...this.tools.values()];
    if (opts?.domain) out = out.filter((t) => t.domain === opts.domain);
    if (opts?.sideEffect) out = out.filter((t) => t.sideEffect === opts.sideEffect);
    if (opts?.scope != null) {
      out = out.filter((t) => !t.scopes || t.scopes.includes(opts.scope as string));
    }
    return out;
  }

  /**
   * 给 LLM 的 function schema 数组（OpenAI 风格，名字已转 wire 形态）。
   *
   * ★ 防护第一层：autonomous 时 send 类**根本不进清单**。
   *   模型不会去用它不知道存在的东西——这是最有效的一层。
   */
  toFunctionSchemas(ctx: InvokeCtx): FunctionSchema[] {
    return [...this.tools.values()]
      .filter((t) => this.isAvailable(t, ctx).ok)
      .filter((t) => !(ctx.autonomous && t.sideEffect === 'send'))
      .map((t) => ({
        name: toWireName(t.name),
        description: t.description,
        parameters: t.inputSchema,
      }));
  }

  /**
   * 统一调用入口：过三道闸后调 tool.run。
   * **任何阶段被拒都返回结构化 error，绝不向外抛** —— 调用方（尤其是 LLM 循环）
   * 需要的是一个能塞回对话的结果，不是一个异常。
   */
  async invoke(name: string, input: unknown, ctx: InvokeCtx = {}): Promise<ToolResult> {
    const tool = this.get(name);
    if (!tool) return { ok: false, error: `未知 tool：${name}` };

    const gate = this.isAvailable(tool, ctx);
    if (!gate.ok) return { ok: false, error: gate.error };

    // ★ 闸 3：send + 自主编排 → 绝不放行
    if (tool.sideEffect === 'send' && ctx.autonomous) {
      return {
        ok: false,
        requiresConfirm: true,
        error: `${tool.name} 会对外发送，禁止自主调用 —— 须人确认`,
      };
    }

    try {
      return await tool.run(input);
    } catch (e) {
      return { ok: false, error: e instanceof Error ? e.message : String(e) };
    }
  }

  /**
   * ★ 防护第二层：给 LLM 循环用的出口，**执行前再查一次副作用等级**。
   *
   * 防的是模型幻觉出一个不存在的名字——它可能猜出 `ticket.submit` 这种看起来
   * 很合理的名字，哪怕你从没给过它。
   */
  async invokeReadOnly(wireName: string, input: unknown, ctx: InvokeCtx = {}): Promise<ToolResult> {
    const tool = this.get(wireName);
    if (!tool) return { ok: false, error: `未知 tool：${wireName}` };
    if (tool.sideEffect !== 'read') {
      return { ok: false, error: `${tool.name} 不是只读 tool，禁止自主调用` };
    }
    return this.invoke(tool.name, input, { ...ctx, autonomous: true });
  }

  private isAvailable(tool: Tool, ctx: InvokeCtx): { ok: boolean; error?: string } {
    if (tool.featureFlag && !ctx.flags?.[tool.featureFlag]) {
      return { ok: false, error: `功能未授权：${tool.featureFlag}` };
    }
    if (tool.scopes && (ctx.scope == null || !tool.scopes.includes(ctx.scope))) {
      return {
        ok: false,
        error: `不在适用场景（需 ${tool.scopes.join('/')}，当前 ${ctx.scope ?? 'null'}）`,
      };
    }
    return { ok: true };
  }
}

/** 全局单例。各域在自己的 `<域>-tools.ts` 里注册。 */
export const toolRegistry = new ToolRegistry();
```

```ts
// src/tools/orchestrator.ts
/**
 * 最小编排器 —— 按「计划」顺序调 tool，把每步结果喂给下一步。
 *
 * 计划是**手写的固定链**：可回归、可单测、出问题能精确定位到哪一步。
 * LLM 生成的动态计划也是同样的结构，但那条路要先解决可预测性，见 README。
 *
 * 安全：编排器一律以 **autonomous 语境**调用 → 注册表会拒绝 send 类。
 *      要对外发送必须由上层人确认路径单独触发，编排器不碰 send。
 */

import type { ToolResult } from './tool';
import { toolRegistry } from './registry';

export interface PlanContext {
  /** 计划的初始输入 */
  seed: Record<string, unknown>;
  /** 已完成步骤的结果，按 tool 名索引（同名后者覆盖前者） */
  byTool: Record<string, ToolResult | undefined>;
  /** 按序号索引 */
  bySeq: (StepOutcome | undefined)[];
}

export interface AgentStep {
  tool: string;
  /** 静态输入；与 buildInput 二选一（buildInput 优先） */
  input?: unknown;
  /**
   * 动态输入：读之前步骤的结果算出本步 input。
   * 返回 undefined = **跳过本步**（前置条件不满足）。
   */
  buildInput?: (ctx: PlanContext) => unknown | undefined;
  /** 拿到本步结果后判断是否提前结束整个计划 */
  stopIf?: (result: ToolResult, ctx: PlanContext) => boolean;
}

export interface AgentPlan {
  name: string;
  steps: AgentStep[];
}

export interface StepOutcome {
  tool: string;
  skipped: boolean;
  result?: ToolResult;
}

export interface PlanRunResult {
  ok: boolean;
  planName: string;
  outcomes: StepOutcome[];
  /** 若因 stopIf 提前结束，记在哪一步停的 */
  stoppedAt?: number;
}

/**
 * 执行一个计划。
 *
 * 任一步返回 ok:false **不一定中止** —— 由计划自己的 stopIf 决定。
 * 默认继续并记录失败：一步查不到东西，后面几步往往仍然有意义。
 */
export async function runPlan(
  plan: AgentPlan,
  seed: Record<string, unknown> = {},
  ctxBase: { scope?: string | null; flags?: Record<string, boolean> } = {},
): Promise<PlanRunResult> {
  const ctx: PlanContext = { seed, byTool: {}, bySeq: [] };
  const outcomes: StepOutcome[] = [];

  for (let i = 0; i < plan.steps.length; i++) {
    const step = plan.steps[i];
    const input = step.buildInput ? step.buildInput(ctx) : step.input;

    if (input === undefined && step.buildInput) {
      const oc: StepOutcome = { tool: step.tool, skipped: true };
      outcomes.push(oc);
      ctx.bySeq[i] = oc;
      continue;
    }

    // autonomous:true → 注册表拒绝 send 类；read / write 正常
    const result = await toolRegistry.invoke(step.tool, input ?? {}, {
      ...ctxBase,
      autonomous: true,
    });

    const oc: StepOutcome = { tool: step.tool, skipped: false, result };
    outcomes.push(oc);
    ctx.bySeq[i] = oc;
    ctx.byTool[step.tool] = result;

    if (step.stopIf?.(result, ctx)) {
      return { ok: true, planName: plan.name, outcomes, stoppedAt: i };
    }
  }

  return {
    ok: !outcomes.some((o) => o.result && !o.result.ok),
    planName: plan.name,
    outcomes,
  };
}
```

---

**设计与踩坑** → [给 agent 的 tool 分级：read / write / send，以及为什么 send 永远不给 LLM](/blog/agent-tool-side-effects)