v0.5.11 添加了情绪管理器
This commit is contained in:
@@ -13,6 +13,7 @@ from .willing_manager import willing_manager
|
|||||||
from nonebot.rule import to_me
|
from nonebot.rule import to_me
|
||||||
from .bot import chat_bot
|
from .bot import chat_bot
|
||||||
from .emoji_manager import emoji_manager
|
from .emoji_manager import emoji_manager
|
||||||
|
from ..moods.moods import MoodManager # 导入情绪管理器
|
||||||
import time
|
import time
|
||||||
from ..utils.statistic import LLMStatistics
|
from ..utils.statistic import LLMStatistics
|
||||||
|
|
||||||
@@ -65,6 +66,11 @@ async def start_background_tasks():
|
|||||||
llm_stats.start()
|
llm_stats.start()
|
||||||
print("\033[1;32m[初始化]\033[0m LLM统计功能已启动")
|
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))
|
asyncio.create_task(emoji_manager.start_periodic_check(interval_MINS=global_config.EMOJI_CHECK_INTERVAL))
|
||||||
await bot_schedule.initialize()
|
await bot_schedule.initialize()
|
||||||
@@ -123,3 +129,9 @@ async def merge_memory_task():
|
|||||||
# await hippocampus.operation_merge_memory(percentage=0.1)
|
# await hippocampus.operation_merge_memory(percentage=0.1)
|
||||||
# print("\033[1;32m[记忆整合]\033[0m 记忆整合完成")
|
# 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()
|
||||||
|
|
||||||
|
|||||||
@@ -7,6 +7,7 @@ from .llm_generator import ResponseGenerator
|
|||||||
from .topic_identifier import topic_identifier
|
from .topic_identifier import topic_identifier
|
||||||
from random import random, choice
|
from random import random, choice
|
||||||
from .emoji_manager import emoji_manager # 导入表情包管理器
|
from .emoji_manager import emoji_manager # 导入表情包管理器
|
||||||
|
from ..moods.moods import MoodManager # 导入情绪管理器
|
||||||
import time
|
import time
|
||||||
import os
|
import os
|
||||||
from .cq_code import CQCode # 导入CQCode模块
|
from .cq_code import CQCode # 导入CQCode模块
|
||||||
@@ -24,6 +25,8 @@ class ChatBot:
|
|||||||
self.gpt = ResponseGenerator()
|
self.gpt = ResponseGenerator()
|
||||||
self.bot = None # bot 实例引用
|
self.bot = None # bot 实例引用
|
||||||
self._started = False
|
self._started = False
|
||||||
|
self.mood_manager = MoodManager.get_instance() # 获取情绪管理器单例
|
||||||
|
self.mood_manager.start_mood_update() # 启动情绪更新
|
||||||
|
|
||||||
self.emoji_chance = 0.2 # 发送表情包的基础概率
|
self.emoji_chance = 0.2 # 发送表情包的基础概率
|
||||||
# self.message_streams = MessageStreamContainer()
|
# self.message_streams = MessageStreamContainer()
|
||||||
@@ -192,9 +195,17 @@ class ChatBot:
|
|||||||
emotion = await self.gpt._get_emotion_tags(raw_content)
|
emotion = await self.gpt._get_emotion_tags(raw_content)
|
||||||
print(f"为 '{response}' 获取到的情感标签为:{emotion}")
|
print(f"为 '{response}' 获取到的情感标签为:{emotion}")
|
||||||
valuedict={
|
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]])
|
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)
|
# willing_manager.change_reply_willing_after_sent(event.group_id)
|
||||||
|
|
||||||
|
|||||||
@@ -57,6 +57,10 @@ class BotConfig:
|
|||||||
enable_advance_output: bool = False # 是否启用高级输出
|
enable_advance_output: bool = False # 是否启用高级输出
|
||||||
enable_kuuki_read: bool = True # 是否启用读空气功能
|
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=[
|
PROMPT_PERSONALITY=[
|
||||||
"曾经是一个学习地质的女大学生,现在学习心理学和脑科学,你会刷贴吧",
|
"曾经是一个学习地质的女大学生,现在学习心理学和脑科学,你会刷贴吧",
|
||||||
@@ -165,6 +169,12 @@ class BotConfig:
|
|||||||
config.build_memory_interval = memory_config.get("build_memory_interval", config.build_memory_interval)
|
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)
|
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:
|
if "groups" in toml_dict:
|
||||||
groups_config = toml_dict["groups"]
|
groups_config = toml_dict["groups"]
|
||||||
|
|||||||
@@ -87,7 +87,7 @@ class CQCode:
|
|||||||
elif self.type == 'face':
|
elif self.type == 'face':
|
||||||
face_id = self.params.get('id', '')
|
face_id = self.params.get('id', '')
|
||||||
# self.translated_plain_text = f"[表情{face_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':
|
elif self.type == 'forward':
|
||||||
self.translated_plain_text = await self.translate_forward()
|
self.translated_plain_text = await self.translate_forward()
|
||||||
else:
|
else:
|
||||||
|
|||||||
@@ -11,6 +11,7 @@ from random import choice
|
|||||||
import numpy as np
|
import numpy as np
|
||||||
import jieba
|
import jieba
|
||||||
from collections import Counter
|
from collections import Counter
|
||||||
|
from ..moods.moods import MoodManager
|
||||||
|
|
||||||
class PromptBuilder:
|
class PromptBuilder:
|
||||||
def __init__(self):
|
def __init__(self):
|
||||||
@@ -49,6 +50,12 @@ class PromptBuilder:
|
|||||||
|
|
||||||
#开始构建prompt
|
#开始构建prompt
|
||||||
|
|
||||||
|
|
||||||
|
#心情
|
||||||
|
mood_manager = MoodManager.get_instance()
|
||||||
|
mood_prompt = mood_manager.get_prompt()
|
||||||
|
|
||||||
|
|
||||||
#日程构建
|
#日程构建
|
||||||
current_date = time.strftime("%Y-%m-%d", time.localtime())
|
current_date = time.strftime("%Y-%m-%d", time.localtime())
|
||||||
current_time = time.strftime("%H:%M:%S", time.localtime())
|
current_time = time.strftime("%H:%M:%S", time.localtime())
|
||||||
@@ -108,13 +115,13 @@ class PromptBuilder:
|
|||||||
|
|
||||||
#激活prompt构建
|
#激活prompt构建
|
||||||
activate_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', '机器人']
|
bot_keywords = ['人机', 'bot', '机器', '入机', 'robot', '机器人']
|
||||||
is_bot = any(keyword in message_txt.lower() for keyword in bot_keywords)
|
is_bot = any(keyword in message_txt.lower() for keyword in bot_keywords)
|
||||||
if is_bot:
|
if is_bot:
|
||||||
is_bot_prompt = '有人说你是人机或者机器人,否定这一事实,攻击或反问他。'
|
is_bot_prompt = '有人说你是人机或者机器人,否定这一事实,攻击或反问他。注意,只有认为你是机器人才回复,否则不要否认'
|
||||||
else:
|
else:
|
||||||
is_bot_prompt = ''
|
is_bot_prompt = ''
|
||||||
|
|
||||||
|
|||||||
230
src/plugins/moods/moods.py
Normal file
230
src/plugins/moods/moods.py
Normal file
@@ -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()
|
||||||
@@ -37,6 +37,11 @@ max_response_length = 1024 # 麦麦回答的最大token数
|
|||||||
build_memory_interval = 300 # 记忆构建间隔 单位秒
|
build_memory_interval = 300 # 记忆构建间隔 单位秒
|
||||||
forget_memory_interval = 300 # 记忆遗忘间隔 单位秒
|
forget_memory_interval = 300 # 记忆遗忘间隔 单位秒
|
||||||
|
|
||||||
|
[mood]
|
||||||
|
mood_update_interval = 1.0 # 情绪更新间隔 单位秒
|
||||||
|
mood_decay_rate = 0.95 # 情绪衰减率
|
||||||
|
mood_intensity_factor = 1.0 # 情绪强度因子
|
||||||
|
|
||||||
[others]
|
[others]
|
||||||
enable_advance_output = true # 是否启用高级输出
|
enable_advance_output = true # 是否启用高级输出
|
||||||
enable_kuuki_read = true # 是否启用读空气功能
|
enable_kuuki_read = true # 是否启用读空气功能
|
||||||
|
|||||||
Reference in New Issue
Block a user