django传统全栈开发一个ChatGPT应用

根据客户需求，开发一个能多人使用的ChatGPT平台，背后使用的是ChatGPT的api_key。

需求

1、可多轮对话

2、可删除对话

3、流式显示对话

4、可多人使用

5、多个api_key均衡使用

在这里插入图片描述

技术分析

第一次接触openai的二次开发，看文档、看文章，技术点如下：

1、不同等级的api_key使用不同的model即模型，普通账号能使用text-davinci-003和gpt-3.5-turbo模型，都是ChatGPT 3.5的；

2、api_key有限流，普通账号限流挺严的，每分钟3次请求或每分钟40000的tokens，意味着需要搭建一个api_key池，维护多个账号，自己写算法动态调节避免被限流。不然少数的几个账号分分钟就能触碰每分钟3次请求的限制；

3、openai是官方提供的sdk，有同步接口，也有异步接口，由于时间短任务中，异步就不考虑了，直接上同步；

4、前端没写过vue，虽然有点跃跃欲试，最后还是选择了熟悉的layui，前端结构化的就不谈了，把功能写出来就完事了；

5、关于api_key，其实还有点，即key的状态，sdk里也没找到什么可用的接口来获取key的剩余额度、有效期等信息，暂时先放一放，让客户自行充值就好了，后面有办法了再解决。

api_key维护

简单来说写了三个类，算法也很简单，使用的数据结构如下：

[# API实现在下方{'key': <API object xxxxxx>, 'counter': 0}, {'key': <API object xxxxxx>, 'counter': 0},...
]

类实现分别为：

1、Singleton 单例的抽象基类

2、API主题类

3、ApiPool代理类

主要由ApiPool对外提供服务，继承抽象基类实现单例，确保全局数据的唯一性。

抽象基类

class Singleton(type):_instance = Nonedef __call__(cls, *args, **kwargs):if cls._instance is None:cls._instance = super().__call__(*args, **kwargs)return cls._instance

本想上redis维护api_key池的，又得多写代码，考虑也就十几号人同时用，要啥自行车，直接写单例模式来维护，上面的抽象基类就是为这个事服务的。

API主题类


class API:# 使用时间间隔为20秒 避免触发限流rqtl = 20 def __init__(self, key):self.key = keyself.__time = time.time() # 初始化时记录时间戳@propertydef last_time(self):return self.__time@last_time.setterdef last_time(self, value: float):self.__time = valuedef __repr__(self):return f'<{self.key} - {self.last_time}>'@propertydef can_use(self):return self.__bool__()def __bool__(self):"""调用时时间差大于20秒可用 反之不可用"""return bool((time.time() - self.last_time) >= API.rqtl)def __call__(self):return self.key

该类主要实现的是api_key是否可用，所有的api_key都保存在数据库，系统启动或重启时，从数据库加载所有的api_key，逐个使用API初始化，并保存时间戳，对外暴露can_use，当调用这个方法时，会使用当前时间戳和记录的时间戳做比，大于等于20秒就使用，在使用时就更新时间戳，所以也暴露了last_time.setter。

ApiPool代理类

class ApiPool(metaclass=Singleton):"""1、从数据库里取出api2、每个api都是API类的实例 每个实例会记录上次使用的时间3、取api使用时 先判断是否can_use 能就取 反之取使用次数最少的"""def __init__(self, query):# django启动或重启时从数据库中加载api_keyself.__lst = self.init(query)def init(self, query):lst = []for api in query:lst.append({'key': API(api.api_key), 'counter': 0})return lst@propertydef lst(self):return self.__lst# 取一个可用的api_keydef get(self):_api = Nonefor api in self.__lst:if api.get('key').can_use:_api = api['key']# 使用一次就+1api['counter'] += 1# 更新时间戳api['key'].last_time = time.time()break# 如果所有的key的时间间隔都未超过20秒# 则使用第一个 因为它的使用次数最少if not _api:api = self.__lst[0]_api = api['key']# 使用一次就+1api['counter'] += 1# 更新时间戳api['key'].last_time = time.time()# 提取后重新排序 counter 升序self.__lst.sort(key=lambda api: api['counter'])return _api# django后台增加api_key或设置为可用时调用def add(self, key):s = False# 存在时不操作for api in self.__lst:_key = api.get('key').keyif key == _key:return s# 不存在时才增加    if isinstance(key, str):self.__lst.append({'key': API(key), 'counter': 0})s = Truereturn s# django后台删除api_key设置为不可用时调用def remove(self, key: str):k = Nonefor api in self.__lst:if api.get('key').key == key:k = apibreakif k:self.__lst.remove(k)return Truereturn Falsedef __repr__(self):return f'<ApiPool {len(self.__lst)}>'# 应对某些情况时使用@propertydef available(self):lst = []for api in self.__lst:if api.get('key').can_use:lst.append(api)return lst

ApiPool对外提供服务，在django启动时就得实例化，在settings.py中初始化不可行，因为那时django的app都未完成初始化，所以最后在某个views.py中实例化，前端请求达到views.py调用openai接口前，先调用get方法拿到一个api_key。演示如下：

# 实例化ApiPool
from . apikey import ApiPool
api_pool = ApiPool(ApiKey.objects.filter(status=True))@login_required
@require_POST
def conversation(request):"""省略其他代码"""key = api_pool.get()if key is None:return JsonResponse({'code': 400, 'msg': '暂无可用的key'})ret = sync_stream_ChatCompletion(messages, uuid, q, key())return StreamingHttpResponse(ret, content_type='application/octet-stream')

前端技术点

前端没使用古老的XMLHttpRequest也没使用jquery.ajax，使用了浏览器原生的fetch(fetch不好的地方就是要两次then才能拿到数据)和后端交互，因为它用来接收steam数据流相对方便些，大概的结构如下：


fetch(url, {options})
.then(response=>{// 判断下响应是否为'application/octet-stream'// 因为后端也写了json的响应再无api_key可用的情况下// 1、'application/octet-stream'时，直接闭包处理let reader = response.body.getReader();function read(){return reader.read().then(//拿到流式数据写到页面)// 因为是流式，所以需要递归调用};return read()// 2、'application/json'时let ret = response.json()function bad(){return ret.then(//友好提示无key可用)};return bad;
})