基于 Python FastAPI 构建企业级轻量大模型流式代理与计费网关

在企业或团队内部落地大模型应用时，直接将上游官方 API Key 分发给前端或业务系统会带来严重的密钥泄露风险与额度滥用。搭建一个轻量、无状态且全兼容 OpenAI 接口规范的转发网关是最佳工程实践。

一、 技术选型与架构设计

我们选用 FastAPI + httpx (异步 HTTP/2 客户端)：

• 标准全兼容：对外暴露 /v1/chat/completions 接口，任何现成客户端（如 NextChat、LobeChat、Cursor）无需修改代码即可直接接入。
• SSE 零缓冲流式透传：利用 StreamingResponse 实时将上游数据包推回客户端，保证首包延迟（TTFB）低于 200ms。
• 令牌切片鉴权：自定义内部虚拟 Token，上游真实 Key 严格保存在网关内存或加密环境变量中。

二、 核心实现源码

from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import StreamingResponse
import httpx
import os

app = FastAPI(title="DevNotes LLM Gateway")

UPSTREAM_API_URL = "https://api.openai.com/v1/chat/completions"
UPSTREAM_API_KEY = os.getenv("OPENAI_API_KEY")
INTERNAL_KEY = os.getenv("GATEWAY_AUTH_KEY", "sk-internal-secret-token")

@app.post("/v1/chat/completions")
async def chat_completions(request: Request):
    auth_header = request.headers.get("Authorization", "")
    if not auth_header.startswith("Bearer ") or auth_header.split(" ")[1] != INTERNAL_KEY:
        raise HTTPException(status_code=401, detail="Invalid Gateway Token")

    body = await request.json()
    is_stream = body.get("stream", False)

    headers = {
        "Authorization": f"Bearer {UPSTREAM_API_KEY}",
        "Content-Type": "application/json"
    }

    client = httpx.AsyncClient(timeout=60.0)
    
    if is_stream:
        async def stream_generator():
            try:
                async with client.stream("POST", UPSTREAM_API_URL, headers=headers, json=body) as resp:
                    async for chunk in resp.aiter_bytes():
                        yield chunk
            finally:
                await client.aclose()

        return StreamingResponse(stream_generator(), media_type="text/event-stream")
    else:
        resp = await client.post(UPSTREAM_API_URL, headers=headers, json=body)
        await client.aclose()
        return resp.json()

三、 部署与生产调优建议

若前方挂载了 Nginx，必须在 Nginx 对应 location 中设置 proxy_buffering off; 和 proxy_read_timeout 300s;，否则流式打字效果会被 Nginx 缓冲区吃掉导致大段卡顿。