从零开发ChatGPT API代理：新手避坑指南与实战代码解析

1次阅读

没有评论

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

去年我们团队直接调用 ChatGPT API 时踩过这些坑：

密钥泄露：前端同学不小心把 API key 提交到 GitHub 公开仓库，导致 $1200 的异常调用费用
突发限流：促销活动时突发流量触发 OpenAI 的 429 错误，整个推荐系统瘫痪 2 小时
成本失控：实习生写的死循环代码调用 GPT- 4 接口，10 分钟烧掉 $800

对比三大主流框架：

SpringBoot：适合 Java 技术栈，但依赖管理较重
FastAPI：异步性能好，但学习曲线略陡
Flask：我们的选择，因为：
Python 生态与 OpenAI SDK 无缝衔接
轻量级扩展机制（一个 route.py 就能搞定代理）
丰富的中间件支持（JWT/ 限流等）

安装关键包：

pip install flask-jwt-extended python-dotenv

.env文件配置：

# 非对称加密更安全
JWT_ALGORITHM=RS256
JWT_PRIVATE_KEY=-----BEGIN RSA PRIVATE KEY-----
...

认证逻辑代码：

@app.route('/auth', methods=['POST'])
def login():
    # 验证用户名密码后签发 token
    access_token = create_access_token(identity=user_id)
    return {'access_token': access_token}

@app.route('/proxy', methods=['POST'])
@jwt_required()  # 自动验证 JWT
def proxy_request():
    current_user = get_jwt_identity()
    # ... 代理逻辑

滑动窗口算法实现：

def check_rate_limit(user, limit=60):
    key = f"rate_limit:{user}"
    pipe = redis.pipeline()
    now = int(time.time())

    # 记录本次请求
    pipe.zadd(key, {now: now})
    # 移除 1 分钟前的记录
    pipe.zremrangebyscore(key, 0, now-60)
    # 获取当前计数
    pipe.zcard(key)
    _, _, current_count = pipe.execute()

    if current_count > limit:
        raise RateLimitExceeded()

关键处理逻辑：

try:
    resp = requests.post(
        'https://api.openai.com/v1/chat/completions',
        headers={'Authorization': f'Bearer {API_KEY}'},
        json=request.json
    )

    # 透传 OpenAI 的限流信息
    if resp.status_code == 429:
        return {
            'error': 'rate_limit',
            'retry_after': resp.headers.get('Retry-After', 30)
        }, 429

except ConnectionError:
    # 自定义重试逻辑
    logger.warning('OpenAI 服务不稳定')

Gunicorn 启动脚本：

gunicorn -w 4 -k gevent --bind 0.0.0.0:5000 --access-logfile - "app:create_app()"

Nginx 关键配置：

location / {
    proxy_buffering on;
    proxy_buffer_size 4k;
    proxy_buffers 8 16k;  # 处理大模型响应

    # 过滤敏感头
    proxy_set_header Authorization "";
}

Prometheus 监控：

from prometheus_flask_exporter import PrometheusMetrics
metrics = PrometheusMetrics(app)
metrics.info('app_info', 'API Proxy Version', version='1.0')

日志过滤：

# 自动脱敏 API key
class SensitiveFilter(logging.Filter):
    def filter(self, record):
        record.msg = re.sub(r'sk-\w{40}', '[REDACTED]', record.msg)
        return True

当使用 GPT- 4 这类高价 API 时，可以尝试：