diff --git a/src/heart_flow/background_tasks.py b/src/heart_flow/background_tasks.py new file mode 100644 index 000000000..8538b4c3e --- /dev/null +++ b/src/heart_flow/background_tasks.py @@ -0,0 +1,204 @@ +import asyncio +import traceback +from typing import Optional, Coroutine, Callable, Any, List + +from src.common.logger import get_module_logger + +# Need manager types for dependency injection +from src.heart_flow.mai_state_manager import MaiStateManager, MaiStateInfo +from src.heart_flow.subheartflow_manager import SubHeartflowManager +from src.heart_flow.interest_logger import InterestLogger + +logger = get_module_logger("background_tasks") + + +class BackgroundTaskManager: + """管理 Heartflow 的后台周期性任务。""" + + def __init__( + self, + mai_state_info: MaiStateInfo, # Needs current state info + mai_state_manager: MaiStateManager, + subheartflow_manager: SubHeartflowManager, + interest_logger: InterestLogger, + update_interval: int, + cleanup_interval: int, + log_interval: int, + inactive_threshold: int, + ): + self.mai_state_info = mai_state_info + self.mai_state_manager = mai_state_manager + self.subheartflow_manager = subheartflow_manager + self.interest_logger = interest_logger + + # Intervals + self.update_interval = update_interval + self.cleanup_interval = cleanup_interval + self.log_interval = log_interval + self.inactive_threshold = inactive_threshold # For cleanup task + + # Task references + self._state_update_task: Optional[asyncio.Task] = None + self._cleanup_task: Optional[asyncio.Task] = None + self._logging_task: Optional[asyncio.Task] = None + self._tasks: List[Optional[asyncio.Task]] = [] # Keep track of all tasks + + async def start_tasks(self): + """启动所有后台任务""" + # 状态更新任务 + if self._state_update_task is None or self._state_update_task.done(): + self._state_update_task = asyncio.create_task( + self._run_state_update_cycle(self.update_interval), name="hf_state_update" + ) + self._tasks.append(self._state_update_task) + logger.debug(f"聊天状态更新任务已启动 间隔:{self.update_interval}s") + else: + logger.warning("状态更新任务已在运行") + + # 清理任务 + if self._cleanup_task is None or self._cleanup_task.done(): + self._cleanup_task = asyncio.create_task(self._run_cleanup_cycle(), name="hf_cleanup") + self._tasks.append(self._cleanup_task) + logger.info(f"清理任务已启动 间隔:{self.cleanup_interval}s 阈值:{self.inactive_threshold}s") + else: + logger.warning("清理任务已在运行") + + # 日志任务 + if self._logging_task is None or self._logging_task.done(): + self._logging_task = asyncio.create_task(self._run_logging_cycle(), name="hf_logging") + self._tasks.append(self._logging_task) + logger.info(f"日志任务已启动 间隔:{self.log_interval}s") + else: + logger.warning("日志任务已在运行") + + # # 初始状态检查 + # initial_state = self.mai_state_info.get_current_state() + # if initial_state != self.mai_state_info.mai_status.OFFLINE: + # logger.info(f"初始状态:{initial_state.value} 触发初始激活检查") + # asyncio.create_task(self.subheartflow_manager.activate_random_subflows_to_chat(initial_state)) + + async def stop_tasks(self): + """停止所有后台任务。 + + 该方法会: + 1. 遍历所有后台任务并取消未完成的任务 + 2. 等待所有取消操作完成 + 3. 清空任务列表 + """ + logger.info("正在停止所有后台任务...") + cancelled_count = 0 + + # 第一步：取消所有运行中的任务 + for task in self._tasks: + if task and not task.done(): + task.cancel() # 发送取消请求 + cancelled_count += 1 + + # 第二步：处理取消结果 + if cancelled_count > 0: + logger.debug(f"正在等待{cancelled_count}个任务完成取消...") + # 使用gather等待所有取消操作完成，忽略异常 + await asyncio.gather(*[t for t in self._tasks if t and t.cancelled()], return_exceptions=True) + logger.info(f"成功取消{cancelled_count}个后台任务") + else: + logger.info("没有需要取消的后台任务") + + # 第三步：清空任务列表 + self._tasks = [] # 重置任务列表 + + async def _run_periodic_loop( + self, task_name: str, interval: int, task_func: Callable[..., Coroutine[Any, Any, None]], **kwargs + ): + """周期性任务主循环""" + while True: + start_time = asyncio.get_event_loop().time() + logger.debug(f"开始执行后台任务: {task_name}") + + try: + await task_func(**kwargs) # 执行实际任务 + except asyncio.CancelledError: + logger.info(f"任务 {task_name} 已取消") + break + except Exception as e: + logger.error(f"任务 {task_name} 执行出错: {e}") + logger.error(traceback.format_exc()) + + # 计算并执行间隔等待 + elapsed = asyncio.get_event_loop().time() - start_time + sleep_time = max(0, interval - elapsed) + if sleep_time < 0.1: # 任务超时处理 + logger.warning(f"任务 {task_name} 超时执行 ({elapsed:.2f}s > {interval}s)") + await asyncio.sleep(sleep_time) + + # 非离线状态时评估兴趣 + if self.mai_state_info.get_current_state() != self.mai_state_info.mai_status.OFFLINE: + await self.subheartflow_manager.evaluate_interest_and_promote() + + logger.debug(f"任务循环结束, 当前状态: {self.mai_state_info.get_current_state().value}") + + async def _perform_state_update_work(self): + """执行状态更新工作""" + previous_status = self.mai_state_info.get_current_state() + next_state = self.mai_state_manager.check_and_decide_next_state(self.mai_state_info) + + state_changed = False + + if next_state is not None: + state_changed = self.mai_state_info.update_mai_status(next_state) + + # 处理保持离线状态的特殊情况 + if not state_changed and next_state == previous_status == self.mai_state_info.mai_status.OFFLINE: + self.mai_state_info.reset_state_timer() + logger.debug("[后台任务] 保持离线状态并重置计时器") + state_changed = True # 触发后续处理 + + if state_changed: + current_state = self.mai_state_info.get_current_state() + await self.subheartflow_manager.enforce_subheartflow_limits(current_state) + + # 状态转换处理 + if ( + previous_status == self.mai_state_info.mai_status.OFFLINE + and current_state != self.mai_state_info.mai_status.OFFLINE + ): + logger.info("[后台任务] 主状态激活，触发子流激活") + await self.subheartflow_manager.activate_random_subflows_to_chat(current_state) + elif ( + current_state == self.mai_state_info.mai_status.OFFLINE + and previous_status != self.mai_state_info.mai_status.OFFLINE + ): + logger.info("[后台任务] 主状态离线，触发子流停用") + await self.subheartflow_manager.deactivate_all_subflows() + + async def _perform_cleanup_work(self): + """执行一轮子心流清理操作。""" + flows_to_stop = self.subheartflow_manager.cleanup_inactive_subheartflows(self.inactive_threshold) + if flows_to_stop: + logger.info(f"[Background Task Cleanup] Attempting to stop {len(flows_to_stop)} inactive flows...") + stopped_count = 0 + for flow_id, reason in flows_to_stop: + if await self.subheartflow_manager.stop_subheartflow(flow_id, f"定期清理: {reason}"): + stopped_count += 1 + logger.info(f"[Background Task Cleanup] Cleanup cycle finished. Stopped {stopped_count} inactive flows.") + else: + logger.debug("[Background Task Cleanup] Cleanup cycle finished. No inactive flows found.") + + async def _perform_logging_work(self): + """执行一轮兴趣日志记录。""" + await self.interest_logger.log_interest_states() + + # --- Specific Task Runners --- # + async def _run_state_update_cycle(self, interval: int): + await self._run_periodic_loop( + task_name="State Update", interval=interval, task_func=self._perform_state_update_work + ) + + async def _run_cleanup_cycle(self): + await self._run_periodic_loop( + task_name="Subflow Cleanup", interval=self.cleanup_interval, task_func=self._perform_cleanup_work + ) + + async def _run_logging_cycle(self): + await self._run_periodic_loop( + task_name="Interest Logging", interval=self.log_interval, task_func=self._perform_logging_work + ) diff --git a/src/heart_flow/heartflow.py b/src/heart_flow/heartflow.py index 3dc9a6b08..4bf3765be 100644 --- a/src/heart_flow/heartflow.py +++ b/src/heart_flow/heartflow.py @@ -1,883 +1,116 @@ -from src.heart_flow.sub_heartflow import SubHeartflow, ChattingObservation, ChatState -from src.plugins.moods.moods import MoodManager +from src.heart_flow.sub_heartflow import SubHeartflow from src.plugins.models.utils_model import LLMRequest from src.config.config import global_config from src.plugins.schedule.schedule_generator import bot_schedule -from src.plugins.utils.prompt_builder import Prompt, global_prompt_manager -import asyncio -from src.common.logger import get_module_logger, LogConfig, HEARTFLOW_STYLE_CONFIG # 修改 -from src.individuality.individuality import Individuality -import time -import random -from typing import Dict, Any, Optional, TYPE_CHECKING -import traceback -import enum -import os # 新增 -import json # 新增 -from src.plugins.chat.chat_stream import chat_manager # 新增 - -# --- Add imports for merged dependencies --- +from src.common.logger import get_module_logger, LogConfig, HEARTFLOW_STYLE_CONFIG +from typing import Any, Optional from src.plugins.heartFC_chat.heartFC_generator import ResponseGenerator from src.do_tool.tool_use import ToolUser -from src.plugins.chat.emoji_manager import emoji_manager # Module instance from src.plugins.person_info.relationship_manager import relationship_manager # Module instance -# --- End imports --- +from src.heart_flow.mai_state_manager import MaiStateInfo, MaiStateManager +from src.heart_flow.subheartflow_manager import SubHeartflowManager +from src.heart_flow.mind import Mind +from src.heart_flow.interest_logger import InterestLogger # Import InterestLogger +from src.heart_flow.background_tasks import BackgroundTaskManager # Import BackgroundTaskManager +# --- End import --- heartflow_config = LogConfig( - # 使用海马体专用样式 console_format=HEARTFLOW_STYLE_CONFIG["console_format"], file_format=HEARTFLOW_STYLE_CONFIG["file_format"], ) logger = get_module_logger("heartflow", config=heartflow_config) -# Type hinting for circular dependency -if TYPE_CHECKING: - from src.heart_flow.sub_heartflow import SubHeartflow, ChatState # Keep SubHeartflow here too - # from src.plugins.heartFC_chat.heartFC_controler import HeartFCController # No longer needed +# Task Intervals (should be in BackgroundTaskManager or config) +CLEANUP_INTERVAL_SECONDS = 1200 +STATE_UPDATE_INTERVAL_SECONDS = 30 -def init_prompt(): - prompt = "" - prompt += "你刚刚在做的事情是：{schedule_info}\n" - prompt += "{personality_info}\n" - prompt += "你想起来{related_memory_info}。" - prompt += "刚刚你的主要想法是{current_thinking_info}。" - prompt += "你还有一些小想法，因为你在参加不同的群聊天，这是你正在做的事情：{sub_flows_info}\n" - prompt += "你现在{mood_info}。" - prompt += "现在你接下去继续思考，产生新的想法，但是要基于原有的主要想法，不要分点输出，" - prompt += "输出连贯的内心独白，不要太长，但是记得结合上述的消息，关注新内容:" - Prompt(prompt, "thinking_prompt") - prompt = "" - prompt += "{personality_info}\n" - prompt += "现在{bot_name}的想法是：{current_mind}\n" - prompt += "现在{bot_name}在qq群里进行聊天，聊天的话题如下：{minds_str}\n" - prompt += "你现在{mood_info}\n" - prompt += """现在请你总结这些聊天内容，注意关注聊天内容对原有的想法的影响，输出连贯的内心独白 - 不要太长，但是记得结合上述的消息，要记得你的人设，关注新内容:""" - Prompt(prompt, "mind_summary_prompt") - - -# --- 新增：从 interest.py 移动过来的常量 --- -LOG_DIRECTORY = "logs/interest" -HISTORY_LOG_FILENAME = "interest_history.log" -CLEANUP_INTERVAL_SECONDS = 1200 # 清理任务运行间隔 (例如：20分钟) - 保持与 interest.py 一致 -INACTIVE_THRESHOLD_SECONDS = 1200 # 不活跃时间阈值 (例如：20分钟) - 保持与 interest.py 一致 -LOG_INTERVAL_SECONDS = 3 # 日志记录间隔 (例如：3秒) - 保持与 interest.py 一致 -# --- 结束新增常量 --- - -# --- 新增：状态更新常量 --- -STATE_UPDATE_INTERVAL_SECONDS = 30 # 状态更新检查间隔（秒） -FIVE_MINUTES = 1 * 60 -FIFTEEN_MINUTES = 5 * 60 -TWENTY_MINUTES = 10 * 60 -# --- 结束新增常量 --- - - -# 新增 ChatStatus 枚举 -class MaiState(enum.Enum): - """ - 聊天状态: - OFFLINE: 不在线：回复概率极低，不会进行任何聊天 - PEEKING: 看一眼手机：回复概率较低，会进行一些普通聊天 - NORMAL_CHAT: 正常聊天：回复概率较高，会进行一些普通聊天和少量的专注聊天 - FOCUSED_CHAT: 专注聊天：回复概率极高，会进行专注聊天和少量的普通聊天 - """ - - OFFLINE = "不在线" - PEEKING = "看一眼手机" - NORMAL_CHAT = "正常聊天" - FOCUSED_CHAT = "专注聊天" - - def get_normal_chat_max_num(self): - if self == MaiState.OFFLINE: - return 0 - elif self == MaiState.PEEKING: - return 1 - elif self == MaiState.NORMAL_CHAT: - return 3 - elif self == MaiState.FOCUSED_CHAT: - return 2 - - def get_focused_chat_max_num(self): - if self == MaiState.OFFLINE: - return 0 - elif self == MaiState.PEEKING: - return 0 - elif self == MaiState.NORMAL_CHAT: - return 1 - elif self == MaiState.FOCUSED_CHAT: - return 2 - - -class MaiStateInfo: - def __init__(self): - # 使用枚举类型初始化状态，默认为正常聊天 - self.mai_status: MaiState = MaiState.OFFLINE - self.mai_status_history = [] # 历史状态，包含 状态，最后时间 - self.last_status_change_time: float = time.time() # 新增：状态最后改变时间 - self.last_5min_check_time: float = time.time() # 新增：上次5分钟规则检查时间 - - self.normal_chatting = [] - self.focused_chatting = [] - - self.mood_manager = MoodManager() - self.mood = self.mood_manager.get_prompt() - - # 新增更新聊天状态的方法 - def update_mai_status(self, new_status: MaiState): - """更新聊天状态""" - if isinstance(new_status, MaiState) and new_status != self.mai_status: # 只有状态实际改变时才更新 - self.mai_status = new_status - current_time = time.time() - self.last_status_change_time = current_time # 更新状态改变时间 - self.last_5min_check_time = current_time # 重置5分钟检查计时器 - # 将新状态和时间戳添加到历史记录 - self.mai_status_history.append((new_status, current_time)) - logger.info(f"麦麦状态更新为: {self.mai_status.value}") - elif not isinstance(new_status, MaiState): - logger.warning(f"尝试设置无效的麦麦状态: {new_status}") - # else: # 状态未改变，不处理 - # pass +# Thresholds (should be in SubHeartflowManager or config) +INACTIVE_THRESHOLD_SECONDS = 1200 +# --- End Constants --- # class Heartflow: + """主心流协调器，负责初始化并协调各个子系统: + - 状态管理 (MaiState) + - 子心流管理 (SubHeartflow) + - 思考过程 (Mind) + - 日志记录 (InterestLogger) + - 后台任务 (BackgroundTaskManager) + """ + def __init__(self): - self.current_mind = "你什么也没想" - self.past_mind = [] - self.current_state: MaiStateInfo = MaiStateInfo() + # 核心状态 + self.current_mind = "什么也没想" # 当前主心流想法 + self.past_mind = [] # 历史想法记录 + + # 状态管理相关 + self.current_state: MaiStateInfo = MaiStateInfo() # 当前状态信息 + self.mai_state_manager: MaiStateManager = MaiStateManager() # 状态决策管理器 + + # 子心流管理 + self.subheartflow_manager: SubHeartflowManager = SubHeartflowManager() # 子心流管理器 + + # LLM模型配置 self.llm_model = LLMRequest( model=global_config.llm_heartflow, temperature=0.6, max_tokens=1000, request_type="heart_flow" ) - self._subheartflows: Dict[Any, "SubHeartflow"] = {} # Update type hint + # 外部依赖模块 + self.gpt_instance = ResponseGenerator() # 响应生成器 + self.tool_user_instance = ToolUser() # 工具使用模块 + self.relationship_manager_instance = relationship_manager # 关系管理模块 - # --- Dependencies moved from HeartFCController --- - self.gpt_instance = ResponseGenerator() - self.mood_manager = ( - MoodManager.get_instance() - ) # Note: MaiStateInfo also has one, consider consolidating later if needed - self.tool_user_instance = ToolUser() - self.emoji_manager_instance = emoji_manager # Module instance - self.relationship_manager_instance = relationship_manager # Module instance - # --- End moved dependencies --- + # 子系统初始化 + self.mind: Mind = Mind(self.subheartflow_manager, self.llm_model) # 思考管理器 + self.interest_logger: InterestLogger = InterestLogger(self.subheartflow_manager) # 兴趣日志记录器 - # --- Background Task Management --- - self._history_log_file_path = os.path.join(LOG_DIRECTORY, HISTORY_LOG_FILENAME) - self._ensure_log_directory() # 初始化时确保目录存在 - self._cleanup_task: Optional[asyncio.Task] = None - self._logging_task: Optional[asyncio.Task] = None - self._state_update_task: Optional[asyncio.Task] = None # 新增：状态更新任务 - # 注意：衰减任务 (_decay_task) 不再需要，衰减在 SubHeartflow 的 InterestChatting 内部处理 - # --- End moved dependencies --- - - def _ensure_log_directory(self): # 新增方法 (从 InterestManager 移动) - """确保日志目录存在""" - # 移除 try-except 块，根据用户要求 - os.makedirs(LOG_DIRECTORY, exist_ok=True) - logger.info(f"Log directory '{LOG_DIRECTORY}' ensured.") - # except OSError as e: - # logger.error(f"Error creating log directory '{LOG_DIRECTORY}': {e}") - - async def _periodic_cleanup_task( - self, interval_seconds: int, max_age_seconds: int - ): # 新增方法 (从 InterestManager 移动和修改) - """后台清理任务的异步函数""" - while True: - await asyncio.sleep(interval_seconds) - logger.info(f"[Heartflow] 运行定期清理 (间隔: {interval_seconds}秒)...") - self.cleanup_inactive_subheartflows(max_age_seconds=max_age_seconds) # 调用 Heartflow 自己的清理方法 - - async def _periodic_log_task(self, interval_seconds: int): # 新增方法 (从 InterestManager 移动和修改) - """后台日志记录任务的异步函数 (记录所有子心流的兴趣历史数据)""" - while True: - await asyncio.sleep(interval_seconds) - try: - current_timestamp = time.time() - all_interest_states = await self.get_all_interest_states() # 获取所有子心流的兴趣状态 - - # 以追加模式打开历史日志文件 - # 移除 try-except IO 块，根据用户要求 - with open(self._history_log_file_path, "a", encoding="utf-8") as f: - count = 0 - # 创建 items 快照以安全迭代 - items_snapshot = list(all_interest_states.items()) - for stream_id, state in items_snapshot: - # 从 chat_manager 获取 group_name - group_name = stream_id # 默认值 - try: - chat_stream = chat_manager.get_stream(stream_id) - if chat_stream and chat_stream.group_info: - group_name = chat_stream.group_info.group_name - elif chat_stream and not chat_stream.group_info: # 处理私聊 - group_name = ( - f"私聊_{chat_stream.user_info.user_nickname}" - if chat_stream.user_info - else stream_id - ) - except Exception: - # 不记录警告，避免刷屏，使用默认 stream_id 即可 - # logger.warning(f"Could not get group name for stream_id {stream_id}: {e}") - pass # 静默处理 - - log_entry = { - "timestamp": round(current_timestamp, 2), - "stream_id": stream_id, - "interest_level": state.get("interest_level", 0.0), # 使用 get 获取，提供默认值 - "group_name": group_name, - "reply_probability": state.get("current_reply_probability", 0.0), # 使用 get 获取 - "is_above_threshold": state.get("is_above_threshold", False), # 使用 get 获取 - } - # 将每个条目作为单独的 JSON 行写入 - f.write(json.dumps(log_entry, ensure_ascii=False) + "\n") - count += 1 - # logger.debug(f"[Heartflow] Successfully appended {count} interest history entries to {self._history_log_file_path}") - - # except IOError as e: - # logger.error(f"[Heartflow] Error writing interest history log to {self._history_log_file_path}: {e}") - except Exception as e: # 保留对其他异常的捕获 - logger.error(f"[Heartflow] Unexpected error during periodic history logging: {e}") - logger.error(traceback.format_exc()) # 记录 traceback - - async def _periodic_state_update_task(self): - """定期检查并更新 Mai 状态""" - while True: - await asyncio.sleep(STATE_UPDATE_INTERVAL_SECONDS) - try: - current_time = time.time() - # 获取更新前的状态 - previous_status = self.current_state.mai_status - current_status = self.current_state.mai_status # 保持此行以进行后续逻辑 - time_in_current_status = current_time - self.current_state.last_status_change_time - time_since_last_5min_check = current_time - self.current_state.last_5min_check_time - next_state = None # 预设下一状态为 None - - # --- 状态转换逻辑 (保持不变) --- - # 1. 通用规则：每5分钟检查 (对于非 OFFLINE 状态) - if time_since_last_5min_check >= FIVE_MINUTES: - self.current_state.last_5min_check_time = current_time # 重置5分钟检查计时器（无论是否切换） - if current_status != MaiState.OFFLINE: - if random.random() < 0.10: # 10% 概率切换到 OFFLINE - logger.debug(f"[Heartflow State] 触发5分钟规则，从 {current_status.value} 切换到 OFFLINE") - next_state = MaiState.OFFLINE # 设置 next_state 而不是直接更新 - # self.current_state.update_mai_status(MaiState.OFFLINE) - # continue # 状态已改变，进入下一轮循环 - - # 2. 状态持续时间规则 (仅在未被5分钟规则覆盖时执行) - if next_state is None: # 仅当5分钟规则未触发切换时检查持续时间 - if current_status == MaiState.OFFLINE: - # OFFLINE 状态下，检查是否已持续5分钟 - if time_in_current_status >= FIVE_MINUTES: - weights = [35, 35, 30] - choices_list = [MaiState.PEEKING, MaiState.NORMAL_CHAT, MaiState.OFFLINE] - next_state_candidate = random.choices(choices_list, weights=weights, k=1)[0] - if next_state_candidate != MaiState.OFFLINE: - next_state = next_state_candidate - logger.debug(f"[Heartflow State] OFFLINE 持续时间达到，切换到 {next_state.value}") - else: - # 保持 OFFLINE，重置计时器以开始新的5分钟计时 - logger.debug("[Heartflow State] OFFLINE 持续时间达到，保持 OFFLINE，重置计时器") - self.current_state.last_status_change_time = current_time - self.current_state.last_5min_check_time = current_time # 保持一致 - # 显式将 next_state 设为 OFFLINE 以便后续处理 - next_state = MaiState.OFFLINE - - elif current_status == MaiState.PEEKING: - if time_in_current_status >= FIVE_MINUTES: # PEEKING 最多持续 5 分钟 - weights = [50, 30, 20] - choices_list = [MaiState.OFFLINE, MaiState.NORMAL_CHAT, MaiState.FOCUSED_CHAT] - next_state = random.choices(choices_list, weights=weights, k=1)[0] - logger.debug(f"[Heartflow State] PEEKING 持续时间达到，切换到 {next_state.value}") - - elif current_status == MaiState.NORMAL_CHAT: - if time_in_current_status >= FIFTEEN_MINUTES: # NORMAL_CHAT 最多持续 15 分钟 - weights = [50, 50] - choices_list = [MaiState.OFFLINE, MaiState.FOCUSED_CHAT] - next_state = random.choices(choices_list, weights=weights, k=1)[0] - logger.debug(f"[Heartflow State] NORMAL_CHAT 持续时间达到，切换到 {next_state.value}") - - elif current_status == MaiState.FOCUSED_CHAT: - if time_in_current_status >= TWENTY_MINUTES: # FOCUSED_CHAT 最多持续 20 分钟 - weights = [80, 20] - choices_list = [MaiState.OFFLINE, MaiState.NORMAL_CHAT] - next_state = random.choices(choices_list, weights=weights, k=1)[0] - logger.debug(f"[Heartflow State] FOCUSED_CHAT 持续时间达到，切换到 {next_state.value}") - # --- 状态转换逻辑结束 --- - - # --- 更新状态并执行相关操作 --- # - if next_state is not None: - # 检查状态是否真的改变了 - if next_state != previous_status: - logger.info(f"[Heartflow] 准备从 {previous_status.value} 转换状态到 {next_state.value}") - self.current_state.update_mai_status(next_state) - - # 在状态改变后，强制执行子心流数量限制 (保持) - await self._enforce_subheartflow_limits(next_state) - - # --- 新增逻辑：根据状态转换调整子心流 --- # - if previous_status == MaiState.OFFLINE and next_state != MaiState.OFFLINE: - logger.info("[Heartflow] 主状态从 OFFLINE 激活，尝试激活子心流到 CHAT 状态。") - await self._activate_random_subflows_to_chat(next_state) - elif next_state == MaiState.OFFLINE and previous_status != MaiState.OFFLINE: - logger.info("[Heartflow] 主状态变为 OFFLINE，停用所有子心流活动。") - await self._deactivate_all_subflows_on_offline() - # --- 结束新增逻辑 --- # - - elif next_state == MaiState.OFFLINE and previous_status == MaiState.OFFLINE: - # 如果决定保持 OFFLINE 状态（例如，因为随机选择或持续时间规则），并且之前已经是 OFFLINE - # 确保计时器被重置 (这在上面的持续时间规则中已处理，但为了清晰再次确认) - if time_in_current_status >= FIVE_MINUTES: - # 确保计时器已在上面重置，这里无需操作，只记录日志 - logger.debug("[Heartflow State] 保持 OFFLINE 状态，计时器已重置。") - pass # 无需状态转换，也无需调用激活/停用逻辑 - - # --- 如果没有确定 next_state (即没有触发任何切换规则) --- # - # logger.debug(f"[Heartflow State] 状态未改变，保持 {current_status.value}") # 减少日志噪音 - - # --- Integrated Interest Evaluation Logic (formerly in Controller loop) --- - if self.current_state.mai_status != MaiState.OFFLINE: - try: - # Use snapshot for safe iteration - subflows_snapshot = list(self._subheartflows.values()) - evaluated_count = 0 - promoted_count = 0 - - for sub_hf in subflows_snapshot: - # Double-check if subflow still exists and is in CHAT state - if sub_hf.subheartflow_id in self._subheartflows and sub_hf.chat_state.chat_status == ChatState.CHAT: - evaluated_count += 1 - if await sub_hf.should_evaluate_reply(): - stream_name = chat_manager.get_stream_name(sub_hf.subheartflow_id) or sub_hf.subheartflow_id - log_prefix = f"[{stream_name}]" - logger.info(f"{log_prefix} 兴趣概率触发，尝试将状态从 CHAT 提升到 FOCUSED") - # set_chat_state handles limit checks and HeartFChatting creation internally - await sub_hf.set_chat_state(ChatState.FOCUSED) - # Check if state actually changed (set_chat_state might block due to limits) - if sub_hf.chat_state.chat_status == ChatState.FOCUSED: - promoted_count += 1 - # else: # No need to log every non-trigger event - # logger.trace(f"[{sub_hf.subheartflow_id}] In CHAT state, but should_evaluate_reply returned False.") - - if evaluated_count > 0: - logger.debug( - f"[Heartflow Interest Eval] Evaluated {evaluated_count} CHAT flows. Promoted {promoted_count} to FOCUSED." - ) - - except Exception as e: - logger.error(f"[Heartflow] 兴趣评估任务出错: {e}") - logger.error(traceback.format_exc()) - # --- End Integrated Interest Evaluation --- - - except Exception as e: - logger.error(f"[Heartflow] 状态更新任务出错: {e}") - logger.error(traceback.format_exc()) - - logger.info(f"当前状态:{self.current_state.mai_status.value}") - - async def get_all_interest_states(self) -> Dict[str, Dict]: # <-- Make async - """获取所有活跃子心流的当前兴趣状态""" - states = {} - # 创建副本以避免在迭代时修改字典 - items_snapshot = list(self._subheartflows.items()) # Make a copy for safe iteration - tasks = [] - results = {} - - # Create tasks to get state concurrently - for stream_id, subheartflow in items_snapshot: - # Ensure subheartflow still exists before creating task - if stream_id in self._subheartflows: - tasks.append(asyncio.create_task(subheartflow.get_interest_state(), name=f"get_state_{stream_id}")) - else: - logger.warning(f"[Heartflow] Subheartflow {stream_id} disappeared before getting state.") - - # Wait for all tasks to complete - if tasks: - done, pending = await asyncio.wait(tasks, timeout=5.0) # Add a timeout - - if pending: - logger.warning(f"[Heartflow] Getting interest states timed out for {len(pending)} tasks.") - for task in pending: - task.cancel() - - for task in done: - stream_id = task.get_name().split("_")[-1] # Extract stream_id from task name - try: - result = task.result() - results[stream_id] = result - except asyncio.CancelledError: - logger.warning(f"[Heartflow] Task to get interest state for {stream_id} was cancelled (timeout).") - except Exception as e: - logger.warning(f"[Heartflow] Error getting interest state for subheartflow {stream_id}: {e}") - - return results - - def cleanup_inactive_subheartflows(self, max_age_seconds=INACTIVE_THRESHOLD_SECONDS): # 修改此方法以使用兴趣时间 - """ - 清理长时间不活跃的子心流记录 (基于兴趣交互时间) - max_age_seconds: 超过此时间未通过兴趣系统交互的将被清理 - """ - current_time = time.time() - keys_to_remove = [] - _initial_count = len(self._subheartflows) - - # 创建副本以避免在迭代时修改字典 - items_snapshot = list(self._subheartflows.items()) - - for subheartflow_id, subheartflow in items_snapshot: - should_remove = False - reason = "" - # 检查 InterestChatting 的最后交互时间 - last_interaction = subheartflow.interest_chatting.last_interaction_time - if max_age_seconds is not None and (current_time - last_interaction) > max_age_seconds: - should_remove = True - reason = ( - f"interest inactive time ({current_time - last_interaction:.0f}s) > max age ({max_age_seconds}s)" - ) - - if should_remove: - keys_to_remove.append(subheartflow_id) - stream_name = chat_manager.get_stream_name(subheartflow_id) or subheartflow_id # 获取流名称 - logger.debug(f"[Heartflow] Marking stream {stream_name} for removal. Reason: {reason}") - - # 标记子心流让其后台任务停止 (如果其后台任务还在运行) - subheartflow.should_stop = True - - if keys_to_remove: - logger.info(f"[Heartflow] 清理识别到 {len(keys_to_remove)} 个不活跃的流。") - for key in keys_to_remove: - if key in self._subheartflows: - # 尝试取消子心流的后台任务 - task_to_cancel = self._subheartflows[key].task - if task_to_cancel and not task_to_cancel.done(): - task_to_cancel.cancel() - logger.debug(f"[Heartflow] Cancelled background task for subheartflow {key}") - # 从字典中删除 - del self._subheartflows[key] - stream_name = chat_manager.get_stream_name(key) or key # 获取流名称 - logger.debug(f"[Heartflow] 移除了流: {stream_name}") - final_count = len(self._subheartflows) # 直接获取当前长度 - logger.info(f"[Heartflow] 清理完成。移除了 {len(keys_to_remove)} 个流。当前数量: {final_count}") - else: - # logger.info(f"[Heartflow] 清理完成。没有流符合移除条件。当前数量: {initial_count}") # 减少日志噪音 - pass - - async def heartflow_start_working(self): - # 启动清理任务 (使用新的 periodic_cleanup_task) - if self._cleanup_task is None or self._cleanup_task.done(): - self._cleanup_task = asyncio.create_task( - self._periodic_cleanup_task( - interval_seconds=CLEANUP_INTERVAL_SECONDS, - max_age_seconds=INACTIVE_THRESHOLD_SECONDS, - ) - ) - logger.info( - f"[Heartflow] 已创建定期清理任务。间隔: {CLEANUP_INTERVAL_SECONDS}s, 不活跃阈值: {INACTIVE_THRESHOLD_SECONDS}s" - ) - else: - logger.warning("[Heartflow] 跳过创建清理任务: 任务已在运行或存在。") - - # 启动日志任务 (使用新的 periodic_log_task) - if self._logging_task is None or self._logging_task.done(): - self._logging_task = asyncio.create_task(self._periodic_log_task(interval_seconds=LOG_INTERVAL_SECONDS)) - logger.info(f"[Heartflow] 已创建定期日志任务。间隔: {LOG_INTERVAL_SECONDS}s") - else: - logger.warning("[Heartflow] 跳过创建日志任务: 任务已在运行或存在。") - - # 新增：启动状态更新任务 - if self._state_update_task is None or self._state_update_task.done(): - self._state_update_task = asyncio.create_task(self._periodic_state_update_task()) - logger.info(f"[Heartflow] 已创建定期状态更新任务。间隔: {STATE_UPDATE_INTERVAL_SECONDS}s") - else: - logger.warning("[Heartflow] 跳过创建状态更新任务: 任务已在运行或存在。") - - # --- 新增：在启动时根据初始状态激活子心流 --- - if self.current_state.mai_status != MaiState.OFFLINE: - logger.info(f"[Heartflow] 初始状态为 {self.current_state.mai_status.value}，执行初始子心流激活检查。") - # 使用 create_task 确保它不会阻塞 heartflow_start_working 的完成 - # 传递当前状态给激活函数，以便它知道激活的限制 - asyncio.create_task(self._activate_random_subflows_to_chat(self.current_state.mai_status)) - # --- 结束新增逻辑 --- - - @staticmethod - async def _update_current_state(): - print("TODO") - - async def do_a_thinking(self): - # logger.debug("麦麦大脑袋转起来了") - - # 开始构建prompt - prompt_personality = "你" - # person - individuality = Individuality.get_instance() - - personality_core = individuality.personality.personality_core - prompt_personality += personality_core - - personality_sides = individuality.personality.personality_sides - # 检查列表是否为空 - if personality_sides: - random.shuffle(personality_sides) - prompt_personality += f",{personality_sides[0]}" - - identity_detail = individuality.identity.identity_detail - # 检查列表是否为空 - if identity_detail: - random.shuffle(identity_detail) - prompt_personality += f",{identity_detail[0]}" - - personality_info = prompt_personality - - current_thinking_info = self.current_mind - mood_info = self.current_state.mood - related_memory_info = "memory" # TODO: 替换为实际的记忆获取逻辑 - try: - sub_flows_info = await self.get_all_subheartflows_minds_summary() # 修改为调用汇总方法 - except Exception as e: - logger.error(f"[Heartflow] 获取子心流想法汇总失败: {e}") - logger.error(traceback.format_exc()) - sub_flows_info = "(获取子心流想法时出错)" # 提供默认值 - - schedule_info = bot_schedule.get_current_num_task(num=4, time_info=True) - - prompt = (await global_prompt_manager.get_prompt_async("thinking_prompt")).format( - schedule_info=schedule_info, # 使用关键字参数确保正确格式化 - personality_info=personality_info, - related_memory_info=related_memory_info, - current_thinking_info=current_thinking_info, - sub_flows_info=sub_flows_info, - mood_info=mood_info, + # 后台任务管理器 (整合所有定时任务) + self.background_task_manager: BackgroundTaskManager = BackgroundTaskManager( + mai_state_info=self.current_state, + mai_state_manager=self.mai_state_manager, + subheartflow_manager=self.subheartflow_manager, + interest_logger=self.interest_logger, + update_interval=STATE_UPDATE_INTERVAL_SECONDS, + cleanup_interval=CLEANUP_INTERVAL_SECONDS, + log_interval=3, # Example: Using value directly, ideally get from config + inactive_threshold=INACTIVE_THRESHOLD_SECONDS, ) - try: - response, reasoning_content = await self.llm_model.generate_response_async(prompt) - if not response: - logger.warning("[Heartflow] 内心独白 LLM 返回空结果。") - response = "(暂时没什么想法...)" # 提供默认想法 - - self.update_current_mind(response) # 更新主心流想法 - logger.info(f"麦麦的总体脑内状态：{self.current_mind}") - - # 更新所有子心流的主心流信息 - items_snapshot = list(self._subheartflows.items()) # 创建快照 - for _, subheartflow in items_snapshot: - subheartflow.main_heartflow_info = response - - except Exception as e: - logger.error(f"[Heartflow] 内心独白获取失败: {e}") - logger.error(traceback.format_exc()) - # 此处不返回，允许程序继续执行，但主心流想法未更新 - - def update_current_mind(self, response): - self.past_mind.append(self.current_mind) - self.current_mind = response - - async def get_all_subheartflows_minds_summary(self): # 重命名并修改 - """获取所有子心流的当前想法，并进行汇总""" - sub_minds_list = [] - # 创建快照 - items_snapshot = list(self._subheartflows.items()) - for _, subheartflow in items_snapshot: - sub_minds_list.append(subheartflow.current_mind) - - if not sub_minds_list: - return "(当前没有活跃的子心流想法)" - - minds_str = "\n".join([f"- {mind}" for mind in sub_minds_list]) # 格式化为列表 - - # 调用 LLM 进行汇总 - return await self.minds_summary(minds_str) - - async def minds_summary(self, minds_str): - """使用 LLM 汇总子心流的想法字符串""" - # 开始构建prompt - prompt_personality = "你" - individuality = Individuality.get_instance() - prompt_personality += individuality.personality.personality_core - if individuality.personality.personality_sides: - prompt_personality += f",{random.choice(individuality.personality.personality_sides)}" # 随机选一个 - if individuality.identity.identity_detail: - prompt_personality += f",{random.choice(individuality.identity.identity_detail)}" # 随机选一个 - - personality_info = prompt_personality - mood_info = self.current_state.mood - bot_name = global_config.BOT_NICKNAME # 使用全局配置中的机器人昵称 - - prompt = (await global_prompt_manager.get_prompt_async("mind_summary_prompt")).format( - personality_info=personality_info, # 使用关键字参数 - bot_name=bot_name, - current_mind=self.current_mind, - minds_str=minds_str, - mood_info=mood_info, - ) - - try: - response, reasoning_content = await self.llm_model.generate_response_async(prompt) - if not response: - logger.warning("[Heartflow] 想法汇总 LLM 返回空结果。") - return "(想法汇总失败...)" - return response - except Exception as e: - logger.error(f"[Heartflow] 想法汇总失败: {e}") - logger.error(traceback.format_exc()) - return "(想法汇总时发生错误...)" - - # --- Add helper method to count subflows by state --- # - def count_subflows_by_state(self, target_state: "ChatState") -> int: - """Counts the number of subheartflows currently in the specified state.""" - count = 0 - # Use items() directly for read-only iteration if thread safety isn't a major concern here - # Or create snapshot if modification during iteration is possible elsewhere - items_snapshot = list(self._subheartflows.items()) - for _, flow in items_snapshot: - # Check if flow still exists in the main dict in case it was removed concurrently - if flow.subheartflow_id in self._subheartflows and flow.chat_state.chat_status == target_state: - count += 1 - return count - - # --- End helper method --- # - async def create_subheartflow(self, subheartflow_id: Any) -> Optional["SubHeartflow"]: - """ - 获取或创建一个新的SubHeartflow实例。 - 创建本身不受限，因为初始状态是ABSENT。 - 限制将在状态转换时检查。 - """ + """获取或创建一个新的SubHeartflow实例 - 委托给 SubHeartflowManager""" + return await self.subheartflow_manager.create_or_get_subheartflow(subheartflow_id, self.current_state) - existing_subheartflow = self._subheartflows.get(subheartflow_id) - if existing_subheartflow: - return existing_subheartflow - - # logger.info(f"[Heartflow] 尝试创建新的 subheartflow: {subheartflow_id}") - try: - subheartflow = SubHeartflow(subheartflow_id, self) - - # 创建并初始化观察对象 - - observation = ChattingObservation(subheartflow_id) - await observation.initialize() - subheartflow.add_observation(observation) - - # 创建并存储后台任务 (SubHeartflow 自己的后台任务) - subheartflow.task = asyncio.create_task(subheartflow.subheartflow_start_working()) - logger.debug(f"[Heartflow] 为 {subheartflow_id} 创建后台任务成功，添加 observation 成功") - # 添加到管理字典 - self._subheartflows[subheartflow_id] = subheartflow - logger.info(f"[Heartflow] 添加 subheartflow {subheartflow_id} 成功") - return subheartflow - - except Exception as e: - logger.error(f"[Heartflow] 创建 subheartflow {subheartflow_id} 失败: {e}") - logger.error(traceback.format_exc()) - return None - - def get_subheartflow(self, observe_chat_id: Any) -> Optional["SubHeartflow"]: + def get_subheartflow(self, subheartflow_id: Any) -> Optional["SubHeartflow"]: """获取指定ID的SubHeartflow实例""" - return self._subheartflows.get(observe_chat_id) + return self.subheartflow_manager.get_subheartflow(subheartflow_id) def get_all_subheartflows_streams_ids(self) -> list[Any]: - """获取当前所有活跃的子心流的 ID 列表""" - return list(self._subheartflows.keys()) + """获取当前所有活跃的子心流的 ID 列表 - 委托给 SubHeartflowManager""" + return self.subheartflow_manager.get_all_subheartflows_ids() - async def _stop_subheartflow(self, subheartflow_id: Any, reason: str): - """停止并移除指定的子心流，确保 HeartFChatting 被关闭""" - if subheartflow_id in self._subheartflows: - subheartflow = self._subheartflows[subheartflow_id] - stream_name = chat_manager.get_stream_name(subheartflow_id) or subheartflow_id - logger.info(f"[Heartflow Limits] 停止子心流 {stream_name}. 原因: {reason}") + async def heartflow_start_working(self): + """启动后台任务""" + await self.background_task_manager.start_tasks() + logger.info("[Heartflow] 后台任务已启动") - # --- 新增：在取消任务和删除前，先设置状态为 ABSENT 以关闭 HeartFChatting --- - try: - if subheartflow.chat_state.chat_status != ChatState.ABSENT: - logger.debug(f"[Heartflow Limits] 将子心流 {stream_name} 状态设置为 ABSENT 以确保资源释放...") - await subheartflow.set_chat_state(ChatState.ABSENT) # 调用异步方法 - else: - logger.debug(f"[Heartflow Limits] 子心流 {stream_name} 已经是 ABSENT 状态。") - except Exception as e: - logger.error(f"[Heartflow Limits] 在停止子心流 {stream_name} 时设置状态为 ABSENT 出错: {e}") - # 即使出错，仍继续尝试停止任务和移除 - # --- 结束新增逻辑 --- + async def stop_working(self): + """停止所有任务和子心流""" + logger.info("[Heartflow] 正在停止任务和子心流...") + await self.background_task_manager.stop_tasks() + await self.subheartflow_manager.deactivate_all_subflows() + logger.info("[Heartflow] 所有任务和子心流已停止") - # 标记停止并取消任务 - subheartflow.should_stop = True - task_to_cancel = subheartflow.task - if task_to_cancel and not task_to_cancel.done(): - task_to_cancel.cancel() - logger.debug(f"[Heartflow Limits] 已取消子心流 {stream_name} 的后台任务") - - # TODO: Ensure controller.stop_heartFC_chat is called if needed - # This is now handled by subheartflow.set_chat_state(ChatState.ABSENT) called in _stop_subheartflow - # from src.plugins.heartFC_chat.heartFC_controler import HeartFCController # Local import to avoid cycle - # controller = HeartFCController.get_instance() - # if controller and controller.is_heartFC_chat_active(subheartflow_id): - # await controller.stop_heartFC_chat(subheartflow_id) - - # 从字典移除 - del self._subheartflows[subheartflow_id] - logger.debug(f"[Heartflow Limits] 已移除子心流: {stream_name}") - return True - return False - - async def _enforce_subheartflow_limits(self, current_mai_state: MaiState): - """根据当前的 MaiState 强制执行 SubHeartflow 数量限制""" - normal_limit = current_mai_state.get_normal_chat_max_num() - focused_limit = current_mai_state.get_focused_chat_max_num() - logger.debug( - f"[Heartflow Limits] 执行限制检查。当前状态: {current_mai_state.value}, Normal上限: {normal_limit}, Focused上限: {focused_limit}" + async def do_a_thinking(self): + """执行一次主心流思考过程""" + schedule_info = bot_schedule.get_current_num_task(num=4, time_info=True) + new_mind = await self.mind.do_a_thinking( + current_main_mind=self.current_mind, mai_state_info=self.current_state, schedule_info=schedule_info ) - - # 分类并统计当前 subheartflows - normal_flows = [] - focused_flows = [] - other_flows = [] # e.g., ABSENT - - # 创建快照以安全迭代 - items_snapshot = list(self._subheartflows.items()) - - for flow_id, flow in items_snapshot: - # 确保 flow 实例仍然存在 (避免在迭代期间被其他任务移除) - if flow_id not in self._subheartflows: - continue - if flow.chat_state.chat_status == ChatState.CHAT: - normal_flows.append((flow_id, flow.last_active_time)) - elif flow.chat_state.chat_status == ChatState.FOCUSED: - focused_flows.append((flow_id, flow.last_active_time)) - else: - other_flows.append((flow_id, flow.last_active_time)) - - logger.debug( - f"[Heartflow Limits] 当前计数 - Normal: {len(normal_flows)}, Focused: {len(focused_flows)}, Other: {len(other_flows)}" - ) - - stopped_count = 0 - - # 检查 Normal (CHAT) 限制 - if len(normal_flows) > normal_limit: - excess_count = len(normal_flows) - normal_limit - logger.info(f"[Heartflow Limits] 检测到 Normal (CHAT) 状态超额 {excess_count} 个。上限: {normal_limit}") - # 按 last_active_time 升序排序 (最不活跃的在前) - normal_flows.sort(key=lambda item: item[1]) - # 停止最不活跃的超额部分 - for i in range(excess_count): - flow_id_to_stop = normal_flows[i][0] - if await self._stop_subheartflow( - flow_id_to_stop, f"Normal (CHAT) 状态超出上限 ({normal_limit})，停止最不活跃的实例" - ): - stopped_count += 1 - - # 重新获取 focused_flows 列表，因为上面的停止操作可能已经改变了状态或移除了实例 - focused_flows = [] - items_snapshot_after_normal = list(self._subheartflows.items()) - for flow_id, flow in items_snapshot_after_normal: - if flow_id not in self._subheartflows: - continue # Double check - if flow.chat_state.chat_status == ChatState.FOCUSED: - focused_flows.append((flow_id, flow.last_active_time)) - - # 检查 Focused (FOCUSED) 限制 - if len(focused_flows) > focused_limit: - excess_count = len(focused_flows) - focused_limit - logger.info( - f"[Heartflow Limits] 检测到 Focused (FOCUSED) 状态超额 {excess_count} 个。上限: {focused_limit}" - ) - # 按 last_active_time 升序排序 - focused_flows.sort(key=lambda item: item[1]) - # 停止最不活跃的超额部分 - for i in range(excess_count): - flow_id_to_stop = focused_flows[i][0] - if await self._stop_subheartflow( - flow_id_to_stop, f"Focused (FOCUSED) 状态超出上限 ({focused_limit})，停止最不活跃的实例" - ): - stopped_count += 1 - - if stopped_count > 0: - logger.info( - f"[Heartflow Limits] 限制执行完成，共停止了 {stopped_count} 个子心流。当前总数: {len(self._subheartflows)}" - ) - else: - logger.debug(f"[Heartflow Limits] 限制检查完成，无需停止子心流。当前总数: {len(self._subheartflows)}") - - # --- 新增方法 --- # - async def _activate_random_subflows_to_chat(self, new_mai_state: MaiState): - """当主状态从 OFFLINE 激活时，随机选择子心流进入 CHAT 状态""" - limit = new_mai_state.get_normal_chat_max_num() - if limit <= 0: - logger.info("[Heartflow Activate] 当前状态不允许 CHAT 子心流，跳过激活。") - return - - # 使用快照进行迭代 - all_flows_snapshot = list(self._subheartflows.values()) - absent_flows = [ - flow - for flow in all_flows_snapshot - if flow.subheartflow_id in self._subheartflows and flow.chat_state.chat_status == ChatState.ABSENT - ] - - num_to_activate = min(limit, len(absent_flows)) - - if num_to_activate <= 0: - logger.info(f"[Heartflow Activate] 没有处于 ABSENT 状态的子心流可供激活至 CHAT (上限: {limit})。") - return - - logger.info( - f"[Heartflow Activate] 将随机选择 {num_to_activate} 个 (上限 {limit}) ABSENT 子心流激活至 CHAT 状态。" - ) - selected_flows = random.sample(absent_flows, num_to_activate) - - activated_count = 0 - for flow in selected_flows: - # 再次检查 flow 是否仍然存在且状态为 ABSENT (以防并发修改) - if ( - flow.subheartflow_id in self._subheartflows - and self._subheartflows[flow.subheartflow_id].chat_state.chat_status == ChatState.ABSENT - ): - stream_name = chat_manager.get_stream_name(flow.subheartflow_id) or flow.subheartflow_id - logger.debug(f"[Heartflow Activate] 正在将子心流 {stream_name} 状态设置为 CHAT。") - # 调用 set_chat_state，它内部会处理日志记录 - await flow.set_chat_state(ChatState.CHAT) - activated_count += 1 - else: - stream_name = chat_manager.get_stream_name(flow.subheartflow_id) or flow.subheartflow_id - logger.warning(f"[Heartflow Activate] 跳过激活子心流 {stream_name}，因为它不再存在或状态已改变。") - - logger.info(f"[Heartflow Activate] 完成激活，成功将 {activated_count} 个子心流设置为 CHAT 状态。") - - async def _deactivate_all_subflows_on_offline(self): - """当主状态变为 OFFLINE 时，停止所有子心流的活动并设置为 ABSENT""" - logger.info("[Heartflow Deactivate] 开始停用所有子心流...") - try: - # 使用 ID 快照进行迭代 - flow_ids_snapshot = list(self._subheartflows.keys()) - deactivated_count = 0 - - for flow_id in flow_ids_snapshot: - subflow = self._subheartflows.get(flow_id) - if not subflow: - continue # Subflow 可能在迭代过程中被清理 - - stream_name = chat_manager.get_stream_name(flow_id) or flow_id - - try: - # 设置状态为 ABSENT - if subflow.chat_state.chat_status != ChatState.ABSENT: - logger.debug(f"[Heartflow Deactivate] 正在将子心流 {stream_name} 状态设置为 ABSENT。") - # 调用 set_chat_state，它会处理日志和状态更新 - subflow.set_chat_state(ChatState.ABSENT) - deactivated_count += 1 - else: - # 如果已经是 ABSENT，则无需再次设置，但记录一下检查 - logger.trace(f"[Heartflow Deactivate] 子心流 {stream_name} 已处于 ABSENT 状态。") - - except Exception as e: - logger.error(f"[Heartflow Deactivate] 停用子心流 {stream_name} 时出错: {e}") - logger.error(traceback.format_exc()) - - logger.info( - f"[Heartflow Deactivate] 完成停用，共将 {deactivated_count} 个子心流设置为 ABSENT 状态 (不包括已是 ABSENT 的)。" - ) - except Exception as e: - logger.error(f"[Heartflow Deactivate] 停用所有子心流时出错: {e}") - logger.error(traceback.format_exc()) + self.past_mind.append(self.current_mind) + self.current_mind = new_mind + logger.info(f"麦麦的总体脑内状态更新为：{self.current_mind[:100]}...") + self.mind.update_subflows_with_main_mind(new_mind) -init_prompt() -# 创建一个全局的管理器实例 heartflow = Heartflow() diff --git a/src/heart_flow/interest_logger.py b/src/heart_flow/interest_logger.py new file mode 100644 index 000000000..d216b6974 --- /dev/null +++ b/src/heart_flow/interest_logger.py @@ -0,0 +1,137 @@ +import asyncio +import time +import json +import os +import traceback +from typing import TYPE_CHECKING, Dict, List + +from src.common.logger import get_module_logger + +# Need chat_manager to get stream names +from src.plugins.chat.chat_stream import chat_manager + +if TYPE_CHECKING: + from src.heart_flow.subheartflow_manager import SubHeartflowManager + from src.heart_flow.sub_heartflow import SubHeartflow # For type hint in get_interest_states + +logger = get_module_logger("interest_logger") + +# Consider moving log directory/filename constants here +LOG_DIRECTORY = "logs/interest" +HISTORY_LOG_FILENAME = "interest_history.log" + + +class InterestLogger: + """负责定期记录所有子心流的兴趣状态到日志文件。""" + + def __init__(self, subheartflow_manager: "SubHeartflowManager"): + self.subheartflow_manager = subheartflow_manager + self._history_log_file_path = os.path.join(LOG_DIRECTORY, HISTORY_LOG_FILENAME) + self._ensure_log_directory() + + def _ensure_log_directory(self): + """确保日志目录存在。""" + try: + os.makedirs(LOG_DIRECTORY, exist_ok=True) + logger.info(f"已确保日志目录 '{LOG_DIRECTORY}' 存在") + except OSError as e: + logger.error(f"创建日志目录 '{LOG_DIRECTORY}' 出错: {e}") + + async def get_all_interest_states(self) -> Dict[str, Dict]: + """并发获取所有活跃子心流的当前兴趣状态。""" + _states = {} + # Get snapshot from the manager + all_flows: List["SubHeartflow"] = self.subheartflow_manager.get_all_subheartflows() + tasks = [] + results = {} + + if not all_flows: + logger.debug("未找到任何子心流状态") + return results + + logger.debug(f"正在获取 {len(all_flows)} 个子心流的兴趣状态...") + for subheartflow in all_flows: + if self.subheartflow_manager.get_subheartflow(subheartflow.subheartflow_id): + tasks.append( + asyncio.create_task( + subheartflow.get_interest_state(), name=f"get_state_{subheartflow.subheartflow_id}" + ) + ) + else: + logger.warning(f"子心流 {subheartflow.subheartflow_id} 在创建任务前已消失") + + if tasks: + done, pending = await asyncio.wait(tasks, timeout=5.0) + + if pending: + logger.warning(f"获取兴趣状态超时，有 {len(pending)} 个任务未完成") + for task in pending: + task.cancel() + + for task in done: + try: + stream_id_str = task.get_name().split("get_state_")[-1] + stream_id = stream_id_str + except IndexError: + logger.error(f"无法从任务名 {task.get_name()} 中提取 stream_id") + continue + + try: + result = task.result() + results[stream_id] = result + except asyncio.CancelledError: + logger.warning(f"获取子心流 {stream_id} 兴趣状态的任务已取消(超时)", exc_info=False) + except Exception as e: + logger.warning(f"获取子心流 {stream_id} 兴趣状态出错: {e}") + + logger.debug(f"成功获取 {len(results)} 个兴趣状态") + return results + + async def log_interest_states(self): + """获取所有子心流的兴趣状态并写入日志文件。""" + logger.debug("开始定期记录兴趣状态...") + try: + current_timestamp = time.time() + all_interest_states = await self.get_all_interest_states() + + if not all_interest_states: + logger.debug("没有获取到任何兴趣状态") + return + + count = 0 + try: + with open(self._history_log_file_path, "a", encoding="utf-8") as f: + items_snapshot = list(all_interest_states.items()) + for stream_id, state in items_snapshot: + group_name = stream_id + try: + chat_stream = chat_manager.get_stream(stream_id) + if chat_stream and chat_stream.group_info: + group_name = chat_stream.group_info.group_name + elif chat_stream and not chat_stream.group_info: + group_name = ( + f"私聊_{chat_stream.user_info.user_nickname}" + if chat_stream.user_info + else stream_id + ) + except Exception as e: + logger.trace(f"无法获取 stream_id {stream_id} 的群组名: {e}") + pass + + log_entry = { + "timestamp": round(current_timestamp, 2), + "stream_id": stream_id, + "interest_level": state.get("interest_level", 0.0), + "group_name": group_name, + "reply_probability": state.get("current_reply_probability", 0.0), + "is_above_threshold": state.get("is_above_threshold", False), + } + f.write(json.dumps(log_entry, ensure_ascii=False) + "\n") + count += 1 + logger.debug(f"成功记录 {count} 条兴趣历史到 {self._history_log_file_path}") + except IOError as e: + logger.error(f"写入兴趣历史日志到 {self._history_log_file_path} 出错: {e}") + + except Exception as e: + logger.error(f"定期记录兴趣历史时发生意外错误: {e}") + logger.error(traceback.format_exc()) diff --git a/src/heart_flow/mai_state_manager.py b/src/heart_flow/mai_state_manager.py new file mode 100644 index 000000000..a15f49bc4 --- /dev/null +++ b/src/heart_flow/mai_state_manager.py @@ -0,0 +1,185 @@ +import enum +import time +import random +from typing import List, Tuple, Optional +from src.common.logger import get_module_logger +from src.plugins.moods.moods import MoodManager + +logger = get_module_logger("mai_state_manager") + + +class MaiState(enum.Enum): + """ + 聊天状态: + OFFLINE: 不在线：回复概率极低，不会进行任何聊天 + PEEKING: 看一眼手机：回复概率较低，会进行一些普通聊天 + NORMAL_CHAT: 正常聊天：回复概率较高，会进行一些普通聊天和少量的专注聊天 + FOCUSED_CHAT: 专注聊天：回复概率极高，会进行专注聊天和少量的普通聊天 + """ + + OFFLINE = "不在线" + PEEKING = "看一眼手机" + NORMAL_CHAT = "正常聊天" + FOCUSED_CHAT = "专注聊天" + + def get_normal_chat_max_num(self): + if self == MaiState.OFFLINE: + return 0 + elif self == MaiState.PEEKING: + return 1 + elif self == MaiState.NORMAL_CHAT: + return 3 + elif self == MaiState.FOCUSED_CHAT: + return 2 + + def get_focused_chat_max_num(self): + if self == MaiState.OFFLINE: + return 0 + elif self == MaiState.PEEKING: + return 0 + elif self == MaiState.NORMAL_CHAT: + return 1 + elif self == MaiState.FOCUSED_CHAT: + return 2 + + +class MaiStateInfo: + def __init__(self): + self.mai_status: MaiState = MaiState.OFFLINE + self.mai_status_history: List[Tuple[MaiState, float]] = [] # 历史状态，包含 状态，时间戳 + self.last_status_change_time: float = time.time() # 状态最后改变时间 + self.last_min_check_time: float = time.time() # 上次1分钟规则检查时间 + + # Mood management is now part of MaiStateInfo + self.mood_manager = MoodManager.get_instance() # Use singleton instance + + def update_mai_status(self, new_status: MaiState) -> bool: + """ + 更新聊天状态。 + + Args: + new_status: 新的 MaiState 状态。 + + Returns: + bool: 如果状态实际发生了改变则返回 True，否则返回 False。 + """ + if new_status != self.mai_status: + self.mai_status = new_status + current_time = time.time() + self.last_status_change_time = current_time + self.last_min_check_time = current_time # Reset 1-min check on any state change + self.mai_status_history.append((new_status, current_time)) + logger.info(f"麦麦状态更新为: {self.mai_status.value}") + return True + else: + return False + + def reset_state_timer(self): + """ + 重置状态持续时间计时器和一分钟规则检查计时器。 + 通常在状态保持不变但需要重新开始计时的情况下调用（例如，保持 OFFLINE）。 + """ + current_time = time.time() + self.last_status_change_time = current_time + self.last_min_check_time = current_time # Also reset the 1-min check timer + logger.debug("MaiStateInfo 状态计时器已重置。") + + def get_mood_prompt(self) -> str: + """获取当前的心情提示词""" + # Delegate to the internal mood manager + return self.mood_manager.get_prompt() + + def get_current_state(self) -> MaiState: + """获取当前的 MaiState""" + return self.mai_status + + +class MaiStateManager: + """管理 Mai 的整体状态转换逻辑""" + + def __init__(self): + # MaiStateManager doesn't hold the state itself, it operates on a MaiStateInfo instance. + pass + + def check_and_decide_next_state(self, current_state_info: MaiStateInfo) -> Optional[MaiState]: + """ + 根据当前状态和规则检查是否需要转换状态，并决定下一个状态。 + + Args: + current_state_info: 当前的 MaiStateInfo 实例。 + + Returns: + Optional[MaiState]: 如果需要转换，返回目标 MaiState；否则返回 None。 + """ + current_time = time.time() + current_status = current_state_info.mai_status + time_in_current_status = current_time - current_state_info.last_status_change_time + time_since_last_min_check = current_time - current_state_info.last_min_check_time + next_state: Optional[MaiState] = None + + if current_status == MaiState.OFFLINE: + logger.info("[麦麦聊天状态] 思考要不要上线看看......") + elif current_status == MaiState.PEEKING: + logger.info("[麦麦聊天状态] 思考要不要继续聊下去......") + elif current_status == MaiState.NORMAL_CHAT: + logger.info("[麦麦聊天状态] 思考要不要继续聊下去......") + elif current_status == MaiState.FOCUSED_CHAT: + logger.info("[麦麦聊天状态] 思考要不要继续聊下去......") + + # 1. 麦麦每分钟都有概率离线 + if time_since_last_min_check >= 60: + if current_status != MaiState.OFFLINE: + if random.random() < 0.03: # 3% 概率切换到 OFFLINE，20分钟有50%的概率还在线 + logger.debug(f"[麦麦聊天状态] 突然不想聊了，从 {current_status.value} 切换到 离线") + next_state = MaiState.OFFLINE + + # 2. 状态持续时间规则 (如果没有自行下线) + if next_state is None: + if current_status == MaiState.OFFLINE: + # OFFLINE 最多保持一分钟 + # 目前是一个调试值，可以修改 + if time_in_current_status >= 60: + weights = [40, 40, 20] + choices_list = [MaiState.PEEKING, MaiState.NORMAL_CHAT, MaiState.OFFLINE] + next_state_candidate = random.choices(choices_list, weights=weights, k=1)[0] + if next_state_candidate != MaiState.OFFLINE: + next_state = next_state_candidate + logger.debug(f"[麦麦聊天状态] 上线！开始 {next_state.name}") + else: + # 继续离线状态 + next_state = MaiState.OFFLINE + + elif current_status == MaiState.PEEKING: + if time_in_current_status >= 120: # PEEKING 最多持续 120 秒 + weights = [70, 20, 10] + choices_list = [MaiState.OFFLINE, MaiState.NORMAL_CHAT, MaiState.FOCUSED_CHAT] + next_state = random.choices(choices_list, weights=weights, k=1)[0] + logger.debug(f"[麦麦聊天状态] 手机看完了，接下来 {next_state.name}") + + elif current_status == MaiState.NORMAL_CHAT: + if time_in_current_status >= 300: # NORMAL_CHAT 最多持续 300 秒 + weights = [50, 50] + choices_list = [MaiState.OFFLINE, MaiState.FOCUSED_CHAT] + next_state = random.choices(choices_list, weights=weights, k=1)[0] + if next_state == MaiState.FOCUSED_CHAT: + logger.debug(f"[麦麦聊天状态] 继续深入聊天， {next_state.name}") + else: + logger.debug(f"[麦麦聊天状态] 聊完了，接下来 {next_state.name}") + + elif current_status == MaiState.FOCUSED_CHAT: + if time_in_current_status >= 600: # FOCUSED_CHAT 最多持续 600 秒 + weights = [80, 20] + choices_list = [MaiState.OFFLINE, MaiState.NORMAL_CHAT] + next_state = random.choices(choices_list, weights=weights, k=1)[0] + logger.debug(f"[麦麦聊天状态] 深入聊天结束，接下来 {next_state.name}") + + # 如果决定了下一个状态，且这个状态与当前状态不同，则返回下一个状态 + if next_state is not None and next_state != current_status: + return next_state + # 如果决定保持 OFFLINE (next_state == MaiState.OFFLINE) 且当前也是 OFFLINE， + # 并且是由于持续时间规则触发的，返回 OFFLINE 以便调用者可以重置计时器 + elif next_state == MaiState.OFFLINE and current_status == MaiState.OFFLINE and time_in_current_status >= 60: + logger.debug("[麦麦聊天状态] 决定保持 OFFLINE (持续时间规则)，返回 OFFLINE 以提示重置计时器。") + return MaiState.OFFLINE # Return OFFLINE to signal caller that timer reset might be needed + else: + return None # 没有状态转换发生或无需重置计时器 diff --git a/src/heart_flow/mind.py b/src/heart_flow/mind.py new file mode 100644 index 000000000..6ca03c212 --- /dev/null +++ b/src/heart_flow/mind.py @@ -0,0 +1,141 @@ +import traceback +from typing import TYPE_CHECKING + +from src.common.logger import get_module_logger +from src.plugins.models.utils_model import LLMRequest +from src.individuality.individuality import Individuality +from src.plugins.utils.prompt_builder import global_prompt_manager +from src.config.config import global_config + +# Need access to SubHeartflowManager to get minds and update them +if TYPE_CHECKING: + from src.heart_flow.subheartflow_manager import SubHeartflowManager + from src.heart_flow.mai_state_manager import MaiStateInfo + +logger = get_module_logger("mind") + + +class Mind: + """封装 Mai 的思考过程，包括生成内心独白和汇总想法。""" + + def __init__(self, subheartflow_manager: "SubHeartflowManager", llm_model: LLMRequest): + self.subheartflow_manager = subheartflow_manager + self.llm_model = llm_model + self.individuality = Individuality.get_instance() + # Main mind state is still managed by Heartflow for now + # self.current_mind = "你什么也没想" + # self.past_mind = [] + + async def do_a_thinking(self, current_main_mind: str, mai_state_info: "MaiStateInfo", schedule_info: str): + """ + 执行一次主心流思考过程，生成新的内心独白。 + + Args: + current_main_mind: 当前的主心流想法。 + mai_state_info: 当前的 Mai 状态信息 (用于获取 mood)。 + schedule_info: 当前的日程信息。 + + Returns: + str: 生成的新的内心独白，如果出错则返回提示信息。 + """ + logger.debug("Mind: 执行思考...") + + # --- 构建 Prompt --- # + personality_info = ( + self.individuality.get_prompt_snippet() + if hasattr(self.individuality, "get_prompt_snippet") + else self.individuality.personality.personality_core + ) + mood_info = mai_state_info.get_mood_prompt() + related_memory_info = "memory" # TODO: Implement memory retrieval + + # Get subflow minds summary via internal method + try: + sub_flows_info = await self._get_subflows_summary(current_main_mind, mai_state_info) + except Exception as e: + logger.error(f"[Mind Thinking] 获取子心流想法汇总失败: {e}") + logger.error(traceback.format_exc()) + sub_flows_info = "(获取子心流想法时出错)" + + # Format prompt + try: + prompt = (await global_prompt_manager.get_prompt_async("thinking_prompt")).format( + schedule_info=schedule_info, + personality_info=personality_info, + related_memory_info=related_memory_info, + current_thinking_info=current_main_mind, # Use passed current mind + sub_flows_info=sub_flows_info, + mood_info=mood_info, + ) + except Exception as e: + logger.error(f"[Mind Thinking] 格式化 thinking_prompt 失败: {e}") + return "(思考时格式化Prompt出错...)" + + # --- 调用 LLM --- # + try: + response, reasoning_content = await self.llm_model.generate_response_async(prompt) + if not response: + logger.warning("[Mind Thinking] 内心独白 LLM 返回空结果。") + response = "(暂时没什么想法...)" + logger.info(f"Mind: 新想法生成: {response[:100]}...") # Log truncated response + return response + except Exception as e: + logger.error(f"[Mind Thinking] 内心独白 LLM 调用失败: {e}") + logger.error(traceback.format_exc()) + return "(思考时调用LLM出错...)" + + async def _get_subflows_summary(self, current_main_mind: str, mai_state_info: "MaiStateInfo") -> str: + """获取所有活跃子心流的想法，并使用 LLM 进行汇总。""" + # 1. Get active minds from SubHeartflowManager + sub_minds_list = self.subheartflow_manager.get_active_subflow_minds() + + if not sub_minds_list: + return "(当前没有活跃的子心流想法)" + + minds_str = "\n".join([f"- {mind}" for mind in sub_minds_list]) + logger.debug(f"Mind: 获取到 {len(sub_minds_list)} 个子心流想法进行汇总。") + + # 2. Call LLM for summary + # --- 构建 Prompt --- # + personality_info = ( + self.individuality.get_prompt_snippet() + if hasattr(self.individuality, "get_prompt_snippet") + else self.individuality.personality.personality_core + ) + mood_info = mai_state_info.get_mood_prompt() + bot_name = global_config.BOT_NICKNAME + + try: + prompt = (await global_prompt_manager.get_prompt_async("mind_summary_prompt")).format( + personality_info=personality_info, + bot_name=bot_name, + current_mind=current_main_mind, # Use main mind passed for context + minds_str=minds_str, + mood_info=mood_info, + ) + except Exception as e: + logger.error(f"[Mind Summary] 格式化 mind_summary_prompt 失败: {e}") + return "(汇总想法时格式化Prompt出错...)" + + # --- 调用 LLM --- # + try: + response, reasoning_content = await self.llm_model.generate_response_async(prompt) + if not response: + logger.warning("[Mind Summary] 想法汇总 LLM 返回空结果。") + return "(想法汇总失败...)" + logger.debug(f"Mind: 子想法汇总完成: {response[:100]}...") + return response + except Exception as e: + logger.error(f"[Mind Summary] 想法汇总 LLM 调用失败: {e}") + logger.error(traceback.format_exc()) + return "(想法汇总时调用LLM出错...)" + + def update_subflows_with_main_mind(self, main_mind: str): + """触发 SubHeartflowManager 更新所有子心流的主心流信息。""" + logger.debug("Mind: 请求更新子心流的主想法信息。") + self.subheartflow_manager.update_main_mind_in_subflows(main_mind) + + +# Note: update_current_mind (managing self.current_mind and self.past_mind) +# remains in Heartflow for now, as Heartflow is the central coordinator holding the main state. +# Mind class focuses solely on the *process* of thinking and summarizing. diff --git a/src/heart_flow/sub_heartflow.py b/src/heart_flow/sub_heartflow.py index 9bbe2f9a5..7b741ac62 100644 --- a/src/heart_flow/sub_heartflow.py +++ b/src/heart_flow/sub_heartflow.py @@ -4,7 +4,7 @@ from src.plugins.moods.moods import MoodManager from src.plugins.models.utils_model import LLMRequest from src.config.config import global_config import time -from typing import Optional, List, Dict, Callable, TYPE_CHECKING +from typing import Optional, List, Dict, Callable import traceback from src.plugins.chat.utils import parse_text_timestamps import enum @@ -17,12 +17,11 @@ from src.plugins.chat.message import MessageRecv from src.plugins.chat.chat_stream import chat_manager import math from src.plugins.heartFC_chat.heartFC_chat import HeartFChatting +from src.plugins.heartFC_chat.normal_chat import NormalChat +from src.plugins.heartFC_chat.normal_chat_generator import ResponseGenerator +from src.do_tool.tool_use import ToolUser +from src.heart_flow.mai_state_manager import MaiStateInfo -# Type hinting for circular dependency -if TYPE_CHECKING: - from .heartflow import Heartflow # Import Heartflow for type hinting - from .sub_heartflow import ChatState # Keep ChatState here too? - from src.plugins.heartFC_chat.heartFC_chat import HeartFChatting # <-- Add for type hint # 定义常量 (从 interest.py 移动过来) MAX_INTEREST = 15.0 @@ -231,7 +230,7 @@ class InterestChatting: class SubHeartflow: - def __init__(self, subheartflow_id, parent_heartflow: "Heartflow"): + def __init__(self, subheartflow_id, mai_states: MaiStateInfo): """子心流初始化函数 Args: @@ -240,24 +239,26 @@ class SubHeartflow: """ # 基础属性 self.subheartflow_id = subheartflow_id - self.parent_heartflow = parent_heartflow - self.bot_name = global_config.BOT_NICKNAME # 机器人昵称 + self.chat_id = subheartflow_id + + self.mai_states = mai_states # 思维状态相关 - self.current_mind = "你什么也没想" # 当前想法 + self.current_mind = "什么也没想" # 当前想法 self.past_mind = [] # 历史想法记录 - self.main_heartflow_info = "" # 主心流信息 # 聊天状态管理 - self.chat_state: ChatStateInfo = ChatStateInfo() # 聊天状态信息 - self.interest_chatting = InterestChatting(state_change_callback=self.set_chat_state) # 兴趣聊天系统 + self.chat_state: ChatStateInfo = ChatStateInfo() # 该sub_heartflow的聊天状态信息 + self.interest_chatting = InterestChatting( + state_change_callback=self.set_chat_state + ) # 该sub_heartflow的兴趣系统 # 活动状态管理 self.last_active_time = time.time() # 最后活跃时间 - self.is_active = False # 是否活跃标志 self.should_stop = False # 停止标志 self.task: Optional[asyncio.Task] = None # 后台任务 - self.heart_fc_instance: Optional["HeartFChatting"] = None # <-- Add instance variable + self.heart_fc_instance: Optional[HeartFChatting] = None # 该sub_heartflow的HeartFChatting实例 + self.normal_chat_instance: Optional[NormalChat] = None # 该sub_heartflow的NormalChat实例 # 观察和知识系统 self.observations: List[ChattingObservation] = [] # 观察列表 @@ -270,110 +271,146 @@ class SubHeartflow: max_tokens=800, request_type="sub_heart_flow", ) + + self.gpt_instance = ResponseGenerator() # 响应生成器 + self.tool_user_instance = ToolUser() # 工具使用模块 + + + self.log_prefix = chat_manager.get_stream_name(self.subheartflow_id) or self.subheartflow_id - async def set_chat_state(self, new_state: "ChatState"): - """更新sub_heartflow的聊天状态，并管理 HeartFChatting 实例""" + async def set_chat_state(self, new_state: "ChatState", current_states_num: tuple = ()): + """更新sub_heartflow的聊天状态，并管理 HeartFChatting 和 NormalChat 实例及任务""" current_state = self.chat_state.chat_status if current_state == new_state: - logger.trace(f"[{self.subheartflow_id}] State already {current_state.value}, no change.") - return # No change needed + logger.trace(f"{self.log_prefix} 状态已为 {current_state.value}, 无需更改。") + return - log_prefix = f"[{chat_manager.get_stream_name(self.subheartflow_id) or self.subheartflow_id}]" - current_mai_state = self.parent_heartflow.current_state.mai_status + log_prefix = self.log_prefix # 使用实例属性 + current_mai_state = self.mai_states.get_current_state() - # --- Entering CHAT state --- + # --- 状态转换逻辑 --- if new_state == ChatState.CHAT: normal_limit = current_mai_state.get_normal_chat_max_num() - current_chat_count = self.parent_heartflow.count_subflows_by_state(ChatState.CHAT) + current_chat_count = current_states_num[1] - if current_chat_count >= normal_limit: + if current_chat_count >= normal_limit and current_state != ChatState.CHAT: # 仅在状态转换时检查限制 logger.debug( - f"{log_prefix} 拒绝从 {current_state.value} 转换到 CHAT。原因：CHAT 状态已达上限 ({normal_limit})。当前数量: {current_chat_count}" + f"{log_prefix} 麦麦不能从 {current_state.value} 转换到 聊天。原因：聊不过来了 ({current_chat_count}/{normal_limit})" ) - return # Block the state transition + return # 阻止状态转换 else: - logger.debug( - f"{log_prefix} 允许从 {current_state.value} 转换到 CHAT (上限: {normal_limit}, 当前: {current_chat_count})" - ) - # If transitioning out of FOCUSED, shut down HeartFChatting first + logger.debug(f"{log_prefix} 麦麦可以进入或保持 聊天 状态 ({current_chat_count}/{normal_limit})") if current_state == ChatState.FOCUSED and self.heart_fc_instance: - logger.info(f"{log_prefix} 从 FOCUSED 转换到 CHAT，正在关闭 HeartFChatting...") - await self.heart_fc_instance.shutdown() + logger.info(f"{log_prefix} 麦麦不再专注聊天，转为随便水水...") + await self.heart_fc_instance.shutdown() # 正确关闭 HeartFChatting self.heart_fc_instance = None + + chat_stream = chat_manager.get_stream(self.chat_id) + self.normal_chat_instance = NormalChat( + chat_stream=chat_stream, + interest_dict=self.get_interest_dict() + ) + await self.normal_chat_instance.start_monitoring_interest() + # NormalChat 启动/停止逻辑将在下面处理 - # --- Entering FOCUSED state --- elif new_state == ChatState.FOCUSED: focused_limit = current_mai_state.get_focused_chat_max_num() - current_focused_count = self.parent_heartflow.count_subflows_by_state(ChatState.FOCUSED) + current_focused_count = current_states_num[2] - if current_focused_count >= focused_limit: + if current_focused_count >= focused_limit and current_state != ChatState.FOCUSED: # 仅在状态转换时检查限制 logger.debug( - f"{log_prefix} 拒绝从 {current_state.value} 转换到 FOCUSED。原因：FOCUSED 状态已达上限 ({focused_limit})。当前数量: {current_focused_count}" + f"{log_prefix} 麦麦不能从 {current_state.value} 转换到 专注的聊天，原因：聊不过来了。({current_focused_count}/{focused_limit})" ) - return # Block the state transition + return # 阻止状态转换 else: - logger.debug( - f"{log_prefix} 允许从 {current_state.value} 转换到 FOCUSED (上限: {focused_limit}, 当前: {current_focused_count})" - ) + logger.debug(f"{log_prefix} 麦麦可以进入或保持 专注聊天 状态 ({current_focused_count}/{focused_limit})") if not self.heart_fc_instance: - logger.info(f"{log_prefix} 状态转为 FOCUSED，创建并初始化 HeartFChatting 实例...") + logger.info(f"{log_prefix} 麦麦准备开始专注聊天...") try: + await self.normal_chat_instance.stop_monitoring_interest() + self.clear_interest_dict() + + logger.info(f"{log_prefix} 停止 NormalChat 兴趣监控成功。") + except Exception as e: + logger.error(f"{log_prefix} 停止 NormalChat 兴趣监控时出错: {e}") + logger.error(traceback.format_exc()) + try: + self.heart_fc_instance = HeartFChatting( - chat_id=self.subheartflow_id, - gpt_instance=self.parent_heartflow.gpt_instance, - tool_user_instance=self.parent_heartflow.tool_user_instance, - emoji_manager_instance=self.parent_heartflow.emoji_manager_instance, + chat_id=self.chat_id, + gpt_instance=self.gpt_instance, + tool_user_instance=self.tool_user_instance, ) - # Initialize and potentially start the loop via add_time if await self.heart_fc_instance._initialize(): - # Give it an initial time boost to start the loop - await self.heart_fc_instance.add_time() - logger.info(f"{log_prefix} HeartFChatting 实例已创建并启动。") + await self.heart_fc_instance.add_time() # 初始化成功后添加初始时间 + logger.info(f"{log_prefix} 麦麦已成功进入专注聊天模式。") else: logger.error( - f"{log_prefix} HeartFChatting 实例初始化失败，状态回滚到 {current_state.value}" + f"{log_prefix} 麦麦不能专注聊天，因为 HeartFChatting 初始化失败了，状态回滚到 {current_state.value}" ) self.heart_fc_instance = None - return # Prevent state change if HeartFChatting fails to init + return # 阻止进入 FOCUSED 状态 + except Exception as e: - logger.error(f"{log_prefix} 创建 HeartFChatting 实例时出错: {e}") + logger.error(f"{log_prefix} 创建麦麦专注聊天实例时出错: {e}") logger.error(traceback.format_exc()) self.heart_fc_instance = None - return # Prevent state change on error + return # 创建实例异常，阻止进入 FOCUSED 状态 else: - logger.warning(f"{log_prefix} 尝试进入 FOCUSED 状态，但 HeartFChatting 实例已存在。") + # 已经是 FOCUSED 状态，或者 heart_fc_instance 已存在但未运行（不太可能） + if not self.heart_fc_instance._loop_active: + logger.warning(f"{log_prefix} HeartFChatting 实例存在但未激活，尝试重新激活...") + await self.heart_fc_instance.add_time() # 尝试添加时间以激活循环 + else: + logger.debug(f"{log_prefix} 麦麦已经在专注聊天中。") + # NormalChat 启动/停止逻辑将在下面处理 - # --- Entering ABSENT state (or any state other than FOCUSED) --- - elif current_state == ChatState.FOCUSED and self.heart_fc_instance: - logger.info(f"{log_prefix} 从 FOCUSED 转换到 {new_state.value}，正在关闭 HeartFChatting...") - await self.heart_fc_instance.shutdown() - self.heart_fc_instance = None + elif new_state == ChatState.ABSENT: + if current_state == ChatState.FOCUSED and self.heart_fc_instance: + logger.info(f"{log_prefix} 麦麦离开专注的聊天，撤退了.....") + await self.heart_fc_instance.shutdown() # 正确关闭 HeartFChatting + self.heart_fc_instance = None + # NormalChat 启动/停止逻辑将在下面处理 - # --- Update state and timestamp if transition is allowed --- # 更新状态必须放在所有检查和操作之后 + # --- 更新状态和最后活动时间 (先更新状态，再根据新状态管理任务)--- self.chat_state.chat_status = new_state self.last_active_time = time.time() - logger.info(f"{log_prefix} 聊天状态从 {current_state.value} 变更为 {new_state.value}") + logger.info(f"{log_prefix} 麦麦的聊天状态从 {current_state.value} 变更为 {new_state.value}") + + # --- 根据新的状态管理 NormalChat 的监控任务 --- + if self.normal_chat_instance: + try: + if new_state == ChatState.ABSENT: + logger.info(f"{log_prefix} 状态变为 ABSENT，停止 NormalChat 兴趣监控...") + await self.normal_chat_instance.stop_monitoring_interest() + else: # CHAT or FOCUSED + logger.info(f"{log_prefix} 状态变为 {new_state.value}，启动或确认 NormalChat 兴趣监控...") + await self.normal_chat_instance.start_monitoring_interest() + except Exception as e: + logger.error(f"{log_prefix} 管理 NormalChat 监控任务时出错 (新状态: {new_state.value}): {e}") + logger.error(traceback.format_exc()) + else: + logger.warning(f"{log_prefix} NormalChat 实例不可用，无法管理其监控任务。") + async def subheartflow_start_working(self): - while True: - if self.should_stop: - logger.info(f"子心流 {self.subheartflow_id} 被标记为停止，正在退出后台任务...") - break + """启动子心流的后台任务 + + 功能说明: + - 负责子心流的主要后台循环 + - 每30秒检查一次停止标志 + """ + logger.info(f"{self.log_prefix} 子心流开始工作...") + + while not self.should_stop: + # 主循环保持简单，只做状态检查 + await asyncio.sleep(30) # 30秒检查一次停止标志 + + logger.info(f"{self.log_prefix} 子心流后台任务已停止。") - # await asyncio.sleep(global_config.sub_heart_flow_update_interval) - await asyncio.sleep(10) - async def ensure_observed(self): - observation = self._get_primary_observation() - if observation: - try: - await observation.observe() - logger.trace(f"[{self.subheartflow_id}] Observation updated before thinking.") - except Exception as e: - logger.error(f"[{self.subheartflow_id}] Error during pre-thinking observation: {e}") - logger.error(traceback.format_exc()) async def do_thinking_before_reply( self, @@ -410,7 +447,7 @@ class SubHeartflow: extra_info_prompt = "无工具信息。\n" individuality = Individuality.get_instance() - prompt_personality = f"你的名字是{self.bot_name}，你" + prompt_personality = f"你的名字是{individuality.bot_nickname}，你" prompt_personality += individuality.personality.personality_core if individuality.personality.personality_sides: @@ -441,7 +478,7 @@ class SubHeartflow: prompt = (await global_prompt_manager.get_prompt_async("sub_heartflow_prompt_before")).format( extra_info=extra_info_prompt, prompt_personality=prompt_personality, - bot_name=self.bot_name, + bot_name=individuality.bot_nickname, current_thinking_info=current_thinking_info, time_now=time_now, chat_observe_info=chat_observe_info, @@ -510,6 +547,54 @@ class SubHeartflow: def get_interest_dict(self) -> Dict[str, tuple[MessageRecv, float, bool]]: return self.interest_chatting.interest_dict + + def clear_interest_dict(self): + self.interest_chatting.interest_dict.clear() + async def shutdown(self): + """安全地关闭子心流及其管理的任务""" + if self.should_stop: + logger.info(f"{self.log_prefix} 子心流已在关闭过程中。") + return + + logger.info(f"{self.log_prefix} 开始关闭子心流...") + self.should_stop = True # 标记为停止，让后台任务退出 + + # 停止 NormalChat 监控 (保持调用，确保清理) + if self.normal_chat_instance: + logger.info(f"{self.log_prefix} 停止 NormalChat 监控任务 (Shutdown)...") + try: + await self.normal_chat_instance.stop_monitoring_interest() + except Exception as e: + logger.error(f"{self.log_prefix} 停止 NormalChat 监控任务时出错 (Shutdown): {e}") + logger.error(traceback.format_exc()) + + # 停止 HeartFChatting (如果存在且正在运行) + if self.heart_fc_instance: + logger.info(f"{self.log_prefix} 关闭 HeartFChatting 实例 (Shutdown)...") + try: + await self.heart_fc_instance.shutdown() + except Exception as e: + logger.error(f"{self.log_prefix} 关闭 HeartFChatting 实例时出错 (Shutdown): {e}") + logger.error(traceback.format_exc()) + self.heart_fc_instance = None # 清理实例引用 + + # 取消可能存在的旧后台任务 (self.task) + if self.task and not self.task.done(): + logger.info(f"{self.log_prefix} 取消子心流主任务 (Shutdown)...") + self.task.cancel() + try: + await asyncio.wait_for(self.task, timeout=1.0) # 给点时间响应取消 + except asyncio.CancelledError: + logger.info(f"{self.log_prefix} 子心流主任务已取消 (Shutdown)。") + except asyncio.TimeoutError: + logger.warning(f"{self.log_prefix} 等待子心流主任务取消超时 (Shutdown)。") + except Exception as e: + logger.error(f"{self.log_prefix} 等待子心流主任务取消时发生错误 (Shutdown): {e}") + + self.task = None # 清理任务引用 + self.chat_state.chat_status = ChatState.ABSENT # 状态重置为不参与 + + logger.info(f"{self.log_prefix} 子心流关闭完成。") init_prompt() diff --git a/src/heart_flow/subheartflow_manager.py b/src/heart_flow/subheartflow_manager.py new file mode 100644 index 000000000..4a3519976 --- /dev/null +++ b/src/heart_flow/subheartflow_manager.py @@ -0,0 +1,379 @@ +import asyncio +import time +import random +from typing import Dict, Any, Optional, List + +# 导入日志模块 +from src.common.logger import get_module_logger + +# 导入聊天流管理模块 +from src.plugins.chat.chat_stream import chat_manager + +# 导入心流相关类 +from src.heart_flow.sub_heartflow import SubHeartflow, ChatState +from src.heart_flow.mai_state_manager import MaiState, MaiStateInfo +from .observation import ChattingObservation + +# 初始化日志记录器 +logger = get_module_logger("subheartflow_manager") + +# 子心流管理相关常量 +INACTIVE_THRESHOLD_SECONDS = 1200 # 子心流不活跃超时时间(秒) + + +class SubHeartflowManager: + """管理所有活跃的 SubHeartflow 实例。""" + + def __init__(self): + self.subheartflows: Dict[Any, "SubHeartflow"] = {} + self._lock = asyncio.Lock() # 用于保护 self.subheartflows 的访问 + + def get_all_subheartflows(self) -> List["SubHeartflow"]: + """获取所有当前管理的 SubHeartflow 实例列表 (快照)。""" + return list(self.subheartflows.values()) + + def get_all_subheartflows_ids(self) -> List[Any]: + """获取所有当前管理的 SubHeartflow ID 列表。""" + return list(self.subheartflows.keys()) + + def get_subheartflow(self, subheartflow_id: Any) -> Optional["SubHeartflow"]: + """获取指定 ID 的 SubHeartflow 实例。""" + # 注意：这里没有加锁，假设读取操作相对安全或在已知上下文中调用 + # 如果并发写操作很多，get 也应该加锁 + subflow = self.subheartflows.get(subheartflow_id) + if subflow: + subflow.last_active_time = time.time() # 获取时更新活动时间 + return subflow + + async def create_or_get_subheartflow( + self, subheartflow_id: Any, mai_states: MaiStateInfo + ) -> Optional["SubHeartflow"]: + """获取或创建指定ID的子心流实例 + + Args: + subheartflow_id: 子心流唯一标识符 + mai_states: 当前麦麦状态信息 + + Returns: + 成功返回SubHeartflow实例，失败返回None + """ + async with self._lock: + # 检查是否已存在该子心流 + if subheartflow_id in self.subheartflows: + subflow = self.subheartflows[subheartflow_id] + if subflow.should_stop: + logger.warning(f"尝试获取已停止的子心流 {subheartflow_id}，正在重新激活") + subflow.should_stop = False # 重置停止标志 + + subflow.last_active_time = time.time() # 更新活跃时间 + logger.debug(f"获取到已存在的子心流: {subheartflow_id}") + return subflow + + # 创建新的子心流实例 + logger.info(f"子心流 {subheartflow_id} 不存在，正在创建...") + try: + # 初始化子心流 + new_subflow = SubHeartflow(subheartflow_id, mai_states) + + # 添加聊天观察者 + observation = ChattingObservation(chat_id=subheartflow_id) + new_subflow.add_observation(observation) + + # 注册子心流 + self.subheartflows[subheartflow_id] = new_subflow + logger.info(f"子心流 {subheartflow_id} 创建成功") + + # 启动后台任务 + asyncio.create_task(new_subflow.subheartflow_start_working()) + + return new_subflow + except Exception as e: + logger.error(f"创建子心流 {subheartflow_id} 失败: {e}", exc_info=True) + return None + + async def stop_subheartflow(self, subheartflow_id: Any, reason: str) -> bool: + """停止指定的子心流并清理资源""" + subheartflow = self.subheartflows.get(subheartflow_id) + if not subheartflow: + return False + + stream_name = chat_manager.get_stream_name(subheartflow_id) or subheartflow_id + logger.info(f"[子心流管理] 正在停止 {stream_name}, 原因: {reason}") + + try: + # 设置状态为ABSENT释放资源 + if subheartflow.chat_state.chat_status != ChatState.ABSENT: + logger.debug(f"[子心流管理] 设置 {stream_name} 状态为ABSENT") + states_num =(self.count_subflows_by_state(ChatState.ABSENT), + self.count_subflows_by_state(ChatState.CHAT), + self.count_subflows_by_state(ChatState.FOCUSED)) + await subheartflow.set_chat_state(ChatState.ABSENT, states_num) + else: + logger.debug(f"[子心流管理] {stream_name} 已是ABSENT状态") + except Exception as e: + logger.error(f"[子心流管理] 设置ABSENT状态失败: {e}") + + # 停止子心流内部循环 + subheartflow.should_stop = True + + # 取消后台任务 + task = subheartflow.task + if task and not task.done(): + task.cancel() + logger.debug(f"[子心流管理] 已取消 {stream_name} 的后台任务") + + # 从管理字典中移除 + if subheartflow_id in self.subheartflows: + del self.subheartflows[subheartflow_id] + logger.debug(f"[子心流管理] 已移除 {stream_name}") + return True + else: + logger.warning(f"[子心流管理] {stream_name} 已被提前移除") + return False + + def cleanup_inactive_subheartflows(self, max_age_seconds=INACTIVE_THRESHOLD_SECONDS): + """识别并返回需要清理的不活跃子心流(id, 原因)""" + current_time = time.time() + flows_to_stop = [] + + for subheartflow_id, subheartflow in list(self.subheartflows.items()): + # 只检查有interest_chatting的子心流 + if hasattr(subheartflow, "interest_chatting") and subheartflow.interest_chatting: + last_interact = subheartflow.interest_chatting.last_interaction_time + if max_age_seconds and (current_time - last_interact) > max_age_seconds: + reason = f"不活跃时间({current_time - last_interact:.0f}s) > 阈值({max_age_seconds}s)" + name = chat_manager.get_stream_name(subheartflow_id) or subheartflow_id + logger.debug(f"[清理] 标记 {name} 待移除: {reason}") + flows_to_stop.append((subheartflow_id, reason)) + + if flows_to_stop: + logger.info(f"[清理] 发现 {len(flows_to_stop)} 个不活跃子心流") + return flows_to_stop + + async def enforce_subheartflow_limits(self, current_mai_state: MaiState): + """根据主状态限制停止超额子心流(优先停不活跃的)""" + normal_limit = current_mai_state.get_normal_chat_max_num() + focused_limit = current_mai_state.get_focused_chat_max_num() + logger.debug(f"[限制] 状态:{current_mai_state.value}, 普通限:{normal_limit}, 专注限:{focused_limit}") + + # 分类统计当前子心流 + normal_flows = [] + focused_flows = [] + for flow_id, flow in list(self.subheartflows.items()): + if flow.chat_state.chat_status == ChatState.CHAT: + normal_flows.append((flow_id, getattr(flow, "last_active_time", 0))) + elif flow.chat_state.chat_status == ChatState.FOCUSED: + focused_flows.append((flow_id, getattr(flow, "last_active_time", 0))) + + logger.debug(f"[限制] 当前数量 - 普通:{len(normal_flows)}, 专注:{len(focused_flows)}") + stopped = 0 + + # 处理普通聊天超额 + if len(normal_flows) > normal_limit: + excess = len(normal_flows) - normal_limit + logger.info(f"[限制] 普通聊天超额({len(normal_flows)}>{normal_limit}), 停止{excess}个") + normal_flows.sort(key=lambda x: x[1]) + for flow_id, _ in normal_flows[:excess]: + if await self.stop_subheartflow(flow_id, f"普通聊天超额(限{normal_limit})"): + stopped += 1 + + # 处理专注聊天超额(需重新统计) + focused_flows = [ + (fid, t) + for fid, f in list(self.subheartflows.items()) + if (t := getattr(f, "last_active_time", 0)) and f.chat_state.chat_status == ChatState.FOCUSED + ] + if len(focused_flows) > focused_limit: + excess = len(focused_flows) - focused_limit + logger.info(f"[限制] 专注聊天超额({len(focused_flows)}>{focused_limit}), 停止{excess}个") + focused_flows.sort(key=lambda x: x[1]) + for flow_id, _ in focused_flows[:excess]: + if await self.stop_subheartflow(flow_id, f"专注聊天超额(限{focused_limit})"): + stopped += 1 + + if stopped: + logger.info(f"[限制] 已停止{stopped}个子心流, 剩余:{len(self.subheartflows)}") + else: + logger.debug(f"[限制] 无需停止, 当前总数:{len(self.subheartflows)}") + + async def activate_random_subflows_to_chat(self, current_mai_state: MaiState): + """主状态激活时，随机选择ABSENT子心流进入CHAT状态""" + limit = current_mai_state.get_normal_chat_max_num() + if limit <= 0: + logger.info("[激活] 当前状态不允许CHAT子心流") + return + + # 获取所有ABSENT状态的子心流 + absent_flows = [flow for flow in self.subheartflows.values() if flow.chat_state.chat_status == ChatState.ABSENT] + + num_to_activate = min(limit, len(absent_flows)) + if num_to_activate <= 0: + logger.info(f"[激活] 无可用ABSENT子心流(限额:{limit}, 可用:{len(absent_flows)})") + return + + logger.info(f"[激活] 随机选择{num_to_activate}个ABSENT子心流进入CHAT状态") + activated_count = 0 + + for flow in random.sample(absent_flows, num_to_activate): + flow_id = flow.subheartflow_id + stream_name = chat_manager.get_stream_name(flow_id) or flow_id + + if flow_id not in self.subheartflows: + logger.warning(f"[激活] 跳过{stream_name}, 子心流已不存在") + continue + + logger.debug(f"[激活] 正在激活子心流{stream_name}") + + states_num =(self.count_subflows_by_state(ChatState.ABSENT), + self.count_subflows_by_state(ChatState.CHAT), + self.count_subflows_by_state(ChatState.FOCUSED)) + + await flow.set_chat_state(ChatState.CHAT, states_num) + + if flow.chat_state.chat_status == ChatState.CHAT: + activated_count += 1 + else: + logger.warning(f"[激活] {stream_name}状态设置失败") + + logger.info(f"[激活] 完成, 成功激活{activated_count}个子心流") + + async def deactivate_all_subflows(self): + """停用所有子心流(主状态变为OFFLINE时调用)""" + logger.info("[停用] 开始停用所有子心流") + flow_ids = list(self.subheartflows.keys()) + + if not flow_ids: + logger.info("[停用] 无活跃子心流") + return + + stopped_count = 0 + for flow_id in flow_ids: + if await self.stop_subheartflow(flow_id, "主状态离线"): + stopped_count += 1 + + logger.info(f"[停用] 完成, 尝试停止{len(flow_ids)}个, 成功{stopped_count}个") + + async def evaluate_interest_and_promote(self): + """评估CHAT状态的子心流兴趣度，满足条件则提升到FOCUSED状态""" + logger.debug("[子心流管理器] 开始兴趣评估周期...") + evaluated_count = 0 + promoted_count = 0 + + # 使用快照安全遍历 + subflows_snapshot = list(self.subheartflows.values()) + + for sub_hf in subflows_snapshot: + flow_id = sub_hf.subheartflow_id + if flow_id in self.subheartflows and self.subheartflows[flow_id].chat_state.chat_status == ChatState.CHAT: + evaluated_count += 1 + stream_name = chat_manager.get_stream_name(flow_id) or flow_id + log_prefix = f"[{stream_name}]" + + should_promote = await sub_hf.should_evaluate_reply() + if should_promote: + logger.info(f"{log_prefix} 兴趣评估触发升级: CHAT -> FOCUSED") + states_num =(self.count_subflows_by_state(ChatState.ABSENT), + self.count_subflows_by_state(ChatState.CHAT), + self.count_subflows_by_state(ChatState.FOCUSED)) + await sub_hf.set_chat_state(ChatState.FOCUSED, states_num) + if ( + self.subheartflows.get(flow_id) + and self.subheartflows[flow_id].chat_state.chat_status == ChatState.FOCUSED + ): + promoted_count += 1 + logger.debug(f"{log_prefix} 成功升级到FOCUSED状态") + else: + logger.info(f"{log_prefix} 升级FOCUSED可能被限制阻止") + + if evaluated_count > 0: + logger.debug(f"[子心流管理器] 评估完成. 评估{evaluated_count}个CHAT流, 升级{promoted_count}个到FOCUSED") + + def count_subflows_by_state(self, state: ChatState) -> int: + """统计指定状态的子心流数量 + + Args: + state: 要统计的聊天状态枚举值 + + Returns: + int: 处于该状态的子心流数量 + """ + count = 0 + # 遍历所有子心流实例 + for subheartflow in self.subheartflows.values(): + # 检查子心流状态是否匹配 + if subheartflow.chat_state.chat_status == state: + count += 1 + return count + + def get_active_subflow_minds(self) -> List[str]: + """获取所有活跃(非ABSENT)子心流的当前想法 + + 返回: + List[str]: 包含所有活跃子心流当前想法的列表 + """ + minds = [] + for subheartflow in self.subheartflows.values(): + # 检查子心流是否活跃(非ABSENT状态) + if subheartflow.chat_state.chat_status != ChatState.ABSENT: + minds.append(subheartflow.current_mind) + return minds + + def update_main_mind_in_subflows(self, main_mind: str): + """更新所有子心流的主心流想法""" + updated_count = sum( + 1 + for _, subheartflow in list(self.subheartflows.items()) + if subheartflow.subheartflow_id in self.subheartflows + ) + logger.debug(f"[子心流管理器] 更新了{updated_count}个子心流的主想法") + + async def deactivate_subflow(self, subheartflow_id: Any): + """停用并移除指定的子心流。""" + async with self._lock: + subflow = self.subheartflows.pop(subheartflow_id, None) + if subflow: + logger.info(f"正在停用 SubHeartflow: {subheartflow_id}...") + try: + # --- 调用 shutdown 方法 --- + await subflow.shutdown() + # --- 结束调用 --- + logger.info(f"SubHeartflow {subheartflow_id} 已成功停用。") + except Exception as e: + logger.error(f"停用 SubHeartflow {subheartflow_id} 时出错: {e}", exc_info=True) + else: + logger.warning(f"尝试停用不存在的 SubHeartflow: {subheartflow_id}") + + async def deactivate_all_subflows(self): + """停用并移除所有子心流。""" + logger.info("正在停用所有 SubHeartflows...") + # 获取当前所有 ID，避免在迭代时修改字典 + all_ids = self.get_all_subheartflows_ids() + tasks = [self.deactivate_subflow(sub_id) for sub_id in all_ids] + await asyncio.gather(*tasks) + logger.info("所有 SubHeartflows 已停用。") + + async def cleanup_inactive_subflows(self, inactive_threshold_seconds: int): + """清理长时间不活跃的子心流。""" + current_time = time.time() + inactive_ids = [] + # 不加锁地迭代，识别不活跃的 ID + for sub_id, subflow in self.subheartflows.items(): + # 检查 last_active_time 是否存在且是数值 + last_active = getattr(subflow, "last_active_time", 0) + if isinstance(last_active, (int, float)): + if current_time - last_active > inactive_threshold_seconds: + inactive_ids.append(sub_id) + logger.info( + f"发现不活跃的 SubHeartflow: {sub_id} (上次活跃: {time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(last_active))})" + ) + else: + logger.warning(f"SubHeartflow {sub_id} 的 last_active_time 无效: {last_active}。跳过清理检查。") + + if inactive_ids: + logger.info(f"准备清理 {len(inactive_ids)} 个不活跃的 SubHeartflows: {inactive_ids}") + # 逐个停用（deactivate_subflow 会加锁） + tasks = [self.deactivate_subflow(sub_id) for sub_id in inactive_ids] + await asyncio.gather(*tasks) + logger.info("不活跃的 SubHeartflows 清理完成。") + # else: + # logger.debug("没有发现不活跃的 SubHeartflows 需要清理。") diff --git a/src/plugins/heartFC_chat/heartFC_chat.py b/src/plugins/heartFC_chat/heartFC_chat.py index e2357f252..476798f42 100644 --- a/src/plugins/heartFC_chat/heartFC_chat.py +++ b/src/plugins/heartFC_chat/heartFC_chat.py @@ -17,8 +17,8 @@ from src.plugins.utils.timer_calculater import Timer # <--- Import Timer # --- Import necessary dependencies directly --- from .heartFC_generator import ResponseGenerator # Assuming this is the type for gpt from src.do_tool.tool_use import ToolUser -from src.plugins.chat.emoji_manager import EmojiManager # Assuming this is the type from ..chat.message_sender import message_manager # <-- Import the global manager +from src.plugins.chat.emoji_manager import emoji_manager # --- End import --- @@ -78,10 +78,8 @@ class HeartFChatting: def __init__( self, chat_id: str, - # 显式依赖注入 gpt_instance: ResponseGenerator, # 文本回复生成器 tool_user_instance: ToolUser, # 工具使用实例 - emoji_manager_instance: EmojiManager, # 表情管理实例 ): """ HeartFChatting 初始化函数 @@ -106,7 +104,6 @@ class HeartFChatting: # 依赖注入存储 self.gpt_instance = gpt_instance # 文本回复生成器 self.tool_user = tool_user_instance # 工具使用实例 - self.emoji_manager = emoji_manager_instance # 表情管理实例 # LLM规划器配置 self.planner_llm = LLMRequest( @@ -659,26 +656,6 @@ class HeartFChatting: logger.error(traceback.format_exc()) return None - # def _cleanup_thinking_message(self, thinking_id: str): - # """Safely removes the thinking message.""" - # log_prefix = self._get_log_prefix() - # try: - # # Access MessageManager directly - # container = await message_manager.get_container(self.stream_id) - # # container.remove_message(thinking_id, msg_type=MessageThinking) # Need to find the message object first - # found_msg = None - # for msg in container.get_all_messages(): - # if isinstance(msg, MessageThinking) and msg.message_info.message_id == thinking_id: - # found_msg = msg - # break - # if found_msg: - # container.remove_message(found_msg) - # logger.debug(f"{log_prefix} Cleaned up thinking message {thinking_id}.") - # else: - # logger.warning(f"{log_prefix} Could not find thinking message {thinking_id} to cleanup.") - # except Exception as e: - # logger.error(f"{log_prefix} Error cleaning up thinking message {thinking_id}: {e}") - # --- 发送器 (Sender) --- # async def _sender( self, @@ -783,12 +760,6 @@ class HeartFChatting: log_prefix = self._get_log_prefix() response_set: Optional[List[str]] = None try: - # --- Generate Response with LLM --- # - # Access gpt instance directly - # logger.debug(f"{log_prefix}[Replier-{thinking_id}] Calling LLM to generate response...") - - # Ensure generate_response has access to current_mind if it's crucial context - # Access gpt_instance directly response_set = await self.gpt_instance.generate_response( current_mind_info=self.sub_hf.current_mind, reason=reason, @@ -902,15 +873,12 @@ class HeartFChatting: return chat = anchor_message.chat_stream - # Access emoji_manager directly - # emoji_manager_instance = self.heartfc_controller.emoji_manager # Removed + if send_emoji: - # Use self.emoji_manager directly - emoji_raw = await self.emoji_manager.get_emoji_for_text(send_emoji) + emoji_raw = await emoji_manager.get_emoji_for_text(send_emoji) else: emoji_text_source = "".join(response_set) if response_set else "" - # Use self.emoji_manager directly - emoji_raw = await self.emoji_manager.get_emoji_for_text(emoji_text_source) + emoji_raw = await emoji_manager.get_emoji_for_text(emoji_text_source) if emoji_raw: emoji_path, _description = emoji_raw diff --git a/src/plugins/heartFC_chat/heartflow_processor.py b/src/plugins/heartFC_chat/heartflow_processor.py index 02d65d30f..44206d8b3 100644 --- a/src/plugins/heartFC_chat/heartflow_processor.py +++ b/src/plugins/heartFC_chat/heartflow_processor.py @@ -12,7 +12,6 @@ from ..chat.chat_stream import chat_manager from ..chat.message_buffer import message_buffer from ..utils.timer_calculater import Timer from src.plugins.person_info.relationship_manager import relationship_manager -from .normal_chat import NormalChat # 定义日志配置 processor_config = LogConfig( @@ -25,7 +24,6 @@ logger = get_module_logger("heartflow_processor", config=processor_config) class HeartFCProcessor: def __init__(self): self.storage = MessageStorage() - self.normal_chat = NormalChat.get_instance() async def process_message(self, message_data: str) -> None: """处理接收到的原始消息数据，完成消息解析、缓冲、过滤、存储、兴趣度计算与更新等核心流程。 @@ -75,11 +73,6 @@ class HeartFCProcessor: logger.error(f"无法为 stream_id {chat.stream_id} 创建或获取 SubHeartflow，中止处理") return - # --- 添加兴趣追踪启动 (现在移动到这里，确保 subheartflow 存在后启动) --- - # 在获取到 chat 对象和确认 subheartflow 后，启动对该聊天流的兴趣监控 - await self.normal_chat.start_monitoring_interest(chat) # start_monitoring_interest 内部需要修改以适应 - # --- 结束添加 --- - message.update_chat_stream(chat) await heartflow.create_subheartflow(chat.stream_id) diff --git a/src/plugins/heartFC_chat/heartflow_prompt_builder.py b/src/plugins/heartFC_chat/heartflow_prompt_builder.py index 53c4e575a..a3c5f6ce1 100644 --- a/src/plugins/heartFC_chat/heartflow_prompt_builder.py +++ b/src/plugins/heartFC_chat/heartflow_prompt_builder.py @@ -226,11 +226,6 @@ class PromptBuilder: "memory_prompt", related_memory_info=related_memory_info ) - # print(f"相关记忆：{related_memory_info}") - - # 日程构建 - # schedule_prompt = f"""你现在正在做的事情是：{bot_schedule.get_current_num_task(num=1, time_info=False)}""" - # 获取聊天上下文 if chat_stream.group_info: chat_in_group = True @@ -292,9 +287,12 @@ class PromptBuilder: logger.debug("开始构建prompt") - schedule_prompt = await global_prompt_manager.format_prompt( - "schedule_prompt", schedule_info=bot_schedule.get_current_num_task(num=1, time_info=False) - ) + if global_config.ENABLE_SCHEDULE_GEN: + schedule_prompt = await global_prompt_manager.format_prompt( + "schedule_prompt", schedule_info=bot_schedule.get_current_num_task(num=1, time_info=False) + ) + else: + schedule_prompt = "" prompt = await global_prompt_manager.format_prompt( "reasoning_prompt_main", diff --git a/src/plugins/heartFC_chat/normal_chat.py b/src/plugins/heartFC_chat/normal_chat.py index 38251c02f..cdb1c2c3f 100644 --- a/src/plugins/heartFC_chat/normal_chat.py +++ b/src/plugins/heartFC_chat/normal_chat.py @@ -1,17 +1,15 @@ import time -import threading # 导入 threading -from random import random -import traceback import asyncio -from typing import List, Dict +import traceback +from random import random +from typing import List, Optional # 导入 Optional + from ..moods.moods import MoodManager from ...config.config import global_config from ..chat.emoji_manager import emoji_manager from .normal_chat_generator import ResponseGenerator from ..chat.message import MessageSending, MessageRecv, MessageThinking, MessageSet from ..chat.message_sender import message_manager -from ..storage.storage import MessageStorage -from ..chat.utils import is_mentioned_bot_in_message from ..chat.utils_image import image_path_to_base64 from ..willing.willing_manager import willing_manager from ..message import UserInfo, Seg @@ -20,8 +18,6 @@ from src.plugins.chat.chat_stream import ChatStream, chat_manager from src.plugins.person_info.relationship_manager import relationship_manager from src.plugins.respon_info_catcher.info_catcher import info_catcher_manager from src.plugins.utils.timer_calculater import Timer -from src.heart_flow.heartflow import heartflow -from src.heart_flow.sub_heartflow import ChatState # 定义日志配置 chat_config = LogConfig( @@ -33,46 +29,34 @@ logger = get_module_logger("normal_chat", config=chat_config) class NormalChat: - _instance = None - _lock = threading.Lock() - _initialized = False + def __init__(self, chat_stream: ChatStream, interest_dict: dict): + """ + 初始化 NormalChat 实例，针对特定的 ChatStream。 - def __new__(cls, *args, **kwargs): - if cls._instance is None: - with cls._lock: - # Double-check locking - if cls._instance is None: - cls._instance = super().__new__(cls) - return cls._instance + Args: + chat_stream (ChatStream): 此 NormalChat 实例关联的聊天流对象。 + """ - def __init__(self): - # 防止重复初始化 - if self._initialized: - return - with self.__class__._lock: # 使用类锁确保线程安全 - if self._initialized: - return - logger.info("正在初始化 NormalChat 单例...") # 添加日志 - self.storage = MessageStorage() - self.gpt = ResponseGenerator() - self.mood_manager = MoodManager.get_instance() - # 用于存储每个 chat stream 的兴趣监控任务 - self._interest_monitoring_tasks: Dict[str, asyncio.Task] = {} - self._initialized = True - logger.info("NormalChat 单例初始化完成。") # 添加日志 + self.chat_stream = chat_stream + self.stream_id = chat_stream.stream_id + self.stream_name = chat_manager.get_stream_name(self.stream_id) or self.stream_id - @classmethod - def get_instance(cls): - """获取 NormalChat 的单例实例。""" - if cls._instance is None: - # 如果实例还未创建（理论上应该在 main 中初始化，但作为备用） - logger.warning("NormalChat 实例在首次 get_instance 时创建。") - cls() # 调用构造函数来创建实例 - return cls._instance + self.interest_dict = interest_dict - @staticmethod - async def _create_thinking_message(message, chat, userinfo, messageinfo): + logger.info(f"[{self.stream_name}] 正在初始化 NormalChat 实例...") + + self.gpt = ResponseGenerator() + self.mood_manager = MoodManager.get_instance() # MoodManager 保持单例 + # 存储此实例的兴趣监控任务 + self._interest_monitoring_task: Optional[asyncio.Task] = None + logger.info(f"[{self.stream_name}] NormalChat 实例初始化完成。") + + # 改为实例方法 + async def _create_thinking_message(self, message: MessageRecv) -> str: """创建思考消息""" + userinfo = message.message_info.user_info + messageinfo = message.message_info + bot_user_info = UserInfo( user_id=global_config.BOT_QQ, user_nickname=global_config.BOT_NICKNAME, @@ -83,7 +67,7 @@ class NormalChat: thinking_id = "mt" + str(thinking_time_point) thinking_message = MessageThinking( message_id=thinking_id, - chat_stream=chat, + chat_stream=self.chat_stream, # 使用 self.chat_stream bot_user_info=bot_user_info, reply=message, thinking_start_time=thinking_time_point, @@ -93,10 +77,12 @@ class NormalChat: return thinking_id - @staticmethod - async def _send_response_messages(message, chat, response_set: List[str], thinking_id) -> MessageSending: + # 改为实例方法 + async def _add_messages_to_manager( + self, message: MessageRecv, response_set: List[str], thinking_id + ) -> Optional[MessageSending]: """发送回复消息""" - container = await message_manager.get_container(chat.stream_id) + container = await message_manager.get_container(self.stream_id) # 使用 self.stream_id thinking_message = None for msg in container.messages[:]: @@ -106,11 +92,11 @@ class NormalChat: break if not thinking_message: - logger.warning(f"[{chat.stream_id}] 未找到对应的思考消息 {thinking_id}，可能已超时被移除") + logger.warning(f"[{self.stream_name}] 未找到对应的思考消息 {thinking_id}，可能已超时被移除") return None thinking_start_time = thinking_message.thinking_start_time - message_set = MessageSet(chat, thinking_id) + message_set = MessageSet(self.chat_stream, thinking_id) # 使用 self.chat_stream mark_head = False first_bot_msg = None @@ -118,7 +104,7 @@ class NormalChat: message_segment = Seg(type="text", data=msg) bot_message = MessageSending( message_id=thinking_id, - chat_stream=chat, + chat_stream=self.chat_stream, # 使用 self.chat_stream bot_user_info=UserInfo( user_id=global_config.BOT_QQ, user_nickname=global_config.BOT_NICKNAME, @@ -141,8 +127,8 @@ class NormalChat: return first_bot_msg - @staticmethod - async def _handle_emoji(message, chat, response): + # 改为实例方法 + async def _handle_emoji(self, message: MessageRecv, response: str): """处理表情包""" if random() < global_config.emoji_chance: emoji_raw = await emoji_manager.get_emoji_for_text(response) @@ -155,7 +141,7 @@ class NormalChat: message_segment = Seg(type="emoji", data=emoji_cq) bot_message = MessageSending( message_id="mt" + str(thinking_time_point), - chat_stream=chat, + chat_stream=self.chat_stream, # 使用 self.chat_stream bot_user_info=UserInfo( user_id=global_config.BOT_QQ, user_nickname=global_config.BOT_NICKNAME, @@ -170,193 +156,151 @@ class NormalChat: ) await message_manager.add_message(bot_message) + # 改为实例方法 (虽然它只用 message.chat_stream, 但逻辑上属于实例) async def _update_relationship(self, message: MessageRecv, response_set): """更新关系情绪""" ori_response = ",".join(response_set) stance, emotion = await self.gpt._get_emotion_tags(ori_response, message.processed_plain_text) await relationship_manager.calculate_update_relationship_value( - chat_stream=message.chat_stream, label=emotion, stance=stance + chat_stream=self.chat_stream, + label=emotion, + stance=stance, # 使用 self.chat_stream ) self.mood_manager.update_mood_from_emotion(emotion, global_config.mood_intensity_factor) - async def _find_interested_message(self, chat: ChatStream) -> None: - # 此函数设计为后台任务，轮询指定 chat 的兴趣消息。 - # 它通常由外部代码在 chat 流活跃时启动。 - stream_id = chat.stream_id # 获取 stream_id - - # logger.info(f"[{stream_id}] 兴趣消息监控任务启动。") # 减少日志 + async def _find_interested_message(self) -> None: + """ + 后台任务方法，轮询当前实例关联chat的兴趣消息 + 通常由start_monitoring_interest()启动 + """ while True: await asyncio.sleep(1) # 每秒检查一次 - subheartflow = heartflow.get_subheartflow(stream_id) - - if not subheartflow or subheartflow.should_stop: - # logger.info(f"[{stream_id}] SubHeartflow 不存在或已停止，兴趣消息监控任务退出。") # 减少日志 + # 检查任务是否已被取消 + if self._interest_monitoring_task is None or self._interest_monitoring_task.cancelled(): + logger.info(f"[{self.stream_name}] 兴趣监控任务被取消或置空，退出") break - interest_dict = subheartflow.get_interest_dict() - items_to_process = list(interest_dict.items()) - + # 获取待处理消息列表 + items_to_process = list(self.interest_dict.items()) if not items_to_process: continue + # 处理每条兴趣消息 for msg_id, (message, interest_value, is_mentioned) in items_to_process: - # --- 在处理前检查 SubHeartflow 的状态 --- # - current_chat_state = subheartflow.chat_state.chat_status - stream_name = chat_manager.get_stream_name(stream_id) or stream_id - - if current_chat_state != ChatState.CHAT: - # 如果不是闲聊状态 (可能是 ABSENT 或 FOCUSED)，则跳过推理聊天 - # logger.debug(f"[{stream_name}] 跳过处理兴趣消息 {msg_id}，因为当前状态为 {current_chat_state.value}") - # 移除消息并继续下一个 - removed_item = interest_dict.pop(msg_id, None) - if removed_item: - # logger.debug(f"[{stream_name}] 已从兴趣字典中移除消息 {msg_id} (因状态跳过)") # 减少日志 - pass - continue # 处理下一条消息 - # --- 结束状态检查 --- # - - # --- 检查 HeartFChatting 是否活跃 (改为检查 SubHeartflow 状态) --- # - is_focused = subheartflow.chat_state.chat_status == ChatState.FOCUSED - - if is_focused: # New check: If the subflow is focused, NormalChat shouldn't process - removed_item = interest_dict.pop(msg_id, None) - if removed_item: - # logger.debug(f"[{stream_name}] SubHeartflow 处于 FOCUSED 状态，已跳过并移除 NormalChat 兴趣消息 {msg_id}") # Reduce noise - pass - continue - # --- 结束检查 --- # - - # 只有当状态为 CHAT 且 HeartFChatting 不活跃 (即 Subflow 不是 FOCUSED) 时才执行以下处理逻辑 try: - await self.normal_normal_chat( - message=message, - chat=chat, - is_mentioned=is_mentioned, - interested_rate=interest_value, + # 处理消息 + await self.normal_response( + message=message, is_mentioned=is_mentioned, interested_rate=interest_value ) except Exception as e: - logger.error(f"[{stream_name}] 处理兴趣消息 {msg_id} 时出错: {e}\n{traceback.format_exc()}") + logger.error(f"[{self.stream_name}] 处理兴趣消息{msg_id}时出错: {e}\n{traceback.format_exc()}") finally: - removed_item = interest_dict.pop(msg_id, None) - if removed_item: - # logger.debug(f"[{stream_name}] 已从兴趣字典中移除消息 {msg_id}") # 减少日志 - pass + self.interest_dict.pop(msg_id, None) + + # 改为实例方法, 移除 chat 参数 + async def normal_response(self, message: MessageRecv, is_mentioned: bool, interested_rate: float) -> None: + # 检查收到的消息是否属于当前实例处理的 chat stream + if message.chat_stream.stream_id != self.stream_id: + logger.error( + f"[{self.stream_name}] normal_response 收到不匹配的消息 (来自 {message.chat_stream.stream_id})，预期 {self.stream_id}。已忽略。" + ) + return - async def normal_normal_chat( - self, message: MessageRecv, chat: ChatStream, is_mentioned: bool, interested_rate: float - ) -> None: timing_results = {} userinfo = message.message_info.user_info messageinfo = message.message_info - stream_id = chat.stream_id - stream_name = chat_manager.get_stream_name(stream_id) or stream_id - # --- 在开始时检查 SubHeartflow 状态 --- # - sub_hf = heartflow.get_subheartflow(stream_id) - if not sub_hf: - logger.warning(f"[{stream_name}] 无法获取 SubHeartflow，无法执行 normal_normal_chat。") - return + reply_probability = 1.0 if is_mentioned else 0.0 # 如果被提及，基础概率为1，否则需要意愿判断 - current_chat_state = sub_hf.chat_state.chat_status - if current_chat_state != ChatState.CHAT: - logger.debug( - f"[{stream_name}] 跳过 normal_normal_chat，因为 SubHeartflow 状态为 {current_chat_state.value} (需要 CHAT)。" - ) - # 可以在这里添加 not_reply_handle 逻辑吗？ 如果不回复，也需要清理意愿。 - # 注意：willing_manager.setup 尚未调用 - willing_manager.setup(message, chat, is_mentioned, interested_rate) # 先 setup - await willing_manager.not_reply_handle(message.message_info.message_id) - willing_manager.delete(message.message_info.message_id) - return - # --- 结束状态检查 --- # - - # --- 接下来的逻辑只在 ChatState.CHAT 状态下执行 --- # - - is_mentioned, reply_probability = is_mentioned_bot_in_message(message) # 意愿管理器：设置当前message信息 - willing_manager.setup(message, chat, is_mentioned, interested_rate) + + willing_manager.setup(message, self.chat_stream, is_mentioned, interested_rate) # 获取回复概率 is_willing = False - if reply_probability != 1: + # 仅在未被提及或基础概率不为1时查询意愿概率 + if reply_probability < 1: # 简化逻辑，如果未提及 (reply_probability 为 0)，则获取意愿概率 is_willing = True reply_probability = await willing_manager.get_reply_probability(message.message_info.message_id) if message.message_info.additional_config: if "maimcore_reply_probability_gain" in message.message_info.additional_config.keys(): reply_probability += message.message_info.additional_config["maimcore_reply_probability_gain"] + reply_probability = min(max(reply_probability, 0), 1) # 确保概率在 0-1 之间 # 打印消息信息 - mes_name = chat.group_info.group_name if chat.group_info else "私聊" + mes_name = self.chat_stream.group_info.group_name if self.chat_stream.group_info else "私聊" current_time = time.strftime("%H:%M:%S", time.localtime(message.message_info.time)) - willing_log = f"[回复意愿:{await willing_manager.get_willing(chat.stream_id):.2f}]" if is_willing else "" + # 使用 self.stream_id + willing_log = f"[回复意愿:{await willing_manager.get_willing(self.stream_id):.2f}]" if is_willing else "" logger.info( f"[{current_time}][{mes_name}]" - f"{chat.user_info.user_nickname}:" + f"{self.chat_stream.user_info.user_nickname}:" # 使用 self.chat_stream f"{message.processed_plain_text}{willing_log}[概率:{reply_probability * 100:.1f}%]" ) do_reply = False + response_set = None # 初始化 response_set if random() < reply_probability: do_reply = True # 回复前处理 await willing_manager.before_generate_reply_handle(message.message_info.message_id) - # 创建思考消息 with Timer("创建思考消息", timing_results): - thinking_id = await self._create_thinking_message(message, chat, userinfo, messageinfo) + thinking_id = await self._create_thinking_message(message) - logger.debug(f"创建捕捉器，thinking_id:{thinking_id}") + logger.debug(f"[{self.stream_name}] 创建捕捉器，thinking_id:{thinking_id}") info_catcher = info_catcher_manager.get_info_catcher(thinking_id) info_catcher.catch_decide_to_response(message) - # 生成回复 - sub_hf = heartflow.get_subheartflow(stream_id) - try: with Timer("生成回复", timing_results): response_set = await self.gpt.generate_response( - sub_hf=sub_hf, message=message, thinking_id=thinking_id, ) info_catcher.catch_after_generate_response(timing_results["生成回复"]) except Exception as e: - logger.error(f"回复生成出现错误：{str(e)} {traceback.format_exc()}") - response_set = None + logger.error(f"[{self.stream_name}] 回复生成出现错误：{str(e)} {traceback.format_exc()}") + response_set = None # 确保出错时 response_set 为 None if not response_set: - logger.info(f"[{chat.stream_id}] 模型未生成回复内容") + logger.info(f"[{self.stream_name}] 模型未生成回复内容") # 如果模型未生成回复，移除思考消息 - container = await message_manager.get_container(chat.stream_id) - # thinking_message = None - for msg in container.messages[:]: # Iterate over a copy + container = await message_manager.get_container(self.stream_id) # 使用 self.stream_id + for msg in container.messages[:]: if isinstance(msg, MessageThinking) and msg.message_info.message_id == thinking_id: - # thinking_message = msg container.messages.remove(msg) - # container.remove_message(msg) # 直接移除 - logger.debug(f"[{chat.stream_id}] 已移除未产生回复的思考消息 {thinking_id}") + logger.debug(f"[{self.stream_name}] 已移除未产生回复的思考消息 {thinking_id}") break - return # 不发送回复 + # 需要在此处也调用 not_reply_handle 和 delete 吗？ + # 如果是因为模型没回复，也算是一种 "未回复" + await willing_manager.not_reply_handle(message.message_info.message_id) + willing_manager.delete(message.message_info.message_id) + return # 不执行后续步骤 - logger.info(f"[{chat.stream_id}] 回复内容: {response_set}") + logger.info(f"[{self.stream_name}] 回复内容: {response_set}") - # 发送回复 + # 发送回复 (不再需要传入 chat) with Timer("消息发送", timing_results): - first_bot_msg = await self._send_response_messages(message, chat, response_set, thinking_id) + first_bot_msg = await self._add_messages_to_manager(message, response_set, thinking_id) - info_catcher.catch_after_response(timing_results["消息发送"], response_set, first_bot_msg) + # 检查 first_bot_msg 是否为 None (例如思考消息已被移除的情况) + if first_bot_msg: + info_catcher.catch_after_response(timing_results["消息发送"], response_set, first_bot_msg) + else: + logger.warning(f"[{self.stream_name}] 思考消息 {thinking_id} 在发送前丢失，无法记录 info_catcher") info_catcher.done_catch() - # 处理表情包 + # 处理表情包 (不再需要传入 chat) with Timer("处理表情包", timing_results): - await self._handle_emoji(message, chat, response_set[0]) + await self._handle_emoji(message, response_set[0]) - # 更新关系情绪 + # 更新关系情绪 (不再需要传入 chat) with Timer("关系更新", timing_results): await self._update_relationship(message, response_set) @@ -364,88 +308,104 @@ class NormalChat: await willing_manager.after_generate_reply_handle(message.message_info.message_id) # 输出性能计时结果 - if do_reply: + if do_reply and response_set: # 确保 response_set 不是 None timing_str = " | ".join([f"{step}: {duration:.2f}秒" for step, duration in timing_results.items()]) trigger_msg = message.processed_plain_text - response_msg = " ".join(response_set) if response_set else "无回复" - logger.info(f"触发消息: {trigger_msg[:20]}... | 推理消息: {response_msg[:20]}... | 性能计时: {timing_str}") - else: + response_msg = " ".join(response_set) + logger.info( + f"[{self.stream_name}] 触发消息: {trigger_msg[:20]}... | 推理消息: {response_msg[:20]}... | 性能计时: {timing_str}" + ) + elif not do_reply: # 不回复处理 await willing_manager.not_reply_handle(message.message_info.message_id) + # else: # do_reply is True but response_set is None (handled above) + # logger.info(f"[{self.stream_name}] 决定回复但模型未生成内容。触发: {message.processed_plain_text[:20]}...") - # 意愿管理器：注销当前message信息 + # 意愿管理器：注销当前message信息 (无论是否回复，只要处理过就删除) willing_manager.delete(message.message_info.message_id) + # 保持 staticmethod, 因为不依赖实例状态, 但需要 chat 对象来获取日志上下文 @staticmethod - def _check_ban_words(text: str, chat, userinfo) -> bool: + def _check_ban_words(text: str, chat: ChatStream, userinfo: UserInfo) -> bool: """检查消息中是否包含过滤词""" + stream_name = chat_manager.get_stream_name(chat.stream_id) or chat.stream_id for word in global_config.ban_words: if word in text: logger.info( - f"[{chat.group_info.group_name if chat.group_info else '私聊'}]{userinfo.user_nickname}:{text}" + f"[{stream_name}][{chat.group_info.group_name if chat.group_info else '私聊'}]" + f"{userinfo.user_nickname}:{text}" ) - logger.info(f"[过滤词识别]消息中含有{word}，filtered") + logger.info(f"[{stream_name}][过滤词识别] 消息中含有 '{word}'，filtered") return True return False + # 保持 staticmethod, 因为不依赖实例状态, 但需要 chat 对象来获取日志上下文 @staticmethod - def _check_ban_regex(text: str, chat, userinfo) -> bool: + def _check_ban_regex(text: str, chat: ChatStream, userinfo: UserInfo) -> bool: """检查消息是否匹配过滤正则表达式""" + stream_name = chat_manager.get_stream_name(chat.stream_id) or chat.stream_id for pattern in global_config.ban_msgs_regex: if pattern.search(text): logger.info( - f"[{chat.group_info.group_name if chat.group_info else '私聊'}]{userinfo.user_nickname}:{text}" + f"[{stream_name}][{chat.group_info.group_name if chat.group_info else '私聊'}]" + f"{userinfo.user_nickname}:{text}" ) - logger.info(f"[正则表达式过滤]消息匹配到{pattern}，filtered") + logger.info(f"[{stream_name}][正则表达式过滤] 消息匹配到 '{pattern.pattern}'，filtered") return True return False - async def start_monitoring_interest(self, chat: ChatStream): - """为指定的 ChatStream 启动兴趣消息监控任务（如果尚未运行）。""" - stream_id = chat.stream_id - if stream_id not in self._interest_monitoring_tasks or self._interest_monitoring_tasks[stream_id].done(): - logger.info(f"为聊天流 {stream_id} 启动兴趣消息监控任务...") - # 创建新任务 - task = asyncio.create_task(self._find_interested_message(chat)) - # 添加完成回调 - task.add_done_callback(lambda t: self._handle_task_completion(stream_id, t)) - self._interest_monitoring_tasks[stream_id] = task - # else: - # logger.debug(f"聊天流 {stream_id} 的兴趣消息监控任务已在运行。") + # 改为实例方法, 移除 chat 参数 + async def start_monitoring_interest(self): + """为此 NormalChat 实例关联的 ChatStream 启动兴趣消息监控任务（如果尚未运行）。""" + if self._interest_monitoring_task is None or self._interest_monitoring_task.done(): + logger.info(f"[{self.stream_name}] 启动兴趣消息监控任务...") + task = asyncio.create_task(self._find_interested_message()) + task.add_done_callback(lambda t: self._handle_task_completion(t)) # 回调现在是实例方法 + self._interest_monitoring_task = task - def _handle_task_completion(self, stream_id: str, task: asyncio.Task): + # 改为实例方法, 移除 stream_id 参数 + def _handle_task_completion(self, task: asyncio.Task): """兴趣监控任务完成时的回调函数。""" + # 检查完成的任务是否是当前实例的任务 + if task is not self._interest_monitoring_task: + logger.warning(f"[{self.stream_name}] 收到一个未知或过时任务的完成回调。") + return + try: # 检查任务是否因异常而结束 exception = task.exception() if exception: - logger.error(f"聊天流 {stream_id} 的兴趣监控任务因异常结束: {exception}") + logger.error(f"[{self.stream_name}] 兴趣监控任务因异常结束: {exception}") logger.error(traceback.format_exc()) # 记录完整的 traceback - else: - logger.info(f"聊天流 {stream_id} 的兴趣监控任务正常结束。") + # else: # 减少日志 + # logger.info(f"[{self.stream_name}] 兴趣监控任务正常结束。") except asyncio.CancelledError: - logger.info(f"聊天流 {stream_id} 的兴趣监控任务被取消。") + logger.info(f"[{self.stream_name}] 兴趣监控任务被取消。") except Exception as e: - logger.error(f"处理聊天流 {stream_id} 任务完成回调时出错: {e}") + logger.error(f"[{self.stream_name}] 处理任务完成回调时出错: {e}") finally: - # 从字典中移除已完成或取消的任务 - if stream_id in self._interest_monitoring_tasks: - del self._interest_monitoring_tasks[stream_id] - logger.debug(f"已从监控任务字典中移除 {stream_id}") + # 标记任务已完成/移除 + if self._interest_monitoring_task is task: # 再次确认是当前任务 + self._interest_monitoring_task = None + logger.debug(f"[{self.stream_name}] 兴趣监控任务已被标记为完成/移除。") - async def stop_monitoring_interest(self, stream_id: str): - """停止指定聊天流的兴趣监控任务。""" - if stream_id in self._interest_monitoring_tasks: - task = self._interest_monitoring_tasks[stream_id] - if task and not task.done(): - task.cancel() # 尝试取消任务 - logger.info(f"尝试取消聊天流 {stream_id} 的兴趣监控任务。") - try: - await task # 等待任务响应取消 - except asyncio.CancelledError: - logger.info(f"聊天流 {stream_id} 的兴趣监控任务已成功取消。") - except Exception as e: - logger.error(f"等待聊天流 {stream_id} 监控任务取消时出现异常: {e}") - # 在回调函数 _handle_task_completion 中移除任务 + # 改为实例方法, 移除 stream_id 参数 + async def stop_monitoring_interest(self): + """停止当前实例的兴趣监控任务。""" + if self._interest_monitoring_task and not self._interest_monitoring_task.done(): + task = self._interest_monitoring_task + logger.info(f"[{self.stream_name}] 尝试取消兴趣监控任务。") + task.cancel() + try: + await task # 等待任务响应取消 + except asyncio.CancelledError: + logger.info(f"[{self.stream_name}] 兴趣监控任务已成功取消。") + except Exception as e: + # 回调函数 _handle_task_completion 会处理异常日志 + logger.warning(f"[{self.stream_name}] 等待监控任务取消时捕获到异常 (可能已在回调中记录): {e}") + finally: + # 确保任务状态更新，即使等待出错 (回调函数也会尝试更新) + if self._interest_monitoring_task is task: + self._interest_monitoring_task = None # else: - # logger.debug(f"聊天流 {stream_id} 没有正在运行的兴趣监控任务可停止。") + # logger.debug(f"[{self.stream_name}] 没有正在运行的兴趣监控任务可停止。") diff --git a/src/plugins/heartFC_chat/normal_chat_generator.py b/src/plugins/heartFC_chat/normal_chat_generator.py index 82bfaffcc..ee133f440 100644 --- a/src/plugins/heartFC_chat/normal_chat_generator.py +++ b/src/plugins/heartFC_chat/normal_chat_generator.py @@ -8,7 +8,6 @@ from ..chat.utils import process_llm_response from ..utils.timer_calculater import Timer from src.common.logger import get_module_logger, LogConfig, LLM_STYLE_CONFIG from src.plugins.respon_info_catcher.info_catcher import info_catcher_manager -from src.heart_flow.sub_heartflow import SubHeartflow # 定义日志配置 llm_config = LogConfig( @@ -41,9 +40,7 @@ class ResponseGenerator: self.current_model_type = "r1" # 默认使用 R1 self.current_model_name = "unknown model" - async def generate_response( - self, sub_hf: SubHeartflow, message: MessageThinking, thinking_id: str - ) -> Optional[Union[str, List[str]]]: + async def generate_response(self, message: MessageThinking, thinking_id: str) -> Optional[Union[str, List[str]]]: """根据当前模型类型选择对应的生成函数""" # 从global_config中获取模型概率值并选择模型 if random.random() < global_config.model_reasoning_probability: @@ -57,9 +54,7 @@ class ResponseGenerator: f"{self.current_model_type}思考:{message.processed_plain_text[:30] + '...' if len(message.processed_plain_text) > 30 else message.processed_plain_text}" ) # noqa: E501 - model_response = await self._generate_response_with_model(sub_hf, message, current_model, thinking_id) - - # print(f"raw_content: {model_response}") + model_response = await self._generate_response_with_model(message, current_model, thinking_id) if model_response: logger.info(f"{global_config.BOT_NICKNAME}的回复是：{model_response}") @@ -70,9 +65,7 @@ class ResponseGenerator: logger.info(f"{self.current_model_type}思考，失败") return None - async def _generate_response_with_model( - self, sub_hf: SubHeartflow, message: MessageThinking, model: LLMRequest, thinking_id: str - ): + async def _generate_response_with_model(self, message: MessageThinking, model: LLMRequest, thinking_id: str): info_catcher = info_catcher_manager.get_info_catcher(thinking_id) if message.chat_stream.user_info.user_cardname and message.chat_stream.user_info.user_nickname: diff --git a/src/plugins/schedule/schedule_generator.py b/src/plugins/schedule/schedule_generator.py index ab1fa4552..761fcb7dc 100644 --- a/src/plugins/schedule/schedule_generator.py +++ b/src/plugins/schedule/schedule_generator.py @@ -73,29 +73,32 @@ class ScheduleGenerator: async def mai_schedule_start(self): """启动日程系统，每5分钟执行一次move_doing，并在日期变化时重新检查日程""" try: - logger.info(f"日程系统启动/刷新时间: {self.start_time.strftime('%Y-%m-%d %H:%M:%S')}") - # 初始化日程 - await self.check_and_create_today_schedule() - # self.print_schedule() + if global_config.ENABLE_SCHEDULE_GEN: + logger.info(f"日程系统启动/刷新时间: {self.start_time.strftime('%Y-%m-%d %H:%M:%S')}") + # 初始化日程 + await self.check_and_create_today_schedule() + # self.print_schedule() - while True: - # print(self.get_current_num_task(1, True)) + while True: + # print(self.get_current_num_task(1, True)) - current_time = datetime.datetime.now(TIME_ZONE) + current_time = datetime.datetime.now(TIME_ZONE) - # 检查是否需要重新生成日程（日期变化） - if current_time.date() != self.start_time.date(): - logger.info("检测到日期变化，重新生成日程") - self.start_time = current_time - await self.check_and_create_today_schedule() - # self.print_schedule() + # 检查是否需要重新生成日程（日期变化） + if current_time.date() != self.start_time.date(): + logger.info("检测到日期变化，重新生成日程") + self.start_time = current_time + await self.check_and_create_today_schedule() + # self.print_schedule() - # 执行当前活动 - # mind_thinking = heartflow.current_state.current_mind + # 执行当前活动 + # mind_thinking = heartflow.current_state.current_mind - await self.move_doing() + await self.move_doing() - await asyncio.sleep(self.schedule_doing_update_interval) + await asyncio.sleep(self.schedule_doing_update_interval) + else: + logger.info("日程系统未启用") except Exception as e: logger.error(f"日程系统运行时出错: {str(e)}")