diff --git a/interest_monitor_gui.py b/interest_monitor_gui.py index c713cc122..97536f701 100644 --- a/interest_monitor_gui.py +++ b/interest_monitor_gui.py @@ -40,7 +40,7 @@ class InterestMonitorApp: # key: stream_id, value: deque([(timestamp, reply_probability), ...]) self.probability_history = {} self.stream_colors = {} # 为每个 stream 分配颜色 - self.stream_display_names = {} # 存储显示名称 (group_name) + self.stream_display_names = {} # 存储显示名称 (group_name) self.selected_stream_id = tk.StringVar() # 用于 Combobox 绑定 # --- 新增：存储其他参数 --- @@ -61,12 +61,11 @@ class InterestMonitorApp: self.single_stream_last_active = tk.StringVar(value="活跃: N/A") self.single_stream_last_interaction = tk.StringVar(value="交互: N/A") - # --- UI 元素 --- # --- 新增：顶部全局信息框架 --- - self.global_info_frame = ttk.Frame(root, padding="5 0 5 5") # 顶部内边距调整 - self.global_info_frame.pack(side=tk.TOP, fill=tk.X, pady=(5, 0)) # 底部外边距为0 + self.global_info_frame = ttk.Frame(root, padding="5 0 5 5") # 顶部内边距调整 + self.global_info_frame.pack(side=tk.TOP, fill=tk.X, pady=(5, 0)) # 底部外边距为0 ttk.Label(self.global_info_frame, text="全局状态:").pack(side=tk.LEFT, padx=(0, 10)) ttk.Label(self.global_info_frame, textvariable=self.latest_mai_state).pack(side=tk.LEFT, padx=5) @@ -75,11 +74,10 @@ class InterestMonitorApp: ttk.Label(self.global_info_frame, text="子流数:").pack(side=tk.LEFT, padx=(10, 0)) ttk.Label(self.global_info_frame, textvariable=self.latest_subflow_count).pack(side=tk.LEFT, padx=5) - # 创建 Notebook (选项卡控件) self.notebook = ttk.Notebook(root) # 修改：fill 和 expand，让 notebook 填充剩余空间 - self.notebook.pack(pady=(5, 0), padx=10, fill=tk.BOTH, expand=1) #顶部外边距改小 + self.notebook.pack(pady=(5, 0), padx=10, fill=tk.BOTH, expand=1) # 顶部外边距改小 # --- 第一个选项卡：所有流 --- self.frame_all = ttk.Frame(self.notebook, padding="5 5 5 5") @@ -87,7 +85,7 @@ class InterestMonitorApp: # 状态标签 (移动到最底部) self.status_label = tk.Label(root, text="Initializing...", anchor="w", fg="grey") - self.status_label.pack(side=tk.BOTTOM, fill=tk.X, padx=10, pady=(0, 5)) # 调整边距 + self.status_label.pack(side=tk.BOTTOM, fill=tk.X, padx=10, pady=(0, 5)) # 调整边距 # Matplotlib 图表设置 (用于第一个选项卡) self.fig = Figure(figsize=(5, 4), dpi=100) @@ -119,10 +117,18 @@ class InterestMonitorApp: self.single_stream_details_frame.pack(side=tk.TOP, fill=tk.X, pady=(0, 5)) ttk.Label(self.single_stream_details_frame, textvariable=self.single_stream_sub_mind).pack(side=tk.LEFT, padx=5) - ttk.Label(self.single_stream_details_frame, textvariable=self.single_stream_chat_state).pack(side=tk.LEFT, padx=5) - ttk.Label(self.single_stream_details_frame, textvariable=self.single_stream_threshold).pack(side=tk.LEFT, padx=5) - ttk.Label(self.single_stream_details_frame, textvariable=self.single_stream_last_active).pack(side=tk.LEFT, padx=5) - ttk.Label(self.single_stream_details_frame, textvariable=self.single_stream_last_interaction).pack(side=tk.LEFT, padx=5) + ttk.Label(self.single_stream_details_frame, textvariable=self.single_stream_chat_state).pack( + side=tk.LEFT, padx=5 + ) + ttk.Label(self.single_stream_details_frame, textvariable=self.single_stream_threshold).pack( + side=tk.LEFT, padx=5 + ) + ttk.Label(self.single_stream_details_frame, textvariable=self.single_stream_last_active).pack( + side=tk.LEFT, padx=5 + ) + ttk.Label(self.single_stream_details_frame, textvariable=self.single_stream_last_interaction).pack( + side=tk.LEFT, padx=5 + ) # Matplotlib 图表设置 (用于第二个选项卡) self.fig_single = Figure(figsize=(5, 4), dpi=100) @@ -176,37 +182,39 @@ class InterestMonitorApp: read_count += 1 try: log_entry = json.loads(line.strip()) - timestamp = log_entry.get("timestamp") # 获取顶层时间戳 + timestamp = log_entry.get("timestamp") # 获取顶层时间戳 # *** 时间过滤 *** if timestamp is None: error_count += 1 - continue # 跳过没有时间戳的行 + continue # 跳过没有时间戳的行 try: entry_timestamp = float(timestamp) if entry_timestamp < time_threshold: continue # 跳过时间过早的条目 except (ValueError, TypeError): error_count += 1 - continue # 跳过时间戳格式错误的行 + continue # 跳过时间戳格式错误的行 # --- 新增：更新顶层信息 (使用最后一个有效行的数据) --- - self.latest_main_mind.set(log_entry.get("main_mind", self.latest_main_mind.get())) # 保留旧值如果缺失 + self.latest_main_mind.set( + log_entry.get("main_mind", self.latest_main_mind.get()) + ) # 保留旧值如果缺失 self.latest_mai_state.set(log_entry.get("mai_state", self.latest_mai_state.get())) self.latest_subflow_count.set(log_entry.get("subflow_count", self.latest_subflow_count.get())) # --- 修改开始：迭代 subflows --- subflows = log_entry.get("subflows") - if not isinstance(subflows, list): # 检查 subflows 是否存在且为列表 + if not isinstance(subflows, list): # 检查 subflows 是否存在且为列表 error_count += 1 - continue # 跳过没有 subflows 或格式无效的行 + continue # 跳过没有 subflows 或格式无效的行 for subflow_entry in subflows: stream_id = subflow_entry.get("stream_id") interest_level = subflow_entry.get("interest_level") # 获取 group_name，如果不存在则回退到 stream_id group_name = subflow_entry.get("group_name", stream_id) - reply_probability = subflow_entry.get("reply_probability") # 获取概率值 + reply_probability = subflow_entry.get("reply_probability") # 获取概率值 # *** 检查必要的字段 *** # 注意：时间戳已在顶层检查过 @@ -218,12 +226,12 @@ class InterestMonitorApp: try: interest_level_float = float(interest_level) except (ValueError, TypeError): - continue # 跳过 interest_level 无效的 subflow + continue # 跳过 interest_level 无效的 subflow # 如果是第一次读到这个 stream_id，则创建 deque if stream_id not in new_stream_history: new_stream_history[stream_id] = deque(maxlen=MAX_HISTORY_POINTS) - new_probability_history[stream_id] = deque(maxlen=MAX_HISTORY_POINTS) # 创建概率 deque + new_probability_history[stream_id] = deque(maxlen=MAX_HISTORY_POINTS) # 创建概率 deque # 检查是否已有颜色，没有则分配 if stream_id not in self.stream_colors: self.stream_colors[stream_id] = self.get_random_color() @@ -235,8 +243,10 @@ class InterestMonitorApp: self.stream_sub_minds[stream_id] = subflow_entry.get("sub_mind", "N/A") self.stream_chat_states[stream_id] = subflow_entry.get("sub_chat_state", "N/A") self.stream_threshold_status[stream_id] = subflow_entry.get("is_above_threshold", False) - self.stream_last_active[stream_id] = subflow_entry.get("last_active_time") # 存储原始时间戳 - self.stream_last_interaction[stream_id] = subflow_entry.get("last_interaction_time") # 存储原始时间戳 + self.stream_last_active[stream_id] = subflow_entry.get("last_active_time") # 存储原始时间戳 + self.stream_last_interaction[stream_id] = subflow_entry.get( + "last_interaction_time" + ) # 存储原始时间戳 # 添加数据点 (使用顶层时间戳) new_stream_history[stream_id].append((entry_timestamp, interest_level_float)) diff --git a/src/heart_flow/background_tasks.py b/src/heart_flow/background_tasks.py index f5784dfa2..7fb63bf66 100644 --- a/src/heart_flow/background_tasks.py +++ b/src/heart_flow/background_tasks.py @@ -46,54 +46,59 @@ class BackgroundTaskManager: self.cleanup_interval = cleanup_interval self.log_interval = log_interval self.inactive_threshold = inactive_threshold # For cleanup task - self.interest_eval_interval = interest_eval_interval # 存储兴趣评估间隔 + self.interest_eval_interval = interest_eval_interval # 存储兴趣评估间隔 # 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._interest_eval_task: Optional[asyncio.Task] = None # 新增兴趣评估任务引用 + self._interest_eval_task: Optional[asyncio.Task] = None # 新增兴趣评估任务引用 self._tasks: List[Optional[asyncio.Task]] = [] # Keep track of all tasks async def start_tasks(self): """启动所有后台任务 - + 功能说明: - 启动核心后台任务: 状态更新、清理、日志记录和兴趣评估 - 每个任务启动前检查是否已在运行 - 将任务引用保存到任务列表 """ - + # 任务配置列表: (任务变量名, 任务函数, 任务名称, 日志级别, 额外日志信息, 任务对象引用属性名) task_configs = [ - (self._state_update_task, - lambda: self._run_state_update_cycle(self.update_interval), - "hf_state_update", - "debug", - f"聊天状态更新任务已启动 间隔:{self.update_interval}s", - "_state_update_task"), - - (self._cleanup_task, - self._run_cleanup_cycle, - "hf_cleanup", - "info", - f"清理任务已启动 间隔:{self.cleanup_interval}s 阈值:{self.inactive_threshold}s", - "_cleanup_task"), - - (self._logging_task, - self._run_logging_cycle, - "hf_logging", - "info", - f"日志任务已启动 间隔:{self.log_interval}s", - "_logging_task"), - + ( + self._state_update_task, + lambda: self._run_state_update_cycle(self.update_interval), + "hf_state_update", + "debug", + f"聊天状态更新任务已启动 间隔:{self.update_interval}s", + "_state_update_task", + ), + ( + self._cleanup_task, + self._run_cleanup_cycle, + "hf_cleanup", + "info", + f"清理任务已启动 间隔:{self.cleanup_interval}s 阈值:{self.inactive_threshold}s", + "_cleanup_task", + ), + ( + self._logging_task, + self._run_logging_cycle, + "hf_logging", + "info", + f"日志任务已启动 间隔:{self.log_interval}s", + "_logging_task", + ), # 新增兴趣评估任务配置 - (self._interest_eval_task, - self._run_interest_eval_cycle, - "hf_interest_eval", - "debug", # 设为debug，避免过多日志 - f"兴趣评估任务已启动 间隔:{self.interest_eval_interval}s", - "_interest_eval_task"), + ( + self._interest_eval_task, + self._run_interest_eval_cycle, + "hf_interest_eval", + "debug", # 设为debug，避免过多日志 + f"兴趣评估任务已启动 间隔:{self.interest_eval_interval}s", + "_interest_eval_task", + ), ] # 统一启动所有任务 @@ -103,9 +108,9 @@ class BackgroundTaskManager: if current_task_var is None or current_task_var.done(): new_task = asyncio.create_task(task_func(), name=task_name) setattr(self, task_attr_name, new_task) # 更新任务变量 - if new_task not in self._tasks: # 避免重复添加 + if new_task not in self._tasks: # 避免重复添加 self._tasks.append(new_task) - + # 根据配置记录不同级别的日志 getattr(logger, log_level)(log_msg) else: @@ -164,7 +169,7 @@ class BackgroundTaskManager: # logger.warning(f"任务 {task_name} 超时执行 ({elapsed:.2f}s > {interval}s)") await asyncio.sleep(sleep_time) - logger.debug(f"任务循环结束: {task_name}") # 调整日志信息 + logger.debug(f"任务循环结束: {task_name}") # 调整日志信息 async def _perform_state_update_work(self): """执行状态更新工作""" @@ -222,6 +227,7 @@ class BackgroundTaskManager: """执行一轮子心流兴趣评估与提升检查。""" # 直接调用 subheartflow_manager 的方法，并传递当前状态信息 await self.subheartflow_manager.evaluate_interest_and_promote(self.mai_state_info) + # --- 结束新增 --- # --- Specific Task Runners --- # @@ -237,14 +243,15 @@ class BackgroundTaskManager: async def _run_logging_cycle(self): await self._run_periodic_loop( - task_name="State Logging", - interval=self.log_interval, - task_func=self._perform_logging_work + task_name="State Logging", interval=self.log_interval, task_func=self._perform_logging_work ) # --- 新增兴趣评估任务运行器 --- async def _run_interest_eval_cycle(self): await self._run_periodic_loop( - task_name="Interest Evaluation", interval=self.interest_eval_interval, task_func=self._perform_interest_eval_work + task_name="Interest Evaluation", + interval=self.interest_eval_interval, + task_func=self._perform_interest_eval_work, ) + # --- 结束新增 --- diff --git a/src/heart_flow/interest_logger.py b/src/heart_flow/interest_logger.py index 15a08b64d..866ae4e01 100644 --- a/src/heart_flow/interest_logger.py +++ b/src/heart_flow/interest_logger.py @@ -60,9 +60,7 @@ class InterestLogger: for subheartflow in all_flows: if self.subheartflow_manager.get_subheartflow(subheartflow.subheartflow_id): tasks.append( - asyncio.create_task( - subheartflow.get_full_state(), name=f"get_state_{subheartflow.subheartflow_id}" - ) + asyncio.create_task(subheartflow.get_full_state(), name=f"get_state_{subheartflow.subheartflow_id}") ) else: logger.warning(f"子心流 {subheartflow.subheartflow_id} 在创建任务前已消失") @@ -107,7 +105,7 @@ class InterestLogger: "main_mind": main_mind, "mai_state": mai_state_name, "subflow_count": len(all_subflow_states), - "subflows": [] + "subflows": [], } if not all_subflow_states: diff --git a/src/heart_flow/sub_heartflow.py b/src/heart_flow/sub_heartflow.py index 0e39946e3..0b17c278c 100644 --- a/src/heart_flow/sub_heartflow.py +++ b/src/heart_flow/sub_heartflow.py @@ -18,6 +18,7 @@ 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.do_tool.tool_use import ToolUser from src.heart_flow.mai_state_manager import MaiStateInfo @@ -109,8 +110,8 @@ class InterestChatting: self.interest_dict: Dict[str, tuple[MessageRecv, float, bool]] = {} self.update_interval = 1.0 - self.start_updates(self.update_interval) # 初始化时启动后台更新任务 - + self.start_updates(self.update_interval) # 初始化时启动后台更新任务 + self.above_threshold = False self.start_hfc_probability = 0.0 @@ -120,36 +121,37 @@ class InterestChatting: async def _calculate_decay(self): """计算兴趣值的衰减 - + 参数: current_time: 当前时间戳 - + 处理逻辑: 1. 计算时间差 2. 处理各种异常情况(负值/零值) 3. 正常计算衰减 4. 更新最后更新时间 """ - + # 处理极小兴趣值情况 if self.interest_level < 1e-9: self.interest_level = 0.0 return - + # 异常情况处理 if self.decay_rate_per_second <= 0: interest_logger.warning(f"衰减率({self.decay_rate_per_second})无效，重置兴趣值为0") self.interest_level = 0.0 return - + # 正常衰减计算 try: decay_factor = math.pow(self.decay_rate_per_second, self.update_interval) self.interest_level *= decay_factor except ValueError as e: - interest_logger.error(f"衰减计算错误: {e} 参数: 衰减率={self.decay_rate_per_second} 时间差={self.update_interval} 当前兴趣={self.interest_level}") + interest_logger.error( + f"衰减计算错误: {e} 参数: 衰减率={self.decay_rate_per_second} 时间差={self.update_interval} 当前兴趣={self.interest_level}" + ) self.interest_level = 0.0 - async def _update_reply_probability(self): self.above_threshold = self.interest_level >= self.trigger_threshold @@ -171,7 +173,7 @@ class InterestChatting: return self.interest_level async def get_state(self) -> dict: - interest = self.interest_level # 直接使用属性值 + interest = self.interest_level # 直接使用属性值 return { "interest_level": round(interest, 2), "last_update_time": self.last_update_time, @@ -181,7 +183,6 @@ class InterestChatting: } async def should_evaluate_reply(self) -> bool: - if self.current_reply_probability > 0: trigger = random.random() < self.current_reply_probability return trigger @@ -195,9 +196,8 @@ class InterestChatting: try: if self.interest_level != 0: await self._calculate_decay() - + await self._update_reply_probability() - # 等待下一个周期或停止事件 await asyncio.wait_for(self._stop_event.wait(), timeout=update_interval) @@ -214,7 +214,6 @@ class InterestChatting: await asyncio.sleep(5) interest_logger.info("InterestChatting 更新循环已停止。") - def start_updates(self, update_interval: float = 1.0): """启动后台更新任务""" if self.update_task is None or self.update_task.done(): @@ -224,12 +223,11 @@ class InterestChatting: else: interest_logger.debug("后台兴趣更新任务已在运行中。") - async def stop_updates(self): """停止后台更新任务""" if self.update_task and not self.update_task.done(): interest_logger.info("正在停止 InterestChatting 后台更新任务...") - self._stop_event.set() # 发送停止信号 + self._stop_event.set() # 发送停止信号 try: # 等待任务结束，设置超时 await asyncio.wait_for(self.update_task, timeout=5.0) @@ -238,7 +236,7 @@ class InterestChatting: interest_logger.warning("停止 InterestChatting 后台任务超时，尝试取消...") self.update_task.cancel() try: - await self.update_task # 等待取消完成 + await self.update_task # 等待取消完成 except asyncio.CancelledError: interest_logger.info("InterestChatting 后台更新任务已被取消。") except Exception as e: @@ -247,6 +245,7 @@ class InterestChatting: self.update_task = None else: interest_logger.debug("InterestChatting 后台更新任务未运行或已完成。") + # --- 结束 新增方法 --- @@ -297,11 +296,11 @@ class SubHeartflow: async def add_time_current_state(self, add_time: float): self.current_state_time += add_time - + async def change_to_state_chat(self): self.current_state_time = 120 self._start_normal_chat() - + async def change_to_state_focused(self): self.current_state_time = 60 self._start_heart_fc_chat() @@ -311,14 +310,14 @@ class SubHeartflow: if self.normal_chat_instance: logger.info(f"{self.log_prefix} 停止 NormalChat 兴趣监控...") try: - await self.normal_chat_instance.stop_chat() # 调用 stop_chat + await self.normal_chat_instance.stop_chat() # 调用 stop_chat except Exception as e: logger.error(f"{self.log_prefix} 停止 NormalChat 监控任务时出错: {e}") logger.error(traceback.format_exc()) async def _start_normal_chat(self) -> bool: """启动 NormalChat 实例及其兴趣监控，确保 HeartFChatting 已停止""" - await self._stop_heart_fc_chat() # 确保专注聊天已停止 + await self._stop_heart_fc_chat() # 确保专注聊天已停止 log_prefix = self.log_prefix try: @@ -334,12 +333,12 @@ class SubHeartflow: logger.info(f"{log_prefix} 创建或更新 NormalChat 实例。") logger.info(f"{log_prefix} 启动 NormalChat 兴趣监控...") - await self.normal_chat_instance.start_chat() # <--- 修正：调用 start_chat + await self.normal_chat_instance.start_chat() # <--- 修正：调用 start_chat return True except Exception as e: logger.error(f"{log_prefix} 启动 NormalChat 时出错: {e}") logger.error(traceback.format_exc()) - self.normal_chat_instance = None # 启动失败，清理实例 + self.normal_chat_instance = None # 启动失败，清理实例 return False async def _stop_heart_fc_chat(self): @@ -357,18 +356,18 @@ class SubHeartflow: async def _start_heart_fc_chat(self) -> bool: """启动 HeartFChatting 实例，确保 NormalChat 已停止""" - await self._stop_normal_chat() # 确保普通聊天监控已停止 - self.clear_interest_dict() # 清理兴趣字典，准备专注聊天 + await self._stop_normal_chat() # 确保普通聊天监控已停止 + self.clear_interest_dict() # 清理兴趣字典，准备专注聊天 log_prefix = self.log_prefix if self.heart_fc_instance: if not self.heart_fc_instance._loop_active: logger.warning(f"{log_prefix} HeartFChatting 实例存在但未激活，尝试重新激活...") - await self.heart_fc_instance.add_time() # 尝试添加时间以激活循环 - return True # 假设 add_time 会处理激活逻辑 + await self.heart_fc_instance.add_time() # 尝试添加时间以激活循环 + return True # 假设 add_time 会处理激活逻辑 else: logger.debug(f"{log_prefix} HeartFChatting 已在运行中。") - return True # 已经在运行 + return True # 已经在运行 logger.info(f"{log_prefix} 麦麦准备开始专注聊天...") try: @@ -381,12 +380,12 @@ class SubHeartflow: return True else: logger.error(f"{log_prefix} HeartFChatting 初始化失败，无法进入专注模式。") - self.heart_fc_instance = None # 初始化失败，清理实例 + self.heart_fc_instance = None # 初始化失败，清理实例 return False except Exception as e: logger.error(f"{log_prefix} 创建或初始化 HeartFChatting 实例时出错: {e}") logger.error(traceback.format_exc()) - self.heart_fc_instance = None # 创建或初始化异常，清理实例 + self.heart_fc_instance = None # 创建或初始化异常，清理实例 return False async def set_chat_state(self, new_state: "ChatState", current_states_num: tuple = ()): @@ -398,7 +397,7 @@ class SubHeartflow: log_prefix = self.log_prefix current_mai_state = self.mai_states.get_current_state() - state_changed = False # 标记状态是否实际发生改变 + state_changed = False # 标记状态是否实际发生改变 # --- 状态转换逻辑 --- if new_state == ChatState.CHAT: @@ -406,8 +405,10 @@ class SubHeartflow: current_chat_count = current_states_num[1] if len(current_states_num) > 1 else 0 if current_chat_count >= normal_limit and current_state != ChatState.CHAT: - logger.debug(f"{log_prefix} 无法从 {current_state.value} 转到 聊天。原因：聊不过来了 ({current_chat_count}/{normal_limit})") - return # 阻止状态转换 + logger.debug( + f"{log_prefix} 无法从 {current_state.value} 转到 聊天。原因：聊不过来了 ({current_chat_count}/{normal_limit})" + ) + return # 阻止状态转换 else: logger.debug(f"{log_prefix} 准备进入或保持 聊天 状态 ({current_chat_count}/{normal_limit})") if await self._start_normal_chat(): @@ -416,15 +417,17 @@ class SubHeartflow: else: logger.error(f"{log_prefix} 启动 NormalChat 失败，无法进入 CHAT 状态。") # 考虑是否需要回滚状态或采取其他措施 - return # 启动失败，不改变状态 + return # 启动失败，不改变状态 elif new_state == ChatState.FOCUSED: focused_limit = current_mai_state.get_focused_chat_max_num() current_focused_count = current_states_num[2] if len(current_states_num) > 2 else 0 - + if current_focused_count >= focused_limit and current_state != ChatState.FOCUSED: - logger.debug(f"{log_prefix} 无法从 {current_state.value} 转到 专注。原因：聊不过来了 ({current_focused_count}/{focused_limit})") - return # 阻止状态转换 + logger.debug( + f"{log_prefix} 无法从 {current_state.value} 转到 专注。原因：聊不过来了 ({current_focused_count}/{focused_limit})" + ) + return # 阻止状态转换 else: logger.debug(f"{log_prefix} 准备进入或保持 专注聊天 状态 ({current_focused_count}/{focused_limit})") if await self._start_heart_fc_chat(): @@ -433,14 +436,13 @@ class SubHeartflow: else: logger.error(f"{log_prefix} 启动 HeartFChatting 失败，无法进入 FOCUSED 状态。") # 启动失败，状态回滚到之前的状态或ABSENT？这里保持不改变 - return # 启动失败，不改变状态 - + return # 启动失败，不改变状态 elif new_state == ChatState.ABSENT: logger.info(f"{log_prefix} 进入 ABSENT 状态，停止所有聊天活动...") await self._stop_normal_chat() await self._stop_heart_fc_chat() - state_changed = True # 总是可以成功转换到 ABSENT + state_changed = True # 总是可以成功转换到 ABSENT # --- 更新状态和最后活动时间 --- if state_changed: @@ -449,7 +451,9 @@ class SubHeartflow: self.last_active_time = time.time() else: # 如果因为某些原因（如启动失败）没有成功改变状态，记录一下 - logger.debug(f"{log_prefix} 尝试将状态从 {current_state.value} 变为 {new_state.value}，但未成功或未执行更改。") + logger.debug( + f"{log_prefix} 尝试将状态从 {current_state.value} 变为 {new_state.value}，但未成功或未执行更改。" + ) async def subheartflow_start_working(self): """启动子心流的后台任务 @@ -462,7 +466,6 @@ class SubHeartflow: while not self.should_stop: await asyncio.sleep(30) # 30秒检查一次停止标志 - logger.info(f"{self.log_prefix} 子心流后台任务已停止。") diff --git a/src/heart_flow/subheartflow_manager.py b/src/heart_flow/subheartflow_manager.py index e19172f45..344eaf772 100644 --- a/src/heart_flow/subheartflow_manager.py +++ b/src/heart_flow/subheartflow_manager.py @@ -18,7 +18,7 @@ from .observation import ChattingObservation subheartflow_manager_log_config = LogConfig( console_format=SUBHEARTFLOW_MANAGER_STYLE_CONFIG["console_format"], - file_format=SUBHEARTFLOW_MANAGER_STYLE_CONFIG["file_format"], + file_format=SUBHEARTFLOW_MANAGER_STYLE_CONFIG["file_format"], ) logger = get_module_logger("subheartflow_manager", config=subheartflow_manager_log_config) @@ -271,7 +271,9 @@ class SubHeartflowManager: current_state_enum = current_mai_state.get_current_state() focused_limit = current_state_enum.get_focused_chat_max_num() if focused_limit <= 0: - logger.debug(f"{log_prefix_manager} 当前状态 ({current_state_enum.value}) 不允许 FOCUSED 子心流, 跳过提升检查。") + logger.debug( + f"{log_prefix_manager} 当前状态 ({current_state_enum.value}) 不允许 FOCUSED 子心流, 跳过提升检查。" + ) return # 获取当前 FOCUSED 状态的数量 (初始值) @@ -280,7 +282,7 @@ class SubHeartflowManager: # 使用快照安全遍历 subflows_snapshot = list(self.subheartflows.values()) - promoted_count = 0 # 记录本次提升的数量 + promoted_count = 0 # 记录本次提升的数量 try: for sub_hf in subflows_snapshot: flow_id = sub_hf.subheartflow_id @@ -300,8 +302,10 @@ class SubHeartflowManager: # 注意：在循环内部再次获取当前数量，因为之前的提升可能已经改变了计数 # 使用已经记录并在循环中更新的 current_focused_count if current_focused_count >= focused_limit: - logger.debug(f"{log_prefix_manager} {log_prefix_flow} 达到专注上限 ({current_focused_count}/{focused_limit}), 无法提升。概率={sub_hf.interest_chatting.start_hfc_probability:.2f}") - continue # 跳过这个子心流，继续检查下一个 + logger.debug( + f"{log_prefix_manager} {log_prefix_flow} 达到专注上限 ({current_focused_count}/{focused_limit}), 无法提升。概率={sub_hf.interest_chatting.start_hfc_probability:.2f}" + ) + continue # 跳过这个子心流，继续检查下一个 # --- 执行提升 --- # 获取当前实例以检查最新状态 (防御性编程) @@ -310,16 +314,18 @@ class SubHeartflowManager: logger.warning(f"{log_prefix_manager} {log_prefix_flow} 尝试提升时状态已改变或实例消失，跳过。") continue - logger.info(f"{log_prefix_manager} {log_prefix_flow} 兴趣评估触发升级 (prob={sub_hf.interest_chatting.start_hfc_probability:.2f}, 上限:{focused_limit}, 当前:{current_focused_count}) -> FOCUSED") + logger.info( + f"{log_prefix_manager} {log_prefix_flow} 兴趣评估触发升级 (prob={sub_hf.interest_chatting.start_hfc_probability:.2f}, 上限:{focused_limit}, 当前:{current_focused_count}) -> FOCUSED" + ) states_num = ( self.count_subflows_by_state(ChatState.ABSENT), - self.count_subflows_by_state(ChatState.CHAT), # 这个值在提升前计算 - current_focused_count, # 这个值在提升前计算 + self.count_subflows_by_state(ChatState.CHAT), # 这个值在提升前计算 + current_focused_count, # 这个值在提升前计算 ) # --- 状态设置 --- - original_state = current_subflow.chat_state.chat_status # 记录原始状态 + original_state = current_subflow.chat_state.chat_status # 记录原始状态 await current_subflow.set_chat_state(ChatState.FOCUSED, states_num) # --- 状态验证 --- @@ -327,15 +333,21 @@ class SubHeartflowManager: if final_subflow: final_state = final_subflow.chat_state.chat_status if final_state == ChatState.FOCUSED: - logger.debug(f"{log_prefix_manager} {log_prefix_flow} 成功从 {original_state.value} 升级到 FOCUSED 状态") + logger.debug( + f"{log_prefix_manager} {log_prefix_flow} 成功从 {original_state.value} 升级到 FOCUSED 状态" + ) promoted_count += 1 # 提升成功后，更新当前专注计数，以便后续检查能使用最新值 current_focused_count += 1 - elif final_state == original_state: # 状态未变 - logger.warning(f"{log_prefix_manager} {log_prefix_flow} 尝试从 {original_state.value} 升级 FOCUSED 失败，状态仍为: {final_state.value} (可能被内部逻辑阻止)") - else: # 状态变成其他了? - logger.warning(f"{log_prefix_manager} {log_prefix_flow} 尝试从 {original_state.value} 升级 FOCUSED 后状态变为 {final_state.value}") - else: # 子心流消失了? + elif final_state == original_state: # 状态未变 + logger.warning( + f"{log_prefix_manager} {log_prefix_flow} 尝试从 {original_state.value} 升级 FOCUSED 失败，状态仍为: {final_state.value} (可能被内部逻辑阻止)" + ) + else: # 状态变成其他了? + logger.warning( + f"{log_prefix_manager} {log_prefix_flow} 尝试从 {original_state.value} 升级 FOCUSED 后状态变为 {final_state.value}" + ) + else: # 子心流消失了? logger.warning(f"{log_prefix_manager} {log_prefix_flow} 升级后验证时子心流 {flow_id} 消失") except Exception as e: diff --git a/src/plugins/heartFC_chat/heartFC_chat.py b/src/plugins/heartFC_chat/heartFC_chat.py index ffed743ab..b34788664 100644 --- a/src/plugins/heartFC_chat/heartFC_chat.py +++ b/src/plugins/heartFC_chat/heartFC_chat.py @@ -73,10 +73,7 @@ class HeartFChatting: 现在由其关联的 SubHeartflow 管理生命周期。 """ - def __init__( - self, - chat_id: str - ): + def __init__(self, chat_id: str): """ HeartFChatting 初始化函数 @@ -195,7 +192,7 @@ class HeartFChatting: # 检查是否满足启动条件：未激活且计时器有时间 if not self._loop_active and self._loop_timer > 0: should_start_loop = True - self._loop_active = True # 在锁内标记为活动，防止重复启动 + self._loop_active = True # 在锁内标记为活动，防止重复启动 if should_start_loop: # 检查是否已有任务在运行（理论上不应该，因为 _loop_active=False） @@ -206,8 +203,8 @@ class HeartFChatting: # 等待旧任务确实被取消 await asyncio.wait_for(self._loop_task, timeout=0.5) except (asyncio.CancelledError, asyncio.TimeoutError): - pass # 忽略取消或超时错误 - self._loop_task = None # 清理旧任务引用 + pass # 忽略取消或超时错误 + self._loop_task = None # 清理旧任务引用 logger.info(f"{log_prefix} 计时器 > 0 且循环未激活，启动主循环...") # 创建新的循环任务 @@ -215,7 +212,7 @@ class HeartFChatting: # 添加完成回调 self._loop_task.add_done_callback(self._handle_loop_completion) # else: - # logger.trace(f"{log_prefix} 不需要启动循环（已激活或计时器为0）") # 可以取消注释以进行调试 + # logger.trace(f"{log_prefix} 不需要启动循环（已激活或计时器为0）") # 可以取消注释以进行调试 def _handle_loop_completion(self, task: asyncio.Task): """当 _run_pf_loop 任务完成时执行的回调。""" diff --git a/src/plugins/heartFC_chat/heartflow_processor.py b/src/plugins/heartFC_chat/heartflow_processor.py index 30ac61c5b..366bb1252 100644 --- a/src/plugins/heartFC_chat/heartflow_processor.py +++ b/src/plugins/heartFC_chat/heartflow_processor.py @@ -69,7 +69,6 @@ class HeartFCProcessor: subheartflow = await heartflow.create_subheartflow(chat.stream_id) - message.update_chat_stream(chat) await heartflow.create_subheartflow(chat.stream_id) diff --git a/src/plugins/heartFC_chat/normal_chat.py b/src/plugins/heartFC_chat/normal_chat.py index 4acacd6d1..c020b4076 100644 --- a/src/plugins/heartFC_chat/normal_chat.py +++ b/src/plugins/heartFC_chat/normal_chat.py @@ -349,7 +349,7 @@ class NormalChat: return False # 改为实例方法, 移除 chat 参数 - + async def start_chat(self): """为此 NormalChat 实例关联的 ChatStream 启动聊天任务（如果尚未运行）。""" if self._chat_task is None or self._chat_task.done(): @@ -358,7 +358,6 @@ class NormalChat: task.add_done_callback(lambda t: self._handle_task_completion(t)) # 回调现在是实例方法 self._chat_task = task - # 改为实例方法, 移除 stream_id 参数 def _handle_task_completion(self, task: asyncio.Task): """兴趣监控任务完成时的回调函数。""" @@ -403,4 +402,3 @@ class NormalChat: # 确保任务状态更新，即使等待出错 (回调函数也会尝试更新) if self._chat_task is task: self._chat_task = None - diff --git a/src/plugins/message/api.py b/src/plugins/message/api.py index 31302f510..fb51539e2 100644 --- a/src/plugins/message/api.py +++ b/src/plugins/message/api.py @@ -148,7 +148,9 @@ class MessageServer(BaseMessageHandler): if self.own_app: # 如果使用自己的 FastAPI 实例，运行 uvicorn 服务器 # 禁用 uvicorn 默认日志和访问日志 - config = uvicorn.Config(self.app, host=self.host, port=self.port, loop="asyncio", log_config=None, access_log=False) + config = uvicorn.Config( + self.app, host=self.host, port=self.port, loop="asyncio", log_config=None, access_log=False + ) self.server = uvicorn.Server(config) await self.server.serve() else: