diff --git a/crazy_functional.py b/crazy_functional.py index c3ee50a..31766f0 100644 --- a/crazy_functional.py +++ b/crazy_functional.py @@ -590,6 +590,20 @@ def get_crazy_functions(): print(trimmed_format_exc()) print('Load function plugin failed') + # try: + # from crazy_functions.互动小游戏 import 随机小游戏 + # function_plugins.update({ + # "随机小游戏": { + # "Group": "智能体", + # "Color": "stop", + # "AsButton": True, + # "Function": HotReload(随机小游戏) + # } + # }) + # except: + # print(trimmed_format_exc()) + # print('Load function plugin failed') + # try: # from crazy_functions.chatglm微调工具 import 微调数据集生成 # function_plugins.update({ diff --git a/crazy_functions/game_fns/game_utils.py b/crazy_functions/game_fns/game_utils.py new file mode 100644 index 0000000..09b6f7a --- /dev/null +++ b/crazy_functions/game_fns/game_utils.py @@ -0,0 +1,35 @@ + +from crazy_functions.json_fns.pydantic_io import GptJsonIO, JsonStringError +from request_llms.bridge_all import predict_no_ui_long_connection +def get_code_block(reply): + import re + pattern = r"```([\s\S]*?)```" # regex pattern to match code blocks + matches = re.findall(pattern, reply) # find all code blocks in text + if len(matches) == 1: + return "```" + matches[0] + "```" # code block + raise RuntimeError("GPT is not generating proper code.") + +def is_same_thing(a, b, llm_kwargs): + from pydantic import BaseModel, Field + class IsSameThing(BaseModel): + is_same_thing: bool = Field(description="determine whether two objects are same thing.", default=False) + + def run_gpt_fn(inputs, sys_prompt, history=[]): + return predict_no_ui_long_connection( + inputs=inputs, llm_kwargs=llm_kwargs, + history=history, sys_prompt=sys_prompt, observe_window=[] + ) + + gpt_json_io = GptJsonIO(IsSameThing) + inputs_01 = "Identity whether the user input and the target is the same thing: \n target object: {a} \n user input object: {b} \n\n\n".format(a=a, b=b) + inputs_01 += "\n\n\n Note that the user may describe the target object with a different language, e.g. cat and 猫 are the same thing." + analyze_res_cot_01 = run_gpt_fn(inputs_01, "", []) + + inputs_02 = inputs_01 + gpt_json_io.format_instructions + analyze_res = run_gpt_fn(inputs_02, "", [inputs_01, analyze_res_cot_01]) + + try: + res = gpt_json_io.generate_output_auto_repair(analyze_res, run_gpt_fn) + return res.is_same_thing + except JsonStringError as e: + return False \ No newline at end of file diff --git a/crazy_functions/multi_stage/multi_stage_utils.py b/crazy_functions/multi_stage/multi_stage_utils.py index f85d35a..1395e79 100644 --- a/crazy_functions/multi_stage/multi_stage_utils.py +++ b/crazy_functions/multi_stage/multi_stage_utils.py @@ -1,6 +1,7 @@ from pydantic import BaseModel, Field from typing import List from toolbox import update_ui_lastest_msg, disable_auto_promotion +from toolbox import CatchException, update_ui, get_conf, select_api_key, get_log_folder from request_llms.bridge_all import predict_no_ui_long_connection from crazy_functions.json_fns.pydantic_io import GptJsonIO, JsonStringError import time @@ -36,6 +37,57 @@ class GptAcademicState(): state.chatbot = chatbot return state -class GatherMaterials(): - def __init__(self, materials) -> None: - materials = ['image', 'prompt'] \ No newline at end of file + +class GptAcademicGameBaseState(): + """ + 1. first init: __init__ -> + """ + def init_game(self, chatbot, lock_plugin): + self.plugin_name = None + self.callback_fn = None + self.delete_game = False + self.step_cnt = 0 + + def lock_plugin(self, chatbot): + if self.callback_fn is None: + raise ValueError("callback_fn is None") + chatbot._cookies['lock_plugin'] = self.callback_fn + self.dump_state(chatbot) + + def get_plugin_name(self): + if self.plugin_name is None: + raise ValueError("plugin_name is None") + return self.plugin_name + + def dump_state(self, chatbot): + chatbot._cookies[f'plugin_state/{self.get_plugin_name()}'] = pickle.dumps(self) + + def set_state(self, chatbot, key, value): + setattr(self, key, value) + chatbot._cookies[f'plugin_state/{self.get_plugin_name()}'] = pickle.dumps(self) + + @staticmethod + def sync_state(chatbot, llm_kwargs, cls, plugin_name, callback_fn, lock_plugin=True): + state = chatbot._cookies.get(f'plugin_state/{plugin_name}', None) + if state is not None: + state = pickle.loads(state) + else: + state = cls() + state.init_game(chatbot, lock_plugin) + state.plugin_name = plugin_name + state.llm_kwargs = llm_kwargs + state.chatbot = chatbot + state.callback_fn = callback_fn + return state + + def continue_game(self, prompt, chatbot, history): + # 游戏主体 + yield from self.step(prompt, chatbot, history) + self.step_cnt += 1 + # 保存状态，收尾 + self.dump_state(chatbot) + # 如果游戏结束，清理 + if self.delete_game: + chatbot._cookies['lock_plugin'] = None + chatbot._cookies[f'plugin_state/{self.get_plugin_name()}'] = None + yield from update_ui(chatbot=chatbot, history=history) diff --git a/crazy_functions/互动小游戏.py b/crazy_functions/互动小游戏.py index e00ef32..a6871b3 100644 --- a/crazy_functions/互动小游戏.py +++ b/crazy_functions/互动小游戏.py @@ -1,159 +1,59 @@ -from toolbox import CatchException, update_ui, get_conf, select_api_key, get_log_folder -from crazy_functions.multi_stage.multi_stage_utils import GptAcademicState +from toolbox import CatchException, update_ui, update_ui_lastest_msg +from crazy_functions.multi_stage.multi_stage_utils import GptAcademicGameBaseState from crazy_functions.crazy_utils import request_gpt_model_in_new_thread_with_ui_alive +from request_llms.bridge_all import predict_no_ui_long_connection +from crazy_functions.game_fns.game_utils import get_code_block, is_same_thing import random -class 小游戏(GptAcademicState): - def __init__(self): - self.need_game_reset = True - self.llm_kwargs = None - super().__init__() + +class MiniGame_ASCII_Art(GptAcademicGameBaseState): - def lock_plugin(self, chatbot): - chatbot._cookies['lock_plugin'] = 'crazy_functions.互动小游戏->谁是卧底' - self.dump_state(chatbot) - - def unlock_plugin(self, chatbot): - self.reset() - chatbot._cookies['lock_plugin'] = None - self.dump_state(chatbot) - - def set_state(self, chatbot, key, value): - return super().set_state(chatbot, key, value) - - def init_game(self, chatbot): - chatbot.get_cookies()['lock_plugin'] = '' - - def clean_up_game(self, chatbot): - chatbot.get_cookies()['lock_plugin'] = None - - def init_player(self): - pass - - def step(self, prompt, chatbot): - pass - - def continue_game(self, prompt, chatbot): - if self.need_game_reset: - self.need_game_reset = False - yield from self.init_game(chatbot) - yield from self.step(prompt, chatbot) - self.dump_state(chatbot) - yield update_ui(chatbot=chatbot, history=[]) - -class 小游戏_谁是卧底_玩家(): - def __init__(self, game_handle, card, llm_model, name) -> None: - self.game_handle = game_handle - self.card = card - self.name = name - self.is_out = False - self.llm_model = llm_model - self.is_human = llm_model == 'human' - self.what_player_has_spoken = [] - - def speek(self, content=None): - if content is None: - assert not self.is_human - speak_what = yield from + def step(self, prompt, chatbot, history): + if self.step_cnt == 0: + chatbot.append(["我画你猜（动物）", "请稍等..."]) else: - self.what_player_has_spoken.append(content) + if prompt.strip() == 'exit': + self.delete_game = True + yield from update_ui_lastest_msg(lastmsg=f"谜底是{self.obj}，游戏结束。", chatbot=chatbot, history=history, delay=0.) + return + chatbot.append([prompt, ""]) + yield from update_ui(chatbot=chatbot, history=history) - def agi_speek(self): - inputs = f'please say something about {self.card}' - res = yield from request_gpt_model_in_new_thread_with_ui_alive( - inputs = inputs, - inputs_show_user=inputs, - llm_kwargs=self.game_handle.llm_kwargs, - chatbot=chatbot, - history=history, - sys_prompt=sys_prompt - ) - pass + if self.step_cnt == 0: + self.lock_plugin(chatbot) + self.cur_task = 'draw' - def vote(self, content=None): - if content is None: - assert not self.is_human - self.vote_who = yield from - else: - try: - self.vote_who = int(content) - except: - self.vote_who = None + if self.cur_task == 'draw': + avail_obj = ["狗","猫","鸟","鱼","老鼠","蛇"] + self.obj = random.choice(avail_obj) + inputs = "I want to play a game called Guess the ASCII art. You can draw the ASCII art and I will try to guess it. " + f"This time you draw a {self.obj}. Note that you must not indicate what you have draw in the text, and you should only produce the ASCII art wrapped by ```. " + raw_res = predict_no_ui_long_connection(inputs=inputs, llm_kwargs=self.llm_kwargs, history=[], sys_prompt="") + self.cur_task = 'identify user guess' + res = get_code_block(raw_res) + history += ['', f'the answer is {self.obj}', inputs, res] + yield from update_ui_lastest_msg(lastmsg=res, chatbot=chatbot, history=history, delay=0.) - def agi_vote(self): - pass - -class 小游戏_谁是卧底(小游戏): - def __init__(self): - self.game_phase = '发言' # 投票 - super().__init__() - - def init_game(self, chatbot): - self.n_players = 3 - self.n_ai_players = self.n_players - 1 - card = "橙子" - undercover_card = "橘子" - llm_model = self.llm_kwargs['llm_model'] - self.players = [ - 小游戏_谁是卧底(self, card, llm_model, str(i)) for i in range(self.n_players) - ] - - undercover = random.randint(0, self.n_players-1) - human = 0 - - self.players[undercover].card = undercover_card - self.players[human].llm_model = 'human' - super().init_game(chatbot) - - def who_is_out(self): - votes = {} - for player in self.players: - if player.is_out: continue - if player.vote is None: continue - if player.vote not in votes: votes[player.vote] = 0 - votes[player.vote] += 1 - max_votes = max(votes.values()) - players_with_max_votes = [player for player, vote_count in votes.items() if vote_count == max_votes] - for player in players_with_max_votes: - print('淘汰了', player.name) - player.is_out = True - return players_with_max_votes - - def step(self, prompt, chatbot): - - if self.game_phase == '发言': - for player in self.players: - if player.is_out: continue - if player.is_human: - player.speek(prompt) - else: - player.speek() - self.game_phase = '投票' - - elif self.game_phase == '投票': - for player in self.players: - if player.is_out: continue - if player.is_human: - player.vote(prompt) - else: - player.vote() - self.who_is_out() - if len([player for player in self.players if not player.is_out]) <= 2: - if sum([player for player in self.players if player.is_undercover]) == 1: - print('卧底获胜') - else: - print('平民获胜') - self.need_game_reset = True - self.game_phase = '发言' - - else: - raise RuntimeError - + elif self.cur_task == 'identify user guess': + if is_same_thing(self.obj, prompt, self.llm_kwargs): + self.delete_game = True + yield from update_ui_lastest_msg(lastmsg="你猜对了！", chatbot=chatbot, history=history, delay=0.) + else: + self.cur_task = 'identify user guess' + yield from update_ui_lastest_msg(lastmsg="猜错了，再试试，输入“exit”获取答案。", chatbot=chatbot, history=history, delay=0.) + @CatchException -def 谁是卧底(prompt, llm_kwargs, plugin_kwargs, chatbot, history, system_prompt, web_port): - # 尚未完成 - history = [] # 清空历史 - state = 小游戏_谁是卧底.get_state(chatbot, 小游戏_谁是卧底) - state.llm_kwargs = llm_kwargs - yield from state.continue_game(prompt, chatbot) +def 随机小游戏(prompt, llm_kwargs, plugin_kwargs, chatbot, history, system_prompt, web_port): + # 清空历史 + history = [] + # 选择游戏 + cls = MiniGame_ASCII_Art + # 如果之前已经初始化了游戏实例，则继续该实例；否则重新初始化 + state = cls.sync_state(chatbot, + llm_kwargs, + cls, + plugin_name='MiniGame_ASCII_Art', + callback_fn='crazy_functions.互动小游戏->随机小游戏', + lock_plugin=True + ) + yield from state.continue_game(prompt, chatbot, history)