手把手:用Python搭建多模型API网关,一键比价省钱又高效

为什么要自己搭API网关

做AI应用开发的人都有一个共同的烦恼:模型太多,API各不相同。

手把手:用Python搭建多模型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有什么技巧?欢迎评论区分享你的方案。


— END —
LC 智趣厅 · 科技与生活的交点
ihygg.cn

滚动至顶部
微信公众号:LC智趣厅

扫码关注微信公众号
LC智趣厅