ChatGPT国内使用指南：技术实现与合规避坑

1次阅读

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

OpenAI 的 ChatGPT 服务目前未对中国大陆地区开放，直接访问 API 会面临以下限制：

IP 地址地理封锁：OpenAI 会拒绝来自中国大陆 IP 的 API 请求
支付限制：国际信用卡绑定困难
合规风险：数据跨境传输可能违反《个人信息保护法》

通过海外服务器搭建反向代理是最基础的解决方案：

# /etc/nginx/conf.d/openai.conf
server {
    listen 443 ssl;
    server_name yourdomain.com;

    ssl_certificate /path/to/cert.pem;
    ssl_certificate_key /path/to/key.pem;

    location /v1/ {
        proxy_pass https://api.openai.com/;
        proxy_set_header Host api.openai.com;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_ssl_server_name on;
        proxy_redirect off;
    }
}

关键配置说明：

必须启用 TLS 1.2+ 加密
建议配置 WAF 规则防止滥用
需要实现 IP 白名单访问控制

直接在客户端调用代理存在风险，建议增加中间层：

# middleware.py
from fastapi import FastAPI, Request, HTTPException
import hmac
import hashlib
import time

app = FastAPI()

SECRET_KEY = "your_shared_secret"
RATE_LIMIT = 5  # 请求 / 秒

@app.middleware("http")
async def verify_request(request: Request, call_next):
    # 1. 签名验证
    signature = request.headers.get("X-Signature")
    timestamp = request.headers.get("X-Timestamp")

    if not signature or not timestamp:
        raise HTTPException(status_code=401)

    expected = hmac.new(SECRET_KEY.encode(), 
        f"{timestamp}{request.url}".encode(),
        hashlib.sha256
    ).hexdigest()

    if not hmac.compare_digest(signature, expected):
        raise HTTPException(status_code=401)

    # 2. 时间窗口检查
    if int(time.time()) - int(timestamp) > 30:
        raise HTTPException(status_code=408)

    # 3. 频率控制
    client_ip = request.client.host
    if not allow_request(client_ip):
        raise HTTPException(status_code=429)

    return await call_next(request)

实现自动重试机制（指数退避）
内置敏感词过滤
支持对话 session 管理
提供 TypeScript 类型定义

# openai_wrapper.py
import os
import httpx
from typing import AsyncGenerator

class OpenAIClient:
    def __init__(self):
        self.base_url = os.getenv("PROXY_ENDPOINT", "https://your-proxy.com/v1")
        self.api_key = os.getenv("OPENAI_KEY")
        self.client = httpx.AsyncClient(timeout=30.0)

    async def chat_completion(
        self, 
        messages: list,
        model: str = "gpt-3.5-turbo"
    ) -> AsyncGenerator[str, None]:
        headers = {"Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }

        data = {
            "model": model,
            "messages": messages,
            "stream": True
        }

        async with self.client.stream(
            "POST",
            f"{self.base_url}/chat/completions",
            headers=headers,
            json=data
        ) as response:
            async for chunk in response.aiter_lines():
                yield chunk

# context_manager.py
from typing import List, Dict

class Conversation:
    def __init__(self, max_history=10):
        self.history: List[Dict] = []
        self.max_history = max_history

    def add_message(self, role: str, content: str):
        self.history.append({"role": role, "content": content})
        if len(self.history) > self.max_history * 2:
            self.history = self.history[-self.max_history:]

    def get_messages(self) -> List[Dict]:
        return self.history.copy()

# content_filter.py
import re
from typing import Optional

class ContentFilter:
    def __init__(self):
        self.patterns = [(r"政治敏感词 1 | 敏感词 2", "***"),
            # 从合规词库加载更多规则
        ]

    def filter(self, text: str) -> Optional[str]:
        if not text:
            return text

        for pattern, repl in self.patterns:
            text = re.sub(pattern, repl, text, flags=re.IGNORECASE)

        if self._check_illegal(text):
            return None

        return text

    def _check_illegal(self, text: str) -> bool:
        # 实现更复杂的合规检查
        return False

# 使用 httpx 连接池
client = httpx.AsyncClient(
    limits=httpx.Limits(
        max_connections=100,
        max_keepalive_connections=20
    ),
    timeout=httpx.Timeout(15.0, read=60.0)
)

对相同 prompt 的请求缓存 60 秒
使用 LRU 缓存算法
对 streaming 响应缓存分块处理

# 使用 asyncio.gather 并行处理
async def batch_process(prompts: List[str]):
    tasks = [client.chat_completion(prompt) for prompt in prompts]
    return await asyncio.gather(*tasks)