' in inputs: inputs = inputs.split('

')[1] inputs = inputs.split("

")[0] elif len(self.history) == 1: inputs = self.history[-1]["content"] del self.history[-1] else: yield chatbot, f"{STANDARD_ERROR_MSG}上下文是空的" return iter = self.predict( inputs, chatbot, stream=stream, use_websearch=use_websearch, files=files, reply_language=reply_language, ) for x in iter: yield x logging.debug("重试完毕") # def reduce_token_size(self, chatbot): # logging.info("开始减少token数量……") # chatbot, status_text = self.next_chatbot_at_once( # summarize_prompt, # chatbot # ) # max_token_count = self.token_upper_limit * REDUCE_TOKEN_FACTOR # num_chat = find_n(self.all_token_counts, max_token_count) # logging.info(f"previous_token_count: {self.all_token_counts}, keeping {num_chat} chats") # chatbot = chatbot[:-1] # self.history = self.history[-2*num_chat:] if num_chat > 0 else [] # self.all_token_counts = self.all_token_counts[-num_chat:] if num_chat > 0 else [] # msg = f"保留了最近{num_chat}轮对话" # logging.info(msg) # logging.info("减少token数量完毕") # return chatbot, msg + "，" + self.token_message(self.all_token_counts if len(self.all_token_counts) > 0 else [0]) def interrupt(self): self.interrupted = True def recover(self): self.interrupted = False def set_token_upper_limit(self, new_upper_limit): self.token_upper_limit = new_upper_limit print(f"token上限设置为{new_upper_limit}") def set_temperature(self, new_temperature): self.temperature = new_temperature def set_top_p(self, new_top_p): self.top_p = new_top_p def set_n_choices(self, new_n_choices): self.n_choices = new_n_choices def set_stop_sequence(self, new_stop_sequence: str): new_stop_sequence = new_stop_sequence.split(",") self.stop_sequence = new_stop_sequence def set_max_tokens(self, new_max_tokens): self.max_generation_token = new_max_tokens def set_presence_penalty(self, new_presence_penalty): self.presence_penalty = new_presence_penalty def set_frequency_penalty(self, new_frequency_penalty): self.frequency_penalty = new_frequency_penalty def set_logit_bias(self, logit_bias): logit_bias = logit_bias.split() bias_map = {} encoding = tiktoken.get_encoding("cl100k_base") for line in logit_bias: word, bias_amount = line.split(":") if word: for token in encoding.encode(word): bias_map[token] = float(bias_amount) self.logit_bias = bias_map def set_user_identifier(self, new_user_identifier): self.user_identifier = new_user_identifier def set_system_prompt(self, new_system_prompt): self.system_prompt = new_system_prompt def set_key(self, new_access_key): if "*" not in new_access_key: self.api_key = new_access_key.strip() msg = i18n("API密钥更改为了") + hide_middle_chars(self.api_key) logging.info(msg) return self.api_key, msg else: return gr.update(), gr.update() def set_single_turn(self, new_single_turn): self.single_turn = new_single_turn def reset(self, remain_system_prompt=False): self.history = [] self.all_token_counts = [] self.interrupted = False self.history_file_path = new_auto_history_filename(self.user_identifier) history_name = self.history_file_path[:-5] choices = [history_name] + get_history_names(self.user_identifier) system_prompt = self.system_prompt if remain_system_prompt else "" return [], self.token_message([0]), gr.Radio.update(choices=choices, value=history_name), system_prompt def delete_first_conversation(self): if self.history: del self.history[:2] del self.all_token_counts[0] return self.token_message() def delete_last_conversation(self, chatbot): if len(chatbot) > 0 and STANDARD_ERROR_MSG in chatbot[-1][1]: msg = "由于包含报错信息，只删除chatbot记录" chatbot = chatbot[:-1] return chatbot, self.history if len(self.history) > 0: self.history = self.history[:-2] if len(chatbot) > 0: msg = "删除了一组chatbot对话" chatbot = chatbot[:-1] if len(self.all_token_counts) > 0: msg = "删除了一组对话的token计数记录" self.all_token_counts.pop() msg = "删除了一组对话" self.auto_save(chatbot) return chatbot, msg def token_message(self, token_lst=None): if token_lst is None: token_lst = self.all_token_counts token_sum = 0 for i in range(len(token_lst)): token_sum += sum(token_lst[: i + 1]) return i18n("Token 计数: ") + f"{sum(token_lst)}" + i18n("，本次对话累计消耗了 ") + f"{token_sum} tokens" def rename_chat_history(self, filename, chatbot, user_name): if filename == "": return gr.update() if not filename.endswith(".json"): filename += ".json" self.delete_chat_history(self.history_file_path, user_name) # 命名重复检测 repeat_file_index = 2 full_path = os.path.join(HISTORY_DIR, user_name, filename) while os.path.exists(full_path): full_path = os.path.join(HISTORY_DIR, user_name, f"{repeat_file_index}_{filename}") repeat_file_index += 1 filename = os.path.basename(full_path) self.history_file_path = filename save_file(filename, self.system_prompt, self.history, chatbot, user_name) return init_history_list(user_name) def auto_name_chat_history(self, name_chat_method, user_question, chatbot, user_name, single_turn_checkbox): if len(self.history) == 2 and not single_turn_checkbox: user_question = self.history[0]["content"] if type(user_question) == list: user_question = user_question[0]["text"] filename = replace_special_symbols(user_question)[:16] + ".json" return self.rename_chat_history(filename, chatbot, user_name) else: return gr.update() def auto_save(self, chatbot): save_file(self.history_file_path, self.system_prompt, self.history, chatbot, self.user_identifier) def export_markdown(self, filename, chatbot, user_name): if filename == "": return if not filename.endswith(".md"): filename += ".md" save_file(filename, self.system_prompt, self.history, chatbot, user_name) def load_chat_history(self, new_history_file_path=None, username=None): logging.debug(f"{self.user_identifier} 加载对话历史中……") if new_history_file_path is not None: if type(new_history_file_path) != str: # copy file from new_history_file_path.name to os.path.join(HISTORY_DIR, self.user_identifier) new_history_file_path = new_history_file_path.name shutil.copyfile(new_history_file_path, os.path.join( HISTORY_DIR, self.user_identifier, os.path.basename(new_history_file_path))) self.history_file_path = os.path.basename(new_history_file_path) else: self.history_file_path = new_history_file_path try: if self.history_file_path == os.path.basename(self.history_file_path): history_file_path = os.path.join( HISTORY_DIR, self.user_identifier, self.history_file_path) else: history_file_path = self.history_file_path if not self.history_file_path.endswith(".json"): history_file_path += ".json" with open(history_file_path, "r", encoding="utf-8") as f: json_s = json.load(f) try: if type(json_s["history"][0]) == str: logging.info("历史记录格式为旧版，正在转换……") new_history = [] for index, item in enumerate(json_s["history"]): if index % 2 == 0: new_history.append(construct_user(item)) else: new_history.append(construct_assistant(item)) json_s["history"] = new_history logging.info(new_history) except: pass if len(json_s["chatbot"]) < len(json_s["history"])//2: logging.info("Trimming corrupted history...") json_s["history"] = json_s["history"][-len(json_s["chatbot"]):] logging.info(f"Trimmed history: {json_s['history']}") logging.debug(f"{self.user_identifier} 加载对话历史完毕") self.history = json_s["history"] return os.path.basename(self.history_file_path), json_s["system"], json_s["chatbot"] except: # 没有对话历史或者对话历史解析失败 logging.info(f"没有找到对话历史记录 {self.history_file_path}") return self.history_file_path, "", [] def delete_chat_history(self, filename, user_name): if filename == "CANCELED": return gr.update(), gr.update(), gr.update() if filename == "": return i18n("你没有选择任何对话历史"), gr.update(), gr.update() if not filename.endswith(".json"): filename += ".json" if filename == os.path.basename(filename): history_file_path = os.path.join(HISTORY_DIR, user_name, filename) else: history_file_path = filename md_history_file_path = history_file_path[:-5] + ".md" try: os.remove(history_file_path) os.remove(md_history_file_path) return i18n("删除对话历史成功"), get_history_list(user_name), [] except: logging.info(f"删除对话历史失败 {history_file_path}") return i18n("对话历史")+filename+i18n("已经被删除啦"), get_history_list(user_name), [] def auto_load(self): filepath = get_history_filepath(self.user_identifier) if not filepath: self.history_file_path = new_auto_history_filename( self.user_identifier) else: self.history_file_path = filepath filename, system_prompt, chatbot = self.load_chat_history() filename = filename[:-5] return filename, system_prompt, chatbot def like(self): """like the last response, implement if needed """ return gr.update() def dislike(self): """dislike the last response, implement if needed """ return gr.update() class Base_Chat_Langchain_Client(BaseLLMModel): def __init__(self, model_name, user_name=""): super().__init__(model_name, user=user_name) self.need_api_key = False self.model = self.setup_model() def setup_model(self): # inplement this to setup the model then return it pass def _get_langchain_style_history(self): history = [SystemMessage(content=self.system_prompt)] for i in self.history: if i["role"] == "user": history.append(HumanMessage(content=i["content"])) elif i["role"] == "assistant": history.append(AIMessage(content=i["content"])) return history def get_answer_at_once(self): assert isinstance( self.model, BaseChatModel), "model is not instance of LangChain BaseChatModel" history = self._get_langchain_style_history() response = self.model.generate(history) return response.content, sum(response.content) def get_answer_stream_iter(self): it = CallbackToIterator() assert isinstance( self.model, BaseChatModel), "model is not instance of LangChain BaseChatModel" history = self._get_langchain_style_history() def thread_func(): self.model(messages=history, callbacks=[ ChuanhuCallbackHandler(it.callback)]) it.finish() t = Thread(target=thread_func) t.start() partial_text = "" for value in it: partial_text += value yield partial_text