AI API 中转站：从野蛮生长到连夜关停，这门"副业"到底靠什么运转

用户请求 → Nginx/Cloudflare（反代 + 限流）→ 官方 API 端点

# ai_proxy.py — 最小 AI API 中转代理示例
# 运行前：pip install flask requests
# 设置环境变量：export OPENAI_API_KEY="sk-你的官方key"
# 启动：python ai_proxy.py
# 测试：curl http://localhost:5000/v1/chat/completions \
#   -H "Authorization: Bearer proxy-key-123" \
#   -H "Content-Type: application/json" \
#   -d '{"model":"gpt-4o-mini","messages":[{"role":"user","content":"hello"}]}'

import os
import time
from flask import Flask, request, Response
import requests

app = Flask(__name__)

# ========== 配置区 ==========
OFFICIAL_BASE = "https://api.openai.com"          # 官方端点
PROXY_API_KEY = os.environ.get("PROXY_API_KEY", "proxy-key-123")  # 给用户分发的 key
OFFICIAL_API_KEY = os.environ.get("OPENAI_API_KEY", "")           # 你的官方 key
RATE_LIMIT = 10  # 每分钟每 IP 最大请求数
# ========== 配置区结束 ==========

# 简易限流：IP → 最近请求时间戳列表
ip_requests: dict[str, list[float]] = {}

def check_rate_limit(ip: str) -> bool:
    now = time.time()
    window = ip_requests.get(ip, [])
    # 只保留最近 60 秒的记录
    window = [t for t in window if now - t < 60]
    ip_requests[ip] = window
    if len(window) >= RATE_LIMIT:
        return False
    window.append(now)
    return True

@app.route("/v1/<path:subpath>", methods=["GET", "POST", "DELETE"])
def proxy(subpath):
    # 1. 验证用户 key
    auth = request.headers.get("Authorization", "")
    if auth.replace("Bearer ", "") != PROXY_API_KEY:
        return Response("invalid proxy key", status=401)

    # 2. 限流
    ip = request.remote_addr
    if not check_rate_limit(ip):
        return Response("rate limit exceeded", status=429)

    # 3. 构造转发请求：替换 Authorization 为官方 key
    headers = {k: v for k, v in request.headers if k.lower() != "host"}
    headers["Authorization"] = f"Bearer {OFFICIAL_API_KEY}"
    headers["Host"] = "api.openai.com"

    url = f"{OFFICIAL_BASE}/v1/{subpath}"
    resp = requests.request(
        method=request.method,
        url=url,
        headers=headers,
        json=request.get_json(silent=True),
        stream=True,  # 支持 SSE 流式响应
    )

    # 4. 原样返回官方响应（含流式）
    excluded = ["content-encoding", "content-length", "transfer-encoding"]
    resp_headers = {k: v for k, v in resp.headers.items() if k.lower() not in excluded}
    return Response(
        resp.content,
        status=resp.status_code,
        headers=resp_headers,
    )

if __name__ == "__main__":
    if not OFFICIAL_API_KEY:
        print("请设置 OPENAI_API_KEY 环境变量")
        os._exit(1)
    app.run(host="0.0.0.0", port=5000)

# /etc/nginx/conf.d/ai-proxy.conf
# 测试：curl https://your-domain.com/v1/chat/completions \
#   -H "Authorization: Bearer your-proxy-key" \
#   -H "Content-Type: application/json" \
#   -d '{"model":"gpt-4o-mini","messages":[{"role":"user","content":"hello"}]}'

server {
    listen 443 ssl;
    server_name your-domain.com;

    ssl_certificate     /etc/ssl/your-domain.crt;
    ssl_certificate_key /etc/ssl/your-domain.key;

    # 限流：每 IP 每秒 5 个请求
    limit_req_zone $binary_remote_addr zone=ai:10m rate=5r/s;
    limit_req zone=ai burst=20 nodelay;

    location /v1/ {
        # 验证 proxy key（用 map 或 auth_basic 模块均可）
        # 这里用简单变量比对示意，生产环境建议用 map
        if ($http_authorization != "Bearer your-proxy-key") {
            return 401;
        }

        proxy_pass https://api.openai.com/v1/;
        proxy_set_header Host api.openai.com;
        proxy_set_header Authorization "Bearer sk-你的官方key";
        proxy_set_header X-Real-IP $remote_addr;

        # SSE 流式支持
        proxy_buffering off;
        proxy_cache off;
        chunked_transfer_encoding on;

        # 超时设置
        proxy_read_timeout 300s;
        proxy_connect_timeout 60s;
    }
}