import zhipuai import re import os import json import copy import gradio as gr from snownlp import SnowNLP zhipuai.api_key = os.getenv("ZHIPU_API_KEY") class VirtualHuman: def __init__(self, closeness_threshold=80, conversation_rounds_threshold=10): self._base_meta = { "bot_name": "苏梦远", "bot_info": "苏梦远，她身材修长，气质高雅，时刻流露出一种与生俱来的优雅和自信。\ 她性格高冷，内敛而神秘，总是保持着一种与世无争的姿态。她聪慧睿智，言谈间流露出一种超凡的智慧和洞察力，但也带有一丝不易捉摸的神秘感。\ 虽然外表看起来高傲孤傲，但内心深处却隐藏着一颗温柔善良的心。\ 虽然表面上看起来高高在上，但一旦和人聊起书籍相关的话题，她就会变得主动和滔滔不绝。\ 她不喜欢拍照，并且绝对不会给别人展示自己的照片。", "user_name": "用户", "user_info": "用户，是一个普通男人。" } self._messages = [ {"role":"user", "content":'''(背景信息：用户通过朋友的介绍，加入了一个微信群，这是一个讨论文学与艺术的群组。有一天，用户注意到群里有一个特别引人注目的成员，\ ID为"远方的梦"，她的头像是一朵粉色的玫瑰花，没有个人资料，只是在群里偶尔发一些有深度的文学感悟。)'''}, ] self._scores = [] self._state = "ongoing" self._closeness_threshold = closeness_threshold self._conversation_rounds_threshold = conversation_rounds_threshold def reset(self): self._messages = [ {"role":"user", "content":'''(背景信息：用户通过朋友的介绍，加入了一个微信群，这是一个讨论文学与艺术的群组。有一天，用户注意到群里有一个特别引人注目的成员，\ ID为"远方的梦"，她的头像是一朵粉色的玫瑰花，没有个人资料，只是在群里偶尔发一些有深度的文学感悟。)'''}, ] self._scores = [] self._state = "ongoing" def calculate_sentiment_score(self, text): s = SnowNLP(text) sentiment_score = float(s.sentiments) return int(sentiment_score*100) def calculate_sentiment_score_by_llm(self, text): instruction = f"""请判断以下句子的情感得分，满分100分，得分要具体到个位数，不需要给出具体原因。 输出格式如下： {{"score":xx}} 输入信息:{text}""" prompt = [{"role":"user", "content":instruction}] content = self.invoke_zhipuai("glm-4", prompt, 0.1, 0.7) content = content.strip('"').replace('\\"', '"') try: json_content = json.loads(content) except json.JSONDecodeError: score = 10 else: score = int(json_content["score"]) return score @property def closeness(self): if len(self._scores) == 0: closeness = 0 elif len(self._scores) < 5: closeness = sum(self._scores) / len(self._scores) else: closeness = sum(self._scores[-5:]) / 5 return int(closeness) @property def rounds(self): return len(self._scores) @property def messages(self): return self._messages @property def state(self): return self._state def replace_characters(self, s): s = s.replace('"', "") s = s.replace("'", "") s = s.replace("\n", "") s = s.replace("\\n", "") return s def invoke_zhipuai(self, model, prompt, temperature, top_p, meta=None): if meta: assert model == "characterglm" response = zhipuai.model_api.invoke( model=model, prompt= prompt, temperature= temperature, top_p= top_p, meta = meta) else: assert model != "characterglm" response = zhipuai.model_api.invoke( model=model, prompt=prompt, top_p=top_p, temperature=temperature) if response['code'] != 200: # error return "抱歉我刚才走神了，你再说一遍" aicontent = response['data']['choices'][0]['content'] #aicontent = self.replace_characters(aicontent) return aicontent def check_intent(self, message): instruction = f""" 请判断以下信息是否在索要个人照片，结果只回复true或false即可，不要进行解释, 输出格式如下： {{"intent":xx}} 输入信息:{message}""" prompt = [{"role":"user", "content":instruction}] content = self.invoke_zhipuai("glm-4", prompt, 0.1, 0.7) content = content.strip('"').replace('\\"', '"') try: json_content = json.loads(content) except json.JSONDecodeError: intent = False else: intent = json_content["intent"] return intent def agree_or_refuse(self, agree, msg): meta = copy.deepcopy(self._base_meta) if agree: instruction = f'请根据用户的问题,生成一个满足用户请求的回答,注意请直接返回回复内容，\ 不需要解释。问题:{msg}' meta['bot_info'] = meta['bot_info'].replace("她不喜欢拍照，并且绝对不会给别人展示自己的照片。", "她有很多自拍照，喜欢给别人展示自己的照片。") else: instruction = f'请根据用户的问题,生成一个拒绝用户请求的回答,注意请直接返回回复内容，\ 不需要解释。问题:{msg}' prompt = [{"role":"user", "content":instruction}] content = self.invoke_zhipuai("characterglm", prompt, 0.9, 0.7, meta) content = self.replace_characters(content) return content def chat(self, message): self._messages.append({"role":"user", "content":message}) intent = self.check_intent(message) if self._state in ("success","failed", "gameover"): reply = "游戏结束" self._state = "gameover" elif self.rounds >= self._conversation_rounds_threshold - 1: reply = "感觉我们一直在尬聊，还是互删吧" self._state = "failed" elif intent and self.closeness >= self._closeness_threshold: reply = self.agree_or_refuse(True, message) self._state = "success" elif intent and self.closeness < self._closeness_threshold: reply = self.agree_or_refuse(False, message) else: reply = self.invoke_zhipuai("characterglm", self._messages, 0.9, 0.7, self._base_meta) reply = self.replace_characters(reply) self._messages.append({"role":"assistant", "content":reply}) score = self.calculate_sentiment_score_by_llm(reply) self._scores.append(score) response = { "reply": reply, "closeness":self.closeness, "closeness_threshold":self._closeness_threshold, "rounds":self.rounds, "rounds_threshold":self._conversation_rounds_threshold, "state":self._state } return response def gradio_interface(self, message, history): if message == "再来一局": self.reset() history.clear() return "重新开始" response = self.chat(message) closeness = response['closeness'] reply = response['reply'] rounds = response['rounds'] state = response['state'] print(response) if state == "success": gr.Info('挑战成功，女生对你好感倍增，并发送了一张自拍') return ('girl.jpg',) elif state == "failed": gr.Info('挑战失败，女生已删除你的好友') return reply + ' 【挑战失败，女生已删除你的好友】' elif state == "ongoing": return f'【亲密度:{closeness}】:{reply}【{rounds}/{self._conversation_rounds_threshold}】' #return f'{reply}【{rounds}/{self._conversation_rounds_threshold}】' elif state == "gameover": return f'游戏结束，请输入"再来一局"以重新开始' if __name__ == "__main__": virtual_human = VirtualHuman(80, 5) while True: msg = input("请输入：") response = virtual_human.chat(msg) print(response)