ChatGPT国内访问的工程化解决方案：从代理配置到API优化

1次阅读

没有评论

共计 2469 个字符，预计需要花费 7 分钟才能阅读完成。

国内开发者访问 ChatGPT API 时，主要面临两个核心问题：

网络稳定性 ：直接访问 OpenAI 服务经常出现连接超时或中断，尤其在晚高峰时段 API 成功率可能低于 60%
合规风险 ：原始 API 响应可能包含需过滤的内容，且用户数据跨境传输需满足《个人信息保护法》要求

优点：
维护成本低（Nginx 标准配置）
支持 TCP 层优化（如 TLS1.3/TCP Fast Open）
易扩展负载均衡
缺点：
需要海外服务器资源

适用场景：
需要长连接保持（如 Chat 场景）
突破企业级防火墙限制
性能损耗：
额外增加 200-300ms 延迟

典型配置：
AWS Lambda + API Gateway
阿里云函数计算
成本提示：
当 QPS>100 时费用可能超过独立服务器

import httpx
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
async def query_chatgpt(prompt):
    async with httpx.AsyncClient(timeout=httpx.Timeout(15.0, connect=3.0),
        limits=httpx.Limits(max_connections=50)
    ) as client:
        try:
            resp = await client.post(
                "https://your-proxy-domain/v1/chat/completions",
                headers={"Authorization": f"Bearer {api_key}",
                    "X-Signature": generate_signature(prompt)  # 下文实现
                },
                json={"model": "gpt-3.5-turbo", "messages": [{"role": "user", "content": prompt}]}
            )
            resp.raise_for_status()
            return resp.json()
        except httpx.ReadTimeout:
            # 触发 tenacity 自动重试
            raise

import hmac
import hashlib

def generate_signature(text):
    secret = os.getenv('SIGNING_SECRET').encode()
    return hmac.new(secret, text.encode(), hashlib.sha256).hexdigest()

Nginx 配置验证：

location /v1/chat/completions {if ($http_x_signature != "$arg_signature") {return 403;}
    proxy_pass https://api.openai.com;
}

from redis import Redis
from datetime import timedelta

redis = Redis.from_url(os.getenv('REDIS_URL'))

def get_cache_key(prompt):
    return f"gpt_cache:{hashlib.md5(prompt.encode()).hexdigest()}"

def get_cached_response(prompt):
    if cached := redis.get(get_cache_key(prompt)):
        return json.loads(cached)
    return None

def set_cache(prompt, response, ttl=timedelta(hours=6)):
    redis.setex(get_cache_key(prompt), ttl, json.dumps(response))

方案	P50 延迟	P99 延迟	成功率
直连 API	1200ms	超时	58%
香港代理	680ms	2200ms	92%
日本代理 + 缓存	350ms*	800ms	99.5%

(* 缓存命中时平均延迟)

upstream openai_backend {
    server 1.1.1.1:443 max_fails=3;
    server 2.2.2.2:443 backup;
    keepalive 20;
}

from fastapi import FastAPI, Request

app = FastAPI()
BANNED_WORDS = [...]  # 从数据库加载

@app.middleware("http")
async def content_filter(request: Request, call_next):
    if any(word in (await request.body()).decode() for word in BANNED_WORDS):
        return JSONResponse(status_code=400, content={"error": "content violation"})
    return await call_next(request)