up

crazy_functions/vt_fns/vt_call_plugin.py

@@ -0,0 +1,114 @@
+from pydantic import BaseModel, Field
+from typing import List
+from toolbox import update_ui_lastest_msg, disable_auto_promotion
+from request_llm.bridge_all import predict_no_ui_long_connection
+from crazy_functions.json_fns.pydantic_io import GptJsonIO, JsonStringError
+import copy, json, pickle, os, sys, time
+
+
+def read_avail_plugin_enum():
+    from crazy_functional import get_crazy_functions
+    plugin_arr = get_crazy_functions()
+    # remove plugins with out explaination
+    plugin_arr = {k:v for k, v in plugin_arr.items() if 'Info' in v}
+    plugin_arr_info = {"F_{:04d}".format(i):v["Info"] for i, v in enumerate(plugin_arr.values(), start=1)}
+    plugin_arr_dict = {"F_{:04d}".format(i):v for i, v in enumerate(plugin_arr.values(), start=1)}
+    plugin_arr_dict_parse = {"F_{:04d}".format(i):v for i, v in enumerate(plugin_arr.values(), start=1)}
+    plugin_arr_dict_parse.update({f"F_{i}":v for i, v in enumerate(plugin_arr.values(), start=1)})
+    prompt = json.dumps(plugin_arr_info, ensure_ascii=False, indent=2)
+    prompt = "\n\nThe defination of PluginEnum:\nPluginEnum=" + prompt
+    return prompt, plugin_arr_dict, plugin_arr_dict_parse
+
+def wrap_code(txt):
+    txt = txt.replace('```','')
+    return f"\n```\n{txt}\n```\n"
+
+def have_any_recent_upload_files(chatbot):
+    _5min = 5 * 60
+    if not chatbot: return False    # chatbot is None
+    most_recent_uploaded = chatbot._cookies.get("most_recent_uploaded", None)
+    if not most_recent_uploaded: return False   # most_recent_uploaded is None
+    if time.time() - most_recent_uploaded["time"] < _5min: return True # most_recent_uploaded is new
+    else: return False  # most_recent_uploaded is too old
+
+def get_recent_file_prompt_support(chatbot):
+    most_recent_uploaded = chatbot._cookies.get("most_recent_uploaded", None)
+    path = most_recent_uploaded['path']
+    prompt =   "\nAdditional Information:\n"
+    prompt =   "In case that this plugin requires a path or a file as argument," 
+    prompt += f"it is important for you to know that the user has recently uploaded a file, located at: `{path}`" 
+    prompt += f"Only use it when necessary, otherwise, you can ignore this file." 
+    return prompt
+
+def get_inputs_show_user(inputs, plugin_arr_enum_prompt):
+    # remove plugin_arr_enum_prompt from inputs string
+    inputs_show_user = inputs.replace(plugin_arr_enum_prompt, "")
+    inputs_show_user += plugin_arr_enum_prompt[:200] + '...'
+    inputs_show_user += '\n...\n'
+    inputs_show_user += '...\n'
+    inputs_show_user += '...}'
+    return inputs_show_user
+
+def execute_plugin(txt, llm_kwargs, plugin_kwargs, chatbot, history, system_prompt, user_intention):
+    plugin_arr_enum_prompt, plugin_arr_dict, plugin_arr_dict_parse = read_avail_plugin_enum()
+    class Plugin(BaseModel):
+        plugin_selection: str = Field(description="The most related plugin from one of the PluginEnum.", default="F_0000")
+        reason_of_selection: str = Field(description="The reason why you should select this plugin.", default="This plugin satisfy user requirement most")
+    # ⭐ ⭐ ⭐ 选择插件
+    yield from update_ui_lastest_msg(lastmsg=f"正在执行任务: {txt}\n\n查找可用插件中...", chatbot=chatbot, history=history, delay=0)
+    gpt_json_io = GptJsonIO(Plugin)
+    gpt_json_io.format_instructions = "The format of your output should be a json that can be parsed by json.loads.\n"
+    gpt_json_io.format_instructions += """Output example: {"plugin_selection":"F_1234", "reason_of_selection":"F_1234 plugin satisfy user requirement most"}\n"""
+    gpt_json_io.format_instructions += "The plugins you are authorized to use are listed below:\n"
+    gpt_json_io.format_instructions += plugin_arr_enum_prompt
+    inputs = "Choose the correct plugin according to user requirements, the user requirement is: \n\n" + \
+             ">> " + txt.rstrip('\n').replace('\n','\n>> ') + '\n\n' + gpt_json_io.format_instructions
+
+    run_gpt_fn = lambda inputs, sys_prompt: predict_no_ui_long_connection(
+        inputs=inputs, llm_kwargs=llm_kwargs, history=[], sys_prompt=sys_prompt, observe_window=[])
+    try:
+        gpt_reply = run_gpt_fn(inputs, "")
+        plugin_sel = gpt_json_io.generate_output_auto_repair(gpt_reply, run_gpt_fn)
+    except JsonStringError:
+        msg = f"抱歉, {llm_kwargs['llm_model']}无法理解您的需求。"
+        msg += "请求的Prompt为：\n" + wrap_code(get_inputs_show_user(inputs, plugin_arr_enum_prompt))
+        msg += "语言模型回复为：\n" + wrap_code(gpt_reply)
+        msg += "\n但您可以尝试再试一次\n"
+        yield from update_ui_lastest_msg(lastmsg=msg, chatbot=chatbot, history=history, delay=2)
+        return
+    if plugin_sel.plugin_selection not in plugin_arr_dict_parse:
+        msg = f"抱歉, 找不到合适插件执行该任务, 或者{llm_kwargs['llm_model']}无法理解您的需求。"
+        msg += f"语言模型{llm_kwargs['llm_model']}选择了不存在的插件：\n" + wrap_code(gpt_reply)
+        msg += "\n但您可以尝试再试一次\n"
+        yield from update_ui_lastest_msg(lastmsg=msg, chatbot=chatbot, history=history, delay=2)
+        return
+    
+    # ⭐ ⭐ ⭐ 确认插件参数
+    if not have_any_recent_upload_files(chatbot):
+        appendix_info = ""
+    else:
+        appendix_info = get_recent_file_prompt_support(chatbot)
+
+    plugin = plugin_arr_dict_parse[plugin_sel.plugin_selection]
+    yield from update_ui_lastest_msg(lastmsg=f"正在执行任务: {txt}\n\n提取插件参数...", chatbot=chatbot, history=history, delay=0)
+    class PluginExplicit(BaseModel):
+        plugin_selection: str = plugin_sel.plugin_selection
+        plugin_arg: str = Field(description="The argument of the plugin.", default="")
+    gpt_json_io = GptJsonIO(PluginExplicit)
+    gpt_json_io.format_instructions += "The information about this plugin is:" + plugin["Info"]
+    inputs = f"A plugin named {plugin_sel.plugin_selection} is selected, " + \
+             "you should extract plugin_arg from the user requirement, the user requirement is: \n\n" + \
+             ">> " + (txt + appendix_info).rstrip('\n').replace('\n','\n>> ') + '\n\n' + \
+             gpt_json_io.format_instructions 
+    run_gpt_fn = lambda inputs, sys_prompt: predict_no_ui_long_connection(
+        inputs=inputs, llm_kwargs=llm_kwargs, history=[], sys_prompt=sys_prompt, observe_window=[])
+    plugin_sel = gpt_json_io.generate_output_auto_repair(run_gpt_fn(inputs, ""), run_gpt_fn)
+
+
+    # ⭐ ⭐ ⭐ 执行插件
+    fn = plugin['Function']
+    fn_name = fn.__name__
+    msg = f'{llm_kwargs["llm_model"]}为您选择了插件: `{fn_name}`\n\n插件说明：{plugin["Info"]}\n\n插件参数：{plugin_sel.plugin_arg}\n\n假如偏离了您的要求，按停止键终止。'
+    yield from update_ui_lastest_msg(lastmsg=msg, chatbot=chatbot, history=history, delay=2)
+    yield from fn(plugin_sel.plugin_arg, llm_kwargs, plugin_kwargs, chatbot, history, system_prompt, -1)
+    return

crazy_functions/vt_fns/vt_modify_config.py

@@ -0,0 +1,81 @@
+from pydantic import BaseModel, Field
+from typing import List
+from toolbox import update_ui_lastest_msg, get_conf
+from request_llm.bridge_all import predict_no_ui_long_connection
+from crazy_functions.json_fns.pydantic_io import GptJsonIO
+import copy, json, pickle, os, sys
+
+
+def modify_configuration_hot(txt, llm_kwargs, plugin_kwargs, chatbot, history, system_prompt, user_intention):
+    ALLOW_RESET_CONFIG, = get_conf('ALLOW_RESET_CONFIG')
+    if not ALLOW_RESET_CONFIG:
+        yield from update_ui_lastest_msg(
+            lastmsg=f"当前配置不允许被修改！如需激活本功能，请在config.py中设置ALLOW_RESET_CONFIG=True后重启软件。", 
+            chatbot=chatbot, history=history, delay=2
+        )
+        return
+
+    # ⭐ ⭐ ⭐ 读取可配置项目条目
+    names = {}
+    from enum import Enum
+    import config
+    for k, v in config.__dict__.items():
+        if k.startswith('__'): continue
+        names.update({k:k})
+        # if len(names) > 20: break   # 限制最多前10个配置项，如果太多了会导致gpt无法理解
+
+    ConfigOptions = Enum('ConfigOptions', names)
+    class ModifyConfigurationIntention(BaseModel):
+        which_config_to_modify: ConfigOptions = Field(description="the name of the configuration to modify, you must choose from one of the ConfigOptions enum.", default=None)
+        new_option_value: str = Field(description="the new value of the option", default=None)
+
+    # ⭐ ⭐ ⭐ 分析用户意图
+    yield from update_ui_lastest_msg(lastmsg=f"正在执行任务: {txt}\n\n读取新配置中", chatbot=chatbot, history=history, delay=0)
+    gpt_json_io = GptJsonIO(ModifyConfigurationIntention)
+    inputs = "Analyze how to change configuration according to following user input, answer me with json: \n\n" + \
+             ">> " + txt.rstrip('\n').replace('\n','\n>> ') + '\n\n' + \
+             gpt_json_io.format_instructions
+    
+    run_gpt_fn = lambda inputs, sys_prompt: predict_no_ui_long_connection(
+        inputs=inputs, llm_kwargs=llm_kwargs, history=[], sys_prompt=sys_prompt, observe_window=[])
+    user_intention = gpt_json_io.generate_output_auto_repair(run_gpt_fn(inputs, ""), run_gpt_fn)
+
+    explicit_conf = user_intention.which_config_to_modify.value
+
+    ok = (explicit_conf in txt)
+    if ok:
+        yield from update_ui_lastest_msg(
+            lastmsg=f"正在执行任务: {txt}\n\n新配置{explicit_conf}={user_intention.new_option_value}", 
+            chatbot=chatbot, history=history, delay=1
+        )
+        yield from update_ui_lastest_msg(
+            lastmsg=f"正在执行任务: {txt}\n\n新配置{explicit_conf}={user_intention.new_option_value}\n\n正在修改配置中", 
+            chatbot=chatbot, history=history, delay=2
+        )
+
+        # ⭐ ⭐ ⭐ 立即应用配置
+        from toolbox import set_conf
+        set_conf(explicit_conf, user_intention.new_option_value)
+
+        yield from update_ui_lastest_msg(
+            lastmsg=f"正在执行任务: {txt}\n\n配置修改完成，重新页面即可生效。", chatbot=chatbot, history=history, delay=1
+        )
+    else:
+        yield from update_ui_lastest_msg(
+            lastmsg=f"失败，如果需要配置{explicit_conf}，您需要明确说明并在指令中提到它。", chatbot=chatbot, history=history, delay=5
+        )
+
+def modify_configuration_reboot(txt, llm_kwargs, plugin_kwargs, chatbot, history, system_prompt, user_intention):
+    ALLOW_RESET_CONFIG, = get_conf('ALLOW_RESET_CONFIG')
+    if not ALLOW_RESET_CONFIG:
+        yield from update_ui_lastest_msg(
+            lastmsg=f"当前配置不允许被修改！如需激活本功能，请在config.py中设置ALLOW_RESET_CONFIG=True后重启软件。", 
+            chatbot=chatbot, history=history, delay=2
+        )
+        return
+
+    yield from modify_configuration_hot(txt, llm_kwargs, plugin_kwargs, chatbot, history, system_prompt, user_intention)
+    yield from update_ui_lastest_msg(
+        lastmsg=f"正在执行任务: {txt}\n\n配置修改完成，五秒后即将重启！若出现报错请无视即可。", chatbot=chatbot, history=history, delay=5
+    )
+    os.execl(sys.executable, sys.executable, *sys.argv)

crazy_functions/vt_fns/vt_state.py

@@ -0,0 +1,28 @@
+import pickle
+
+class VoidTerminalState():
+    def __init__(self):
+        self.reset_state()
+
+    def reset_state(self):
+        self.has_provided_explaination = False
+ 
+    def lock_plugin(self, chatbot):
+        chatbot._cookies['lock_plugin'] = 'crazy_functions.虚空终端->虚空终端'
+        chatbot._cookies['plugin_state'] = pickle.dumps(self)
+
+    def unlock_plugin(self, chatbot):
+        self.reset_state()
+        chatbot._cookies['lock_plugin'] = None
+        chatbot._cookies['plugin_state'] = pickle.dumps(self)
+
+    def set_state(self, chatbot, key, value):
+        setattr(self, key, value)
+        chatbot._cookies['plugin_state'] = pickle.dumps(self)
+
+    def get_state(chatbot):
+        state = chatbot._cookies.get('plugin_state', None)
+        if state is not None:   state = pickle.loads(state)
+        else:                   state = VoidTerminalState()
+        state.chatbot = chatbot
+        return state