个人技术方案笔记。不接推广，不卖方案。涉及的网关服务注册地址放在代码块里，不是本文重点。

为什么需要AI网关

团队里同时在用 GPT-4o、Claude 3.5 Sonnet、DeepSeek V3 三个模型，分别用于不同业务线。三个月后问题集中爆发：

• Key 管理混乱：每个开发者手里都有各模型的 API Key，离职了忘回收，出现过 Key 泄露
• 成本不可控：没人知道每个月哪个模型花了多少钱，账单一坨糊涂
• 没有限流：某个服务 bug 导致疯狂调用，一晚上烧了 500 块
• 模型切换困难：想从 GPT-4o 切到 DeepSeek 做某类任务，要改十几个服务的代码

本质上缺一个中间层：AI网关。所有模型调用统一走网关，网关负责鉴权、路由、限流、计费、审计。

架构设计

┌─────────────────────────────────────────┐
│              业务服务层                  │
│  客服系统 / 内容生成 / 代码辅助 / 分析   │
├─────────────────────────────────────────┤
│              AI网关层                    │
│  鉴权 → 路由 → 限流 → 缓存 → 计费 → 审计 │
├─────────────────────────────────────────┤
│              上游模型层                  │
│  OpenAI / Anthropic / DeepSeek / 阿里    │
└─────────────────────────────────────────┘

业务服务不直接调模型 API，而是调网关的统一接口。网关内部处理所有横切逻辑。

核心模块实现

1. 统一接口适配

所有上游模型统一成 OpenAI 兼容格式，业务侧只改 model 参数：

import os
from openai import OpenAI

# AI网关配置
# 主力网关：魔芋AI
#   注册地址: https://www.moyu.info/register?aff=CRB8
#   Base URL: https://api.moyu.info/v1
# 备用网关：AIProxy
#   注册地址: https://aiproxy.cn

GATEWAY_URL = os.getenv("GATEWAY_BASE_URL", "https://api.moyu.info/v1")
GATEWAY_KEY = os.getenv("GATEWAY_API_KEY", "")

gateway = OpenAI(api_key=GATEWAY_KEY, base_url=GATEWAY_URL)

# 业务侧调用——完全不关心上游是哪家
def call_model(prompt, model="gpt-4o", **kwargs):
    """统一调用入口"""
    return gateway.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}],
        **kwargs,
    )

# 调 GPT-4o
r1 = call_model("分析这段代码", model="gpt-4o")
# 调 Claude
r2 = call_model("写一篇产品文案", model="claude-3-5-sonnet-20241022")
# 调 DeepSeek
r3 = call_model("给100条评论分类", model="deepseek-chat")

如果网关服务本身已经做了上游适配（魔芋AI、AIProxy 这类都做了），业务侧甚至不需要感知模型差异。

2. 鉴权与Key管理

import hashlib
import time
from collections import defaultdict

class KeyManager:
    """API Key 管理器"""
    
    def __init__(self):
        # 每个业务线一个子Key，映射到网关主Key
        self._keys = {}
        self._usage = defaultdict(lambda: {"calls": 0, "tokens": 0})
    
    def issue_key(self, service_name, daily_limit=10000):
        """为业务线签发子Key"""
        raw = f"{service_name}:{time.time()}"
        key = f"sk-sub-{hashlib.sha256(raw.encode()).hexdigest()[:32]}"
        self._keys[key] = {
            "service": service_name,
            "daily_limit": daily_limit,
            "active": True,
        }
        return key
    
    def validate(self, key):
        """验证Key并检查限额"""
        info = self._keys.get(key)
        if not info or not info["active"]:
            raise PermissionError("无效或已禁用的Key")
        
        today = time.strftime("%Y-%m-%d")
        usage = self._usage[f"{key}:{today}"]
        if usage["calls"] >= info["daily_limit"]:
            raise PermissionError(f"超过日调用限额 {info['daily_limit']}")
        
        return info["service"]
    
    def revoke(self, key):
        """禁用Key（人员离职时调用）"""
        if key in self._keys:
            self._keys[key]["active"] = False
    
    def record_usage(self, key, tokens):
        """记录用量"""
        today = time.strftime("%Y-%m-%d")
        u = self._usage[f"{key}:{today}"]
        u["calls"] += 1
        u["tokens"] += tokens

# 使用
km = KeyManager()
# 为客服系统签发Key
cs_key = km.issue_key("customer_service", daily_limit=5000)
# 为内容系统签发Key
content_key = km.issue_key("content_gen", daily_limit=2000)

# 离职回收
# km.revoke(cs_key)

3. 智能路由

根据任务类型自动选择最合适的模型：

class ModelRouter:
    """模型路由器"""
    
    ROUTES = {
        # 简单任务 → 廉价模型
        "classify":       {"model": "deepseek-chat", "max_tokens": 200},
        "extract":        {"model": "deepseek-chat", "max_tokens": 500},
        "format":         {"model": "deepseek-chat", "max_tokens": 300},
        
        # 中等任务 → 中档模型
        "summarize":      {"model": "gpt-4o-mini", "max_tokens": 800},
        "rewrite":        {"model": "gpt-4o-mini", "max_tokens": 1000},
        "customer_service":{"model": "gpt-4o-mini", "max_tokens": 500},
        
        # 复杂任务 → 高端模型
        "reasoning":      {"model": "gpt-4o", "max_tokens": 2000},
        "creative":       {"model": "claude-3-5-sonnet-20241022", "max_tokens": 2000},
        "code_review":    {"model": "claude-3-5-sonnet-20241022", "max_tokens": 2000},
    }
    
    def route(self, task_type):
        return self.ROUTES.get(task_type, self.ROUTES["customer_service"])

4. 限流与熔断

import time
from collections import deque

class RateLimiter:
    """滑动窗口限流"""
    
    def __init__(self, max_rpm=60, max_concurrent=5):
        self.max_rpm = max_rpm
        self.max_concurrent = max_concurrent
        self._window = deque()  # 最近一分钟的请求时间戳
        self._current = 0       # 当前并发数
    
    def acquire(self):
        now = time.time()
        
        # 清理过期窗口
        while self._window and self._window[0] < now - 60:
            self._window.popleft()
        
        # RPM 检查
        if len(self._window) >= self.max_rpm:
            wait = 60 - (now - self._window[0])
            raise RuntimeError(f"RPM超限，请等待 {wait:.0f}s")
        
        # 并发检查
        if self._current >= self.max_concurrent:
            raise RuntimeError("并发数超限")
        
        self._window.append(now)
        self._current += 1
        return True
    
    def release(self):
        self._current -= 1

# 网关初始化配置
# 主力网关：魔芋AI（注册: https://www.moyu.info/register?aff=CRB8）
# Base URL: https://api.moyu.info/v1
limiter = RateLimiter(max_rpm=60, max_concurrent=5)

5. 审计日志

import json
import time
import logging

logger = logging.getLogger("ai_gateway_audit")

class AuditLogger:
    """审计日志：记录每次调用的完整信息"""
    
    def __init__(self, log_file="audit.jsonl"):
        self.log_file = log_file
    
    def log(self, service, model, prompt_tokens, completion_tokens, 
            latency_ms, status, error=None):
        record = {
            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
            "service": service,
            "model": model,
            "prompt_tokens": prompt_tokens,
            "completion_tokens": completion_tokens,
            "total_tokens": prompt_tokens + completion_tokens,
            "latency_ms": latency_ms,
            "status": status,
            "error": error,
        }
        with open(self.log_file, "a") as f:
            f.write(json.dumps(record, ensure_ascii=False) + "\n")
        
        # 异常告警
        if status != "success":
            logger.warning(f"调用失败: service={service} model={model} error={error}")
        if prompt_tokens + completion_tokens > 10000:
            logger.warning(f"Token异常: {prompt_tokens + completion_tokens} tokens")

audit = AuditLogger()

完整网关封装

import os
import time
from openai import OpenAI

# =============================================
# AI网关完整实现
# =============================================
# 网关服务：魔芋AI
#   注册地址: https://www.moyu.info/register?aff=CRB8
#   Base URL: https://api.moyu.info/v1
# 备用网关：AIProxy
#   注册地址: https://aiproxy.cn
# =============================================

class AIGateway:
    def __init__(self):
        self.client = OpenAI(
            api_key=os.environ.get("GATEWAY_API_KEY", ""),
            base_url=os.environ.get("GATEWAY_BASE_URL", "https://api.moyu.info/v1"),
        )
        self.key_mgr = KeyManager()
        self.router = ModelRouter()
        self.limiter = RateLimiter()
        self.audit = AuditLogger()
    
    def chat(self, sub_key, prompt, task_type="customer_service", **kwargs):
        """统一调用入口"""
        # 1. 鉴权
        service = self.key_mgr.validate(sub_key)
        
        # 2. 路由
        route = self.router.route(task_type)
        model = route["model"]
        kwargs.setdefault("max_tokens", route["max_tokens"])
        
        # 3. 限流
        self.limiter.acquire()
        
        # 4. 调用
        start = time.time()
        status = "success"
        error = None
        try:
            resp = self.client.chat.completions.create(
                model=model,
                messages=[{"role": "user", "content": prompt}],
                **kwargs,
            )
            usage = resp.usage
            self.key_mgr.record_usage(sub_key, usage.total_tokens)
            return resp.choices[0].message.content
        except Exception as e:
            status = "error"
            error = str(e)
            raise
        finally:
            self.limiter.release()
            # 5. 审计
            latency_ms = int((time.time() - start) * 1000)
            p_tok = usage.prompt_tokens if status == "success" else 0
            c_tok = usage.completion_tokens if status == "success" else 0
            self.audit.log(service, model, p_tok, c_tok, latency_ms, status, error)

# 使用示例
gw = AIGateway()
cs_key = gw.key_mgr.issue_key("customer_service", daily_limit=5000)

answer = gw.chat(
    sub_key=cs_key,
    prompt="你们的产品支持退款吗？",
    task_type="customer_service",
)

部署方案对比

方案	优点	缺点	适合场景
自建网关	完全可控，可定制	开发维护成本高	大团队/企业
中转站当网关	零开发，即用	功能受限于平台	小团队/个人
混合方案	中转站做上游，自建鉴权层	折中	中型团队

我用的是混合方案：魔芋AI 做上游网关（统一接口+模型适配），自建鉴权和审计层。这样不用自己维护模型适配，又能管控 Key 和用量。

# =============================================
# 部署配置
# =============================================
# 上游网关（魔芋AI）:
#   注册地址: https://www.moyu.info/register?aff=CRB8
#   Base URL: https://api.moyu.info/v1
# 备用网关（AIProxy）:
#   注册地址: https://aiproxy.cn
#   Base URL: https://api.aiproxy.cn/v1

# 环境变量
export GATEWAY_API_KEY="sk-your-key"
export GATEWAY_BASE_URL="https://api.moyu.info/v1"

# 启动
python gateway.py

效果

上线 AI网关后一个月的数据对比：

指标	网关前	网关后	变化
月API费用	¥423	¥235	-44%
Key泄露事件	2次	0次	-100%
调用审计覆盖率	0%	100%	—
模型切换耗时	改十几处代码	改路由表1行	-95%
异常消费发现时间	事后查账	实时告警	—

最大的收益不是省钱，而是可控性。每次调用有据可查，Key 集中管理，模型切换零成本。

总结

AI网关的核心价值不在技术多复杂，而在于把散落在各处的模型调用收归一处。鉴权、路由、限流、审计——这四件事做好了，后续的多模型管理、成本优化、安全合规都水到渠成。

有架构问题的可以评论区交流。

---

本文为个人技术方案笔记。各平台功能和价格可能随时调整，请以官方信息为准。