unified

构建一个你自己的统一 API 代理

📅 2026-08-19 00:00:00 +0000 UTC 👁 2041 浏览 📂 unified

⚠️ 待更新·2026-08-29核验 · 更新时间待核验 · 本文信息可能已过期，请以官方文档为准更新时间：2026-08-29 · 核验状态：待更新 · 官方溯源待补

背景

读完前 14 篇,你大概想自己搭一个试试。本篇给一个 100 行内的最小可用网关:接受 OpenAI 形状请求,按 model 字段路由到任意 OpenAI 兼容上游,支持流式透传与简单降级。生产可用前需补:鉴权、限流、metrics,但作为原型足够清晰。

最小架构

入口:POST /v1/chat/completions,接受标准 OpenAI 请求体。
路由表:YAML 配置文件,model -> (base_url, api_key, fallback)。
客户端:httpx.AsyncClient 池化连接。
流式:原样转发 text/event-stream,逐 chunk 透传。
降级:主路由 5xx 时切到 fallback 指定的备用 Provider。

代码示例

# gateway.py - 最小可用统一 API 代理
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import StreamingResponse
import httpx, yaml, json

app = FastAPI()
cfg = yaml.safe_load(open("providers.yaml"))
CLIENT = httpx.AsyncClient(timeout=120)

@app.post("/v1/chat/completions")
async def chat(req: Request):
    body = await req.json()
    model = body.get("model", "")
    route = next((r for r in cfg["routes"] if r["model"] == model), None)
    if not route:
        raise HTTPException(404, f"model {model} not configured")
    # build headers
    headers = {"Authorization": f"Bearer {route['key']}",
               "Content-Type": "application/json"}
    # streaming passthrough
    if body.get("stream"):
        async def gen():
            async with CLIENT.stream("POST", f"{route['base']}/chat/completions",
                                      json=body, headers=headers) as r:
                async for line in r.aiter_lines():
                    yield line + "\n"
        return StreamingResponse(gen(), media_type="text/event-stream")
    # non-streaming with fallback
    try:
        r = await CLIENT.post(f"{route['base']}/chat/completions",
                              json=body, headers=headers)
        return r.json()
    except (httpx.HTTPError, httpx.TimeoutException):
        if route.get("fallback"):
            body["model"] = route["fallback"]
            return await chat(req)
        raise HTTPException(502, "upstream failed")

配套 providers.yaml:

routes:
  - model: groq/llama-3.3-70b
    base: https://api.groq.com/openai/v1
    key: gsk_xxx
    fallback: openrouter/llama-3.3-70b:free
  - model: openrouter/llama-3.3-70b:free
    base: https://openrouter.ai/api/v1
    key: sk-or-xxx

启动:uvicorn gateway:app --port 8000,客户端把 base_url 指到 http://localhost:8000/v1 即可。

生产化加固清单

最小网关上线后,要把它推到"生产可用"还要补 7 件事:(1) 鉴权:JWT 或 API Key 校验;(2) 限流:每 client_id 令牌桶;(3) metrics:Prometheus /metrics 端点;(4) 日志:结构化 JSON + PII 脱敏;(5) 链路追踪:OpenTelemetry trace;(6) 缓存:Redis 精确层;(7) 健康检查:/healthz 与 /readyz 分离。前 6 项做完,基本能扛 1000 QPS;第 7 项是 Kubernetes 部署的必备。

最佳实践与下一步

加鉴权:用 FastAPI Depends 校验客户端 token,否则你的网关会被白嫖。
加限流:slowapi 或自实现令牌桶,每 client_id 每分钟 30 次。
加 metrics:Prometheus /metrics 端点暴露延迟与成功率。
加缓存:Redis 精确缓存层,命中率立刻翻倍。
加 metrics 看板:Grafana 模板一接,运维体验就接近商业网关。
压力测试:上线前用 wrk 或 vegeta 压测,获取极限 QPS 与 P99 延迟。
回归测试:每次改动都跑一遍"调用 → 流式 → 降级"三场景的集成测试,确保不退化。

最小可用网关 → 加鉴权 → 加限流 → 加 metrics → 加缓存 → 加降级链,每一步都是把第 1-14 篇的某一个概念落到代码上。

⚠️ Pending Update · 2026-08-29 Verification · Content may be outdated, please refer to official docs Updated: 2026-08-29 · Status: Pending Verification

Background

After reading the previous 14 articles, you probably want to try building one yourself. This article gives a minimal-viable gateway in under 100 lines: it accepts OpenAI-shape requests, routes by the model field to any OpenAI-compatible upstream, supports streaming passthrough, and does simple fallback. Before production use, add auth, rate limiting, and metrics — but as a prototype it is clear enough.

Minimal Architecture

Ingress: POST /v1/chat/completions, accepts the standard OpenAI request body.
Routing table: a YAML config file mapping model -> (base_url, api_key, fallback).
Client: a pooled httpx.AsyncClient.
Streaming: pass through text/event-stream chunk by chunk.
Fallback: on primary 5xx, switch to the fallback provider.

Code Example

# gateway.py - minimal viable unified API proxy
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import StreamingResponse
import httpx, yaml

app = FastAPI()
cfg = yaml.safe_load(open("providers.yaml"))
CLIENT = httpx.AsyncClient(timeout=120)

@app.post("/v1/chat/completions")
async def chat(req: Request):
    body = await req.json()
    model = body.get("model", "")
    route = next((r for r in cfg["routes"] if r["model"] == model), None)
    if not route:
        raise HTTPException(404, f"model {model} not configured")
    headers = {"Authorization": f"Bearer {route['key']}",
               "Content-Type": "application/json"}
    # streaming passthrough
    if body.get("stream"):
        async def gen():
            async with CLIENT.stream("POST",
                    f"{route['base']}/chat/completions",
                    json=body, headers=headers) as r:
                async for line in r.aiter_lines():
                    yield line + "\n"
        return StreamingResponse(gen(), media_type="text/event-stream")
    # non-streaming with fallback
    try:
        r = await CLIENT.post(f"{route['base']}/chat/completions",
                              json=body, headers=headers)
        return r.json()
    except (httpx.HTTPError, httpx.TimeoutException):
        if route.get("fallback"):
            body["model"] = route["fallback"]
            return await chat(req)
        raise HTTPException(502, "upstream failed")

Companion providers.yaml:

routes:
  - model: groq/llama-3.3-70b
    base: https://api.groq.com/openai/v1
    key: gsk_xxx
    fallback: openrouter/llama-3.3-70b:free
  - model: openrouter/llama-3.3-70b:free
    base: https://openrouter.ai/api/v1
    key: sk-or-xxx

Launch with uvicorn gateway:app --port 8000, point the client's base_url to http://localhost:8000/v1, and you are ready.

Production Hardening Checklist

After the minimal gateway ships, pushing it to "production-ready" requires seven additions: (1) auth — JWT or API key validation; (2) rate limiting — token bucket per client_id; (3) metrics — Prometheus /metrics endpoint; (4) logs — structured JSON with PII scrubbing; (5) tracing — OpenTelemetry trace; (6) cache — Redis exact layer; (7) health checks — /healthz and /readyz separated. Complete the first six and the gateway handles around 1000 QPS; the seventh is mandatory for Kubernetes deployment.

Best Practices and Next Steps

Add auth: use FastAPI Depends to validate the client token, otherwise your gateway will be free-ridden.
Add rate limiting: slowapi or a self-implemented token bucket, 30 calls/min per client_id.
Add metrics: a Prometheus /metrics endpoint for latency and success rate.
Add a cache: a Redis exact-cache layer doubles hit rate immediately.
Add a dashboard: a Grafana template gets ops experience close to a commercial gateway.
Load test: before going live, run wrk or vegeta to learn your peak QPS and P99 latency.
Regression tests: after every change, run integration tests covering "call → stream → fallback" so behavior never regresses silently.

Minimal viable gateway → add auth → add rate limiting → add metrics → add cache → add degradation chain. Each step lands one of the concepts from articles 1-14 in code.