diff --git a/src/plugins/chat/__init__.py b/src/plugins/chat/__init__.py index f7da8ba96..e2dfdadb0 100644 --- a/src/plugins/chat/__init__.py +++ b/src/plugins/chat/__init__.py @@ -13,6 +13,7 @@ from .willing_manager import willing_manager from nonebot.rule import to_me from .bot import chat_bot from .emoji_manager import emoji_manager +from ..moods.moods import MoodManager # 导入情绪管理器 import time from ..utils.statistic import LLMStatistics @@ -65,6 +66,11 @@ async def start_background_tasks(): llm_stats.start() print("\033[1;32m[初始化]\033[0m LLM统计功能已启动") + # 初始化并启动情绪管理器 + mood_manager = MoodManager.get_instance() + mood_manager.start_mood_update(update_interval=global_config.mood_update_interval) + print("\033[1;32m[初始化]\033[0m 情绪管理器已启动") + # 只启动表情包管理任务 asyncio.create_task(emoji_manager.start_periodic_check(interval_MINS=global_config.EMOJI_CHECK_INTERVAL)) await bot_schedule.initialize() @@ -122,4 +128,10 @@ async def merge_memory_task(): # print("\033[1;32m[记忆整合]\033[0m 开始整合") # await hippocampus.operation_merge_memory(percentage=0.1) # print("\033[1;32m[记忆整合]\033[0m 记忆整合完成") + +@scheduler.scheduled_job("interval", seconds=30, id="print_mood") +async def print_mood_task(): + """每30秒打印一次情绪状态""" + mood_manager = MoodManager.get_instance() + mood_manager.print_mood_status() diff --git a/src/plugins/chat/bot.py b/src/plugins/chat/bot.py index 4306c0f9d..910f6c02a 100644 --- a/src/plugins/chat/bot.py +++ b/src/plugins/chat/bot.py @@ -7,6 +7,7 @@ from .llm_generator import ResponseGenerator from .topic_identifier import topic_identifier from random import random, choice from .emoji_manager import emoji_manager # 导入表情包管理器 +from ..moods.moods import MoodManager # 导入情绪管理器 import time import os from .cq_code import CQCode # 导入CQCode模块 @@ -24,6 +25,8 @@ class ChatBot: self.gpt = ResponseGenerator() self.bot = None # bot 实例引用 self._started = False + self.mood_manager = MoodManager.get_instance() # 获取情绪管理器单例 + self.mood_manager.start_mood_update() # 启动情绪更新 self.emoji_chance = 0.2 # 发送表情包的基础概率 # self.message_streams = MessageStreamContainer() @@ -192,9 +195,17 @@ class ChatBot: emotion = await self.gpt._get_emotion_tags(raw_content) print(f"为 '{response}' 获取到的情感标签为:{emotion}") valuedict={ - 'happy':0.5,'angry':-1,'sad':-0.5,'surprised':0.5,'disgusted':-1.5,'fearful':-0.25,'neutral':0.25 + 'happy': 0.5, + 'angry': -1, + 'sad': -0.5, + 'surprised': 0.2, + 'disgusted': -1.5, + 'fearful': -0.7, + 'neutral': 0.1 } await relationship_manager.update_relationship_value(message.user_id, relationship_value=valuedict[emotion[0]]) + # 使用情绪管理器更新情绪 + self.mood_manager.update_mood_from_emotion(emotion[0], global_config.mood_intensity_factor) # willing_manager.change_reply_willing_after_sent(event.group_id) diff --git a/src/plugins/chat/config.py b/src/plugins/chat/config.py index 5c3c0b27a..0186001ab 100644 --- a/src/plugins/chat/config.py +++ b/src/plugins/chat/config.py @@ -56,6 +56,10 @@ class BotConfig: enable_advance_output: bool = False # 是否启用高级输出 enable_kuuki_read: bool = True # 是否启用读空气功能 + + mood_update_interval: float = 1.0 # 情绪更新间隔 单位秒 + mood_decay_rate: float = 0.95 # 情绪衰减率 + mood_intensity_factor: float = 0.7 # 情绪强度因子 # 默认人设 PROMPT_PERSONALITY=[ @@ -164,6 +168,12 @@ class BotConfig: memory_config = toml_dict["memory"] config.build_memory_interval = memory_config.get("build_memory_interval", config.build_memory_interval) config.forget_memory_interval = memory_config.get("forget_memory_interval", config.forget_memory_interval) + + if "mood" in toml_dict: + mood_config = toml_dict["mood"] + config.mood_update_interval = mood_config.get("mood_update_interval", config.mood_update_interval) + config.mood_decay_rate = mood_config.get("mood_decay_rate", config.mood_decay_rate) + config.mood_intensity_factor = mood_config.get("mood_intensity_factor", config.mood_intensity_factor) # 群组配置 if "groups" in toml_dict: diff --git a/src/plugins/chat/cq_code.py b/src/plugins/chat/cq_code.py index 4427ecb4a..af2651ecf 100644 --- a/src/plugins/chat/cq_code.py +++ b/src/plugins/chat/cq_code.py @@ -87,7 +87,7 @@ class CQCode: elif self.type == 'face': face_id = self.params.get('id', '') # self.translated_plain_text = f"[表情{face_id}]" - self.translated_plain_text = f"[{emojimapper.get(int(face_id), "表情")}]" + self.translated_plain_text = f"[{emojimapper.get(int(face_id), '表情')}]" elif self.type == 'forward': self.translated_plain_text = await self.translate_forward() else: diff --git a/src/plugins/chat/prompt_builder.py b/src/plugins/chat/prompt_builder.py index 3b7894f56..7c398583a 100644 --- a/src/plugins/chat/prompt_builder.py +++ b/src/plugins/chat/prompt_builder.py @@ -11,6 +11,7 @@ from random import choice import numpy as np import jieba from collections import Counter +from ..moods.moods import MoodManager class PromptBuilder: def __init__(self): @@ -49,6 +50,12 @@ class PromptBuilder: #开始构建prompt + + #心情 + mood_manager = MoodManager.get_instance() + mood_prompt = mood_manager.get_prompt() + + #日程构建 current_date = time.strftime("%Y-%m-%d", time.localtime()) current_time = time.strftime("%H:%M:%S", time.localtime()) @@ -108,13 +115,13 @@ class PromptBuilder: #激活prompt构建 activate_prompt = '' - activate_prompt = f"以上是群里正在进行的聊天,{memory_prompt} 现在昵称为 '{sender_name}' 的用户说的:{message_txt}。引起了你的注意,你和他{relation_prompt},你想要{relation_prompt_2}。" + activate_prompt = f"以上是群里正在进行的聊天,{memory_prompt} 现在昵称为 '{sender_name}' 的用户说的:{message_txt}。引起了你的注意,你和他{relation_prompt},{mood_prompt},你想要{relation_prompt_2}。" #检测机器人相关词汇 bot_keywords = ['人机', 'bot', '机器', '入机', 'robot', '机器人'] is_bot = any(keyword in message_txt.lower() for keyword in bot_keywords) if is_bot: - is_bot_prompt = '有人说你是人机或者机器人,否定这一事实,攻击或反问他。' + is_bot_prompt = '有人说你是人机或者机器人,否定这一事实,攻击或反问他。注意,只有认为你是机器人才回复,否则不要否认' else: is_bot_prompt = '' diff --git a/src/plugins/moods/moods.py b/src/plugins/moods/moods.py new file mode 100644 index 000000000..eee5da54c --- /dev/null +++ b/src/plugins/moods/moods.py @@ -0,0 +1,230 @@ +import math +import time +import threading +from typing import Dict, Tuple, Optional +from dataclasses import dataclass +from ..chat.config import global_config + +@dataclass +class MoodState: + valence: float # 愉悦度 (-1 到 1) + arousal: float # 唤醒度 (0 到 1) + text: str # 心情文本描述 + +class MoodManager: + _instance = None + _lock = threading.Lock() + + def __new__(cls): + with cls._lock: + if cls._instance is None: + cls._instance = super().__new__(cls) + cls._instance._initialized = False + return cls._instance + + def __init__(self): + # 确保初始化代码只运行一次 + if self._initialized: + return + + self._initialized = True + + # 初始化心情状态 + self.current_mood = MoodState( + valence=0.0, + arousal=0.5, + text="平静" + ) + + # 从配置文件获取衰减率 + self.decay_rate_valence = 1 - global_config.mood_decay_rate # 愉悦度衰减率 + self.decay_rate_arousal = 1 - global_config.mood_decay_rate # 唤醒度衰减率 + + # 上次更新时间 + self.last_update = time.time() + + # 线程控制 + self._running = False + self._update_thread = None + + # 情绪词映射表 (valence, arousal) + self.emotion_map = { + 'happy': (0.8, 0.6), # 高愉悦度,中等唤醒度 + 'angry': (-0.7, 0.8), # 负愉悦度,高唤醒度 + 'sad': (-0.6, 0.3), # 负愉悦度,低唤醒度 + 'surprised': (0.4, 0.9), # 中等愉悦度,高唤醒度 + 'disgusted': (-0.8, 0.5), # 高负愉悦度,中等唤醒度 + 'fearful': (-0.7, 0.7), # 负愉悦度,高唤醒度 + 'neutral': (0.0, 0.5), # 中性愉悦度,中等唤醒度 + } + + # 情绪文本映射表 + self.mood_text_map = { + # 第一象限:高唤醒,正愉悦 + (0.5, 0.7): "兴奋", + (0.3, 0.8): "快乐", + # 第二象限:高唤醒,负愉悦 + (-0.5, 0.7): "愤怒", + (-0.3, 0.8): "焦虑", + # 第三象限:低唤醒,负愉悦 + (-0.5, 0.3): "悲伤", + (-0.3, 0.2): "疲倦", + # 第四象限:低唤醒,正愉悦 + (0.5, 0.3): "放松", + (0.3, 0.2): "平静" + } + + @classmethod + def get_instance(cls) -> 'MoodManager': + """获取MoodManager的单例实例""" + if cls._instance is None: + cls._instance = MoodManager() + return cls._instance + + def start_mood_update(self, update_interval: float = 1.0) -> None: + """ + 启动情绪更新线程 + :param update_interval: 更新间隔(秒) + """ + if self._running: + return + + self._running = True + self._update_thread = threading.Thread( + target=self._continuous_mood_update, + args=(update_interval,), + daemon=True + ) + self._update_thread.start() + + def stop_mood_update(self) -> None: + """停止情绪更新线程""" + self._running = False + if self._update_thread and self._update_thread.is_alive(): + self._update_thread.join() + + def _continuous_mood_update(self, update_interval: float) -> None: + """ + 持续更新情绪状态的线程函数 + :param update_interval: 更新间隔(秒) + """ + while self._running: + self._apply_decay() + self._update_mood_text() + time.sleep(update_interval) + + def _apply_decay(self) -> None: + """应用情绪衰减""" + current_time = time.time() + time_diff = current_time - self.last_update + + # 应用衰减公式 + self.current_mood.valence *= math.pow(1 - self.decay_rate_valence, time_diff) + self.current_mood.arousal *= math.pow(1 - self.decay_rate_arousal, time_diff) + + # 确保值在合理范围内 + self.current_mood.valence = max(-1.0, min(1.0, self.current_mood.valence)) + self.current_mood.arousal = max(0.0, min(1.0, self.current_mood.arousal)) + + self.last_update = current_time + + def update_mood_from_text(self, text: str, valence_change: float, arousal_change: float) -> None: + """根据输入文本更新情绪状态""" + + self.current_mood.valence += valence_change + self.current_mood.arousal += arousal_change + + # 限制范围 + self.current_mood.valence = max(-1.0, min(1.0, self.current_mood.valence)) + self.current_mood.arousal = max(0.0, min(1.0, self.current_mood.arousal)) + + self._update_mood_text() + + def set_mood_text(self, text: str) -> None: + """直接设置心情文本""" + self.current_mood.text = text + + def _update_mood_text(self) -> None: + """根据当前情绪状态更新文本描述""" + closest_mood = None + min_distance = float('inf') + + for (v, a), text in self.mood_text_map.items(): + distance = math.sqrt( + (self.current_mood.valence - v) ** 2 + + (self.current_mood.arousal - a) ** 2 + ) + if distance < min_distance: + min_distance = distance + closest_mood = text + + if closest_mood: + self.current_mood.text = closest_mood + + def update_mood_by_user(self, user_id: str, valence_change: float, arousal_change: float) -> None: + """根据用户ID更新情绪状态""" + + # 这里可以根据用户ID添加特定的权重或规则 + weight = 1.0 # 默认权重 + + self.current_mood.valence += valence_change * weight + self.current_mood.arousal += arousal_change * weight + + # 限制范围 + self.current_mood.valence = max(-1.0, min(1.0, self.current_mood.valence)) + self.current_mood.arousal = max(0.0, min(1.0, self.current_mood.arousal)) + + self._update_mood_text() + + def get_prompt(self) -> str: + """根据当前情绪状态生成提示词""" + + base_prompt = f"当前心情:{self.current_mood.text}。" + + # 根据情绪状态添加额外的提示信息 + if self.current_mood.valence > 0.5: + base_prompt += "你现在心情很好," + elif self.current_mood.valence < -0.5: + base_prompt += "你现在心情不太好," + + if self.current_mood.arousal > 0.7: + base_prompt += "情绪比较激动。" + elif self.current_mood.arousal < 0.3: + base_prompt += "情绪比较平静。" + + return base_prompt + + def get_current_mood(self) -> MoodState: + """获取当前情绪状态""" + return self.current_mood + + def print_mood_status(self) -> None: + """打印当前情绪状态""" + print(f"\033[1;35m[情绪状态]\033[0m 愉悦度: {self.current_mood.valence:.2f}, " + f"唤醒度: {self.current_mood.arousal:.2f}, " + f"心情: {self.current_mood.text}") + + def update_mood_from_emotion(self, emotion: str, intensity: float = 1.0) -> None: + """ + 根据情绪词更新心情状态 + :param emotion: 情绪词(如'happy', 'sad'等) + :param intensity: 情绪强度(0.0-1.0) + """ + if emotion not in self.emotion_map: + return + + valence_change, arousal_change = self.emotion_map[emotion] + + # 应用情绪强度 + valence_change *= intensity + arousal_change *= intensity + + # 更新当前情绪状态 + self.current_mood.valence += valence_change + self.current_mood.arousal += arousal_change + + # 限制范围 + self.current_mood.valence = max(-1.0, min(1.0, self.current_mood.valence)) + self.current_mood.arousal = max(0.0, min(1.0, self.current_mood.arousal)) + + self._update_mood_text() diff --git a/templete/bot_config_template.toml b/templete/bot_config_template.toml index e6246be07..95e14fc0b 100644 --- a/templete/bot_config_template.toml +++ b/templete/bot_config_template.toml @@ -37,6 +37,11 @@ max_response_length = 1024 # 麦麦回答的最大token数 build_memory_interval = 300 # 记忆构建间隔 单位秒 forget_memory_interval = 300 # 记忆遗忘间隔 单位秒 +[mood] +mood_update_interval = 1.0 # 情绪更新间隔 单位秒 +mood_decay_rate = 0.95 # 情绪衰减率 +mood_intensity_factor = 1.0 # 情绪强度因子 + [others] enable_advance_output = true # 是否启用高级输出 enable_kuuki_read = true # 是否启用读空气功能