Merge pull request #325 from RockChinQ/fix-289-full-default-compatibility

[Feat] 完善情景预设相关内容

config-template.py

@@ -80,12 +80,11 @@ default_prompt = {
 }
 
 # 情景预设格式
-# 参考值：旧版本方式：default | 完整情景：full_scenario
+# 参考值：默认方式：normal | 完整情景：full_scenario
-# 旧版本的格式为上述default_prompt中的内容，或prompts目录下的文件名
+# 默认方式 的格式为上述default_prompt中的内容，或prompts目录下的文件名
-#
+# 完整情景方式 的格式为JSON，在scenario目录下的JSON文件中列出对话的每个回合，编写方法见scenario/default-template.json
-# 完整情景预设的格式为JSON，在scenario目录下的JSON文件中列出对话的每个回合，编写方法见scenario/default-template.json
+#     编写方法请查看：https://github.com/RockChinQ/QChatGPT/wiki/%E5%8A%9F%E8%83%BD%E4%BD%BF%E7%94%A8#%E9%A2%84%E8%AE%BE%E6%96%87%E5%AD%97full_scenario%E6%A8%A1%E5%BC%8F
-# 编写方法请查看：https://github.com/RockChinQ/QChatGPT/wiki/%E5%8A%9F%E8%83%BD%E4%BD%BF%E7%94%A8#%E9%A2%84%E8%AE%BE%E6%96%87%E5%AD%97full_scenario%E6%A8%A1%E5%BC%8F
+preset_mode = "normal"
-preset_mode = "default"
 
 # 群内响应规则
 # 符合此消息的群内消息即使不包含at机器人也会响应

main.py

@@ -192,8 +192,7 @@ def start(first_time_init=False):
         import pkg.qqbot.manager
         import pkg.openai.dprompt
         
-        pkg.openai.dprompt.read_prompt_from_file()
+        pkg.openai.dprompt.register_all()
-        pkg.openai.dprompt.read_scenario_from_file()
 
         # 主启动流程
         database = pkg.database.manager.DatabaseManager()

override-all.json

@@ -17,7 +17,7 @@
     "default_prompt": {
         "default": "如果我之后想获取帮助，请你说“输入!help获取帮助”"
     },
-    "preset_mode": "default",
+    "preset_mode": "normal",
     "response_rules": {
         "at": true,
         "prefix": [
@@ -42,7 +42,7 @@
     "baidu_secret_key": "",
     "inappropriate_message_tips": "[百度云]请珍惜机器人，当前返回内容不合规",
     "encourage_sponsor_at_start": true,
-    "prompt_submit_length": 1024,
+    "prompt_submit_length": 2048,
     "completion_api_params": {
         "model": "gpt-3.5-turbo",
         "temperature": 0.9,
@@ -63,7 +63,9 @@
     "retry_times": 3,
     "hide_exce_info_to_user": false,
     "alter_tip_message": "出错了，请稍后再试",
-    "pool_num": 10,
+    "sys_pool_num": 8,
+    "admin_pool_num": 2,
+    "user_pool_num": 6,
     "session_expire_time": 1200,
     "rate_limitation": 60,
     "rate_limit_strategy": "wait",

pkg/openai/dprompt.py

@@ -1,121 +1,145 @@
 # 多情景预设值管理
 import json
 import logging
+import config
+import os
 
-__current__ = "default"
+# __current__ = "default"
-"""当前默认使用的情景预设的名称
+# """当前默认使用的情景预设的名称
 
-由管理员使用`!default <名称>`指令切换
+# 由管理员使用`!default <名称>`指令切换
-"""
+# """
 
-__prompts_from_files__ = {}
+# __prompts_from_files__ = {}
-"""从文件中读取的情景预设值"""
+# """从文件中读取的情景预设值"""
 
-__scenario_from_files__ = {}
+# __scenario_from_files__ = {}
 
 
-def read_prompt_from_file():
+__universal_first_reply__ = "ok, I'll follow your commands."
-    """从文件读取预设值"""
+"""通用首次回复"""
-    # 读取prompts/目录下的所有文件，以文件名为键，文件内容为值
-    # 保存在__prompts_from_files__中
-    global __prompts_from_files__
-    import os
-
-    __prompts_from_files__ = {}
-    for file in os.listdir("prompts"):
-        with open(os.path.join("prompts", file), encoding="utf-8") as f:
-            __prompts_from_files__[file] = f.read()
 
 
-def read_scenario_from_file():
+class ScenarioMode:
-    """从JSON文件读取情景预设"""
+    """情景预设模式抽象类"""
-    global __scenario_from_files__
-    import os
 
-    __scenario_from_files__ = {}
+    using_prompt_name = "default"
-    for file in os.listdir("scenario"):
+    """新session创建时使用的prompt名称"""
-        if file == "default-template.json":
+
-            continue
+    prompts: dict[str, list] = {}
-        with open(os.path.join("scenario", file), encoding="utf-8") as f:
+
-            __scenario_from_files__[file] = json.load(f)
+    def __init__(self):
+        logging.debug("prompts: {}".format(self.prompts))
+
+    def list(self) -> dict[str, list]:
+        """获取所有情景预设的名称及内容"""
+        return self.prompts
+
+    def get_prompt(self, name: str) -> tuple[list, str]:
+        """获取指定情景预设的名称及内容"""
+        for key in self.prompts:
+            if key.startswith(name):
+                return self.prompts[key], key
+        raise Exception("没有找到情景预设: {}".format(name))
+
+    def set_using_name(self, name: str) -> str:
+        """设置默认情景预设"""
+        for key in self.prompts:
+            if key.startswith(name):
+                self.using_prompt_name = key
+                return key
+        raise Exception("没有找到情景预设: {}".format(name))
+    
+    def get_full_name(self, name: str) -> str:
+        """获取完整的情景预设名称"""
+        for key in self.prompts:
+            if key.startswith(name):
+                return key
+        raise Exception("没有找到情景预设: {}".format(name))
+
+    def get_using_name(self) -> str:
+        """获取默认情景预设"""
+        return self.using_prompt_name
 
 
-def get_prompt_dict() -> dict:
+class NormalScenarioMode(ScenarioMode):
-    """获取预设值字典"""
+    """普通情景预设模式"""
-    import config
-    default_prompt = config.default_prompt
-    if type(default_prompt) == str:
-        default_prompt = {"default": default_prompt}
-    elif type(default_prompt) == dict:
-        pass
-    else:
-        raise TypeError("default_prompt must be str or dict")
 
-    # 将文件中的预设值合并到default_prompt中
+    def __init__(self):
-    for key in __prompts_from_files__:
+        global __universal_first_reply__
-        default_prompt[key] = __prompts_from_files__[key]
+        # 加载config中的default_prompt值
+        if type(config.default_prompt) == str:
+            self.using_prompt_name = "default"
+            self.prompts = {"default": [
+                {
+                    "role": "user",
+                    "content": config.default_prompt
+                },{
+                    "role": "assistant",
+                    "content": __universal_first_reply__
+                }
+            ]}
         
-    return default_prompt
+        elif type(config.default_prompt) == dict:
-
+            for key in config.default_prompt:
-
+                self.prompts[key] = [
-def set_current(name):
-    global __current__
-    for key in get_prompt_dict():
-        if key.lower().startswith(name.lower()):
-            __current__ = key
-            return
-    raise KeyError("未找到情景预设: " + name)
-
-
-def get_current():
-    global __current__
-    return __current__
-
-
-def set_to_default():
-    global __current__
-    default_dict = get_prompt_dict()
-
-    if "default" in default_dict:
-        __current__ = "default"
-    else:
-        __current__ = list(default_dict.keys())[0]
-
-
-def get_prompt(name: str = None) -> list:
-    global __scenario_from_files__
-    import config
-    preset_mode = config.preset_mode
-
-    """获取预设值"""
-    if name is None:
-        name = get_current()
-
-    # JSON预设方式
-    if preset_mode == 'full_scenario':
-        import os
-
-        for key in __scenario_from_files__:
-            if key.lower().startswith(name.lower()):
-                logging.debug('成功加载情景预设从JSON文件: {}'.format(key))
-                return __scenario_from_files__[key]['prompt']
-        
-    # 默认预设方式
-    elif preset_mode == 'default':
-
-        default_dict = get_prompt_dict()
-
-        for key in default_dict:
-            if key.lower().startswith(name.lower()):
-                return [
                     {
                         "role": "user",
-                        "content": default_dict[key]
+                        "content": config.default_prompt[key]
-                    },
+                    },{
-                    {
                         "role": "assistant",
-                        "content": "好的。"
+                        "content": __universal_first_reply__
                     }
                 ]
 
-    raise KeyError("未找到默认情景预设: " + name)
+        # 从prompts/目录下的文件中载入
+        # 遍历文件
+        for file in os.listdir("prompts"):
+            with open(os.path.join("prompts", file), encoding="utf-8") as f:
+                self.prompts[file] = [
+                    {
+                        "role": "user",
+                        "content": f.read()
+                    },{
+                        "role": "assistant",
+                        "content": __universal_first_reply__
+                    }
+                ]
+
+
+class FullScenarioMode(ScenarioMode):
+    """完整情景预设模式"""
+
+    def __init__(self):
+        """从json读取所有"""
+        # 遍历scenario/目录下的所有文件，以文件名为键，文件内容中的prompt为值
+        for file in os.listdir("scenario"):
+            if file == "default-template.json":
+                continue
+            with open(os.path.join("scenario", file), encoding="utf-8") as f:
+                self.prompts[file] = json.load(f)["prompt"]
+
+        super().__init__()
+
+
+scenario_mode_mapping = {}
+"""情景预设模式名称与对象的映射"""
+
+
+def register_all():
+    """注册所有情景预设模式，不使用装饰器，因为装饰器的方式不支持热重载"""
+    global scenario_mode_mapping
+    scenario_mode_mapping = {
+        "normal": NormalScenarioMode(),
+        "full_scenario": FullScenarioMode()
+    }
+
+
+def mode_inst() -> ScenarioMode:
+    """获取指定名称的情景预设模式对象"""
+    import config
+
+    if config.preset_mode == "default":
+        config.preset_mode = "normal"
+
+    return scenario_mode_mapping[config.preset_mode]

pkg/openai/session.py

@@ -141,9 +141,9 @@ class Session:
         import pkg.openai.dprompt as dprompt
 
         if use_default is None:
-            use_default = dprompt.get_current()
+            use_default = dprompt.mode_inst().get_using_name()
 
-        current_default_prompt = dprompt.get_prompt(use_default)
+        current_default_prompt, _ = dprompt.mode_inst().get_prompt(use_default)
         return current_default_prompt
 
     def __init__(self, name: str):

pkg/qqbot/cmds/session.py

@@ -7,6 +7,7 @@ import pkg.openai.session
 import pkg.utils.context
 import config
 
+
 @command(
     "reset",
     "重置当前会话",
@@ -16,7 +17,7 @@ import config
 )
 def cmd_reset(cmd: str, params: list, session_name: str,
               text_message: str, launcher_type: str, launcher_id: int,
-                sender_id: int, is_admin: bool) -> list:
+              sender_id: int, is_admin: bool) -> list:
     """重置会话"""
     reply = []
 
@@ -24,8 +25,12 @@ def cmd_reset(cmd: str, params: list, session_name: str,
         pkg.openai.session.get_session(session_name).reset(explicit=True)
         reply = ["[bot]会话已重置"]
     else:
-        pkg.openai.session.get_session(session_name).reset(explicit=True, use_prompt=params[0])
+        try:
-        reply = ["[bot]会话已重置，使用场景预设:{}".format(params[0])]
+            import pkg.openai.dprompt as dprompt
+            pkg.openai.session.get_session(session_name).reset(explicit=True, use_prompt=params[0])
+            reply = ["[bot]会话已重置，使用场景预设:{}".format(dprompt.mode_inst().get_full_name(params[0]))]
+        except Exception as e:
+            reply = ["[bot]会话重置失败，错误信息：{}".format(e)]
 
     return reply
 
@@ -39,7 +44,7 @@ def cmd_reset(cmd: str, params: list, session_name: str,
 )
 def cmd_last(cmd: str, params: list, session_name: str,
              text_message: str, launcher_type: str, launcher_id: int,
-                sender_id: int, is_admin: bool) -> list:
+             sender_id: int, is_admin: bool) -> list:
     """切换到前一次会话"""
     reply = []
     result = pkg.openai.session.get_session(session_name).last_session()
@@ -52,6 +57,7 @@ def cmd_last(cmd: str, params: list, session_name: str,
 
     return reply
 
+
 @command(
     "next",
     "切换到后一次会话",
@@ -61,7 +67,7 @@ def cmd_last(cmd: str, params: list, session_name: str,
 )
 def cmd_next(cmd: str, params: list, session_name: str,
              text_message: str, launcher_type: int, launcher_id: int,
-                sender_id: int, is_admin: bool) -> list:
+             sender_id: int, is_admin: bool) -> list:
     """切换到后一次会话"""
     reply = []
 
@@ -84,13 +90,13 @@ def cmd_next(cmd: str, params: list, session_name: str,
     False
 )
 def cmd_prompt(cmd: str, params: list, session_name: str,
-                text_message: str, launcher_type: str, launcher_id: int,
+               text_message: str, launcher_type: str, launcher_id: int,
-                 sender_id: int, is_admin: bool) -> list:
+               sender_id: int, is_admin: bool) -> list:
     """获取当前会话的前文"""
     reply = []
 
     msgs = ""
-    session:list = pkg.openai.session.get_session(session_name).prompt
+    session: list = pkg.openai.session.get_session(session_name).prompt
     for msg in session:
         if len(params) != 0 and params[0] in ['-all', '-a']:
             msgs = msgs + "{}: {}\n\n".format(msg['role'], msg['content'])
@@ -111,8 +117,8 @@ def cmd_prompt(cmd: str, params: list, session_name: str,
     False
 )
 def cmd_list(cmd: str, params: list, session_name: str,
-              text_message: str, launcher_type: str, launcher_id: int,
+             text_message: str, launcher_type: str, launcher_id: int,
-               sender_id: int, is_admin: bool) -> list:
+             sender_id: int, is_admin: bool) -> list:
     """列出当前会话的所有历史记录"""
     reply = []
 
@@ -143,12 +149,12 @@ def cmd_list(cmd: str, params: list, session_name: str,
                 pkg.openai.session.get_session(session_name).persistence()
             if len(msg) >= 2:
                 reply_str += "#{} 创建:{} {}\n".format(i + page * 10,
-                                                        datetime_obj.strftime("%Y-%m-%d %H:%M:%S"),
+                                                       datetime_obj.strftime("%Y-%m-%d %H:%M:%S"),
-                                                        msg[0]['content'])
+                                                       msg[0]['content'])
             else:
                 reply_str += "#{} 创建:{} {}\n".format(i + page * 10,
-                                                        datetime_obj.strftime("%Y-%m-%d %H:%M:%S"),
+                                                       datetime_obj.strftime("%Y-%m-%d %H:%M:%S"),
-                                                        "无内容")
+                                                       "无内容")
             if results[i]['create_timestamp'] == pkg.openai.session.get_session(
                     session_name).create_timestamp:
                 current = i + page * 10
@@ -172,8 +178,8 @@ def cmd_list(cmd: str, params: list, session_name: str,
     False
 )
 def cmd_resend(cmd: str, params: list, session_name: str,
-                text_message: str, launcher_type: str, launcher_id: int,
+               text_message: str, launcher_type: str, launcher_id: int,
-                 sender_id: int, is_admin: bool) -> list:
+               sender_id: int, is_admin: bool) -> list:
     """重新获取上一次问题的回复"""
     reply = []
 
@@ -183,7 +189,7 @@ def cmd_resend(cmd: str, params: list, session_name: str,
     mgr = pkg.utils.context.get_qqbot_manager()
 
     reply = pkg.qqbot.message.process_normal_message(to_send, mgr, config,
-                                                        launcher_type, launcher_id, sender_id)
+                                                     launcher_type, launcher_id, sender_id)
 
     return reply
 
@@ -196,8 +202,8 @@ def cmd_resend(cmd: str, params: list, session_name: str,
     False
 )
 def cmd_del(cmd: str, params: list, session_name: str,
-             text_message: str, launcher_type: str, launcher_id: int,
+            text_message: str, launcher_type: str, launcher_id: int,
-              sender_id: int, is_admin: bool) -> list:
+            sender_id: int, is_admin: bool) -> list:
     """删除当前会话的历史记录"""
     reply = []
 
@@ -226,28 +232,37 @@ def cmd_del(cmd: str, params: list, session_name: str,
 )
 def cmd_default(cmd: str, params: list, session_name: str,
                 text_message: str, launcher_type: str, launcher_id: int,
-                 sender_id: int, is_admin: bool) -> list:
+                sender_id: int, is_admin: bool) -> list:
     """操作情景预设"""
     reply = []
 
     if len(params) == 0:
         # 输出目前所有情景预设
         import pkg.openai.dprompt as dprompt
-        reply_str = "[bot]当前所有情景预设:\n\n"
+        reply_str = "[bot]当前所有情景预设({}模式):\n\n".format(config.preset_mode)
-        for key,value in dprompt.get_prompt_dict().items():
-            reply_str += "  - {}: {}\n".format(key,value)
 
-        reply_str += "\n当前默认情景预设:{}\n".format(dprompt.get_current())
+        prompts = dprompt.mode_inst().list()
-        reply_str += "请使用!default <情景预设>来设置默认情景预设"
+
+        for key in prompts:
+            pro = prompts[key]
+            reply_str += "名称: {}".format(key)
+
+            for r in pro:
+                reply_str += "\n   - [{}]: {}".format(r['role'], r['content'])
+
+            reply_str += "\n\n"
+
+        reply_str += "\n当前默认情景预设:{}\n".format(dprompt.mode_inst().get_using_name())
+        reply_str += "请使用 !default <情景预设名称> 来设置默认情景预设"
         reply = [reply_str]
-    elif len(params) >0  and is_admin:
+    elif len(params) > 0 and is_admin:
         # 设置默认情景
         import pkg.openai.dprompt as dprompt
         try:
-            dprompt.set_current(params[0])
+            full_name = dprompt.mode_inst().set_using_name(params[0])
-            reply = ["[bot]已设置默认情景预设为:{}".format(dprompt.get_current())]
+            reply = ["[bot]已设置默认情景预设为:{}".format(full_name)]
-        except KeyError:
+        except Exception as e:
-            reply = ["[bot]err: 未找到情景预设:{}".format(params[0])]
+            reply = ["[bot]err: {}".format(e)]
     else:
         reply = ["[bot]err: 仅管理员可设置默认情景预设"]
 
@@ -262,13 +277,14 @@ def cmd_default(cmd: str, params: list, session_name: str,
     True
 )
 def cmd_delhst(cmd: str, params: list, session_name: str,
-                text_message: str, launcher_type: str, launcher_id: int,
+               text_message: str, launcher_type: str, launcher_id: int,
-                 sender_id: int, is_admin: bool) -> list:
+               sender_id: int, is_admin: bool) -> list:
     """删除指定会话的所有历史记录"""
     reply = []
 
     if len(params) == 0:
-        reply = ["[bot]err:请输入要删除的会话名: group_<群号> 或者 person_<QQ号>, 或使用 !delhst all 删除所有会话的历史记录"]
+        reply = [
+            "[bot]err:请输入要删除的会话名: group_<群号> 或者 person_<QQ号>, 或使用 !delhst all 删除所有会话的历史记录"]
     else:
         if params[0] == "all":
             pkg.utils.context.get_database_manager().delete_all_session_history()