Merge branch 'main' of https://github.com/Moemu/Muice-Chatbot

Muice.py

@@ -1,25 +1,32 @@
-import logging,time,random,json,os
+import json
+import logging
+import os
+import random
+import time
 
 
-class Muice():
-    '''
+class Muice:
+    """
     Muice交互类
-    '''
+    """
 
-
-    def __init__(self, Model, read_memory_from_file:bool=True, known_topic_probability:float=0.003, time_topic_probability:float=0.75):
-        self.model = Model
+    def __init__(self, model, read_memory_from_file: bool = True, known_topic_probability: float = 0.003,
+                 time_topic_probability: float = 0.75):
+        self.reply = None
+        self.user_text = None
+        self.history = None
+        self.user_qq = None
+        self.model = model
         self.read_memory_from_file = read_memory_from_file
         self.known_topic_probability = known_topic_probability
         self.time_topic_probability = time_topic_probability
-        self.known_topic = ['（分享一下你的一些想法）','（创造一个新话题）']
-        self.time_topic = {'07':'（发起一个早晨问候）','12':'（发起一个中午问候）','18':'（发起一个傍晚问候）','00':'（发起一个临睡问候）'}
+        self.known_topic = ['（分享一下你的一些想法）', '（创造一个新话题）']
+        self.time_topic = {'07': '（发起一个早晨问候）', '12': '（发起一个中午问候）', '18': '（发起一个傍晚问候）',
+                           '00': '（发起一个临睡问候）'}
         self.time_topics = self.time_topic.copy()
 
-
-
-    def ask(self, text: str, user_qq:int) -> list:
-        '''发送信息'''
+    def ask(self, text: str, user_qq: int) -> list:
+        """发送信息"""
         self.user_qq = str(user_qq)
         if self.read_memory_from_file:
             self.history = self.get_recent_chat_memory()
@@ -33,69 +40,71 @@ def ask(self, text: str, user_qq:int) -> list:
         logging.info(f'模型调用时长: {end_time - start_time} s')
         return self.reply
 
-    def CreateANewTopic(self , LastTime):
-        '''
+    def create_a_new_topic(self, last_time):
+        """
         主动发起对话
-        '''
+        """
         current_time = time.strftime("%H:%M", time.localtime())
-        TimeDifference = time.time() - LastTime
-        if TimeDifference < 60 * 60:
+        time_difference = time.time() - last_time
+        if time_difference < 60 * 60:
             return None
         if random.random() < self.time_topic_probability:
-            for hour,topic in self.time_topic.items():
-                event_time = hour + ':' + str(random.randint(0,59))
+            for hour, topic in self.time_topic.items():
+                event_time = hour + ':' + str(random.randint(0, 59))
                 if event_time == current_time:
                     del self.time_topic[hour]
                     return topic
-        if not current_time.split(':')[0] in ['23','00','01','02','03','04','05','06'] and random.random() < self.known_topic_probability:
+        if not current_time.split(':')[0] in ['23', '00', '01', '02', '03', '04', '05',
+                                              '06'] and random.random() < self.known_topic_probability:
             return random.choice(self.known_topic)
         if len(self.time_topic) <= 3 and not time.strftime("%H", time.localtime()) in self.time_topics.keys():
             self.time_topic = self.time_topics.copy()
         return None
-
 
     def finish_ask(self, reply: list):
-        '''结束对话并保存记忆'''
+        """
+        结束对话并保存记忆
+        """
         reply = "".join(reply)
         self.save_chat_memory(reply)
 
     def get_recent_chat_memory(self):
-        '''
+        """
         获取最近一条记忆
-        '''
+        """
         try:
             with open(f'./memory/{self.user_qq}.json', 'r', encoding='utf-8') as f:
                 data = f.readlines()
                 return json.loads(data[-1])['history']
         except:
             return []
 
-    def save_chat_memory(self, reply:str):
-        '''
+    def save_chat_memory(self, reply: str):
+        """
         保存至记忆数据库
-        '''
+        """
         if not os.path.isdir('memory'):
             os.mkdir('memory')
         with open(f'./memory/{self.user_qq}.json', 'a', encoding='utf-8') as f:
             json.dump({'prompt': self.user_text, 'completion': reply, 'history': self.history}, f, ensure_ascii=False)
             f.write('\n')
 
     def remove_last_chat_memory(self):
-        '''
+        """
         删除最后一条记忆
-        '''
+        """
         with open(f'./memory/{self.user_qq}.json', 'r', encoding='utf-8') as f:
             data = f.readlines()
             del data[-1]
         with open(f'./memory/{self.user_qq}.json', 'w', encoding='utf-8') as f:
             f.writelines(data)
 
     def refresh(self):
-        '''
+        """
         刷新对话
-        '''
+        """
         logging.info("Start refresh")
         self.remove_last_chat_memory()
         self.history = self.get_recent_chat_memory()
         response = self.model.ask(self.user_text, self.history)
-        return response
+        return response

Tools.py

@@ -1,10 +1,10 @@
 import re
-from typing import List
+
 
 def divide_sentences(text: str) -> list:
-    '''
+    """
     切分一段句子
-    '''
+    """
     sentences = re.findall(r'.*?[~。！？…]+', text)
     if len(sentences) == 0:
         return [text]

api.py

@@ -1,10 +1,13 @@
-'''
+"""
 引用自：https://github.com/THUDM/ChatGLM2-6B/blob/main/api.py
-'''
+"""
+import datetime
+import json
+import os
+import torch
+import uvicorn
 from fastapi import FastAPI, Request
 from transformers import AutoTokenizer, AutoModel, AutoConfig
-import uvicorn, json, datetime, os
-import torch
 
 DEVICE = "cuda"
 DEVICE_ID = "0"
@@ -70,4 +73,4 @@ async def create_item(request: Request):
     # tokenizer = AutoTokenizer.from_pretrained(model_path, trust_remote_code=True)
     # model = load_model_on_gpus(model_path, num_gpus=2)
     model.eval()
-    uvicorn.run(app, host='0.0.0.0', port=8000, workers=1)
+    uvicorn.run(app, host='0.0.0.0', port=8000, workers=1)

command.py

@@ -1,39 +1,43 @@
-import os,shutil,sys
+import os
+import shutil
 
-class Command():
-    '''
+
+class Command:
+    """
     指令类
-    '''
-    def __init__(self,Muice):
-        self.Muice = Muice
+    """
+
+    def __init__(self, muice):
+        self.Muice = muice
         self.commands = {}
-        
-    def register_command(self,command:str,command_function):
-        '''
+
+    def register_command(self, command: str, command_function):
+        """
         注册命令
         Args:
             command: 命令字符串
             command_function: 所执行的函数
-        '''
+        """
         self.commands[command] = command_function
 
-    def run(self,command:str) -> str:
-        '''
+    def run(self, command: str) -> str:
+        """
         执行命令，返回命令命令执行结果或命令不存在的提示
-        '''
-        if command in self.commands:  
-        # 调用与命令关联的函数  
-            return self.commands[command]()  
+        """
+        if command in self.commands:
+            # 调用与命令关联的函数
+            return self.commands[command]()
         return self.no_command()
-    
+
     def no_command(self):
         return "没有当前命令"
-    #用于后期其他功能扩展
+
+    # 用于后期其他功能扩展
 
     def load_default_command(self):
-        '''
+        """
         加载默认命令
-        '''
+        """
         default_commands = {
             '/help': self.default_help,
             '/refresh': self.refresh,
@@ -44,30 +48,30 @@ def load_default_command(self):
         for command, function in default_commands.items():
             self.register_command(command, function)
 
-
     def default_help(self):
-        help_text = " /clean 清空本轮对话历史 \n /help 显示所有可用的命令列表 \n /refresh 刷新本次对话回复 \n /reset 重置所有对话数据(将存档对话数据) \n /undo 撤销上一次对话"
+        help_text = ("/clean 清空本轮对话历史 \n "
+                     "/help 显示所有可用的命令列表 \n "
+                     "/refresh 刷新本次对话回复 \n "
+                     "/reset 重置所有对话数据(将存档对话数据) \n "
+                     "/undo 撤销上一次对话")
         return help_text
-    
+
     def refresh(self):
         reply = self.Muice.refresh()
         self.Muice.save_chat_memory(reply)
         return reply
-    
+
     def clean(self):
         self.Muice.history = []
         return "cleaned"
-    
+
     def reset(self):
-        shutil.copy(f'./memory/{self.Muice.user_qq}.json','./memory/chat_memory_backup.json')
+        shutil.copy(f'./memory/{self.Muice.user_qq}.json', './memory/chat_memory_backup.json')
         os.remove(f'./memory/{self.Muice.user_qq}.json')
         self.Muice.history = []
         return "reseted"
-    
+
     def undo(self):
         self.Muice.remove_last_chat_memory()
         self.Muice.history = self.Muice.get_recent_chat_memory()
         return "undoed"
-
-
-

llm/api.py

@@ -1,14 +1,16 @@
 import requests as r
 import json
 
-class llm():
-    '''
+
+class llm:
+    """
     使用已经存在的API, 这个API应该接受prompt和history两个参数, 返回response
-    '''
-    def __init__(self,url:str,*args,**kwargs):
+    """
+
+    def __init__(self, url: str, *args, **kwargs):
         self.url = url
 
-    def ask(self,user_text:str, history:list,):
-        response = r.post(self.url,json={"prompt": user_text, "history": history})
+    def ask(self, user_text: str, history: list, ):
+        response = r.post(self.url, json={"prompt": user_text, "history": history})
         response = json.loads(response.text)
-        return response['response']
+        return response['response']

llm/llmtuner.py

@@ -1,22 +1,24 @@
 from llmtuner.chat import ChatModel
 
-class llm():
-    '''
+
+class llm:
+    """
     使用LLaMA-Factory方案加载, 适合通过其他微调方案微调的模型加载
-    '''
-    def __init__(self, model_name_or_path:str, adapter_name_or_path:str):
+    """
+
+    def __init__(self, model_name_or_path: str, adapter_name_or_path: str):
         self.model = ChatModel(dict(
-        model_name_or_path = model_name_or_path,
-        adapter_name_or_path = adapter_name_or_path,
-        template="qwen"
+            model_name_or_path=model_name_or_path,
+            adapter_name_or_path=adapter_name_or_path,
+            template="qwen"
         ))
-        
-    def ask(self,user_text:str, history:list,):
+
+    def ask(self, user_text: str, history: list, ):
         messages = []
-        if history != []:
+        if history:
             for chat in history:
-                messages.append({"role": "user", "content":chat[0]})
-                messages.append({"role": "assistant", "content":chat[1]})
+                messages.append({"role": "user", "content": chat[0]})
+                messages.append({"role": "assistant", "content": chat[1]})
         messages.append({"role": "user", "content": user_text})
         response = self.model.chat(messages)
-        return response[0].response_text
+        return response[0].response_text

llm/rwkv-api.py

@@ -1,22 +1,24 @@
 import requests as r
 import json
 
-class llm():
-    '''
+
+class llm:
+    """
     通过RWKV-RUNNER的api服务, 使用第三方RWKV模型
-    '''
-    def __init__(self,url:str,*args,**kwargs):
-        self.url = url+"/chat/completions"
+    """
+
+    def __init__(self, url: str, *args, **kwargs):
+        self.url = url + "/chat/completions"
 
-    def ask(self,user_text:str, history:list,):
+    def ask(self, user_text: str, history: list, ):
         messages = []
-        if history != []:
+        if history:
             for chat in history:
-                messages.append({"role": "user", "content":chat[0], "raw": False})
-                messages.append({"role": "assistant", "content":chat[1], "raw": False})
+                messages.append({"role": "user", "content": chat[0], "raw": False})
+                messages.append({"role": "assistant", "content": chat[1], "raw": False})
 
         messages.append({"role": "user", "content": user_text, "raw": False})
-        response = r.post(self.url,json={
+        response = r.post(self.url, json={
             "frequency_penalty": 1,
             "max_tokens": 1000,
             "messages": messages,
@@ -26,7 +28,7 @@ def ask(self,user_text:str, history:list,):
             "stream": False,
             "temperature": 1,
             "top_p": 0.3
-            })
-        
+        })
+
         response = json.loads(response.text)
-        return response['choices'][0]["message"]["content"].lstrip()
+        return response['choices'][0]["message"]["content"].lstrip()