共计 2385 个字符,预计需要花费 6 分钟才能阅读完成。
背景痛点
国内开发者想要使用 ChatGPT 面临两大难题:一是 OpenAI 的 API 服务在国内无法直接访问,二是内容合规性要求严格。这导致很多企业项目在接入 AI 能力时遇到阻碍。

技术方案对比
目前主流有三种接入方式:
- 官方 API 直连
- 优点:功能完整,更新及时
-
缺点:需要海外服务器,存在封号风险
-
开源模型部署
- 优点:完全自主可控
-
缺点:需要大量计算资源,效果略差
-
API 中转服务
- 优点:国内可用,可定制过滤
- 缺点:需要自建基础设施
核心实现
Flask 中转服务搭建
from flask import Flask, request, jsonify
import jwt
from functools import wraps
app = Flask(__name__)
SECRET_KEY = 'your-secret-key'
# JWT 鉴权装饰器
def token_required(f):
@wraps(f)
def decorated(*args, **kwargs):
token = request.headers.get('Authorization')
if not token:
return jsonify({'error': 'Token is missing'}), 403
try:
data = jwt.decode(token, SECRET_KEY, algorithms=['HS256'])
except:
return jsonify({'error': 'Token is invalid'}), 403
return f(*args, **kwargs)
return decorated
流量控制实现
使用 Redis 实现令牌桶算法:
import redis
from datetime import timedelta
r = redis.Redis(host='localhost', port=6379)
def check_rate_limit(user_id):
key = f'rate_limit:{user_id}'
current = r.get(key)
if current and int(current) > 100: # 每分钟 100 次限制
return False
r.incr(key)
r.expire(key, timedelta(minutes=1))
return True
敏感词过滤模块
基于 AC 自动机算法实现:
from ahocorasick import Automaton
# 初始化敏感词库
automaton = Automaton()
for idx, word in enumerate(sensitive_words):
automaton.add_word(word, (idx, word))
automaton.make_automaton()
# 检查文本
def check_sensitive(text):
for end_index, (insert_order, original_value) in automaton.iter(text):
return False
return True
完整代码示例
API 调用示例
import requests
from tenacity import retry, stop_after_attempt, wait_exponential
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
def call_chatgpt(prompt):
headers = {
'Authorization': 'Bearer your_token',
'Content-Type': 'application/json'
}
data = {
'model': 'gpt-3.5-turbo',
'messages': [{'role': 'user', 'content': prompt}]
}
try:
response = requests.post(
'https://your-proxy-domain.com/v1/chat/completions',
headers=headers,
json=data,
timeout=10
)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
print(f'Request failed: {e}')
raise
Docker 部署配置
FROM python:3.9-slim
WORKDIR /app
COPY . .
RUN pip install -r requirements.txt
EXPOSE 5000
CMD ["gunicorn", "-w 4", "-b :5000", "app:app"]
生产环境考量
性能测试
使用 Locust 进行压力测试,4 核 8G 服务器可支持:
- 500 RPS 短文本请求
- 平均延迟 < 300ms
- 99 分位延迟 < 800ms
数据脱敏方案
- 日志中的 API 密钥自动替换为 *
- 用户输入中的手机号 / 邮箱自动模糊处理
- 数据库存储前进行 AES 加密
避坑指南
常见 403 错误
- 检查 JWT token 是否过期
- 确认请求头 Content-Type 设置为 application/json
- 验证 IP 是否被 OpenAI 封禁
对话上下文管理
推荐实现方案:
- 使用 Redis 存储最近 5 轮对话
- 每次请求携带 session_id
- 设置 TTL 自动过期
扩展思考
结合 LangChain 构建知识库:
- 使用 FAISS 向量数据库存储企业文档
- 通过 embedding 实现语义搜索
- 将搜索结果作为 context 注入 prompt
学习路线
- 掌握 Flask/Django 框架
- 学习 Redis 等中间件
- 了解 NLP 基础知识
- 深入理解 RESTful API 设计
- 研究微服务架构
这套方案已经在多个项目中验证可行,既解决了访问问题,又确保了合规性。建议先在小规模场景测试,再逐步扩大应用范围。
正文完
