自建 AI 网关：一套代码同时接入 GPT-4o / Claude / Gemini，账号被封也不慌

最近 V2EX 上有人贴出 OpenAI 账号被封的帖子，底下一片「我也是」。更麻烦的是：你的项目代码、公司的内部工具，全硬编码了 api.openai.com。封号那一刻，不是换个 key 的事，是整套服务直接瘫痪。

反直觉的结论：你真正需要的不是更稳定的 OpenAI 账号，而是一个不依赖任何单一服务商的调用层。

这篇文章教你用 100 行 Python 搭一个本地 AI 网关路由器，实现：

按模型名自动分发到不同后端
任意一家挂了自动 fallback
统一 OpenAI 格式，上层代码零改动

为什么要自己搞网关

先说问题根源。大多数人的 AI 调用栈长这样：


业务代码 → openai.ChatCompletion.create(model="gpt-4o") → api.openai.com

一旦账号被封、访问受限、或者你想换 Claude 试试效果，就得全局搜代码改 endpoint。项目越大，改得越痛苦。

理想的结构应该是：


业务代码 → 本地网关 → 按策略路由 → GPT / Claude / Gemini / DeepSeek

上层只认一个地址，后端随便换。

动手：用 FastAPI 搭最小可用网关

需要：Python 3.10+，安装依赖：


pip install fastapi uvicorn httpx python-dotenv

新建 gateway.py：


import os

import httpx

from fastapi import FastAPI, Request, HTTPException

from fastapi.responses import StreamingResponse

import json



app = FastAPI()



# ---- 路由表：按 model 前缀决定转发到哪个后端 ----

# 这里统一用无量Api，它聚合了 300+ 模型，一个 key 全搞定

BACKEND_URL = "https://api2everything.xyz/v1"

API_KEY = os.getenv("WULIANG_API_KEY", "your-key-here")



# 如果你有多个 key，可以在这里配置轮询或 fallback

FALLBACK_KEYS = [

    os.getenv("WULIANG_API_KEY_1", API_KEY),

    os.getenv("WULIANG_API_KEY_2", API_KEY),

]



def pick_key(attempt: int) -> str:

    return FALLBACK_KEYS[attempt % len(FALLBACK_KEYS)]



@app.post("/v1/chat/completions")

async def proxy_chat(request: Request):

    body = await request.json()

    model = body.get("model", "gpt-4o")



    # 可选：在这里做模型别名映射

    model_alias = {

        "fast":   "gpt-4o-mini",

        "smart":  "claude-opus-4-5",

        "cheap":  "deepseek-chat",

        "vision": "gemini-2.0-flash",

    }

    body["model"] = model_alias.get(model, model)



    headers = {

        "Content-Type": "application/json",

    }



    is_stream = body.get("stream", False)

    last_error = None



    for attempt in range(len(FALLBACK_KEYS)):

        headers["Authorization"] = f"Bearer {pick_key(attempt)}"

        try:

            if is_stream:

                return await _stream_request(body, headers)

            else:

                return await _normal_request(body, headers)

        except Exception as e:

            last_error = e

            continue  # 自动 fallback 到下一个 key



    raise HTTPException(status_code=502, detail=f"All backends failed: {last_error}")





async def _normal_request(body: dict, headers: dict):

    async with httpx.AsyncClient(timeout=60) as client:

        resp = await client.post(

            f"{BACKEND_URL}/chat/completions",

            json=body,

            headers=headers,

        )

        resp.raise_for_status()

        return resp.json()





async def _stream_request(body: dict, headers: dict):

    async def generator():

        async with httpx.AsyncClient(timeout=120) as client:

            async with client.stream(

                "POST",

                f"{BACKEND_URL}/chat/completions",

                json=body,

                headers=headers,

            ) as resp:

                async for chunk in resp.aiter_bytes():

                    yield chunk



    return StreamingResponse(generator(), media_type="text/event-stream")





# 透传 /v1/models 接口，方便调试

@app.get("/v1/models")

async def list_models():

    headers = {"Authorization": f"Bearer {API_KEY}"}

    async with httpx.AsyncClient(timeout=10) as client:

        resp = await client.get(f"{BACKEND_URL}/models", headers=headers)

        return resp.json()

启动：


export WULIANG_API_KEY="sk-xxxxxxxx"

uvicorn gateway:app --host 0.0.0.0 --port 8080

接入验证

本地网关起来之后，你的业务代码一行都不用改逻辑，只改 base_url：


from openai import OpenAI



client = OpenAI(

    api_key="any-string",          # 网关自己管 key，这里随便填

    base_url="http://localhost:8080/v1",

)



# 用别名：自动路由到 deepseek-chat，最便宜

resp = client.chat.completions.create(

    model="cheap",

    messages=[{"role": "user", "content": "解释一下 Python GIL"}],

)

print(resp.choices[0].message.content)



# 用别名：自动路由到 claude-opus-4-5，适合复杂推理

resp = client.chat.completions.create(

    model="smart",

    messages=[{"role": "user", "content": "帮我做这段代码的架构 review"}],

    stream=True,

)

for chunk in resp:

    print(chunk.choices[0].delta.content or "", end="", flush=True)

用 curl 快速冒烟：


curl http://localhost:8080/v1/chat/completions \

  -H "Content-Type: application/json" \

  -d '{"model":"fast","messages":[{"role":"user","content":"hi"}]}'

进阶：加一层成本追踪

生产环境里你肯定想知道每天烧了多少钱。在 gateway.py 里加个中间件：


import time

from collections import defaultdict



cost_log = defaultdict(float)



@app.middleware("http")

async def log_usage(request: Request, call_next):

    start = time.time()

    response = await call_next(request)

    elapsed = time.time() - start



    if request.url.path == "/v1/chat/completions":

        # 粗略记录请求次数，精细化可解析 usage 字段

        cost_log["requests"] += 1

        cost_log["latency_total"] += elapsed



    return response



@app.get("/stats")

def stats():

    return dict(cost_log)

关于后端选哪家

上面代码里我用的是无量Api（api2everything.xyz），原因很直接：

1. 聚合了 300+ 模型，GPT-4o、Claude Opus、Gemini、DeepSeek 一个 key 全覆盖，不用管各家账号的死活

2. 价格比官方便宜约 65%，SVIP 再打 7.7 折——同样的预算能跑更多请求

3. 完全 OpenAI 格式兼容，把上面代码的 BACKEND_URL 换成它就行，没有任何适配成本

4. 注册送 ¥1 余额，余额永久不过期，拿来测试很合适

国内直连，延迟比自己挂代理打 OpenAI 稳定很多，这对生产服务来说不是小事。

注册地址：https://api2everything.xyz/?utm_source=v2ex&utm_medium=article&utm_campaign=daily&aff=wqUb

小结

账号被封这件事会反复发生，与其每次手忙脚乱，不如一次性把调用层抽出来。上面这个网关的核心逻辑不超过 100 行，部署在本地或者一台小 VPS 上，后端随时换，上层代码永远不动。

有问题欢迎评论区聊，模型路由策略、流量限速、多租户 key 管理这些进阶需求都可以展开写。觉得有用的话点个赞，下次继续。