搭建一个 AI API 网关——统一管理多个模型 API、自动降级、成本统计

多个模型 API → 网关（统一入口）→ 路由 + 限流 + 统计 + 降级 → 返回结果

# aigateway.py
import time, os, hashlib, json
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from openai import OpenAI
from dotenv import load_dotenv
load_dotenv()

app = FastAPI(title="AI API Gateway")

# 模型配置
MODELS = {
    "deepseek": OpenAI(api_key=os.getenv("DEEPSEEK_API_KEY"), base_url="https://api.deepseek.com/v1"),
    "doubao": OpenAI(api_key=os.getenv("DOUBAO_API_KEY"), base_url="https://ark.cn-beijing.volces.com/api/v3"),
}

MODEL_MAP = {"deepseek": "deepseek-chat", "doubao": "doubao-2.1-pro"}

# 降级链：主模型挂了自动切备选
FALLBACK = {"doubao": "deepseek", "deepseek": "doubao"}

# 限流：每 IP 每分钟最多 30 次
rate_limits = {}

PRICES = {"deepseek": 2, "doubao": 6}  # ¥/M tokens

class ChatRequest(BaseModel):
    model: str = "deepseek"
    messages: list
    stream: bool = False

def check_rate(ip: str) -> bool:
    now = time.time()
    key = f"{ip}:{int(now//60)}"
    rate_limits[key] = rate_limits.get(key, 0) + 1
    return rate_limits[key] = today_start // 60}}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, port=8000)

pip install fastapi openai uvicorn python-dotenv
python aigateway.py

# 调用方式跟 OpenAI 完全一样，只改 base_url
from openai import OpenAI

client = OpenAI(
    api_key="any",  # 网关不验证 key
    base_url="http://localhost:8000/v1",
)

resp = client.chat.completions.create(
    model="deepseek",  # 指定要用的模型
    messages=[{"role": "user", "content": "你好"}],
)
print(resp.choices[0].message.content)