Spaces:
Build error
Build error
| 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 | |
| import pickle | |
| 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 | |
| class GptAcademicState(): | |
| def __init__(self): | |
| self.reset() | |
| def reset(self): | |
| pass | |
| def dump_state(self, chatbot): | |
| 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, cls=None): | |
| state = chatbot._cookies.get('plugin_state', None) | |
| if state is not None: state = pickle.loads(state) | |
| elif cls is not None: state = cls() | |
| else: state = GptAcademicState() | |
| state.chatbot = chatbot | |
| return state | |
| 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) | |
| 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) | |