AI 网关与 LLM API 治理实战：从混乱到可控，我踩过的 3 个坑
结论先行：用 Kong + 自定义插件构建 AI 网关，能解决 LLM API 的认证、限流、成本控制和监控问题；但如果你直接套用传统 API 网关方案，必踩坑。本文记录我在生产环境中从零搭建 LLM API 治理体系的完整过程，包括 3 个关键踩坑点和最终方案。
为什么需要 AI 网关？
三个月前，我们团队接了一个 AI 客服项目，后端调用 OpenAI、Claude 和本地部署的 LLaMA 模型。第一个月就让我焦头烂额：
- 开发团队各自申请 API Key，月底账单 3 万刀，没人说得清谁用了多少
- 某个测试脚本死循环调用，直接打爆 OpenAI 的 Rate Limit，导致线上服务中断 15 分钟
- 无法追踪每个请求的 Token 消耗，成本分摊全靠拍脑袋
传统 API 网关（如 Nginx、Kong）只能做 HTTP 层面的限流和认证，但 LLM API 治理需要按 Token 计费、按模型路由、按用户隔离。所以我决定基于 Kong 搭建一个 AI 网关层。
第一步：选型与架构
我对比了三个方案：
| 方案 | 优点 | 缺点 |
|---|---|---|
| Kong + 自定义插件 | 成熟、可控 | 需要 Lua 开发 |
| Apache APISIX | 插件生态丰富 | 学习曲线陡 |
| 自建 Python 代理 | 灵活 | 性能差、无现成限流 |
最终选择了 Kong 3.4 + 自定义 Lua 插件，因为团队已有 Kong 运维经验，而且 Lua 插件的性能远高于 Python 代理，这点在压测中得到了验证。
架构图：
用户 → Kong (AI 网关) → LLM Provider (OpenAI/Claude/本地模型)
↑
自定义插件: auth、rate-limit-by-token、cost-tracker
第二步：核心实现（附代码）
2.1 安装 Kong 并启动
# 使用 Docker 快速启动
docker run -d --name kong \
-e "KONG_DATABASE=off" \
-e "KONG_DECLARATIVE_CONFIG=/kong/declarative/kong.yml" \
-v $(pwd)/kong.yml:/kong/declarative/kong.yml \
-p 8000:8000 \
kong:3.4
2.2 配置路由和服务
kong.yml 核心配置：
_format_version: "3.0"
services:
- name: openai-service
url: https://api.openai.com/v1
routes:
- name: openai-chat
paths:
- /v1/chat/completions
methods:
- POST
plugins:
- name: ai-auth # 自定义认证插件
- name: ai-rate-limit # 自定义 Token 限流插件
- name: ai-cost-tracker # 自定义成本追踪插件
2.3 自定义插件：Token 级限流（踩坑 1）
踩坑 1：传统 QPS 限流不适用于 LLM
最初我用 Kong 内置的 rate-limiting 插件，按每分钟 100 次请求限流。结果发现这完全是个坑：
- 用户 A 请求
gpt-4（单次消耗 2000 Token），用户 B 请求gpt-3.5（单次消耗 200 Token） - 同样 100 次请求，A 消耗 20 万 Token，B 只消耗 2 万 Token
- 成本差异 10 倍，但限流策略完全一样
解决方案：解析请求体中的 model 和 max_tokens，动态计算 Token 消耗并限流。
Lua 插件核心逻辑：
-- plugins/ai-rate-limit/handler.lua
local AIRateLimit = {
PRIORITY = 1000,
VERSION = "1.0.0",
}
function AIRateLimit:access(conf)
local body = kong.request.get_body()
if not body or not body.model then
return kong.response.exit(400, { error = "Missing model in request body" })
end
-- 根据模型获取 Token 单价和最大限制
local model_config = {
["gpt-4"] = { price_per_token = 0.03, max_tokens = 8000 },
["gpt-3.5-turbo"] = { price_per_token = 0.002, max_tokens = 4096 },
}
local config = model_config[body.model]
if not config then
return kong.response.exit(400, { error = "Unsupported model" })
end
-- 估算本次请求消耗的 Token 数
local estimated_tokens = #body.messages * 100 -- 简化估算
if body.max_tokens then
estimated_tokens = estimated_tokens + body.max_tokens
end
-- 按 Token 限流：每个用户每天最多 100000 Token
local user_id = kong.request.get_header("X-User-Id")
local key = "user:" .. user_id .. ":tokens:" .. os.date("%Y%m%d")
local current_tokens = kong.cache:get(key)
if current_tokens and current_tokens + estimated_tokens > 100000 then
return kong.response.exit(429, { error = "Token limit exceeded" })
end
-- 更新计数器
kong.cache:set(key, current_tokens + estimated_tokens, 86400)
end
return AIRateLimit
注意：Token 估算很粗糙，生产环境建议用
tiktoken库精确计算。我后来改用 Python 侧服务做预计算，通过 Kong 的pre-function插件调用。
2.4 成本追踪：实时记录每笔消耗（踩坑 2）
踩坑 2：异步日志导致数据丢失
我最初用 Kong 的 log-serializer 插件把请求日志写到 Kafka，然后消费端做成本统计。结果发现：
- 生产环境 QPS 200+，Kafka 偶尔背压，日志丢失 5%
- 月底对账发现成本少了 3000 刀，这个教训太贵了
解决方案：同步写 Redis + 定时刷盘到数据库。
-- plugins/ai-cost-tracker/handler.lua
function AICostTracker:log(conf)
local body = kong.request.get_body()
local response_body = kong.response.get_body()
if not body or not response_body then
return
end
-- 从响应体提取实际 Token 消耗
local usage = response_body.usage
if not usage then
return
end
local cost_record = {
user_id = kong.request.get_header("X-User-Id"),
model = body.model,
prompt_tokens = usage.prompt_tokens,
completion_tokens = usage.completion_tokens,
total_tokens = usage.total_tokens,
cost = calculate_cost(body.model, usage.total_tokens),
timestamp = os.time(),
request_id = kong.request.get_header("X-Request-Id"),
}
-- 同步写入 Redis，确保不丢失
local redis = require "resty.redis"
local red = redis:new()
red:set_timeout(1000)
local ok, err = red:connect("127.0.0.1", 6379)
if not ok then
kong.log.err("Failed to connect to Redis: ", err)
return
end
local key = "cost:" .. os.date("%Y%m%d") .. ":" .. cost_record.user_id
red:rpush(key, cjson.encode(cost_record))
red:expire(key, 86400 * 3)
end
2.5 多 Provider 路由与降级（踩坑 3）
踩坑 3：OpenAI 故障时，手动切换太慢
某天 OpenAI 出现 5xx 故障，我们花了 12 分钟手动修改 Kong 路由切换到 Claude。期间线上服务全部中断，老板的脸色比 OpenAI 的响应码还难看。
解决方案：实现自动降级 + 健康检查。
# 在 kong.yml 中添加备用服务
services:
- name: openai-service
url: https://api.openai.com/v1
# ... 主服务
- name: claude-service
url: https://api.anthropic.com/v1
# ... 备用服务
- name: local-llama-service
url: http://localhost:8080/v1
# ... 本地服务
plugins:
- name: ai-fallback
config:
# 主服务 5xx 时自动切换
primary: openai-service
fallbacks:
- claude-service
- local-llama-service
retry_interval: 60 # 60 秒后重试主服务
Lua 插件实现自动降级：
function AIFallback:access(conf)
local primary = conf.primary
local fallbacks = conf.fallbacks
-- 检查主服务健康状态
local healthy = kong.cache:get("health:" .. primary)
if healthy == false then
-- 尝试备用服务
for _, fallback in ipairs(fallbacks) do
local fb_healthy = kong.cache:get("health:" .. fallback)
if fb_healthy ~= false then
kong.service.set_upstream(fallback)
return
end
end
-- 所有服务都不可用
return kong.response.exit(503, { error = "All providers unavailable" })
end
end
第三步：监控与告警
部署后，我添加了三个关键监控指标：
- Token 消耗趋势：按小时、按用户、按模型
- Provider 可用性：5xx 率 > 1% 触发告警
- 成本异常：单用户日消耗超过阈值自动限流
用 Prometheus + Grafana 实现：
# prometheus.yml
scrape_configs:
- job_name: 'kong'
metrics_path: '/metrics'
static_configs:
- targets: ['localhost:8001']
最终效果
运行两个月后，数据对比让我松了一口气：
| 指标 | 治理前 | 治理后 |
|---|---|---|
| 月度成本 | $32,000 | $18,500（降低 42%） |
| 故障恢复时间 | 12 分钟 | < 30 秒 |
| 成本可追溯性 | 无 | 每笔请求可查 |
| API Key 泄露风险 | 高 | 低（统一网关认证） |
最大收益不是省钱，而是失控感消失了。现在团队可以放心地让每个开发直接调用 LLM，因为网关层已经兜底了。
延伸思考
- Streaming 响应：我目前的方案不支持 SSE（Server-Sent Events），因为 Kong 的 Lua 插件在流式响应中无法完整捕获 Token 消耗。如果你需要实时流式输出，建议在应用层做日志，或者改用 Envoy 的 WASM 插件。
- 多租户隔离：按用户做 Token 池还是按团队？我们最终选择了按用户，但大客户需要单独配置配额。
- 自定义模型定价：如果你用本地模型，成本计算更复杂（GPU 时间 vs Token 数）。我目前按请求时长估算，但不够精确。
- 未来方向：考虑用 eBPF 做更细粒度的网络监控，或者集成 LLM 专用的安全检测（如 Prompt Injection 防护）。
最后一句：不要迷信任何现成的 AI 网关产品，你的业务场景才是最好的架构师。踩坑不可怕，可怕的是不记录。
评论已关闭！