ChatGPT合租方案全解析：从零搭建高可用共享服务

1次阅读

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

个人开发者在使用 ChatGPT API 时常常面临资源限制问题。官方 API 的定价模式对高频使用者不够友好，尤其是当需要处理大量请求时，成本会迅速上升。相比之下，自建合租服务可以显著降低成本，同时提供更好的资源利用率。

资源限制：个人账号的 API 调用频率和并发数有限，难以满足团队或项目需求。
费用高昂：高频使用下，单月费用可能远超预算，尤其是对于小型团队或个人开发者。
企业版门槛：官方企业版虽然提供更多资源，但价格较高，且对小型团队不够灵活。

使用 Nginx 作为反向代理，可以实现请求的路由和限流。以下是一个简单的配置示例：

http {
    limit_req_zone $binary_remote_addr zone=chatgpt:10m rate=5r/s;

    server {
        listen 80;
        server_name chatgpt.example.com;

        location /api {
            limit_req zone=chatgpt burst=10 nodelay;
            proxy_pass http://backend;
        }
    }
}

在 Kubernetes 中，可以通过 ResourceQuota 实现资源隔离，确保每个租户的资源使用不会影响其他租户。

apiVersion: v1
kind: ResourceQuota
metadata:
  name: chatgpt-quota
spec:
  hard:
    cpu: "10"
    memory: 20Gi
    pods: "10"

使用 JWT 进行鉴权，可以确保每个请求都带有合法的身份信息。以下是一个 Python 代码示例：

from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import jwt

app = FastAPI()
security = HTTPBearer()

SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"

def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
    try:
        payload = jwt.decode(credentials.credentials, SECRET_KEY, algorithms=[ALGORITHM])
        return payload
    except jwt.PyJWTError:
        raise HTTPException(status_code=403, detail="Invalid token")

@app.get("/api")
async def protected_route(payload: dict = Depends(verify_token)):
    return {"message": "Access granted"}

以下是一个带有自动重试机制的 API 调用封装类，使用 Python 异步实现：

import aiohttp
import asyncio
from typing import Optional

class ChatGPTAPI:
    def __init__(self, api_key: str, max_retries: int = 3):
        self.api_key = api_key
        self.max_retries = max_retries

    async def call_api(self, prompt: str) -> Optional[dict]:
        url = "https://api.openai.com/v1/chat/completions"
        headers = {"Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        data = {
            "model": "gpt-3.5-turbo",
            "messages": [{"role": "user", "content": prompt}]
        }

        async with aiohttp.ClientSession() as session:
            for attempt in range(self.max_retries):
                try:
                    async with session.post(url, headers=headers, json=data) as response:
                        if response.status == 200:
                            return await response.json()
                        elif response.status == 429:
                            await asyncio.sleep(2 ** attempt)
                        else:
                            response.raise_for_status()
                except Exception as e:
                    print(f"Attempt {attempt + 1} failed: {e}")
                    if attempt == self.max_retries - 1:
                        raise
                    await asyncio.sleep(2 ** attempt)
        return None

使用 Prometheus 监控 API 调用费用，可以设计如下指标：

metrics:
  - name: chatgpt_api_calls_total
    type: counter
    help: Total number of ChatGPT API calls
    labels:
      - user_id
      - endpoint
  - name: chatgpt_api_cost_total
    type: counter
    help: Total cost of ChatGPT API calls
    labels:
      - user_id
      - endpoint

Kubernetes 的 Horizontal Pod Autoscaler（HPA）可以根据 CPU 或内存使用率自动调整 Pod 数量。以下是一个 HPA 配置示例：

apiVersion: autoscaling/v2beta2
kind: HorizontalPodAutoscaler
metadata:
  name: chatgpt-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: chatgpt-deployment
  minReplicas: 2
  maxReplicas: 10
  metrics:
    - type: Resource
      resource:
        name: cpu
        target:
          type: Utilization
          averageUtilization: 80

令牌桶算法：适用于突发流量，允许短时间内超过平均速率。
漏桶算法：严格限制速率，平滑流量，适用于稳定流量场景。

触发条件：高频调用、相同 IP 大量请求、异常请求内容。
规避方法：合理设置请求间隔，使用多个 API Key 轮询，避免发送敏感内容。

为每个用户分配独立的会话 ID。
使用 Redis 存储会话上下文，确保隔离。

使用 Terraform 部署最小验证环境：

安装 Terraform 和 Kubernetes 集群。
创建以下 Terraform 配置文件：

provider "kubernetes" {config_path = "~/.kube/config"}

resource "kubernetes_deployment" "chatgpt" {
  metadata {name = "chatgpt"}

  spec {
    replicas = 2

    selector {
      match_labels = {app = "chatgpt"}
    }

    template {
      metadata {
        labels = {app = "chatgpt"}
      }

      spec {
        container {
          image = "your-chatgpt-image"
          name  = "chatgpt"
        }
      }
    }
  }
}