给多Agent加上预算限制与人工审批

theAuth 是面向 AI 智能体和人类用户的开源认证库。在 GitHub 上为仓库点亮 Star · 阅读文档 · 运行快速入门示例 · theauth.dev

有一次，我的一个规划智能体在周五晚上把任务分派给了六个工作智能体。到了周六早上，其中一个工作智能体还在不断重试一项失败的摘要任务，而模型服务商的控制台上出现了一条我不想看到的费用记录。没有人入侵系统，所有凭据都有效。系统只是完全不知道，“有效”和“负担得起”是两个不同的问题。

那个周末让我明白，智能体集群除了身份认证，还需要三项控制措施：范围明确、会过期的授权；真正能阻止调用的支出上限；以及在执行不可撤销的操作之前，由人工把关。

本指南会使用 theAuth 构建这三项控制措施。它是面向 AI 智能体和人类用户的开源认证库，在 npm 上的包名是 @glinr/theauth。完成后，你将拥有一个向工作智能体委派限定范围、短期访问权限的规划智能体，一个在某个阈值发出警告、在另一个阈值阻止调用的预算护栏，一套针对破坏性操作的审批流程，以及一份按委派链汇总的成本报告。

我在之前的一篇文章中已经介绍了子智能体委派的基础知识。这篇文章从那里接着讲：委派链建立之后，如何处理费用和高风险操作。

这是 theAuth 系列指南中的第 6 篇，共 8 篇。它可以独立阅读，你可以直接从这里开始。本文会用到第 5 篇指南中的智能体身份。如果你已经知道如何创建智能体，就从这里开始。

面向人类用户开发？从第 1 篇指南开始。面向 AI 智能体开发？从第 5 篇指南开始。每篇指南都会链接到所涉及概念的文档页面。

请注意最后一列。只有委派和权限约束会在 authorize() 内部检查。预算和审批都是需要你主动调用的模块。我会反复强调这一点，因为大多数问题都源于对这一点的错误假设。

你需要 Node 20 或更新版本、一个包管理器，以及一个 TypeScript 项目。你还需要了解 theAuth 中的智能体身份是什么。如果不了解，请先阅读智能体身份文档，花五分钟就够了。

pnpm add @glinr/theauth

如果你想从一个可运行的应用开始，而不是从空项目起步，快速入门提供了只需一条命令的脚手架工具。下面的示例使用 SQLite，因此你可以在任何环境中运行它们。

开始之前，先坦诚说明一下适用范围。如果你希望在自己的进程和数据库中，集中管理身份、限定范围的权限和审计记录，theAuth 很适合。theAuth 不是计费系统。它不会阻止服务商向你收费，也不会在网络层限制支出。这里的一切之所以有效，是因为你的代码会先询问这个库，并遵守它给出的结果。

第 1 步：创建智能体集群

从一个实例和三个智能体开始。规划智能体持有较广泛的权限。摘要智能体初始没有任何权限，之后会通过委派获得访问权限。清理智能体持有自己的一组有限权限，其中包括需要审批的删除权限。

import { createTheAuth } from '@glinr/theauth';

const theauth = await createTheAuth({
  database: { provider: 'sqlite', url: 'theauth.db' },
  approval: {
    ttl: 600, // seconds
    onApprovalNeeded: async (request) => {
      console.log(
        `[approval] ${request.agentId} wants to ${request.action} ${request.resource} (${request.id})`,
      );
    },
  },
});

const planner = await theauth.agent.create({
  ownerId: 'user-123',
  name: 'planner',
  type: 'autonomous',
  permissions: [
    { resource: 'mcp:github:*', actions: ['read', 'write', 'comment'] },
    { resource: 'mcp:linear:*', actions: ['read', 'write'] },
  ],
});

const summarizer = await theauth.agent.create({
  ownerId: 'user-123',
  name: 'summarizer',
  type: 'delegated',
  permissions: [],
});

ownerId 必须对应一条真实存在的用户记录，所以请先创建或查找这个用户。智能体身份文档介绍了完整的生命周期，包括令牌轮换。每个智能体还会获得一个 bearer token，它只会显示一次，存储时保存的是哈希值。看到它时，就立即存入你的密钥存储系统。

为什么摘要智能体使用 delegated 类型？它表达了设计意图：这个智能体的存在，是为了借助委派而来的访问权限完成一项任务，随后退出。

第 2 步：委派满足任务所需的最小授权

现在，规划智能体将 GitHub issue 的读取权限交给摘要智能体，有效期为 30 分钟，并且不允许继续向下委派。

const chain = await theauth.delegate({
  fromAgent: planner.id,
  toAgent: summarizer.id,
  permissions: [{ resource: 'mcp:github:issues', actions: ['read'] }],
  expiresAt: new Date(Date.now() + 30 * 60_000),
  maxDepth: 1,
});

console.log(chain.id, chain.depth); // dlg_..., 1

委派文档中的三个细节，能为你省下不少调试时间。

首先是子集规则。规划智能体必须持有它要委派的每一项权限。如果请求的资源范围更广，或者包含它没有的操作权限，就会抛出错误。这种失败是合理的：即使规划智能体遭到入侵，也无法凭空创建自己从未拥有过的权限。

其次，每次调用都会单独检查 maxDepth。后续的委派跳转不会继承它作为深度上限。要阻止再次委派，请在代码中的每一次 delegate() 调用里都传入较小的 maxDepth。如果某处依赖默认值 3，委派链就可能比你计划的更深。

第三，子集检查使用的是发起委派的智能体自身的权限。仅持有委派访问权限的智能体，无法将这些权限继续传递下去。如果你希望中间的智能体能够再次委派，就要为它自身配置一份相同的权限。我觉得这一点有些出人意料，但它采取了更保守的安全策略。

检查摘要智能体实际能够执行哪些操作：

const allowed = await theauth.authorize(summarizer.id, {
  action: 'read',
  resource: 'mcp:github:issues',
});
// allowed.allowed === true

const denied = await theauth.authorize(summarizer.id, {
  action: 'write',
  resource: 'mcp:github:issues',
});
// denied.allowed === false

authorize() 会先检查智能体自身的权限，再检查委派而来的权限。你也可以直接列出它的委派权限集：

const effective = await theauth.delegation.getEffectivePermissions(summarizer.id);
console.log(effective);

当任务出问题时，你会希望只需一次调用，就能终止整个委派子树。

await theauth.delegation.revoke(chain.id);

撤销一个委派环节，也会撤销从接收权限的智能体发起的所有有效委派链。如果摘要智能体曾继续向下委派，那些委派环节也会一并失效。下一次 authorize() 调用会返回 allowed: false。需要记住一个限制：撤销不会中断已经在执行的操作。一个耗时较长的 HTTP 请求仍会继续运行，直到完成。

第 3 步：在每次 LLM 调用之前检查预算

委派控制的是智能体可以访问什么，至于它可以花多少钱，委派并不负责。为此，theAuth 在 theauth.policies 上提供了预算策略。请先读一遍预算策略文档，因为它的工作模型有一些特定规则。

这里有一点很容易被忽略：authorize() 不会检查预算策略。除非你的代码在 LLM 调用前调用 checkBudget()，并在调用后调用 recordUsage()，否则预算策略不会发挥任何作用。

为摘要智能体创建两项策略，一项软限制，一项硬限制：

await theauth.policies.create({
  agentId: summarizer.id,
  limits: { maxTokensCostPerDay: 800 },
  action: 'warn',
});

await theauth.policies.create({
  agentId: summarizer.id,
  limits: { maxTokensCostPerDay: 1000, maxCallsPerDay: 500 },
  action: 'block',
});

达到 800 个单位时，warn 策略会触发，checkBudget() 仍然返回 allowed: true，同时附带该策略，方便你通知相关人员。达到 1000 时，block 策略会返回 allowed: false。

现在，把模型调用封装起来，确保不会忘记检查预算：

async function guardedCall<T>(
  agentId: string,
  estimatedCost: number,
  run: () => Promise<{ value: T; actualCost: number }>,
): Promise<T> {
  const check = await theauth.policies.checkBudget(agentId, estimatedCost);

if (!check.allowed) {
    throw new Error(`Budget blocked for ${agentId}: ${check.reason}`);
  }
  if (check.reason) {
    console.warn(`[budget] soft limit reached for ${agentId}: ${check.reason}`);
  }

const { value, actualCost } = await run();
  await theauth.policies.recordUsage(agentId, actualCost);
  return value;
}

用它包裹任意模型调用：

const summary = await guardedCall(summarizer.id, 50, async () => {
  const result = await llm.complete('Summarize the open issues.');
  return { value: result.text, actualCost: result.usage.totalTokens };
});

llm 对象在这里代表你自己的客户端。这个封装只需要拿到一个数值。

action 字段有四个可选值：warn、throttle、block 和 revoke。在当前代码中，throttle、block 和 revoke 的行为完全相同，都会返回 allowed: false。revoke 不会替你撤销智能体的令牌。如果你想让智能体彻底停用，需要自行调用 theauth.agent.revoke()。

检查时只会根据 agentId 匹配策略。只设置 userId 或 tenantId 的策略没有指定智能体，因此会应用于所有智能体。这两个字段只能作为 list() 的筛选条件，无法限定预算检查的作用范围。如果你计划按租户设置支出上限，请为每个智能体分别创建策略，并设置租户字段用于筛选，而不要指望它负责匹配。多租户文档介绍了租户之间如何组织和协作。

另外，计数器不会自行重置，需要你安排定时任务。

// UTC midnight cron
const { reset } = await theauth.policies.resetDaily();
console.log(`Reset ${reset} policies`);

// First of the month
await theauth.policies.resetMonthly();

这两个调用都会作用于所有策略。当用量重新降到限额以下时，已触发的策略会恢复为活跃状态。

如果你只想严格限制每个智能体每小时的调用次数，完全不需要预算逻辑，那么限流页面介绍的工具更简单。

第 4 步：不可逆操作必须经过人工审批

预算限制的是用量，审批把关的是操作。有些操作应该先由人点头：删除生产环境文件、转移资金、扩大权限。

theAuth 将其建模为权限约束 requireApproval。给清理智能体授予读取权限，以及需要人工审批的删除权限：

const cleaner = await theauth.agent.create({
  ownerId: 'user-123',
  name: 'file-cleaner',
  type: 'autonomous',
  permissions: [
    { resource: 'file:prod-data/*', actions: ['read'] },
    {
      resource: 'file:prod-data/*',
      actions: ['delete'],
      constraints: { requireApproval: true },
    },
  ],
});

当清理智能体尝试删除时，authorize() 会拒绝请求并说明原因。你的应用识别出这个原因后，就发起审批请求。审批文档描述的也是同一套流程。

async function tryDelete(path: string) {
  const attempt = {
    action: 'delete',
    resource: `file:prod-data/${path}`,
    arguments: { path: `/prod/${path}` },
  };

const result = await theauth.authorize(cleaner.id, attempt);

if (result.allowed) {
    return { done: true as const };
  }

if (result.reason?.includes('requires human approval')) {
    const request = await theauth.approval.request({
      agentId: cleaner.id,
      userId: 'user-123',
      ...attempt,
    });
    return { done: false as const, approvalId: request.id };
  }

throw new Error(`Denied: ${result.reason}`);
}

智能体不会阻塞等待。它拿到 approvalId 后就继续执行。审批人可以在几分钟或几小时后回复，只要没有超过你设置的 TTL（默认是 5 分钟，上面的配置将其提高到了 600 秒）。

theAuth 会存储请求，但不会发送通知。你需要通过 onApprovalNeeded（第 1 步中已展示）或 webhookUrl，决定如何通知审批人。

const theauth = await createTheAuth({
  database: { provider: 'sqlite', url: 'theauth.db' },
  approval: {
    ttl: 300,
    webhookUrl: process.env.APPROVAL_WEBHOOK_URL,
  },
});

这个 webhook 只是一个普通的 POST 请求，携带 approval_needed 事件和请求正文。它没有签名请求头，也不会重试。把它当作提醒，而不是可靠送达的保证。即使发送失败，请求仍然已被存储，定期调用 listPending() 扫描就能找到它。如果接收这个 webhook 的处理程序前面没有你自己的验证层，就不要在其中放置任何敏感信息。

批准、拒绝，然后真正执行操作

你的后端提供两个路由。传入审批人的身份，让记录能够说明是谁做出了决定。

app.post('/approvals/:id/approve', async (req, res) => {
  const updated = await theauth.approval.approve(req.params.id, req.user.email);
  res.json({ status: updated.status });
});

app.post('/approvals/:id/deny', async (req, res) => {
  const updated = await theauth.approval.deny(req.params.id, req.user.email);
  res.json({ status: updated.status });
});

接下来是容易让人踩坑的地方。审批只是记录，不是绕过权限检查的通行证。人工批准后，对该删除操作调用 authorize()，仍然会返回 allowed: false，因为权限依然带有 requireApproval 约束。它不会查询审批记录。

状态变更后，由你的代码自行执行操作：

async function finishDelete(approvalId: string, path: string) {
  const request = await theauth.approval.get(approvalId);

if (request?.status === 'approved') {
    await deleteFile(path); // your own code path, not authorize()
    return 'deleted';
  }
  if (request?.status === 'denied') return 'denied';
  return 'still pending or expired';
}

让这个函数成为唯一执行删除的地方。如果还有另一条代码路径可以不检查审批状态就执行删除，那么你做出来的只是一个勾选框，而不是一道关卡。

还有两个小细节。请求状态不是 pending 时，approve() 和 deny() 都会抛出异常。而且 approve() 不会检查 expiresAt。超过 TTL 的请求在你运行清理之前，会一直保持 pending 状态，所以要安排定时清理：

const { expired } = await theauth.approval.cleanup();
console.log(`Expired ${expired} stale approval requests`);

再配上一个通过 listPending(userId) 列出待审批请求的界面，审批人就能在批准之前，清楚地看到智能体想做什么，以及操作携带的参数。

第 5 步：追踪每一美元的归属

预算策略只会累计你传入的数值。它不会告诉你哪个工具花掉了这些钱，也不会告诉你某项任务属于哪条委派链。要获得这些信息，你需要成本归因。请参阅成本归因页面。

这个模块是独立的。你需要使用数据库句柄来创建它：

import { createCostAttributionModule } from '@glinr/theauth/auth';

const costs = createCostAttributionModule(theauth.db, {
  currency: 'USD',
  retentionDays: 90,
  alertThresholds: { warn: 5.0, critical: 20.0 },
  onAlert: async (alert) => {
    console.warn(
      `[cost] ${alert.type} agent=${alert.agentId} spent=$${alert.currentCostUsd.toFixed(4)} limit=$${alert.threshold} period=${alert.period}`,
    );
    if (alert.type === 'budget_exceeded') {
      await theauth.agent.revoke(alert.agentId);
    }
  },
});

记录每次调用时带上委派链 ID，这样无论是哪个子智能体产生的支出，都能汇总到这项任务的成本中：

await costs.recordCost({
  agentId: summarizer.id,
  tool: 'openai:gpt-4o-mini',
  inputTokens: 1200,
  outputTokens: 300,
  costUsd: 0.0004,
  delegationChainId: chain.id,
  metadata: { job: 'nightly-issue-digest' },
});

之后，查询整条委派链的成本：

const report = await costs.getDelegationChainCost(chain.id);
if (report.success) {
  console.log(report.data.totalCostUsd.toFixed(4));
  console.log(report.data.byTool);
}

每个方法都会返回一个 Result 对象，因此读取 data 之前要先检查 success。其他实用的查询方法还有 getAgentCost()、getOwnerCost() 和 getTopAgentsByCost(10)。最后一个方法回答的，正是那个星期六我一直在问的问题：到底是哪个智能体干的。

需要提前考虑的告警行为

告警采用电平触发机制。只要支出仍高于阈值，每次调用 recordCost() 都会再次触发告警。如果你通过 onAlert 呼叫人工处理，先做好去重。用智能体和告警类型作为键的简单内存集合，在单进程下就能奏效；一旦运行多个进程，就需要共享存储。

warn 和 critical 阈值检查的是过去 24 小时的滚动窗口。budget_exceeded 告警则会将自然月支出，与该智能体所有策略中最小的 maxTokensCostPerMonth 进行比较。

成本模块的 checkBudget() 与策略模块中的同名函数是两个独立的函数：

const status = await costs.checkBudget(summarizer.id);

if (status.success && !status.data.withinBudget) {
  throw new Error('Monthly cost budget reached');
}

它只读取每月 token 成本限额，忽略策略动作。它还会跳过没有匹配 agentId 的策略。务必统一单位。如果你向 recordCost() 传入美元，向 recordUsage() 传入 token 数量，这两个系统的结果就完全对不上。为限额选定一个单位（我使用美元），并以这个单位向两个模块提供数据。

第 6 步：用一个统一的护栏把关所有调用

这是我实际交付的代码结构。一个函数依次完成预算检查、模型调用、成本记录和用量记录。智能体只调用这个函数。

async function runModelCall(opts: {
  agentId: string;
  chainId: string;
  tool: string;
  estimateUsd: number;
  call: () => Promise<{ text: string; inTok: number; outTok: number; usd: number }>;
}): Promise<string> {
  const policy = await theauth.policies.checkBudget(opts.agentId, opts.estimateUsd);
  if (!policy.allowed) throw new Error(`Policy blocked: ${policy.reason}`);

const monthly = await costs.checkBudget(opts.agentId);
  if (monthly.success && !monthly.data.withinBudget) {
    throw new Error('Monthly cost budget reached');
  }

const out = await opts.call();

await costs.recordCost({
    agentId: opts.agentId,
    tool: opts.tool,
    inputTokens: out.inTok,
    outputTokens: out.outTok,
    costUsd: out.usd,
    delegationChainId: opts.chainId,
  });
  await theauth.policies.recordUsage(opts.agentId, out.usd);

return out.text;
}

没错，这里有两次预算检查。它们读取的是不同的计数器，我宁愿其中任意一个触发时就阻止调用，也不想凌晨两点才发现漏洞。重复记账，是这两个原本就没有设计为共享计数器的模块所带来的代价。

有一种竞态需要了解。两个并行调用可能在任意一个记录用量之前，都通过 checkBudget()，因此严格限额仍可能被突破，超出的金额相当于正在执行的调用所产生的成本。如果这对你很重要，就按智能体串行执行调用，或者传入留有充足余量的 estimateUsd。

第 7 步：调用其他智能体的智能体

到这里为止，整个智能体集群都运行在同一个进程中。一旦智能体分布在不同服务里，就需要相互进行身份认证。theAuth 为 Google Agent-to-Agent 协议提供了 A2A 服务端和客户端，A2A 文档页面对两者都有说明。

服务端使用同一套身份系统验证调用方的 bearer token：

import { createAgentCard, createA2AServer } from '@glinr/theauth/a2a';

const card = createAgentCard({
  agent: { id: summarizer.id, name: summarizer.name, type: summarizer.type },
  url: process.env.A2A_PUBLIC_URL!,
  description: 'Summarizes issues and threads',
  version: '1.0.0',
  skills: [
    {
      id: 'summarize',
      name: 'Summarize',
      description: 'Condenses a thread into key points',
      tags: ['summary'],
    },
  ],
  securitySchemes: {
    bearer: { type: 'http', scheme: 'bearer', bearerFormat: 'JWT' },
  },
  security: [{ bearer: [] }],
});

const server = createA2AServer({
  agentCard: card,
  handler: { onMessage: handleSummarizeMessage },
  authenticate: async (request) => {
    const token = request.headers.get('Authorization')?.replace('Bearer ', '');
    if (!token) return null;
    const caller = await theauth.agent.validateToken(token);
    return caller?.id ?? null;
  },
  onAudit: async (event) => {
    console.log('[a2a]', event.method, event.agentId, event.success);
  },
});

将 server.handleRequest(req) 挂载到 /a2a 和 /.well-known/agent.json。handleSummarizeMessage 函数由你实现：它接收消息并返回一个任务对象。在函数内部调用第 6 步中的 runModelCall，这样远程请求就会受到与本地请求相同的预算护栏约束。

文档中有两点提醒。如果省略 authenticate，服务端就会接受未经身份认证的调用。另外，默认的任务存储是内存中的 map，因此重启后任务就会消失。实际使用时，请传入你自己的 taskStore。

A2A 身份认证能够证明调用方是谁，但本身并不携带委托授权。如果远程智能体需要以受限权限执行操作，就在你这一侧创建委托，并像之前一样，将费用记到它的委托链 ID 下。

一次故障的完整过程

下面看看出问题时，这些组件如何协同工作。假设摘要智能体在 02:10 陷入了重试循环。

循环不断调用 runModelCall()。每一轮都会执行 checkBudget()、发起调用、将费用记到委托链下，并记录用量。达到 800 单位时，warn 策略会附上一条原因，日志记录器则输出一条软限制日志。没人醒着看日志，所以情况没有任何变化。

当 24 小时内的支出达到 5 美元时，费用模块触发 warn 告警。达到 20 美元时，触发 critical 告警。你的 onAlert 处理函数会通知值班人员，而且每记录一笔费用都会重复触发告警，因此通知需要去重，否则就会响上 40 次。

达到 1000 单位时，block 策略返回 allowed: false，护栏随即抛出异常。循环在下一轮迭代时终止。智能体仍在运行，仍持有委托授权，也仍然能通过 authorize()。停止的只有支出。

这个区别很重要。停止支出和停止访问是两种不同的控制手段。如果你还想取消访问权限，可以通过 theauth.delegation.revoke(chain.id) 撤销委托链，或通过 theauth.agent.revoke() 撤销智能体。我会让 budget_exceeded 告警触发智能体撤销，而把撤销委托链留给人工主动执行，避免一项合理的大型任务因为某个异常频发的夜晚而失去整棵委托树。

再对比一下审批流程。如果同一个失控循环尝试执行删除操作，第一次尝试会创建一条请求。如果循环反复重试删除，每次尝试都会创建一条新请求，除非你保存了 approvalId，并先调用 get() 检查。维护一个从待执行操作到其待审批请求的映射，重复尝试时返回已有的 ID。否则，审批人员打开队列时，就会看到 300 条一模一样的记录。

先测试护栏，再信任它

只有亲眼看到预算护栏阻止调用，我才会相信它。写一个临时脚本，故意耗尽 3 单位的预算。

const probe = await theauth.agent.create({
  ownerId: 'user-123',
  name: 'budget-probe',
  type: 'autonomous',
  permissions: [],
});

await theauth.policies.create({
  agentId: probe.id,
  limits: { maxTokensCostPerDay: 3 },
  action: 'block',
});

for (let i = 1; i <= 5; i++) {
  const check = await theauth.policies.checkBudget(probe.id, 1);
  console.log(`call ${i}: allowed=${check.allowed}`);
  if (check.allowed) await theauth.policies.recordUsage(probe.id, 1);
}

你应该看到第 1 到第 3 次调用通过，第 4 和第 5 次调用被拒绝。如果第 4 次调用通过了，说明你的计量单位或 recordUsage() 调用有问题。在将护栏接入实际智能体之前，先修好它。

对审批也做同样的测试。创建一条请求，确认 listPending() 能返回它，批准该请求，再确认 get() 报告的状态为 approved。然后，对同一个删除操作调用 authorize()，观察它拒绝授权。亲眼见过一次这样的拒绝，就能避免以后写出错误的执行路径。

在生产环境中运行

演示结束后，养成几个习惯，才能让这套配置持续可靠地运作。

安排四个定时任务。在 UTC 午夜运行 resetDaily()，在每月 1 日运行 resetMonthly()。每 5 分钟运行一次 approval.cleanup()，让过期的请求转为 expired 状态。每晚运行 costs.cleanup()，让费用事件按设定的保留期限清理。

将 authorize() 返回的 auditId 与你自己的请求 ID 一起写入日志。当有人问智能体为什么做了某件事时，审计记录能提供决策信息，而你可以补上相关上下文。

每周使用 getTopAgentsByCost(10) 检查支出最高的智能体。在我的集群中，有一个智能体的费用比另外九个加起来还高，而每次找出的罪魁祸首都和我猜的不一样。

最后，委托授权的有效期应与任务时长一致，而不是按一天来设置。30 分钟的任务就授予 30 分钟的权限。授权到期时，访问权限也随之失效，你无需记得再做任何处理。

故障排查与注意事项

智能体达到上限后仍在继续花钱

检查你的代码是否在模型调用之前执行了 checkBudget()，并对 allowed: false 作出相应处理。这个库不会拦截你的 HTTP 客户端。十有八九都是这个原因。

针对单个用户的策略