AI Gateway 保姆级教程：API 网关统一管理大模型限流/鉴权/缓存，Token成本直降30%！

文章链接：https://blog.csdn.net/qq_43201350/article/details/163477707

2026年，企业的 AI 调用量正以月均 30% 的速度增长。当一个团队开始同时调 5 个大模型、有 20 个内部系统在消费 AI 能力时，你会遇到什么？Token 超预算、API Key 泄露、重复请求吃掉成本——而这一切，都可以在网关层解决。

一、为什么需要AIGateway？

1.1 直调大模型的五个坑

想象你现在是这样调 AI 的：

前端 ──→ 后端 ──→ 模型 A 的 API
          └──→ 模型 B 的 API
          └──→ 模型 C 的 API

问题清单：

痛点	表现	后果
API Key 裸奔	每个后端服务都存一份 Key	泄露面 = 服务数 N
Token 超预算	某业务线一个月跑了 50 万 Token	月底账单暴击
重复调用无缓存	5 个用户问同一个问题，调 5 次模型	白花 4 份钱
无流量管控	突发 1000 QPS 直打模型 API	被限流/封号
多模型难切换	换模型要改代码、重新部署	一周起步

1.2 加一层 AIGateway之后

前端 ──→ 后端 ──→ [AI Gateway] ──→ 模型 A
                          ├──→ 模型 B
                          ├──→ 模型 C
                          ├── 限流 / 鉴权 / 缓存
                          ├── Token 用量统计
                          └── Key 统一管理

网关层统一解决的问题，恰好就是上面五个坑。下面逐一用 Demo 演示。

二、整体架构设计

先搭一个可运行的 Demo 环境。

2.1 架构图

┌─────────────────────────────────────────────────────┐
│                      AI Gateway                      │
│  ┌──────────────────────────────────────────────┐   │
│  │              路由层 (Route)                   │   │
│  │  /ai/chat  →  模型集群转发                     │   │
│  │  /ai/image →  图像模型                         │   │
│  └──────────────────────────────────────────────┘   │
│  ┌──────────┐ ┌──────────┐ ┌──────────────────┐    │
│  │ Key-Auth │ │ 限流模块  │ │   缓存模块        │    │
│  │ 插件     │ │ 插件     │ │   插件            │    │
│  └──────────┘ └──────────┘ └──────────────────┘    │
│  ┌──────────────────────────────────────────────┐   │
│  │              可观测层                         │   │
│  │  耗时记录 / Token 计数 / 错误率                 │   │
│  └──────────────────────────────────────────────┘   │
└─────────────────────────────────────────────────────┘
          │            │            │
     ┌────┴────┐  ┌───┴────┐  ┌───┴────┐
     │ 模型 S  │  │ 模型 Q │  │ 模型 H │
     │ (对话)  │  │ (代码) │  │ (图像) │
     └─────────┘  └────────┘  └────────┘

2.2 目录结构

ai-gateway-demo/
├── docker-compose.yml        # 网关 + 后端 + Mock AI 服务
├── config/
│   └── gateway-config.yaml   # 网关路由配置
├── backend/
│   └── app.py                # 业务后端 Demo
├── mock-ai/
│   ├── chat.py               # Mock 对话模型
│   ├── code.py               # Mock 代码模型
│   └── image.py              # Mock 图像模型
└── scripts/
    ├── setup.sh              # 一键初始化
    └── pressure_test.py      # 压测脚本

2.3 docker-compose.yml

version: '3.8'
 
services:
  # =========== 配置中心 ===========
  etcd:
    image: bitnami/etcd:3.5
    environment:
      - ALLOW_NONE_AUTHENTICATION=yes
      - ETCD_ADVERTISE_CLIENT_URLS=http://0.0.0.0:2379
    ports:
      - "2379:2379"
 
  # =========== AI Gateway ===========
  gateway:
    image: apache/apisix:3.10.0-centos
    depends_on:
      - etcd
    volumes:
      - ./config/gateway-config.yaml:/usr/local/apisix/conf/config.yaml
    ports:
      - "9080:9080"    # 对外网关端口
      - "9180:9180"    # Admin API
    environment:
      - APISIX_STAND_ALONE=false
 
  # =========== 业务后端 ===========
  backend:
    build: ./backend
    ports:
      - "9090:9090"
    environment:
      - GATEWAY_URL=http://gateway:9080
 
  # =========== Mock AI 服务 ===========
  mock-chat:
    build: ./mock-ai
    environment:
      - MODEL_TYPE=chat
    ports:
      - "9101:9101"
 
  mock-code:
    build: ./mock-ai
    environment:
      - MODEL_TYPE=code
    ports:
      - "9102:9102"
 
  mock-image:
    build: ./mock-ai
    environment:
      - MODEL_TYPE=image
    ports:
      - "9103:9103"

2.4MockAI 服务

三个 Mock 服务结构相同，用一个统一入口：

# mock-ai/app.py
from flask import Flask, request, jsonify
import time
import os
import hashlib
 
app = Flask(__name__)
MODEL_TYPE = os.getenv("MODEL_TYPE", "chat")
 
# 模拟不同模型的价格（元/千Token）
PRICING = {
    "chat":  {"input": 0.002, "output": 0.006},
    "code":  {"input": 0.001, "output": 0.004},
    "image": {"input": 0.008, "output": 0.020},
}
 
@app.route('/v1/chat/completions', methods=['POST'])
def chat_completions():
    data = request.json
    messages = data.get("messages", [])
    model_name = data.get("model", "demo-model")
 
    # ---- 模拟 Token 计数 ----
    input_text = "".join([m.get("content", "") for m in messages])
    input_tokens = len(input_text) // 2
    output_tokens = 150  # 固定模拟输出
 
    # ---- 模拟耗时 ----
    delay = 0.3 + (input_tokens * 0.001)
    time.sleep(min(delay, 3.0))
 
    # ---- 根据请求内容生成确定性"缓存指纹" ----
    request_fingerprint = hashlib.md5(input_text.encode()).hexdigest()[:8]
 
    return jsonify({
        "id": f"chatcmpl-{request_fingerprint}",
        "object": "chat.completion",
        "model": model_name,
        "choices": [{
            "index": 0,
            "message": {
                "role": "assistant",
                "content": f"[{MODEL_TYPE.upper()} DEMO] 您的请求已处理。"
                           f"输入约 {input_tokens} tokens，"
                           f"对话内容摘要：{input_text[:50]}..."
            },
            "finish_reason": "stop"
        }],
        "usage": {
            "prompt_tokens": input_tokens,
            "completion_tokens": output_tokens,
            "total_tokens": input_tokens + output_tokens,
            "estimated_cost_cny": round(
                PRICING[MODEL_TYPE]["input"] * input_tokens / 1000 +
                PRICING[MODEL_TYPE]["output"] * output_tokens / 1000, 6
            )
        }
    })
 
if __name__ == '__main__':
    port = {"chat": 9101, "code": 9102, "image": 9103}[MODEL_TYPE]
    app.run(host="0.0.0.0", port=port)

三、场景一：统一APIKey 管理

3.1 问题

每个服务各自保存模型 API Key，泄露风险高。统一由网关管理 Key，下游服务不用接触。

3.2 网关配置

# 1. 在网关中创建消费者，存储模型 API Key
curl -s http://127.0.0.1:9180/admin-api/consumers -X PUT -d '
{
  "username": "ai-consumer-demo",
  "plugins": {
    "key-auth": {
      "key": "demo-ai-api-key-2026"
    }
  }
}'

# 2. 配置上游——三个 AI 模型
# 对话模型
curl -s http://127.0.0.1:9180/admin-api/upstreams/ai-chat-upstream -X PUT -d '
{
  "type": "roundrobin",
  "nodes": {
    "mock-chat:9101": 1
  }
}'
 
# 代码模型
curl -s http://127.0.0.1:9180/admin-api/upstreams/ai-code-upstream -X PUT -d '
{
  "type": "roundrobin",
  "nodes": {
    "mock-code:9102": 1
  }
}'
 
# 图像模型
curl -s http://127.0.0.1:9180/admin-api/upstreams/ai-image-upstream -X PUT -d '
{
  "type": "roundrobin",
  "nodes": {
    "mock-image:9103": 1
  }
}'

# 3. 创建路由，绑定 key-auth 插件
curl -s http://127.0.0.1:9180/admin-api/routes/ai-chat-route -X PUT -d '
{
  "uri": "/ai/chat/*",
  "upstream_id": "ai-chat-upstream",
  "plugins": {
    "key-auth": {},
    "proxy-rewrite": {
      "uri": "/v1/chat/completions"
    }
  }
}'
 
curl -s http://127.0.0.1:9180/admin-api/routes/ai-code-route -X PUT -d '
{
  "uri": "/ai/code/*",
  "upstream_id": "ai-code-upstream",
  "plugins": {
    "key-auth": {},
    "proxy-rewrite": {
      "uri": "/v1/chat/completions"
    }
  }
}'
 
curl -s http://127.0.0.1:9180/admin-api/routes/ai-image-route -X PUT -d '
{
  "uri": "/ai/image/*",
  "upstream_id": "ai-image-upstream",
  "plugins": {
    "key-auth": {},
    "proxy-rewrite": {
      "uri": "/v1/chat/completions"
    }
  }
}'

3.3 验证

# 无 Key → 401
curl -s http://127.0.0.1:9080/ai/chat -X POST \
  -H "Content-Type: application/json" \
  -d '{"model":"demo-model","messages":[{"role":"user","content":"你好"}]}'
# 返回: {"message":"Missing API key found in request"}
 
# 带 Key → 成功
curl -s http://127.0.0.1:9080/ai/chat -X POST \
  -H "Content-Type: application/json" \
  -H "apiKey: demo-ai-api-key-2026" \
  -d '{"model":"demo-model","messages":[{"role":"user","content":"你好"}]}'
# 返回: 正常的 AI 响应 + usage 信息

效果：下游 20 个服务不再各自持有模型 Key，所有 Key 只在网关层管理。换 Key 时更新网关配置即可，下游零感知。

四、场景二：Token用量限流——防月底账单暴击

4.1 问题

某业务小哥写了个循环调 AI 的脚本，一晚上跑了 200 万 Token。第二天账单出来，成本是上个月的三倍。

4.2 三层限流策略

层级	维度	限制	目的
消费者级	每个 API Key	每小时 10 万 Token	防止单业务超量
路由级	每个模型	每秒 20 QPS	防止打爆下游
IP 级	每个调用方 IP	每分钟 30 次	防止脚本滥用

4.3 配置实现

网关 G 支持组合插件。我们用 limit-count + limit-req + 自定义 Token 计数实现：

# === 消费者级 Token 限额 ===
# 为消费者 ai-consumer-demo 添加限流
curl -s http://127.0.0.1:9180/admin-api/consumers/ai-consumer-demo -X PUT -d '
{
  "username": "ai-consumer-demo",
  "plugins": {
    "key-auth": {
      "key": "demo-ai-api-key-2026"
    },
    "limit-count": {
      "count": 100,
      "time_window": 3600,
      "key_type": "consumer_name",
      "rejected_code": 429,
      "rejected_msg": "Token 用量超限，当前小时已消耗 10 万 Token，请稍后再试"
    },
    "limit-req": {
      "rate": 5,
      "burst": 3,
      "key_type": "consumer_name",
      "rejected_code": 429,
      "rejected_msg": "请求频率过高，请降低并发"
    }
  }
}'

# === 路由级 QPS 限制 ===
curl -s http://127.0.0.1:9180/admin-api/routes/ai-chat-route -X PATCH -d '
{
  "plugins": {
    "key-auth": {},
    "proxy-rewrite": {
      "uri": "/v1/chat/completions"
    },
    "limit-req": {
      "rate": 20,
      "burst": 5,
      "key": "remote_addr",
      "rejected_code": 429,
      "rejected_msg": "模型请求繁忙，请稍后重试"
    }
  }
}'

4.4 验证

# scripts/pressure_test.py
import requests
import time
import threading
 
GATEWAY_URL = "http://127.0.0.1:9080/ai/chat"
API_KEY = "demo-ai-api-key-2026"
 
def send_request(idx):
    try:
        resp = requests.post(GATEWAY_URL, headers={
            "Content-Type": "application/json",
            "apiKey": API_KEY
        }, json={
            "model": "demo-model",
            "messages": [{"role": "user", "content": f"压测请求 #{idx}"}]
        }, timeout=10)
        return idx, resp.status_code, resp.json().get("message", "ok")[:60]
    except Exception as e:
        return idx, 0, str(e)
 
# 模拟 50 并发
results = []
threads = []
for i in range(50):
    t = threading.Thread(
        target=lambda idx=i: results.append(send_request(idx))
    )
    threads.append(t)
    t.start()
 
for t in threads:
    t.join()
 
# 统计
success = sum(1 for r in results if r[1] == 200)
rate_limited = sum(1 for r in results if r[1] == 429)
print(f"50 并发结果: 成功 {success}, 被限流 {rate_limited}")

输出示例：

50 并发结果: 成功 8, 被限流 42

限流生效——只有 8 个请求通过，其余 42 个被 429 拦截，模型侧毫发无伤。

五、场景三：问答缓存——不花钱重复回答

5.1 问题

同一个问题被多次提问，每次都调大模型。尤其典型场景：

新手引导 FAQ：“怎么开始用这个平台？”——每天被问 200 次
客服机器人：“退款政策是什么？”——高频问题
代码助手：“写一个快排”——千篇一律

每次调用成本 0.01 ~ 0.1 元，日积月累就是数千元浪费。

5.2 解决方案：网关层缓存

网关 G 没有内置 AI 语义缓存插件，但我们可以写一个：

-- 文件: gateway-plugins/demo-ai-cache.lua
local core = require("apisix.core")
local redis = require("resty.redis")
local cjson = require("cjson")
local md5 = require("resty.md5")
 
local plugin_name = "demo-ai-cache"
 
local schema = {
    type = "object",
    properties = {
        ttl = {type = "integer", minimum = 1, default = 3600},      -- 秒
        cache_key_fields = {
            type = "array",
            items = {type = "string"},
            default = {"model", "messages"}
        },
        redis_host = {type = "string", default = "127.0.0.1"},
        redis_port = {type = "integer", default = 6379},
        similar_threshold = {type = "number", default = 0.85}       -- 相似度阈值
    }
}
 
local _M = {
    version = 0.1,
    priority = 900,   -- 高优先级，先检查缓存
    name = plugin_name,
    schema = schema
}
 
function _M.check_schema(conf)
    return core.schema.check(schema, conf)
end
 
-- 生成缓存 Key
local function build_cache_key(conf, ctx)
    local body = core.request.get_body()
    if not body then return nil end
 
    local params, err = cjson.decode(body)
    if not params then return nil end
 
    -- 提取用于缓存判定的字段
    local cache_parts = {}
    for _, field in ipairs(conf.cache_key_fields) do
        if params[field] then
            table.insert(cache_parts, cjson.encode(params[field]))
        end
    end
 
    if #cache_parts == 0 then return nil end
 
    local raw_key = table.concat(cache_parts, "|")
    -- 对原文做规范化：去空格、小写化——提升命中率
    raw_key = raw_key:gsub("%s+", ""):lower()
    local key_digest = md5.sumhex(raw_key)
    return "ai_cache:" .. key_digest
end
 
-- Access 阶段：查缓存
function _M.access(conf, ctx)
    local cache_key = build_cache_key(conf, ctx)
    if not cache_key then return end
 
    local red = redis:new()
    red:set_timeouts(1000, 1000, 1000)
 
    local ok, err = red:connect(conf.redis_host, conf.redis_port)
    if not ok then
        core.log.warn("demo-ai-cache: redis 连接失败, 跳过缓存, err=", err)
        return
    end
 
    local cached, err = red:get(cache_key)
    if cached then
        core.log.info("demo-ai-cache: 缓存命中! key=", cache_key)
 
        -- 记录缓存命中指标
        local metric = core.metrics.new()
        metric:inc("ai_cache_hit", 1, {plugin = plugin_name})
 
        -- 直接返回缓存内容，终止后续处理
        core.response.set_header("X-AI-Cache", "HIT")
        core.response.set_header("X-AI-Cache-Key", cache_key)
        return 200, cjson.decode(cached)
    end
 
    -- 标记需要缓存
    ctx.var.demo_ai_cache_key = cache_key
    red:close()
end
 
-- Body Filter 阶段：写缓存
function _M.body_filter(conf, ctx)
    if not ctx.var.demo_ai_cache_key then return end
    if ngx.status ~= 200 then return end
 
    local body = ngx.arg[1] or ""
    local body_data = cjson.decode(body)
    if not body_data then return end
 
    -- 只缓存成功响应
    if not body_data.choices or #body_data.choices == 0 then return end
 
    local red = redis:new()
    red:set_timeouts(1000, 1000, 1000)
 
    local ok, err = red:connect(conf.redis_host, conf.redis_port)
    if not ok then
        core.log.warn("demo-ai-cache: redis 连接失败, 跳过写缓存")
        return
    end
 
    -- 写入 Redis，设置 TTL
    red:setex(ctx.var.demo_ai_cache_key, conf.ttl, cjson.encode(body_data))
    core.log.info("demo-ai-cache: 写入缓存 key=", ctx.var.demo_ai_cache_key, " ttl=", conf.ttl)
    red:close()
 
    ngx.arg[1] = body  -- 保持原始响应不变
end
 
return _M

5.3 挂载插件

curl -s http://127.0.0.1:9180/admin-api/routes/ai-chat-route -X PATCH -d '
{
  "plugins": {
    "key-auth": {},
    "proxy-rewrite": {"uri": "/v1/chat/completions"},
    "limit-req": {"rate": 20, "burst": 5, "key": "remote_addr", "rejected_code": 429},
    "demo-ai-cache": {
      "ttl": 1800,
      "cache_key_fields": ["model", "messages", "temperature"],
      "redis_host": "redis",
      "redis_port": 6379
    }
  }
}'

5.4 验证

# 第一次调用——缓存未命中，走模型
curl -s http://127.0.0.1:9080/ai/chat -X POST \
  -H "Content-Type: application/json" \
  -H "apiKey: demo-ai-api-key-2026" \
  -d '{"model":"demo-model","messages":[{"role":"user","content":"解释什么是微服务架构"}]}' \
  -D headers1.txt | head -20
 
# 第二次调用同样内容——缓存命中
curl -s http://127.0.0.1:9080/ai/chat -X POST \
  -H "Content-Type: application/json" \
  -H "apiKey: demo-ai-api-key-2026" \
  -d '{"model":"demo-model","messages":[{"role":"user","content":"解释什么是微服务架构"}]}' \
  -D headers2.txt
 
# 检查响应头
grep "X-AI-Cache" headers2.txt
# 输出: X-AI-Cache: HIT

响应头中有 X-AI-Cache: HIT ，说明是从缓存返回的，零调用成本。

5.5 缓存效果估算

场景	日请求量	命中率	日节省调用	日节省成本(元)
客服 FAQ	5000	40%	2000	~20
新人引导	3000	35%	1050	~10
代码助手	10000	25%	2500	~25
合计	18000	—	5550	~55

一个月就能省下 1600+ 元 ，而且响应延迟从 1.5s 降到 10ms。

六、场景四：多模型灰度切换

6.1 问题

想从模型 A 切换到模型 B，但是不敢直接切——怕 B 在某个场景下效果不好。怎么办？

答案：网关层做流量权重分配。

6.2 实现

# 创建加权上游：70% 走模型 A，30% 走模型 B
curl -s http://127.0.0.1:9180/admin-api/upstreams/ai-chat-canary-upstream -X PUT -d '
{
  "type": "chash",
  "hash_on": "header",
  "key": "X-User-Id",
  "nodes": {
    "mock-chat:9101": 7,     # 模型A 权重 7 (70%)
    "mock-code:9102": 3      # 模型B 权重 3 (30%)
  }
}'
 
# 切换路由到灰度上游
curl -s http://127.0.0.1:9180/admin-api/routes/ai-chat-route -X PATCH -d '
{
  "upstream_id": "ai-chat-canary-upstream"
}'

hash_on: header + key: X-User-Id ：同一用户始终打到同一模型，确保体验一致性
70:30 权重分配：灰度比例可动态调整
观察 30% 用户的反馈，确认模型 B 没问题后，逐步把比例调到 100:0

6.3 验证灰度效果

# 模拟不同用户的请求
for i in $(seq 1 10); do
  RESPONSE=$(curl -s http://127.0.0.1:9080/ai/chat -X POST \
    -H "Content-Type: application/json" \
    -H "apiKey: demo-ai-api-key-2026" \
    -H "X-User-Id: user-$i" \
    -d "{\"model\":\"demo-model\",\"messages\":[{\"role\":\"user\",\"content\":\"test $i\"}]}")
 
  # 从响应中提取模型标识
  MODEL=$(echo $RESPONSE | grep -o '\[.*DEMO\]' | head -1)
  echo "用户 user-$i → $MODEL"
done

输出示例：

用户 user-1  → [CHAT DEMO]
用户 user-2  → [CODE DEMO]
用户 user-3  → [CHAT DEMO]
用户 user-4  → [CHAT DEMO]
用户 user-5  → [CODE DEMO]
...

观察到请求被分流到了不同模型——灰度发布成功。

七、场景五：Token 成本可视化

7.1 为什么需要可视化

网关是所有 AI 请求的必经之路，可以零侵入地记录每笔调用的 Token 消耗。不需要改任何业务代码。

7.2 用网关的日志插件记录 Token

# 配置 http-logger，将 AI 调用日志推送至分析后端
curl -s http://127.0.0.1:9180/admin-api/routes/ai-chat-route -X PATCH -d '
{
  "plugins": {
    "key-auth": {},
    "proxy-rewrite": {"uri": "/v1/chat/completions"},
    "limit-req": {"rate": 20, "burst": 5, "key": "remote_addr", "rejected_code": 429},
    "demo-ai-cache": {"ttl": 1800, "redis_host": "redis"},
    "http-logger": {
      "uri": "http://analytics-backend:9090/api/ai-usage-log",
      "batch_max_size": 50,
      "inactive_timeout": 3,
      "concat_method": "new_line"
    },
    "response-rewrite": {
      "headers": {
        "set": {
          "X-Upstream-Latency": "$upstream_response_time",
          "X-Gateway-Latency": "$request_time"
        }
      }
    }
  }
}'

7.3 分析后端

# analytics-backend/app.py
from flask import Flask, request, jsonify
import sqlite3
from datetime import datetime, timedelta
 
app = Flask(__name__)
 
def init_db():
    conn = sqlite3.connect("ai_usage.db")
    conn.execute("""
        CREATE TABLE IF NOT EXISTS usage_logs (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            consumer TEXT,
            model TEXT,
            input_tokens INTEGER,
            output_tokens INTEGER,
            total_tokens INTEGER,
            cost_cny REAL,
            cache_hit INTEGER DEFAULT 0,
            latency_ms REAL,
            api_key TEXT,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
    """)
    conn.commit()
    return conn
 
init_db()
 
@app.route('/api/ai-usage-log', methods=['POST'])
def receive_logs():
    """接收网关推送的日志"""
    logs = request.get_data(as_text=True).strip().split("\n")
    conn = sqlite3.connect("ai_usage.db")
    count = 0
 
    for line in logs:
        try:
            data = json.loads(line)
            usage = data.get("response", {}).get("body", {}).get("usage", {})
 
            conn.execute(
                """INSERT INTO usage_logs 
                   (consumer, model, input_tokens, output_tokens, 
                    total_tokens, cost_cny, cache_hit, latency_ms, api_key)
                   VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""",
                (
                    data.get("consumer", "unknown"),
                    data.get("request", {}).get("body", {}).get("model", "unknown"),
                    usage.get("prompt_tokens", 0),
                    usage.get("completion_tokens", 0),
                    usage.get("total_tokens", 0),
                    usage.get("estimated_cost_cny", 0),
                    1 if "HIT" in str(data.get("response", {}).get("headers", {})) else 0,
                    data.get("latency", 0),
                    data.get("api_key", "***")[:8] + "***"
                )
            )
            count += 1
        except Exception:
            continue
 
    conn.commit()
    conn.close()
    return jsonify({"received": count}), 200
 
@app.route('/api/usage-report', methods=['GET'])
def usage_report():
    """获取用量报告"""
    period = request.args.get("period", "today")  # today / week / month
 
    now = datetime.now()
    if period == "today":
        start = now.replace(hour=0, minute=0, second=0)
    elif period == "week":
        start = now - timedelta(days=7)
    else:
        start = now - timedelta(days=30)
 
    conn = sqlite3.connect("ai_usage.db")
    conn.row_factory = sqlite3.Row
 
    # 按模型汇总
    by_model = conn.execute(
        """SELECT model, 
                  SUM(total_tokens) as tokens,
                  SUM(cost_cny) as cost,
                  COUNT(*) as calls,
                  SUM(cache_hit) as cache_hits
           FROM usage_logs 
           WHERE created_at >= ? 
           GROUP BY model
           ORDER BY cost DESC""",
        (start,)
    ).fetchall()
 
    # 按消费者汇总
    by_consumer = conn.execute(
        """SELECT consumer,
                  SUM(total_tokens) as tokens,
                  SUM(cost_cny) as cost,
                  COUNT(*) as calls
           FROM usage_logs 
           WHERE created_at >= ? 
           GROUP BY consumer
           ORDER BY cost DESC""",
        (start,)
    ).fetchall()
 
    conn.close()
 
    return jsonify({
        "period": period,
        "total_cost": round(sum(r["cost"] for r in by_model), 4),
        "total_tokens": sum(r["tokens"] for r in by_model),
        "total_calls": sum(r["calls"] for r in by_model),
        "by_model": [dict(r) for r in by_model],
        "by_consumer": [dict(r) for r in by_consumer]
    })
 
if __name__ == '__main__':
    app.run(host="0.0.0.0", port=9090)

7.4 日报效果

curl -s http://127.0.0.1:9090/api/usage-report?period=today | python -m json.tool

{
  "period": "today",
  "total_cost": 12.567,
  "total_tokens": 1850000,
  "total_calls": 3200,
  "by_model": [
    {"model": "demo-model", "tokens": 1200000, "cost": 8.24, "calls": 2100, "cache_hits": 420},
    {"model": "demo-code-model", "tokens": 500000, "cost": 3.12, "calls": 800, "cache_hits": 150},
    {"model": "demo-image-model", "tokens": 150000, "cost": 1.207, "calls": 300, "cache_hits": 60}
  ],
  "by_consumer": [
    {"consumer": "ai-consumer-demo", "tokens": 1500000, "cost": 10.0, "calls": 2600},
    {"consumer": "demo-customer-service", "tokens": 350000, "cost": 2.567, "calls": 600}
  ]
}

现在你知道：今天花了 12.57 元、其中缓存帮你省了约 20% 的调用。这些数字可以直接配告警——当日成本超过预算时发钉钉/飞书通知。

八、完整路由配置汇总

最后把所有能力串成一条完整路由：

curl -s http://127.0.0.1:9180/admin-api/routes/ai-chat-route -X PUT -d '
{
  "uri": "/ai/chat/*",
  "name": "ai-chat-complete",
  "methods": ["POST"],
  "upstream_id": "ai-chat-canary-upstream",
  "plugins": {
    "key-auth": {},
    "proxy-rewrite": {
      "uri": "/v1/chat/completions"
    },
    "limit-req": {
      "rate": 20,
      "burst": 5,
      "key": "remote_addr",
      "rejected_code": 429,
      "rejected_msg": "请求繁忙，请稍后重试"
    },
    "limit-count": {
      "count": 100,
      "time_window": 3600,
      "key_type": "consumer_name",
      "rejected_code": 429,
      "rejected_msg": "小时 Token 用量已超限"
    },
    "demo-ai-cache": {
      "ttl": 1800,
      "redis_host": "redis",
      "redis_port": 6379
    },
    "http-logger": {
      "uri": "http://analytics-backend:9090/api/ai-usage-log",
      "batch_max_size": 50
    },
    "response-rewrite": {
      "headers": {
        "set": {
          "X-Gateway": "ai-gateway-demo",
          "X-Upstream-Latency": "$upstream_response_time"
        }
      }
    }
  }
}'

这一条路由同时具备： 鉴权 + 限流 + 缓存 + 日志 + 灰度发布 。

九、总结

9.1 AI Gateway 的核心价值

能力	无网关	有网关
API Key 管理	散落各处，泄露风险高	网关统一管控
Token 预算	月底才发现超支	实时限流 + 告警
重复调用	每次都花钱	答过的直接缓存
模型切换	改代码发版	网关改配置秒级生效
成本可视	零散、难统计	日报自动出
安全审计	无从下手	每笔调用可追溯

9.2 一个请求的完整生命周期

用户请求 → [Key-Auth] → [限流检查] → [缓存检查]
                                        ├─ 命中 → 直接返回
                                        └─ 未命中 → [路由到上游]
                                                    │
                                                  [模型响应]
                                                    │
                                          [body_filter 写缓存]
                                                    │
                                          [http-logger 记日志]
                                                    │
                                                 返回用户

9.3 写在最后

AI Gateway 不是银弹，但它解决了一个根本问题： 让 AI 能力的消耗变得可控、可观测、可治理 。

从"每个服务各自扯一根线到模型 API"，变成"所有流量经过网关再到模型"，你获得的不是复杂度，而是 对 AI 成本的掌控力 。

写在最后：AI Gateway 的玩法远不止这些

上面讲的限流、鉴权、缓存，只是 AI Gateway 的入门三件套。

在实际生产中，你还需要面对这些问题：

🔸 多个大模型怎么统一管理？一个入口自动分发 GPT / Claude / 国产模型，按成本和延迟智能切换
🔸 用户构造恶意 Prompt 怎么防？在网关层拦截，不让危险请求打到模型上
🔸 Token 成本怎么精细化控制？按消费者、按路由、按时间段分别设预算上限
🔸 模型超时了怎么办？自动降级到备用模型，而不是让用户对着报错发呆

这些问题的答案，都在我的付费专栏里。

🎯 专栏《云原生 API 网关从入门到生产》包含：

✅ 10 篇实战教程（选型 / 部署 / 路由 / 插件 / 鉴权 / 负载均衡 / 性能调优……）
✅ 完整代码仓库（一键 clone，所有配置直接能用）
✅ 生产上线 Checklist（50 项，PDF 可打印）
✅ 3 次真实线上事故复盘手册（根因分析 + 应急 SOP）
✅ Grafana 大盘 JSON + K8s YAML 全集
✅ 网关面试题精选 20 道
✅ 付费读者优先答疑

👉 点击订阅《云原生 API 网关从入门到生产》19.9 元

如果觉得这篇免费文章对你有帮助，点个赞 + 收藏，也是对我最大的支持。