diff --git a/src/main.py b/src/main.py index c60379208..14dc04355 100644 --- a/src/main.py +++ b/src/main.py @@ -56,8 +56,9 @@ class MainSystem: self.mood_manager.start_mood_update(update_interval=global_config.mood_update_interval) logger.success("情绪管理器启动成功") - # 检查并清除person_info冗余字段 + # 检查并清除person_info冗余字段，启动个人习惯推断 await person_info_manager.del_all_undefined_field() + asyncio.create_task(person_info_manager.personal_habit_deduction()) # 启动愿望管理器 await willing_manager.ensure_started() diff --git a/src/plugins/chat/message_buffer.py b/src/plugins/chat/message_buffer.py new file mode 100644 index 000000000..1d0291a1c --- /dev/null +++ b/src/plugins/chat/message_buffer.py @@ -0,0 +1,187 @@ +from ..person_info.person_info import person_info_manager +from src.common.logger import get_module_logger +import asyncio +from dataclasses import dataclass, field +from .message import MessageRecv +from ..message.message_base import BaseMessageInfo +import hashlib +from typing import Dict +from collections import OrderedDict +import random +import time +from ..config.config import global_config + +logger = get_module_logger("message_buffer") + +@dataclass +class CacheMessages: + message: MessageRecv + cache_determination: asyncio.Event = field(default_factory=asyncio.Event) # 判断缓冲是否产生结果 + result: str = "U" + + +class MessageBuffer: + def __init__(self): + self.buffer_pool: Dict[str, OrderedDict[str, CacheMessages]] = {} + self.lock = asyncio.Lock() + + def get_person_id_(self, platform:str, user_id:str, group_id:str): + """获取唯一id""" + group_id = group_id or "私聊" + key = f"{platform}_{user_id}_{group_id}" + return hashlib.md5(key.encode()).hexdigest() + + async def start_caching_messages(self, message:MessageRecv): + """添加消息，启动缓冲""" + if not global_config.message_buffer: + person_id = person_info_manager.get_person_id(message.message_info.user_info.platform, + message.message_info.user_info.user_id) + asyncio.create_task(self.save_message_interval(person_id, message.message_info)) + return + person_id_ = self.get_person_id_(message.message_info.platform, + message.message_info.user_info.user_id, + message.message_info.group_info.group_id) + + async with self.lock: + if person_id_ not in self.buffer_pool: + self.buffer_pool[person_id_] = OrderedDict() + + # 标记该用户之前的未处理消息 + for cache_msg in self.buffer_pool[person_id_].values(): + if cache_msg.result == "U": + cache_msg.result = "F" + cache_msg.cache_determination.set() + logger.debug(f"被新消息覆盖信息id: {cache_msg.message.message_info.message_id}") + + # 查找最近的处理成功消息(T) + recent_F_count = 0 + for msg_id in reversed(self.buffer_pool[person_id_]): + msg = self.buffer_pool[person_id_][msg_id] + if msg.result == "T": + break + elif msg.result == "F": + recent_F_count += 1 + + # 判断条件：最近T之后有超过3-5条F + if (recent_F_count >= random.randint(3, 5)): + new_msg = CacheMessages(message=message, result="T") + new_msg.cache_determination.set() + self.buffer_pool[person_id_][message.message_info.message_id] = new_msg + logger.debug(f"快速处理消息(已堆积{recent_F_count}条F): {message.message_info.message_id}") + return + + # 添加新消息 + self.buffer_pool[person_id_][message.message_info.message_id] = CacheMessages(message=message) + + # 启动3秒缓冲计时器 + person_id = person_info_manager.get_person_id(message.message_info.user_info.platform, + message.message_info.user_info.user_id) + asyncio.create_task(self.save_message_interval(person_id, message.message_info)) + asyncio.create_task(self._debounce_processor(person_id_, + message.message_info.message_id, + person_id)) + + async def _debounce_processor(self, person_id_: str, message_id: str, person_id: str): + """等待3秒无新消息""" + interval_time = await person_info_manager.get_value(person_id, "msg_interval") + if not isinstance(interval_time, (int, str)) or not str(interval_time).isdigit(): + logger.debug("debounce_processor无效的时间") + return + interval_time = max(0.5, int(interval_time) / 1000) + await asyncio.sleep(interval_time) + + async with self.lock: + if (person_id_ not in self.buffer_pool or + message_id not in self.buffer_pool[person_id_]): + logger.debug(f"消息已被清理，msgid: {message_id}") + return + + cache_msg = self.buffer_pool[person_id_][message_id] + if cache_msg.result == "U": + cache_msg.result = "T" + cache_msg.cache_determination.set() + + + async def query_buffer_result(self, message:MessageRecv) -> bool: + """查询缓冲结果，并清理""" + if not global_config.message_buffer: + return True + person_id_ = self.get_person_id_(message.message_info.platform, + message.message_info.user_info.user_id, + message.message_info.group_info.group_id) + + + async with self.lock: + user_msgs = self.buffer_pool.get(person_id_, {}) + cache_msg = user_msgs.get(message.message_info.message_id) + + if not cache_msg: + logger.debug(f"查询异常，消息不存在，msgid: {message.message_info.message_id}") + return False # 消息不存在或已清理 + + try: + await asyncio.wait_for(cache_msg.cache_determination.wait(), timeout=10) + result = cache_msg.result == "T" + + if result: + async with self.lock: # 再次加锁 + # 清理所有早于当前消息的已处理消息， 收集所有早于当前消息的F消息的processed_plain_text + keep_msgs = OrderedDict() + combined_text = [] + found = False + type = "text" + is_update = True + for msg_id, msg in self.buffer_pool[person_id_].items(): + if msg_id == message.message_info.message_id: + found = True + type = msg.message.message_segment.type + combined_text.append(msg.message.processed_plain_text) + continue + if found: + keep_msgs[msg_id] = msg + elif msg.result == "F": + # 收集F消息的文本内容 + if (hasattr(msg.message, 'processed_plain_text') + and msg.message.processed_plain_text): + if msg.message.message_segment.type == "text": + combined_text.append(msg.message.processed_plain_text) + elif msg.message.message_segment.type != "text": + is_update = False + elif msg.result == "U": + logger.debug(f"异常未处理信息id： {msg.message.message_info.message_id}") + + # 更新当前消息的processed_plain_text + if combined_text and combined_text[0] != message.processed_plain_text and is_update: + if type == "text": + message.processed_plain_text = "".join(combined_text) + logger.debug(f"整合了{len(combined_text)-1}条F消息的内容到当前消息") + elif type == "emoji": + combined_text.pop() + message.processed_plain_text = "".join(combined_text) + message.is_emoji = False + logger.debug(f"整合了{len(combined_text)-1}条F消息的内容，覆盖当前emoji消息") + + self.buffer_pool[person_id_] = keep_msgs + return result + except asyncio.TimeoutError: + logger.debug(f"查询超时消息id： {message.message_info.message_id}") + return False + + async def save_message_interval(self, person_id:str, message:BaseMessageInfo): + message_interval_list = await person_info_manager.get_value(person_id, "msg_interval_list") + now_time_ms = int(round(time.time() * 1000)) + if len(message_interval_list) < 1000: + message_interval_list.append(now_time_ms) + else: + message_interval_list.pop(0) + message_interval_list.append(now_time_ms) + data = { + "platform" : message.platform, + "user_id" : message.user_info.user_id, + "nickname" : message.user_info.user_nickname, + "konw_time" : int(time.time()) + } + await person_info_manager.update_one_field(person_id, "msg_interval_list", message_interval_list, data) + + +message_buffer = MessageBuffer() \ No newline at end of file diff --git a/src/plugins/chat_module/reasoning_chat/reasoning_chat.py b/src/plugins/chat_module/reasoning_chat/reasoning_chat.py index 0163a306e..13f055185 100644 --- a/src/plugins/chat_module/reasoning_chat/reasoning_chat.py +++ b/src/plugins/chat_module/reasoning_chat/reasoning_chat.py @@ -17,6 +17,7 @@ from ...message import UserInfo, Seg from src.common.logger import get_module_logger, CHAT_STYLE_CONFIG, LogConfig from ...chat.chat_stream import chat_manager from ...person_info.relationship_manager import relationship_manager +from ...chat.message_buffer import message_buffer # 定义日志配置 chat_config = LogConfig( @@ -143,6 +144,8 @@ class ReasoningChat: userinfo = message.message_info.user_info messageinfo = message.message_info + # 消息加入缓冲池 + await message_buffer.start_caching_messages(message) # logger.info("使用推理聊天模式") @@ -172,6 +175,17 @@ class ReasoningChat: timer2 = time.time() timing_results["记忆激活"] = timer2 - timer1 + # 查询缓冲器结果，会整合前面跳过的消息，改变processed_plain_text + buffer_result = await message_buffer.query_buffer_result(message) + if not buffer_result: + if message.message_segment.type == "text": + logger.info(f"触发缓冲，已炸飞消息：{message.processed_plain_text}") + elif message.message_segment.type == "image": + logger.info("触发缓冲，已炸飞表情包/图片") + elif message.message_segment.type == "seglist": + logger.info("触发缓冲，已炸飞消息列") + return + is_mentioned = is_mentioned_bot_in_message(message) # 计算回复意愿 diff --git a/src/plugins/chat_module/think_flow_chat/think_flow_chat.py b/src/plugins/chat_module/think_flow_chat/think_flow_chat.py index 5fc752c4c..9c2eebd94 100644 --- a/src/plugins/chat_module/think_flow_chat/think_flow_chat.py +++ b/src/plugins/chat_module/think_flow_chat/think_flow_chat.py @@ -18,6 +18,7 @@ from src.heart_flow.heartflow import heartflow from src.common.logger import get_module_logger, CHAT_STYLE_CONFIG, LogConfig from ...chat.chat_stream import chat_manager from ...person_info.relationship_manager import relationship_manager +from ...chat.message_buffer import message_buffer # 定义日志配置 chat_config = LogConfig( @@ -163,6 +164,8 @@ class ThinkFlowChat: userinfo = message.message_info.user_info messageinfo = message.message_info + # 消息加入缓冲池 + await message_buffer.start_caching_messages(message) # 创建聊天流 chat = await chat_manager.get_or_create_stream( @@ -197,8 +200,20 @@ class ThinkFlowChat: timing_results["记忆激活"] = timer2 - timer1 logger.debug(f"记忆激活: {interested_rate}") + # 查询缓冲器结果，会整合前面跳过的消息，改变processed_plain_text + buffer_result = await message_buffer.query_buffer_result(message) + if not buffer_result: + if message.message_segment.type == "text": + logger.info(f"触发缓冲，已炸飞消息：{message.processed_plain_text}") + elif message.message_segment.type == "image": + logger.info("触发缓冲，已炸飞表情包/图片") + elif message.message_segment.type == "seglist": + logger.info("触发缓冲，已炸飞消息列") + return + is_mentioned = is_mentioned_bot_in_message(message) + # 计算回复意愿 current_willing_old = willing_manager.get_willing(chat_stream=chat) # current_willing_new = (heartflow.get_subheartflow(chat.stream_id).current_state.willing - 5) / 4 diff --git a/src/plugins/config/config.py b/src/plugins/config/config.py index 680c2bafb..b1e6299d5 100644 --- a/src/plugins/config/config.py +++ b/src/plugins/config/config.py @@ -162,6 +162,7 @@ class BotConfig: emoji_chance: float = 0.2 # 发送表情包的基础概率 thinking_timeout: int = 120 # 思考时间 max_response_length: int = 1024 # 最大回复长度 + message_buffer: bool = True # 消息缓冲器 ban_words = set() ban_msgs_regex = set() @@ -505,6 +506,8 @@ class BotConfig: if config.INNER_VERSION in SpecifierSet(">=0.0.11"): config.max_response_length = msg_config.get("max_response_length", config.max_response_length) + if config.INNER_VERSION in SpecifierSet(">=1.1.4"): + config.message_buffer = msg_config.get("message_buffer", config.message_buffer) def memory(parent: dict): memory_config = parent["memory"] diff --git a/src/plugins/moods/moods.py b/src/plugins/moods/moods.py index 98fd61952..3d3feadf1 100644 --- a/src/plugins/moods/moods.py +++ b/src/plugins/moods/moods.py @@ -237,7 +237,7 @@ class MoodManager: old_arousal = self.current_mood.arousal old_mood = self.current_mood.text - valence_change *= relationship_manager.gain_coefficient[relationship_manager.positive_feedback_value] + valence_change = relationship_manager.feedback_to_mood(valence_change) # 应用情绪强度 valence_change *= intensity diff --git a/src/plugins/person_info/person_info.py b/src/plugins/person_info/person_info.py index f940c0fca..0bf15e177 100644 --- a/src/plugins/person_info/person_info.py +++ b/src/plugins/person_info/person_info.py @@ -2,8 +2,14 @@ from src.common.logger import get_module_logger from ...common.database import db import copy import hashlib -from typing import Any, Callable, Dict, TypeVar -T = TypeVar('T') # 泛型类型 +from typing import Any, Callable, Dict +import datetime +import asyncio +import numpy +import matplotlib.pyplot as plt +from pathlib import Path +import pandas as pd + """ PersonInfoManager 类方法功能摘要： @@ -15,6 +21,7 @@ PersonInfoManager 类方法功能摘要： 6. get_values - 批量获取字段值（任一字段无效则返回空字典） 7. del_all_undefined_field - 清理全集合中未定义的字段 8. get_specific_value_list - 根据指定条件，返回person_id,value字典 +9. personal_habit_deduction - 定时推断个人习惯 """ logger = get_module_logger("person_info") @@ -30,6 +37,8 @@ person_info_default = { # "impression" : None, # "gender" : Unkown, "konw_time" : 0, + "msg_interval": 3000, + "msg_interval_list": [] } # 个人信息的各项与默认值在此定义，以下处理会自动创建/补全每一项 class PersonInfoManager: @@ -108,8 +117,9 @@ class PersonInfoManager: if document and field_name in document: return document[field_name] else: - logger.debug(f"获取{person_id}的{field_name}失败，已返回默认值{person_info_default[field_name]}") - return person_info_default[field_name] + default_value = copy.deepcopy(person_info_default[field_name]) + logger.debug(f"获取{person_id}的{field_name}失败，已返回默认值{default_value}") + return default_value async def get_values(self, person_id: str, field_names: list) -> dict: """获取指定person_id文档的多个字段值，若不存在该字段，则返回该字段的全局默认值""" @@ -133,7 +143,10 @@ class PersonInfoManager: result = {} for field in field_names: - result[field] = document.get(field, person_info_default[field]) if document else person_info_default[field] + result[field] = copy.deepcopy( + document.get(field, person_info_default[field]) + if document else person_info_default[field] + ) return result @@ -209,5 +222,76 @@ class PersonInfoManager: except Exception as e: logger.error(f"数据库查询失败: {str(e)}", exc_info=True) return {} + + async def personal_habit_deduction(self): + """启动个人信息推断，每天根据一定条件推断一次""" + try: + while(1): + await asyncio.sleep(60) + current_time = datetime.datetime.now() + logger.info(f"个人信息推断启动: {current_time.strftime('%Y-%m-%d %H:%M:%S')}") -person_info_manager = PersonInfoManager() \ No newline at end of file + # "msg_interval"推断 + msg_interval_map = False + msg_interval_lists = await self.get_specific_value_list( + "msg_interval_list", + lambda x: isinstance(x, list) and len(x) >= 100 + ) + for person_id, msg_interval_list_ in msg_interval_lists.items(): + try: + time_interval = [] + for t1, t2 in zip(msg_interval_list_, msg_interval_list_[1:]): + delta = t2 - t1 + if delta < 8000 and delta > 0: # 小于8秒 + time_interval.append(delta) + + if len(time_interval) > 30: + time_interval.sort() + + # 画图(log) + msg_interval_map = True + log_dir = Path("logs/person_info") + log_dir.mkdir(parents=True, exist_ok=True) + plt.figure(figsize=(10, 6)) + + time_series = pd.Series(time_interval) + + # 绘制直方图 + plt.hist(time_series, bins=50, density=True, alpha=0.4, color='pink', label='Histogram') + + # 绘制KDE曲线（使用相同的实际数据） + time_series.plot(kind='kde', color='mediumpurple', linewidth=1, label='Density') + + plt.grid(True, alpha=0.2) + plt.xlim(0, 8000) + plt.title(f"Message Interval Distribution (User: {person_id[:8]}...)") + plt.xlabel("Interval (ms)") + plt.ylabel("Density") + plt.legend(framealpha=0.9, facecolor='white') + + img_path = log_dir / f"interval_distribution_{person_id[:8]}.png" + plt.savefig(img_path) + plt.close() + # 画图 + + filtered_intervals = [t for t in time_interval if t >= 500] + if len(filtered_intervals) > 25: + msg_interval = int(round(numpy.percentile(filtered_intervals, 80))) + await self.update_one_field(person_id, "msg_interval", msg_interval) + logger.debug(f"用户{person_id}的msg_interval已经被更新为{msg_interval}") + except Exception as e: + logger.debug(f"处理用户{person_id}msg_interval推断时出错: {str(e)}") + continue + + # 其他... + + if msg_interval_map: + logger.info("已保存分布图到: logs/person_info") + logger.info(f"个人信息推断结束: {current_time.strftime('%Y-%m-%d %H:%M:%S')}") + await asyncio.sleep(86400) + + except Exception as e: + logger.error(f"个人信息推断运行时出错: {str(e)}") + logger.exception("详细错误信息：") + +person_info_manager = PersonInfoManager() diff --git a/src/plugins/person_info/relationship_manager.py b/src/plugins/person_info/relationship_manager.py index 707dbbe51..9bbcf4e19 100644 --- a/src/plugins/person_info/relationship_manager.py +++ b/src/plugins/person_info/relationship_manager.py @@ -63,7 +63,15 @@ class RelationshipManager: value += value * mood_gain logger.info(f"当前relationship增益系数：{mood_gain:.3f}") return value - + + def feedback_to_mood(self, mood_value): + """对情绪的反馈""" + coefficient = self.gain_coefficient[abs(self.positive_feedback_value)] + if (mood_value > 0 and self.positive_feedback_value > 0 + or mood_value < 0 and self.positive_feedback_value < 0): + return mood_value*coefficient + else: + return mood_value/coefficient async def calculate_update_relationship_value(self, chat_stream: ChatStream, label: str, stance: str) -> None: """计算并变更关系值 diff --git a/template/bot_config_template.toml b/template/bot_config_template.toml index 1906b00e5..1ab846203 100644 --- a/template/bot_config_template.toml +++ b/template/bot_config_template.toml @@ -1,5 +1,5 @@ [inner] -version = "1.1.3" +version = "1.1.4" #以下是给开发人员阅读的，一般用户不需要阅读 @@ -72,6 +72,7 @@ max_context_size = 12 # 麦麦获得的上文数量，建议12，太短太长都 emoji_chance = 0.2 # 麦麦使用表情包的概率 thinking_timeout = 60 # 麦麦最长思考时间，超过这个时间的思考会放弃 max_response_length = 256 # 麦麦回答的最大token数 +message_buffer = true # 启用消息缓冲器？启用此项以解决消息的拆分问题，但会使麦麦的回复延迟 ban_words = [ # "403","张三" ] @@ -236,4 +237,4 @@ pri_out = 1.26 name = "Qwen/Qwen2.5-32B-Instruct" provider = "SILICONFLOW" pri_in = 1.26 -pri_out = 1.26 \ No newline at end of file +pri_out = 1.26