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

这篇具有很好参考价值的文章主要介绍了django传统全栈开发一个ChatGPT应用。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

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

需求

1、可多轮对话

2、可删除对话

3、流式显示对话

4、可多人使用

5、多个api_key均衡使用

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

技术分析

第一次接触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 = None

    def __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 = key
        self.__time = time.time() # 初始化时记录时间戳

    @property
    def last_time(self):
        return self.__time

    @last_time.setter
    def last_time(self, value: float):
        self.__time = value
    
    def __repr__(self):
        return f'<{self.key} - {self.last_time}>'
    
    @property
    def 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、从数据库里取出api
    2、每个api都是API类的实例 每个实例会记录上次使用的时间
    3、取api使用时 先判断是否can_use 能就取 反之取使用次数最少的
    """
    
    def __init__(self, query):
        # django启动或重启时从数据库中加载api_key
        self.__lst = self.init(query)

    def init(self, query):
        lst = []
        for api in query:
            lst.append(
                {'key': API(api.api_key), 'counter': 0}
            )
        return lst

    @property
    def lst(self):
        return self.__lst
    
    # 取一个可用的api_key
    def get(self):
        _api = None
        for api in self.__lst:
            if api.get('key').can_use:
                _api = api['key']
                # 使用一次就+1
                api['counter'] += 1
                # 更新时间戳
                api['key'].last_time = time.time()
                break
        
        # 如果所有的key的时间间隔都未超过20秒
        # 则使用第一个 因为它的使用次数最少
        if not _api:
            api = self.__lst[0]
            _api = api['key']
            # 使用一次就+1
            api['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').key
            if key == _key:
                return s
        # 不存在时才增加    
        if isinstance(key, str):
            self.__lst.append({'key': API(key), 'counter': 0})
            s = True

        return s
    
    # django后台删除api_key设置为不可用时调用
    def remove(self, key: str):
        k = None
        for api in self.__lst:
            if api.get('key').key == key:
                k = api
                break
        if k:
            self.__lst.remove(k)
            return True
        return False

    def __repr__(self):
        return f'<ApiPool {len(self.__lst)}>'
    
    # 应对某些情况时使用
    @property
    def 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;
})