feat: 支持多场景预设切换

This commit is contained in:
Rock Chin
2023-02-19 11:46:12 +08:00
parent c35fd9c4b7
commit d0bcf6940a
7 changed files with 122 additions and 11 deletions

View File

@@ -43,10 +43,34 @@ admin_qq = 0
# 情景预设(机器人人格)
# 每个会话的预设信息,影响所有会话,无视指令重置
# 可以通过这个字段指定某些情况的回复,可直接用自然语言描述指令
# 例如: 如果我之后想获取帮助,请你说“输入!help获取帮助”
# 例如:
# default_prompt = "如果我之后想获取帮助,请你说“输入!help获取帮助”"
# 这样用户在不知所措的时候机器人就会提示其输入!help获取帮助
# 可参考 https://github.com/PlexPt/awesome-chatgpt-prompts-zh
default_prompt = "如果我之后想获取帮助,请你说“输入!help获取帮助”"
#
# 如果需要多个情景预设,并在运行期间方便切换,请使用字典的形式填写,例如
# default_prompt = {
# "default": "如果我之后想获取帮助,请你说“输入!help获取帮助”",
# "linux-terminal": "我想让你充当 Linux 终端。我将输入命令,您将回复终端应显示的内容。",
# "en-dict": "我想让你充当英英词典,对于给出的英文单词,你要给出其中文意思以及英文解释,并且给出一个例句,此外不要有其他反馈。",
# }
#
# 在使用期间即可通过指令:
# !reset [名称]
# 来使用指定的情景预设重置会话
# 例如:
# !reset linux-terminal
# 若不指定名称,则使用默认情景预设
#
# 也可以使用指令:
# !default <名称>
# 将指定的情景预设设置为默认情景预设
# 例如:
# !default linux-terminal
# 之后的会话重置时若不指定名称则使用linux-terminal情景预设
default_prompt = {
"default": "如果我之后想获取帮助,请你说“输入!help获取帮助”",
}
# 群内响应规则
# 符合此消息的群内消息即使不包含at机器人也会响应

51
pkg/openai/dprompt.py Normal file
View File

@@ -0,0 +1,51 @@
# 多情景预设值管理
__current__ = "default"
def get_prompt_dict() -> dict:
"""获取预设值字典"""
import config
default_prompt = config.default_prompt
if type(default_prompt) == str:
return {"default": default_prompt}
elif type(default_prompt) == dict:
return default_prompt
else:
raise TypeError("default_prompt must be str or dict")
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) -> str:
"""获取预设值"""
if name is None:
name = get_current()
default_dict = get_prompt_dict()
for key in default_dict:
if key.lower().startswith(name.lower()):
return default_dict[key]
raise KeyError("未找到情景预设: " + name)

View File

@@ -89,13 +89,21 @@ class Session:
logging.debug('{},lock release successfully,{}'.format(self.name, self.response_lock))
# 从配置文件获取会话预设信息
def get_default_prompt(self):
def get_default_prompt(self, use_default: str=None):
config = pkg.utils.context.get_config()
import pkg.openai.dprompt as dprompt
if use_default is None:
current_default_prompt = dprompt.get_prompt(dprompt.get_current())
else:
current_default_prompt = dprompt.get_prompt(use_default)
user_name = config.user_name if hasattr(config, 'user_name') and config.user_name != '' else 'You'
bot_name = config.bot_name if hasattr(config, 'bot_name') and config.bot_name != '' else 'Bot'
return (user_name + ":{}\n".format(config.default_prompt) + bot_name + ":好的\n") \
if hasattr(config, 'default_prompt') and config.default_prompt != '' else ''
return (user_name + ":{}\n".format(current_default_prompt) + bot_name + ":好的\n") \
if current_default_prompt != '' else ''
def __init__(self, name: str):
self.name = name
@@ -242,8 +250,8 @@ class Session:
self.prompt)
# 重置session
def reset(self, explicit: bool = False, expired: bool = False, schedule_new: bool = True):
if self.prompt != self.get_default_prompt():
def reset(self, explicit: bool = False, expired: bool = False, schedule_new: bool = True, use_prompt: str = None):
if not self.prompt.endswith(':好的\n'):
self.persistence()
if explicit:
# 触发插件事件
@@ -259,7 +267,7 @@ class Session:
if expired:
pkg.utils.context.get_database_manager().set_session_expired(self.name, self.create_timestamp)
self.prompt = self.get_default_prompt()
self.prompt = self.get_default_prompt(use_prompt)
self.create_timestamp = int(time.time())
self.last_interact_timestamp = int(time.time())
self.just_switched_to_exist_session = False

View File

@@ -189,6 +189,7 @@ def register(name: str, description: str, version: str, author: str):
__current_registering_plugin__ = name
# print("registering plugin: n='{}', d='{}', v={}, a='{}'".format(name, description, version, author))
host.__plugins__[name] = {
"name": name,
"description": description,

View File

@@ -23,6 +23,8 @@ def wrapper_dict_from_plugin_list() -> dict:
def apply_switch(switch: dict):
"""将开关数据应用到插件列表中"""
# print("将开关数据应用到插件列表中")
# print(switch)
for plugin_name in switch:
host.__plugins__[plugin_name]["enabled"] = switch[plugin_name]["enabled"]

View File

@@ -171,8 +171,12 @@ def process_command(session_name: str, text_message: str, mgr, config,
if cmd == 'help':
reply = ["[bot]" + config.help_message]
elif cmd == 'reset':
pkg.openai.session.get_session(session_name).reset(explicit=True)
reply = ["[bot]会话已重置"]
if len(params) == 0:
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])
reply = ["[bot]会话已重置,使用场景预设:{}".format(params[0])]
elif cmd == 'last':
result = pkg.openai.session.get_session(session_name).last_session()
if result is None:
@@ -279,6 +283,27 @@ def process_command(session_name: str, text_message: str, mgr, config,
reply = plugin_operation(cmd, params, True
if (launcher_type == 'person' and launcher_id == config.admin_qq)
else False)
elif cmd == 'default':
if len(params) == 0:
# 输出目前所有情景预设
import pkg.openai.dprompt as dprompt
reply_str = "[bot]当前所有情景预设:\n\n"
for key,value in dprompt.get_prompt_dict().items():
reply_str += " - {}: {}\n".format(key,value)
reply_str += "\n当前默认情景预设:{}\n".format(dprompt.get_current())
reply_str += "请使用!default <情景预设>来设置默认情景预设"
reply = [reply_str]
elif len(params) >0 and launcher_type == 'person' and launcher_id == config.admin_qq:
# 设置默认情景
import pkg.openai.dprompt as dprompt
try:
dprompt.set_current(params[0])
reply = ["[bot]已设置默认情景预设为:{}".format(dprompt.get_current())]
except KeyError:
reply = ["[bot]err: 未找到情景预设:{}".format(params[0])]
else:
reply = ["[bot]err: 仅管理员可设置默认情景预设"]
elif cmd == 'reload' and launcher_type == 'person' and launcher_id == config.admin_qq:
def reload_task():
pkg.utils.reloader.reload_all()

View File

@@ -27,7 +27,7 @@ def go(func, args=()):
# 检查消息是否符合泛响应匹配机制
def check_response_rule(text: str) -> (bool, str):
def check_response_rule(text: str) -> tuple[bool, str]:
config = pkg.utils.context.get_config()
if not hasattr(config, 'response_rules'):
return False, ''