virtual_human.py

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)