做AI应用开发的人都有一个共同的烦恼：模型太多，API各不相同。
你可能用DeepSeek做翻译、用通义千问做摘要、用Claude做代码生成——每个模型都有自己的SDK、定价模式、速率限制。切换起来像在不同的App之间跳来跳去，不仅效率低，还很难做成本管控。
本文教你用50行Python代码搭建一个轻量级多模型API网关，实现统一调用、自动比价、请求缓存三大功能。
准备工作
你需要：
- • Python 3.10+
- • 至少两个AI模型的API Key（推荐DeepSeek + 通义千问，都有免费额度）
- •
pip install httpx（异步HTTP客户端）
第一步：创建项目结构
mkdir ai-gateway && cd ai-gateway
touch gateway.py config.py
第二步：配置模型信息
在 config.py 中定义模型列表：
MODELS = {
"deepseek-v3": {
"provider": "deepseek",
"endpoint": "https://api.deepseek.com/v1/chat/completions",
"api_key": "***",
"price_per_1k_input": 0.00014, # 美元
"price_per_1k_output": 0.00028,
},
"qwen-plus": {
"provider": "alibaba",
"endpoint": "https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions",
"api_key": "***",
"price_per_1k_input": 0.0005,
"price_per_1k_output": 0.002,
},
}
第三步：编写网关核心逻辑
创建 gateway.py：
import httpx
import hashlib
import json
from config import MODELS
# 简易缓存（生产环境建议用Redis）
cache = {}
async def call_model(model_name, messages, temperature=0.7):
"""统一的模型调用接口"""
model = MODELS.get(model_name)
if not model:
return {"error": f"未知模型: {model_name}"}
# 生成缓存键
cache_key = hashlib.md5(
json.dumps({"model": model_name, "messages": messages}).encode()
).hexdigest()
# 命中缓存直接返回
if cache_key in cache:
print(f"[CACHE HIT] {model_name}")
return cache[cache_key]
# 构造请求
headers = {"Authorization": f"Bearer {model['api_key']}"}
payload = {
"model": model_name,
"messages": messages,
"temperature": temperature,
}
async with httpx.AsyncClient(timeout=60) as client:
resp = await client.post(model["endpoint"], json=payload, headers=headers)
data = resp.json()
# 计算成本
usage = data.get("usage", {})
input_tokens = usage.get("prompt_tokens", 0)
output_tokens = usage.get("completion_tokens", 0)
cost = (input_tokens * model["price_per_1k_input"] / 1000 +
output_tokens * model["price_per_1k_output"] / 1000)
result = {
"model": model_name,
"content": data["choices"][0]["message"]["content"],
"tokens": {"input": input_tokens, "output": output_tokens},
"cost_usd": round(cost, 6),
}
cache[cache_key] = result
return result
async def smart_call(messages, preferred_model=None):
"""智能调用：优先指定模型，失败则自动降级"""
if preferred_model:
try:
return await call_model(preferred_model, messages)
except Exception as e:
print(f"[FALLBACK] {preferred_model} 失败: {e}")
# 自动选最便宜的模型
sorted_models = sorted(
MODELS.items(),
key=lambda x: x[1]["price_per_1k_input"]
)
for name, _ in sorted_models:
try:
return await call_model(name, messages)
except Exception:
continue
return {"error": "所有模型调用失败"}
第四步：实战测试
新建 test.py 来验证你的网关。这个测试文件会调用"指定模型"和"智能路由"两种模式，让你直观感受网关的工作原理：
import asyncio
from gateway import call_model, smart_call
async def main():
# 方式一：指定模型
result = await call_model("deepseek-v3", [
{"role": "user", "content": "用三句话解释什么是注意力机制"}
])
print(f"模型: {result['model']}")
print(f"消耗: {result['cost_usd']}美元")
print(f"回复: {result['content'][:100]}...\n")
# 方式二：智能调用（自动选最便宜的可用模型）
result2 = await smart_call([
{"role": "user", "content": "写一首关于人工智能的五言绝句"}
])
print(f"自动选择: {result2['model']}, 成本: {result2['cost_usd']}美元")
asyncio.run(main())
运行测试：
python3 test.py
预期输出类似：
模型: deepseek-v3
消耗: 0.000266美元
回复: 注意力机制是Transformer架构的核心创新...
自动选择: deepseek-v3, 成本: 0.000294美元
进阶优化方向
这个网关已经可以满足日常使用，但如果你想让它在生产环境落地，可以考虑三个方向：
- 1. 真正的比价路由：在
smart_call 中加入价格比较逻辑，同时向多个模型发送请求，取最快+最便宜的结果。 - 2. 流式输出支持：用
httpx.stream() 实现Server-Sent Events，让前端逐字显示回复，用户体验大幅提升。 - 3. 用量仪表盘：将每次调用的token消耗和成本写入SQLite数据库，然后用Streamlit做一个可视化面板，团队的AI支出一目了然。
- 4. 负载均衡：当某个模型API响应变慢时，自动将请求路由到备用模型。这对于依赖AI能力的生产环境尤为重要——API偶尔的抖动不应该影响你的服务可用性。
为什么不直接用现成方案
你可能会问：市面上已经有LiteLLM、One API等成熟的AI网关项目，为什么要自己写？
答案是学习成本和控制力。现成方案功能强大但配置复杂，动辄几十个环境变量和YAML配置文件。对于个人开发者或小团队来说，50行自研代码的好处是：你清楚地知道每一行在做什么，出了问题能在五分钟内定位和修复。而且，自建网关可以无缝嵌入你现有的项目架构——不需要引入一个重量级的外部依赖。
如果你后续想升级到生产级方案，这个自研网关也给了你一个完美的"过渡期"——先用起来，等真正遇到瓶颈时再迁移到LiteLLM等专业工具也不迟。
核心思路：API网关的本质是"把复杂度关在盒子里"——让上层应用只关心"我想问AI什么"，而不是"我要调用哪个模型的哪个版本的哪家API"。
🔥 关注LC智趣厅，每周推送实用的AI开发教程。
👇 你平时管理多个模型API有什么技巧？欢迎评论区分享你的方案。
