按成本约束AI Agent接口

系列文章：AI in TypeScript——全套 5 本，从第一次调用 LLM 到让 Agent 上线生产环境——五本都在这里

我的项目：Hermes IDE | GitHub——一款为使用 Claude Code 和其他 AI 编程工具交付产品的开发者打造的 IDE

关于我：xgabriel.com | GitHub

标准的限流机制统计的是请求次数。当每个请求的成本大致相同时，这种方式很有效——每分钟 100 个请求，就相当于 100 个单位的负载。

但 Agent endpoint 并不是这样。一个请求可能只经过两轮对话，成本不到一美分；另一个请求却可能执行二十轮、读取六份文档，成本高达两欧元。相同的 route、相同的 method，成本却相差两个数量级。

如果按请求次数限流，要么会毫无意义地限制那些低成本请求，要么会放任少数高成本请求在午饭前就花光你整个月的预算。

限制真正会耗尽的三种资源

资金。也就是你向服务提供商支付的费用。这才是那个会终结职业生涯的指标。

并发量。包括服务提供商允许的并行请求数，以及你自己的 worker pool 容量。超过上限后，上游会返回 429，而这在用户眼中体现为延迟。

请求数。它仍然值得保留，作为一种粗粒度的滥用防护手段，只是不应成为主要控制方式。

export type Limits = {
  costUsdPerHour: number;
  costUsdPerDay: number;
  concurrent: number;
  requestsPerMinute: number;
};

export const LIMITS: Record<Plan, Limits> = {
  free:  { costUsdPerHour: 0.50, costUsdPerDay: 2,  concurrent: 1, requestsPerMinute: 5 },
  pro:   { costUsdPerHour: 5,    costUsdPerDay: 40, concurrent: 3, requestsPerMinute: 60 },
  team:  { costUsdPerHour: 40,   costUsdPerDay: 300, concurrent: 10, requestsPerMinute: 300 },
};

同时设置每小时和每日限额。每小时限额可以阻止一小时内失控的循环；每日限额则可以阻止那种持续消耗、却始终不会触发小时限额的缓慢支出。

问题在于：执行结束前，你并不知道成本

其他限流器都可以在工作开始前扣减额度。但在这里，只有等运行结束后才能知道成本。

做法是先预留一笔预估费用，再根据实际费用进行结算：

export async function reserve(userId: string, estUsd: number) {
  const key = `cost:${userId}:${hourBucket()}`;
  const used = await redis.incrByFloat(key, estUsd);
  await redis.expire(key, 7200);

  if (used > LIMITS[planOf(userId)].costUsdPerHour) {
    await redis.incrByFloat(key, -estUsd);          // give it back
    throw new RateLimited("cost", await ttl(key));
  }
  return { key, estUsd };
}

export async function settle(r: Reservation, actualUsd: number) {
  await redis.incrByFloat(r.key, actualUsd - r.estUsd);   // correct the difference
}

incrByFloat 是原子操作，因此并发请求不可能在本应只有一个请求通过时同时通过检查。预估值应当偏保守——可以取运行成本中位数的两倍左右，因为低估会让一批请求在完成结算前一起穿透限制。

应在 finally 中完成结算，包括执行失败的情况。一次运行即使消耗了一欧元后超时，这一欧元也已经实实在在地花掉了：

const r = await reserve(user.id, estimate(intent));
try {
  const out = await runAgent(input, ctx);
  await settle(r, out.costUsd);
  return out;
} catch (err) {
  await settle(r, ctx.spentSoFar());          // partial spend still counts
  throw err;
}

并发控制需要的是 slot，而不是 counter

const SLOT_TTL = 300;                          // longer than any run

export async function acquire(userId: string): Promise<Slot> {
  const limit = LIMITS[planOf(userId)].concurrent;
  const id = crypto.randomUUID();
  const key = `slots:${userId}`;

  const n = await redis.zcard(key);
  if (n >= limit) throw new RateLimited("concurrency");

  await redis.zadd(key, Date.now() + SLOT_TTL * 1000, id);
  await redis.expire(key, SLOT_TTL * 2);
  return { key, id };
}

export async function release(s: Slot) {
  await redis.zrem(s.key, s.id);
}

这里应使用以过期时间为 score 的 sorted set，而不是普通 counter。因为进程如果在运行中途被终止，就不会再对 counter 执行递减；几次崩溃之后，用户就会永久处于已达到限额的状态。

在计数前，先清理已经过期的成员：

await redis.zremrangebyscore(key, 0, Date.now());

这样可以自我修复，而 counter 做不到。

用排队代替拒绝

对于一个原本就需要 30 秒才能完成的 endpoint，返回 429 并不划算。如果客户端无论如何都要等待，不如让它在队列中等待，这样你还能控制处理顺序。

const job = await queue.add("agent-run", { userId, input }, {
  priority: user.plan === "free" ? 10 : 1,
  jobId: `${userId}:${hash(input)}`,          // dedupe identical retries
});

return res.status(202).json({
  runId: job.id,
  statusUrl: `/runs/${job.id}`,
});

返回 202 和一个轮询 URL，可以把限流从一次失败转化为一次等待。通过 jobId 对相同 payload 去重，还能吸收重复提交，否则同一任务会产生两次费用。

worker 负责执行全局并发限制。无论当前有多少活跃用户，这才是保护服务提供商并发额度的机制：

new Worker("agent-run", handler, { concurrency: 8 });

告诉客户端发生了什么

res.set({
  "RateLimit-Policy": `${limits.costUsdPerHour};w=3600;unit=usd`,
  "RateLimit-Remaining": remaining.toFixed(3),
  "RateLimit-Reset": String(resetSeconds),
  "Retry-After": String(resetSeconds),
});
return res.status(429).json({
  error: "cost_limit_exceeded",
  message: `Hourly budget of $${limits.costUsdPerHour} reached. Resets in ${mins} minutes.`,
  upgradeUrl: "/billing",
});

同时提供机器可读的错误码和人类可读的提示信息。客户端根据错误码执行分支逻辑，而提示信息则展示在 UI 中。Retry-After 可以阻止客户端在明知已触发限制后仍不断发起请求。

人们经常忘记的两个限制

单次运行成本。前面的限制针对的是用户在一段时间内的累计使用量。单次运行同样需要设置上限，否则一个异常循环可能在九十秒内耗尽整小时的预算：

if (ctx.budget.spent > flag.maxCostUsdPerRun) throw new BudgetExceeded();

全局成本。汇总所有用户的支出，作为最后一道防线，防止你自己的限流器出现 bug：

const global = await redis.incrByFloat(`cost:global:${hourBucket()}`, estUsd);
if (global > GLOBAL_HOURLY_USD) {
  metrics.increment("agent.global_limit_hit");
  throw new ServiceBusy();
}

可以将它设置为正常峰值的三倍左右。它本不应该被触发，而真正触发的那一天，它会成为你写过的最划算的告警。

观察限流器的运行情况

metrics.increment("ratelimit.block", 1, { reason, plan });
metrics.histogram("ratelimit.est_error", actualUsd - estUsd, { intent });
metrics.gauge("ratelimit.util", used / limits.costUsdPerHour, { plan });

上线后，首先应该检查的是 est_error。如果它持续呈负偏态，说明你预留的费用过高，阻止了那些实际上仍有预算的用户；如果呈正偏态，则说明一批请求会在结算完成前穿透限制。

还要按触发原因分析拦截情况：如果主要受并发限制，就增加 worker；如果付费方案主要受成本限制，那么这更像是一个定价问题，而不是工程问题。

《AI That Ships》讲解了 AI 功能的生产运维：基于成本的限流、队列、单次运行预算，以及那些能告诉你限额究竟是在保护系统，还是仅仅在惹恼用户的指标。

完整系列位于 xgabriel.com/ai-in-typescript。

如需采取进一步措施，你可以考虑屏蔽此人和／或举报滥用行为。