前几篇一直在聊 AI API Token 的成本、安全、预算、告警和权限隔离。

这篇继续往下延伸，换一个更贴近工程落地的角度：

一次 AI 请求从前端发起，到后端处理，再到 API 网关转发模型，中间到底应该经过哪些检查？

很多项目刚开始接入 AI 时，流程可能非常简单：

前端提交内容 -> 后端调用模型 -> 返回结果

看起来没问题。

但项目一旦上线，就不能只考虑“能不能调通”。
还需要考虑：

这个用户有没有权限？
这个 Token 是否有效？
这个 Key 是否超额？
输入内容是否过长？
是否命中缓存？
是否需要限流？
调用失败后要不要扣量？
日志怎么记录？
异常怎么追踪？

如果这些逻辑都散落在业务代码里，后期维护会很麻烦。

所以这篇就围绕一次完整的 AI API 请求，拆一下比较合理的调用链路。

---

一、最简单的调用链路

很多项目最开始可能是这样写的。

前端：

async function sendMessage(message) {
  const res = await fetch("/api/chat", {
    method: "POST",
    headers: {
      "Content-Type": "application/json"
    },
    body: JSON.stringify({ message })
  });

  return await res.json();
}

后端：

app.post("/api/chat", async (req, res) => {
  const result = await fetch("https://api.example.com/v1/chat/completions", {
    method: "POST",
    headers: {
      "Authorization": `Bearer ${process.env.AI_API_KEY}`,
      "Content-Type": "application/json"
    },
    body: JSON.stringify({
      model: "chat-model",
      messages: [
        { role: "user", content: req.body.message }
      ]
    })
  });

  const data = await result.json();
  res.json(data);
});

这个版本适合 Demo。

但放到生产环境里，它缺少很多保护：

没有用户权限校验
没有输入长度限制
没有 Token 额度检查
没有请求频率限制
没有日志记录
没有错误兜底
没有成本统计
没有缓存
没有请求去重

所以项目一复杂，就需要把调用链路拆细。

---

二、生产环境里，一次请求至少要经过几层？

我比较推荐的链路是：

前端页面
  ↓
业务后端
  ↓
权限校验
  ↓
输入校验
  ↓
额度检查
  ↓
缓存检查
  ↓
限流检查
  ↓
AI API 网关
  ↓
上游模型服务
  ↓
结果处理
  ↓
日志记录
  ↓
返回前端

看起来步骤很多，但大多数都可以封装成中间件或统一方法。

业务代码不需要每次都重复写，只要走统一调用入口即可。

---

三、第一层：前端不要直接持有真实 Token

前端只负责提交用户输入，不应该保存真实 AI API Token。

不要这样写：

fetch("https://api.example.com/v1/chat/completions", {
  headers: {
    "Authorization": "Bearer sk-xxxx"
  }
});

因为浏览器里的请求用户都能看到。
只要打开开发者工具，就能看到接口地址和请求头。

更合理的方式是：

前端 -> 自己的业务后端 -> AI API 网关 -> 模型服务

前端请求自己的后端：

await fetch("/api/ai/chat", {
  method: "POST",
  headers: {
    "Content-Type": "application/json"
  },
  body: JSON.stringify({
    message: userInput
  })
});

真实 Token 放在后端或网关里，不暴露给浏览器。

---

四、第二层：用户权限检查

后端收到请求后，第一件事不是调用模型，而是确认用户有没有权限。

比如：

function checkUserPermission(user, taskType) {
  if (!user) {
    throw new Error("请先登录");
  }

  if (!user.permissions.includes(taskType)) {
    throw new Error("当前账号无权使用该功能");
  }
}

不同用户可以有不同权限：

免费用户：只能使用基础问答
普通会员：可以使用总结、改写
高级会员：可以使用长文分析
团队用户：可以使用批量任务
管理员：可以查看统计和日志

这样可以避免所有用户都直接访问高消耗功能。

---

五、第三层：输入内容校验

AI API 的成本和输入长度关系很大。

所以在调用模型前，要先检查输入内容。

例如：

const MAX_MESSAGE_LENGTH = 5000;

function validateInput(message) {
  if (!message || !message.trim()) {
    throw new Error("输入内容不能为空");
  }

  if (message.length > MAX_MESSAGE_LENGTH) {
    throw new Error("输入内容过长，请缩短后再提交");
  }
}

如果是知识库问答，还要限制检索上下文：

function buildContext(chunks) {
  return chunks
    .slice(0, 4)
    .map(item => item.content.slice(0, 1200))
    .join("\n\n");
}

如果是聊天应用，还要限制历史轮数：

function keepRecentMessages(messages, maxCount = 8) {
  return messages.slice(-maxCount);
}

这一步看似简单，但能避免很多 Token 浪费。

---

六、第四层：额度检查

在真正调用模型前，需要先检查用户、项目或 Key 的额度。

比如用户每日额度：

function checkUserQuota(user) {
  if (user.usedTokensToday >= user.dailyTokenLimit) {
    throw new Error("今日额度已用完，请明天再试");
  }
}

项目额度：

function checkProjectQuota(project) {
  if (project.usedTokensToday >= project.dailyTokenLimit) {
    throw new Error("项目今日额度已达上限");
  }
}

Key 额度：

function checkKeyQuota(apiKey) {
  if (apiKey.usedTokensToday >= apiKey.dailyTokenLimit) {
    throw new Error("当前 API Key 今日额度已达上限");
  }
}

实际项目里，通常会同时存在多层额度：

用户额度
团队额度
项目额度
API Key 额度
任务额度
全局额度

这样才能避免某个用户或某个任务把整体额度耗光。

---

七、第五层：缓存检查

不是所有请求都需要重新调用模型。

有些任务非常适合缓存：

文章摘要
关键词提取
标题生成
文档分类
商品文案生成
批量翻译

比如同一篇文章多次生成摘要，可以先根据内容生成 hash：

import crypto from "crypto";

function createCacheKey(text) {
  return crypto
    .createHash("sha256")
    .update(text)
    .digest("hex");
}

调用前先查缓存：

async function getSummary(article) {
  const cacheKey = createCacheKey(article.content);
  const cached = await cache.get(cacheKey);

  if (cached) {
    return {
      fromCache: true,
      data: cached
    };
  }

  const result = await callAI(article.content);
  await cache.set(cacheKey, result);

  return {
    fromCache: false,
    data: result
  };
}

缓存命中时，不需要消耗新的 Token。

这对批量任务尤其有用。

---

八、中间插一句工具实践

如果只是一个小项目，上面这些权限、额度、缓存和日志逻辑可以自己慢慢写。

但项目多了之后，比如同时有聊天应用、知识库问答、批量摘要、团队成员共享、不同模型接入，统一管理就会更省心。

我最近体验的斑马 API 这类 AI API 统一入口工具，比较适合把 Key 管理、用量统计、模型接入和团队共享放到一个地方处理。
比如一个项目一个 Key，一个批量任务一个 Key，再结合 Token 消耗记录去排查问题，会比所有业务直接拿上游 Key 调用更清楚。

目前新用户有体验权益，邀请新用户也有额外体验时长。https://bmapi.020212.xyz/register?aff=YU55ECFS8AF2

---

九、第六层：限流检查

限流和额度不一样。

额度关注的是总量，限流关注的是短时间内的频率。

例如：

每分钟最多 60 次请求
每个用户每分钟最多 10 次
批量任务每分钟最多 30 次
开发 Key 每分钟最多 5 次

一个简单的限流逻辑可以这样理解：

function checkRateLimit(keyStats) {
  if (keyStats.requestsInLastMinute > keyStats.limitPerMinute) {
    throw new Error("请求过于频繁，请稍后再试");
  }
}

不同类型的 Key 应该有不同限流策略：

prod_chat_api：较高频率
batch_summary_job：中等频率
dev_local_test：低频率
temp_test_key：低频率

这样可以避免某个脚本短时间内打满请求。

---

十、第七层：调用 AI API 网关

前面的校验都通过后，才真正调用 AI 服务。

业务后端可以统一封装一个方法：

async function callAIGateway({ apiKey, model, messages, taskType }) {
  const response = await fetch(process.env.AI_GATEWAY_BASE_URL + "/chat/completions", {
    method: "POST",
    headers: {
      "Authorization": `Bearer ${apiKey.value}`,
      "Content-Type": "application/json"
    },
    body: JSON.stringify({
      model,
      messages,
      metadata: {
        taskType
      }
    })
  });

  if (!response.ok) {
    throw new Error(`AI gateway error: ${response.status}`);
  }

  return await response.json();
}

业务侧不直接调用上游模型，而是统一走网关。

这样后续换模型、换 Key、换上游服务时，业务代码改动更少。

---

十一、第八层：结果处理

模型返回结果后，不要直接原样返回前端。

可以做一些基础处理：

检查返回是否为空
检查格式是否符合预期
提取真正需要的字段
过滤异常内容
记录输出长度
必要时做结构化解析

例如分类任务只需要一个标签：

function parseCategoryResult(result) {
  const text = result.content.trim();

  const allowed = ["咨询", "投诉", "建议", "其他"];

  if (!allowed.includes(text)) {
    return "其他";
  }

  return text;
}

如果你要求模型返回 JSON，也要做好解析失败处理：

function safeParseJSON(text) {
  try {
    return JSON.parse(text);
  } catch {
    return null;
  }
}

AI 输出不一定永远稳定，所以业务侧要有兜底。

---

十二、第九层：Token 扣量

这里有一个容易忽略的问题：什么时候扣 Token？

常见方式有两种。

1. 请求前预扣

先根据输入估算 Token，提前扣除一部分额度。

优点是可以防止用户额度不足还继续请求。
缺点是估算不一定准确。

2. 请求后实扣

模型返回后，根据实际消耗扣除 Token。

优点是准确。
缺点是请求过程中如果用户额度不足，可能已经产生调用成本。

实际项目里，可以结合使用：

请求前：检查剩余额度是否大于最低门槛
请求后：按实际 usage 扣除
失败时：根据是否产生 usage 决定是否扣量

例如：

function deductTokens(account, usage) {
  account.usedTokensToday += usage.total_tokens;
}

如果模型返回了 usage 信息，就按实际值记录：

{
  "prompt_tokens": 1200,
  "completion_tokens": 300,
  "total_tokens": 1500
}

---

十三、失败请求要不要扣量？

这个问题要看失败发生在哪一层。

1. 校验失败

比如输入太长、额度不足、权限不够。

这种请求没有调用模型，不应该扣 Token。

不扣量
记录失败原因即可

2. 网关前失败

比如限流、缓存命中、参数错误。

如果没有请求上游模型，也不应该扣 Token。

3. 上游模型已处理但返回失败

如果上游已经处理了请求，并产生了 usage，那么可能仍然需要记录消耗。

例如：

请求超时
返回格式异常
客户端中断
网关返回错误但上游已有消耗

这种情况要看实际 usage 是否存在。

所以日志里最好区分：

业务失败
校验失败
网关失败
上游失败
已产生 Token 的失败
未产生 Token 的失败

---

十四、第十层：日志记录

日志是后期排查的基础。

一条完整 AI 请求日志可以包含：

{
  "request_id": "req_20250101_001",
  "user_id": "u_1001",
  "project": "kb_qa",
  "api_key_name": "prod_kb_qa",
  "task_type": "knowledge_qa",
  "model": "chat-model",
  "prompt_tokens": 1200,
  "completion_tokens": 300,
  "total_tokens": 1500,
  "latency_ms": 2200,
  "status": "success",
  "error_code": null,
  "cache_hit": false,
  "created_at": "2025-01-01 10:00:00"
}

注意不要记录完整 Token。

可以记录：

api_key_name
api_key_id
token_prefix

不要记录：

完整 Authorization
完整 API Token
完整上游密钥

---

十五、一次完整调用的伪代码

把上面的逻辑串起来，可以写成类似这样：

async function handleAIRequest({ user, project, apiKey, message }) {
  const requestId = createRequestId();
  const startTime = Date.now();

  try {
    checkUserPermission(user, "chat");
    validateInput(message);
    checkUserQuota(user);
    checkProjectQuota(project);
    checkKeyQuota(apiKey);
    checkRateLimit(apiKey);

    const cacheKey = createCacheKey(message);
    const cached = await cache.get(cacheKey);

    if (cached) {
      await writeLog({
        requestId,
        userId: user.id,
        project: project.name,
        apiKeyName: apiKey.name,
        status: "success",
        cacheHit: true,
        totalTokens: 0,
        latencyMs: Date.now() - startTime
      });

      return cached;
    }

    const result = await callAIGateway({
      apiKey,
      model: "chat-model",
      messages: [
        { role: "user", content: message }
      ],
      taskType: "chat"
    });

    const usage = result.usage || {
      prompt_tokens: 0,
      completion_tokens: 0,
      total_tokens: 0
    };

    deductTokens(user, usage);
    deductTokens(project, usage);
    deductTokens(apiKey, usage);

    await cache.set(cacheKey, result);

    await writeLog({
      requestId,
      userId: user.id,
      project: project.name,
      apiKeyName: apiKey.name,
      status: "success",
      cacheHit: false,
      promptTokens: usage.prompt_tokens,
      completionTokens: usage.completion_tokens,
      totalTokens: usage.total_tokens,
      latencyMs: Date.now() - startTime
    });

    return result;
  } catch (error) {
    await writeLog({
      requestId,
      userId: user?.id,
      project: project?.name,
      apiKeyName: apiKey?.name,
      status: "failed",
      errorMessage: error.message,
      latencyMs: Date.now() - startTime
    });

    throw error;
  }
}

这只是一个示例，但它体现了一个思路：

AI 调用不要散着写
尽量统一入口
统一做权限、额度、缓存、限流、日志和扣量

---

十六、不同任务应该走不同链路策略

不是所有 AI 请求都应该用同一套策略。

普通聊天

需要控制历史上下文
需要限制单次输入
需要记录用户用量
响应速度比较重要

知识库问答

需要限制检索片段数量
需要记录上下文 Token
需要控制历史对话
需要关注引用准确性

批量任务

需要单独 Key
需要单独额度
需要限速
需要支持中断和恢复
需要避免重复处理

文档分析

需要分段处理
需要异步执行
需要控制单段长度
需要记录每段消耗

所以统一入口不是所有任务完全一样，而是把公共能力统一，把任务差异通过配置控制。

---

十七、一个实际例子：批量摘要任务

批量摘要任务可以这样设计：

Key：batch_article_summary
每日额度：500,000 tokens
限流：每分钟 30 次
缓存：按文章内容 hash
输入限制：单篇文章最多 8000 字
失败策略：最多重试 1 次
日志：记录每篇文章消耗

调用流程：

读取文章
生成内容 hash
检查是否已处理
检查批量 Key 额度
检查限流
调用模型生成摘要
记录 Token
保存结果
继续下一篇

这样即使任务异常，也不会影响线上聊天或知识库问答。

---

十八、一个实际例子：知识库问答

知识库问答可以这样设计：

Key：prod_kb_qa
每日额度：2,000,000 tokens
检索片段：最多 4 段
单段长度：最多 1200 字
历史对话：最多 6 条
输出限制：尽量不超过 800 字
日志：记录检索片段数量和 Token 消耗

调用流程：

用户提问
校验用户权限
检索知识库
截断上下文
拼接 prompt
检查项目额度
调用模型
记录 prompt_tokens 和 completion_tokens
返回答案

这种方式比无限制拼接上下文要稳定很多。

---

十九、一个实际例子：AI 写作工具

AI 写作工具可以按功能分级：

标题生成：低消耗
摘要生成：中低消耗
段落改写：中等消耗
长文扩写：高消耗
批量生成：高消耗

当 Token 预算正常时，所有功能可用。

当预算达到 90% 时：

暂停批量生成
限制长文扩写
保留标题生成和摘要生成

当预算达到 100% 时：

只保留低消耗功能
提示高消耗功能明日恢复

这样比直接全站不可用更友好。

---

二十、总结

AI API 调用不应该只是简单地把请求转发给模型。

从工程角度看，一次完整请求最好包含：

用户权限校验
输入内容校验
额度检查
缓存检查
限流检查
统一网关调用
结果解析
Token 扣量
日志记录
异常处理

这些能力不一定一开始全部做完，但项目只要进入长期运行阶段，就会越来越需要。

尤其是下面几类场景：

团队多人使用
多个项目共用 AI 能力
存在批量任务
存在知识库问答
存在长文本处理
需要会员或额度体系
需要统计成本
需要排查异常消耗

如果没有统一调用链路，后期每个项目都各写一套逻辑，很容易混乱。

比较理想的方式是：

业务代码只管发起 AI 任务
统一入口负责 Token、额度、权限、日志和模型分发

这样业务侧更轻，后期治理也更容易。