互动游戏

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({

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

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:
+class GptAcademicGameBaseState():
-        materials = ['image', 'prompt']
+    """
+    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)

crazy_functions/互动小游戏.py

@@ -1,159 +1,59 @@
-from toolbox import CatchException, update_ui, get_conf, select_api_key, get_log_folder
+from toolbox import CatchException, update_ui, update_ui_lastest_msg
-from crazy_functions.multi_stage.multi_stage_utils import GptAcademicState
+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):
+class MiniGame_ASCII_Art(GptAcademicGameBaseState):
-        self.need_game_reset = True
-        self.llm_kwargs = None
-        super().__init__()
     
-    def lock_plugin(self, chatbot):
+    def step(self, prompt, chatbot, history):
-        chatbot._cookies['lock_plugin'] = 'crazy_functions.互动小游戏->谁是卧底'
+        if self.step_cnt == 0:  
-        self.dump_state(chatbot)
+            chatbot.append(["我画你猜（动物）", "请稍等..."])
-
-    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
         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):
+        if self.step_cnt == 0:
-        inputs = f'please say something about {self.card}'
+            self.lock_plugin(chatbot)
-        res = yield from request_gpt_model_in_new_thread_with_ui_alive(
+            self.cur_task = 'draw'
-            inputs = inputs,
-            inputs_show_user=inputs,
-            llm_kwargs=self.game_handle.llm_kwargs,
-            chatbot=chatbot,
-            history=history,
-            sys_prompt=sys_prompt
-        )
-        pass
 
-    def vote(self, content=None):
+        if self.cur_task == 'draw':
-        if content is None:
+            avail_obj = ["狗","猫","鸟","鱼","老鼠","蛇"]
-            assert not self.is_human
+            self.obj = random.choice(avail_obj)
-            self.vote_who = yield from
+            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 ```. "
-        else:
+            raw_res = predict_no_ui_long_connection(inputs=inputs, llm_kwargs=self.llm_kwargs, history=[], sys_prompt="")
-            try:
+            self.cur_task = 'identify user guess'
-                self.vote_who = int(content)
+            res = get_code_block(raw_res)
-            except:
+            history += ['', f'the answer is {self.obj}', inputs, res]
-                self.vote_who = None
+            yield from update_ui_lastest_msg(lastmsg=res, chatbot=chatbot, history=history, delay=0.)
 
-    def agi_vote(self):
+        elif self.cur_task == 'identify user guess':
-        pass
+            if is_same_thing(self.obj, prompt, self.llm_kwargs):
-
+                self.delete_game = True
-class 小游戏_谁是卧底(小游戏):
+                yield from update_ui_lastest_msg(lastmsg="你猜对了！", chatbot=chatbot, history=history, delay=0.)
-    def __init__(self):
+            else:
-        self.game_phase = '发言' # 投票
+                self.cur_task = 'identify user guess'
-        super().__init__()
+                yield from update_ui_lastest_msg(lastmsg="猜错了，再试试，输入“exit”获取答案。", chatbot=chatbot, history=history, delay=0.)
-
+            
-    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
-        
 
 @CatchException
-def 谁是卧底(prompt, llm_kwargs, plugin_kwargs, chatbot, history, system_prompt, web_port):
+def 随机小游戏(prompt, llm_kwargs, plugin_kwargs, chatbot, history, system_prompt, web_port):
-    # 尚未完成
+    # 清空历史
-    history = []    # 清空历史
+    history = []
-    state = 小游戏_谁是卧底.get_state(chatbot, 小游戏_谁是卧底)
+    # 选择游戏
-    state.llm_kwargs = llm_kwargs
+    cls = MiniGame_ASCII_Art
-    yield from state.continue_game(prompt, chatbot)
+    # 如果之前已经初始化了游戏实例，则继续该实例；否则重新初始化
+    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)