搭建一个 AI API 网关——统一管理多个模型 API 的调用、限流和成本统计用着多个模型的 APIKey 散落在各处、成本不好统计、某个 API 挂了没备选。这篇文章做个AI API 网关统一入口管理所有模型调用。网关做什么多个模型 API → 网关统一入口→ 路由 限流 统计 降级 → 返回结果核心实现# aigateway.pyimporttime,os,hashlib,jsonfromfastapiimportFastAPI,HTTPException,Requestfromfastapi.responsesimportStreamingResponsefrompydanticimportBaseModelfromopenaiimportOpenAIfromdotenvimportload_dotenv load_dotenv()appFastAPI(titleAI API Gateway)# 模型配置MODELS{deepseek:OpenAI(api_keyos.getenv(DEEPSEEK_API_KEY),base_urlhttps://api.deepseek.com/v1),doubao:OpenAI(api_keyos.getenv(DOUBAO_API_KEY),base_urlhttps://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 tokensclassChatRequest(BaseModel):model:strdeepseekmessages:liststream:boolFalsedefcheck_rate(ip:str)-bool:nowtime.time()keyf{ip}:{int(now//60)}rate_limits[key]rate_limits.get(key,0)1returnrate_limits[key]30app.post(/v1/chat/completions)asyncdefchat(req:ChatRequest,request:Request):# 限流检查iprequest.headers.get(X-Forwarded-For,request.client.host)ifnotcheck_rate(ip):raiseHTTPException(429,请求太频繁)modelreq.model clientMODELS.get(model)ifnotclient:raiseHTTPException(400,f未知模型:{model})try:respclient.chat.completions.create(modelMODEL_MAP[model],messagesreq.messages,temperature0.7,streamreq.stream,)exceptException:# 降级到备选模型fallbackFALLBACK.get(model)iffallback:clientMODELS[fallback]respclient.chat.completions.create(modelMODEL_MAP[fallback],messagesreq.messages,temperature0.7,streamreq.stream,)ifreq.stream:asyncdefgen():forchunkinresp:ifchunk.choices[0].delta.content:yieldfdata:{chunk.choices[0].delta.content}\n\nyielddata: [DONE]\n\nreturnStreamingResponse(gen(),media_typetext/event-stream)# 记录成本tokensresp.usage.total_tokens costtokens/1_000_000*PRICES.get(model,2)return{choices:[{message:{content:resp.choices[0].message.content}}],usage:{total_tokens:tokens},cost_yuan:round(cost,4),}app.get(/stats)asyncdefstats():查看今日调用统计。nowtime.time()today_startint(now)-int(now)%86400return{rate_limits:{k:vfork,vinrate_limits.items()ifint(k.split(:)[1])today_start//60}}if__name____main__:importuvicorn uvicorn.run(app,port8000)使用方式pipinstallfastapi openai uvicorn python-dotenv python aigateway.py# 调用方式跟 OpenAI 完全一样只改 base_urlfromopenaiimportOpenAI clientOpenAI(api_keyany,# 网关不验证 keybase_urlhttp://localhost:8000/v1,)respclient.chat.completions.create(modeldeepseek,# 指定要用的模型messages[{role:user,content:你好}],)print(resp.choices[0].message.content)网关提供的能力功能说明 模型路由一个入口调用多个模型️ 限流保护每个 IP 每分钟最多 30 次 自动降级主模型挂了自动切备选 成本统计每次调用返回费用 OpenAI 兼容客户端代码不用改总结AI API 网关的核心价值统一管理、自动降级、成本可控。一个 FastAPI 服务、150 行代码解决所有模型 API 的管理问题。