feat：基于不同心流单独的发送器实例，反正能跑，但我也不知道能不能跑

睡觉

src/do_tool/tool_use.py

@@ -3,10 +3,11 @@ from src.config.config import global_config
 import json
 from src.common.logger import get_module_logger, TOOL_USE_STYLE_CONFIG, LogConfig
 from src.do_tool.tool_can_use import get_all_tool_definitions, get_tool_instance
-from src.heart_flow.sub_heartflow import SubHeartflow
 import traceback
 from src.plugins.person_info.relationship_manager import relationship_manager
 from src.plugins.chat.utils import parse_text_timestamps
+from src.plugins.chat.chat_stream import ChatStream
+from src.heart_flow.observation import ChattingObservation
 
 tool_use_config = LogConfig(
     # 使用消息发送专用样式
@@ -23,7 +24,7 @@ class ToolUser:
         )
 
     @staticmethod
-    async def _build_tool_prompt(message_txt: str, subheartflow: SubHeartflow = None):
+    async def _build_tool_prompt(message_txt: str, chat_stream: ChatStream = None, observation: ChattingObservation = None):
         """构建工具使用的提示词
 
         Args:
@@ -34,8 +35,8 @@ class ToolUser:
             str: 构建好的提示词
         """
 
-        if subheartflow:
+        if observation:
-            mid_memory_info = subheartflow.observations[0].mid_memory_info
+            mid_memory_info = observation.mid_memory_info
             # print(f"intol111111111111111111111111111111111222222222222mid_memory_info：{mid_memory_info}")
 
         # 这些信息应该从调用者传入，而不是从self获取
@@ -102,14 +103,14 @@ class ToolUser:
             logger.error(f"执行工具调用时发生错误: {str(e)}")
             return None
 
-    async def use_tool(self, message_txt: str, sub_heartflow: SubHeartflow = None):
+    async def use_tool(self, message_txt: str, chat_stream: ChatStream = None, observation: ChattingObservation = None):
         """使用工具辅助思考，判断是否需要额外信息
 
         Args:
             message_txt: 用户消息文本
             sender_name: 发送者名称
             chat_stream: 聊天流对象
-            sub_heartflow: 子心流对象（可选）
+            observation: 观察对象（可选）
 
         Returns:
             dict: 工具使用结果，包含结构化的信息
@@ -118,7 +119,8 @@ class ToolUser:
             # 构建提示词
             prompt = await self._build_tool_prompt(
                 message_txt=message_txt,
-                subheartflow=sub_heartflow,
+                chat_stream=chat_stream,
+                observation=observation,
             )
 
             # 定义可用工具

src/heart_flow/heartflow.py

@@ -20,7 +20,6 @@ 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
-from src.plugins.heartFC_chat.heartflow_message_sender import MessageManager
 # --- End imports ---
 
 heartflow_config = LogConfig(
@@ -66,9 +65,9 @@ LOG_INTERVAL_SECONDS = 3  # 日志记录间隔 (例如：3秒) - 保持与 inter
 
 # --- 新增：状态更新常量 ---
 STATE_UPDATE_INTERVAL_SECONDS = 30 # 状态更新检查间隔（秒）
-FIVE_MINUTES = 3 * 60
+FIVE_MINUTES = 1 * 60
-FIFTEEN_MINUTES = 6 * 60
+FIFTEEN_MINUTES = 5 * 60
-TWENTY_MINUTES = 9 * 60
+TWENTY_MINUTES = 10 * 60
 # --- 结束新增常量 ---
 
 
@@ -111,7 +110,7 @@ class MaiState(enum.Enum):
 class MaiStateInfo:
     def __init__(self):
 
-        # 使用枚举类型初始化状态，默认为不在线
+        # 使用枚举类型初始化状态，默认为正常聊天
         self.mai_status: MaiState = MaiState.OFFLINE
         self.mai_status_history = [] # 历史状态，包含 状态，最后时间
         self.last_status_change_time: float = time.time() # 新增：状态最后改变时间
@@ -157,7 +156,6 @@ class Heartflow:
         self.tool_user_instance = ToolUser()
         self.emoji_manager_instance = emoji_manager # Module instance
         self.relationship_manager_instance = relationship_manager # Module instance
-        self.message_manager_instance = MessageManager() # Instantiate the message manager
         # --- End moved dependencies ---
 
         # --- Background Task Management ---
@@ -463,7 +461,13 @@ class Heartflow:
         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():
@@ -646,12 +650,24 @@ class Heartflow:
         return list(self._subheartflows.keys())
 
     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}")
 
+            # --- 新增：在取消任务和删除前，先设置状态为 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}")
+                 # 即使出错，仍继续尝试停止任务和移除
+            # --- 结束新增逻辑 ---
+
             # 标记停止并取消任务
             subheartflow.should_stop = True
             task_to_cancel = subheartflow.task

src/heart_flow/sub_heartflow.py

@@ -16,11 +16,13 @@ from ..plugins.utils.prompt_builder import Prompt, global_prompt_manager
 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
 
 # Type hinting for circular dependency
 if TYPE_CHECKING:
     from .heartflow import Heartflow, MaiState # 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
@@ -230,15 +232,38 @@ class InterestChatting:
 
 class SubHeartflow:
     def __init__(self, subheartflow_id, parent_heartflow: 'Heartflow'):
+        """子心流初始化函数
+        
+        Args:
+            subheartflow_id: 子心流唯一标识符
+            parent_heartflow: 父级心流实例
+        """
+        # 基础属性
         self.subheartflow_id = subheartflow_id
         self.parent_heartflow = parent_heartflow
-
+        self.bot_name = global_config.BOT_NICKNAME  # 机器人昵称
-        self.current_mind = "你什么也没想"
+        
-        self.past_mind = []
+        # 思维状态相关
-        self.chat_state: ChatStateInfo = ChatStateInfo()
+        self.current_mind = "你什么也没想"  # 当前想法
-
+        self.past_mind = []  # 历史想法记录
-        self.interest_chatting = InterestChatting(state_change_callback=self.set_chat_state)
+        self.main_heartflow_info = ""  # 主心流信息
-
+        
+        # 聊天状态管理
+        self.chat_state: ChatStateInfo = ChatStateInfo()  # 聊天状态信息
+        self.interest_chatting = InterestChatting(state_change_callback=self.set_chat_state)  # 兴趣聊天系统
+        
+        # 活动状态管理
+        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.observations: List[ChattingObservation] = []  # 观察列表
+        self.running_knowledges = []  # 运行中的知识
+        
+        # LLM模型配置
         self.llm_model = LLMRequest(
             model=global_config.llm_sub_heartflow,
             temperature=global_config.llm_sub_heartflow["temp"],
@@ -246,33 +271,19 @@ class SubHeartflow:
             request_type="sub_heart_flow",
         )
 
-        self.main_heartflow_info = ""
+    async def set_chat_state(self, new_state: 'ChatState'):
-
+        """更新sub_heartflow的聊天状态，并管理 HeartFChatting 实例"""
-        self.last_active_time = time.time()
-        self.should_stop = False
-        self.task: Optional[asyncio.Task] = None
-
-        self.is_active = False
-
-        self.observations: List[ChattingObservation] = []
-
-        self.running_knowledges = []
-
-        self.bot_name = global_config.BOT_NICKNAME
-        logger.info(f"SubHeartflow {self.subheartflow_id} created with initial state: {self.chat_state.chat_status.value}")
-
-    def set_chat_state(self, new_state: 'ChatState'):
-        """更新sub_heartflow的聊天状态"""
 
         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
 
         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
 
-        # --- Limit Check before entering CHAT state --- #
+        # --- Entering CHAT state ---
         if new_state == ChatState.CHAT:
-            current_mai_state = self.parent_heartflow.current_state.mai_status
             normal_limit = current_mai_state.get_normal_chat_max_num()
             current_chat_count = self.parent_heartflow.count_subflows_by_state(ChatState.CHAT)
 
@@ -281,15 +292,61 @@ class SubHeartflow:
                 return # Block the state transition
             else:
                 logger.debug(f"{log_prefix} 允许从 {current_state.value} 转换到 CHAT (上限: {normal_limit}, 当前: {current_chat_count})" )
+                # If transitioning out of FOCUSED, shut down HeartFChatting first
+                if current_state == ChatState.FOCUSED and self.heart_fc_instance:
+                    logger.info(f"{log_prefix} 从 FOCUSED 转换到 CHAT，正在关闭 HeartFChatting...")
+                    await self.heart_fc_instance.shutdown()
+                    self.heart_fc_instance = None
 
-        # 如果检查通过或目标状态不是CHAT，则进行状态变更
+        # --- 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)
+
+            if current_focused_count >= focused_limit:
+                logger.debug(f"{log_prefix} 拒绝从 {current_state.value} 转换到 FOCUSED。原因：FOCUSED 状态已达上限 ({focused_limit})。当前数量: {current_focused_count}")
+                return # Block the state transition
+            else:
+                logger.debug(f"{log_prefix} 允许从 {current_state.value} 转换到 FOCUSED (上限: {focused_limit}, 当前: {current_focused_count})" )
+                if not self.heart_fc_instance:
+                    logger.info(f"{log_prefix} 状态转为 FOCUSED，创建并初始化 HeartFChatting 实例...")
+                    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,
+                        )
+                        # 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 实例已创建并启动。")
+                        else:
+                            logger.error(f"{log_prefix} HeartFChatting 实例初始化失败，状态回滚到 {current_state.value}")
+                            self.heart_fc_instance = None
+                            return # Prevent state change if HeartFChatting fails to init
+                    except Exception as e:
+                        logger.error(f"{log_prefix} 创建 HeartFChatting 实例时出错: {e}")
+                        logger.error(traceback.format_exc())
+                        self.heart_fc_instance = None
+                        return # Prevent state change on error
+
+                else:
+                    logger.warning(f"{log_prefix} 尝试进入 FOCUSED 状态，但 HeartFChatting 实例已存在。")
+
+        # --- 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
+
+
+        # --- Update state and timestamp if transition is allowed --- # 更新状态必须放在所有检查和操作之后
         self.chat_state.chat_status = new_state
-        # 状态变更时更新最后活跃时间
+        self.last_active_time = time.time()
-        self.last_active_time = time.time()  
         logger.info(f"{log_prefix} 聊天状态从 {current_state.value} 变更为 {new_state.value}")
 
-        # TODO: 考虑从FOCUSED状态转出时是否需要停止HeartFChatting
-        # 这部分逻辑可能更适合放在Heartflow的_stop_subheartflow或HeartFCController的循环中处理
 
     async def subheartflow_start_working(self):
         while True:
@@ -297,7 +354,8 @@ class SubHeartflow:
                 logger.info(f"子心流 {self.subheartflow_id} 被标记为停止，正在退出后台任务...")
                 break
 
-            await asyncio.sleep(global_config.sub_heart_flow_update_interval)
+            # await asyncio.sleep(global_config.sub_heart_flow_update_interval)
+            await asyncio.sleep(10)
 
     async def ensure_observed(self):
         observation = self._get_primary_observation()

src/main.py

@@ -9,7 +9,7 @@ from .plugins.willing.willing_manager import willing_manager
 from .plugins.chat.chat_stream import chat_manager
 from .heart_flow.heartflow import heartflow
 from .plugins.memory_system.Hippocampus import HippocampusManager
-from .plugins.chat.messagesender import message_manager
+from .plugins.chat.message_sender import message_manager
 from .plugins.storage.storage import MessageStorage
 from .config.config import global_config
 from .plugins.chat.bot import chat_bot
@@ -101,9 +101,9 @@ class MainSystem:
         logger.success("个体特征初始化成功")
 
         try:
-            # 启动 Heartflow 的 MessageManager (负责消息发送/排队)
+            # 启动全局消息管理器 (负责消息发送/排队)
-            await heartflow.message_manager_instance.start()
+            await message_manager.start()
-            logger.success("心流消息管理器启动成功")
+            logger.success("全局消息管理器启动成功")
 
             # 启动心流系统主循环
             asyncio.create_task(heartflow.heartflow_start_working())

src/plugins/chat/__init__.py

@@ -1,7 +1,7 @@
 from .emoji_manager import emoji_manager
 from ..person_info.relationship_manager import relationship_manager
 from .chat_stream import chat_manager
-from .messagesender import message_manager
+from .message_sender import message_manager
 from ..storage.storage import MessageStorage
 
 

src/plugins/chat/message.py

@@ -290,6 +290,7 @@ class MessageSending(MessageProcessBase):
         is_head: bool = False,
         is_emoji: bool = False,
         thinking_start_time: float = 0,
+        apply_set_reply_logic: bool = False,
     ):
         # 调用父类初始化
         super().__init__(
@@ -306,6 +307,7 @@ class MessageSending(MessageProcessBase):
         self.reply_to_message_id = reply.message_info.message_id if reply else None
         self.is_head = is_head
         self.is_emoji = is_emoji
+        self.apply_set_reply_logic = apply_set_reply_logic
 
     def set_reply(self, reply: Optional["MessageRecv"] = None) -> None:
         """设置回复消息"""

src/plugins/chat/message_sender.py

@@ -0,0 +1,343 @@
+# src/plugins/chat/message_sender.py
+import asyncio
+import time
+from typing import Dict, List, Optional, Union
+
+from src.common.logger import get_module_logger
+# from ...common.database import db # 数据库依赖似乎不需要了，注释掉
+from ..message.api import global_api
+from .message import MessageSending, MessageThinking, MessageSet
+
+from ..storage.storage import MessageStorage
+from ...config.config import global_config
+from .utils import truncate_message, calculate_typing_time, count_messages_between
+
+from src.common.logger import LogConfig, SENDER_STYLE_CONFIG
+
+# 定义日志配置
+sender_config = LogConfig(
+    # 使用消息发送专用样式
+    console_format=SENDER_STYLE_CONFIG["console_format"],
+    file_format=SENDER_STYLE_CONFIG["file_format"],
+)
+
+logger = get_module_logger("msg_sender", config=sender_config)
+
+
+class MessageSender:
+    """发送器 (不再是单例)"""
+
+    def __init__(self):
+        self.message_interval = (0.5, 1)  # 消息间隔时间范围(秒)
+        self.last_send_time = 0
+        self._current_bot = None
+
+    def set_bot(self, bot):
+        """设置当前bot实例"""
+        pass
+
+    async def send_via_ws(self, message: MessageSending) -> None:
+        """通过 WebSocket 发送消息"""
+        try:
+            await global_api.send_message(message)
+        except Exception as e:
+            logger.error(f"WS发送失败: {e}")
+            raise ValueError(f"未找到平台：{message.message_info.platform} 的url配置，请检查配置文件") from e
+
+    async def send_message(
+        self,
+        message: MessageSending,
+    ) -> None:
+        """发送消息（核心发送逻辑）"""
+
+        # --- 添加计算打字和延迟的逻辑 (从 heartflow_message_sender 移动并调整) ---
+        typing_time = calculate_typing_time(
+            input_string=message.processed_plain_text,
+            thinking_start_time=message.thinking_start_time,
+            is_emoji=message.is_emoji,
+        )
+        # logger.trace(f"{message.processed_plain_text},{typing_time},计算输入时间结束") # 减少日志
+        await asyncio.sleep(typing_time)
+        # logger.trace(f"{message.processed_plain_text},{typing_time},等待输入时间结束") # 减少日志
+        # --- 结束打字延迟 ---
+
+        message_json = message.to_dict()
+        message_preview = truncate_message(message.processed_plain_text)
+
+        try:
+            end_point = global_config.api_urls.get(message.message_info.platform, None)
+            if end_point:
+                try:
+                    await global_api.send_message_rest(end_point, message_json)
+                except Exception as e:
+                    logger.error(f"REST发送失败: {str(e)}")
+                    logger.info(f"[{message.chat_stream.stream_id}] 尝试使用WS发送")
+                    await self.send_via_ws(message)
+            else:
+                await self.send_via_ws(message)
+            logger.success(f"发送消息   '{message_preview}'   成功") # 调整日志格式
+        except Exception as e:
+            logger.error(f"发送消息   '{message_preview}'   失败: {str(e)}")
+
+
+class MessageContainer:
+    """单个聊天流的发送/思考消息容器"""
+
+    def __init__(self, chat_id: str, max_size: int = 100):
+        self.chat_id = chat_id
+        self.max_size = max_size
+        self.messages: List[Union[MessageThinking, MessageSending]] = [] # 明确类型
+        self.last_send_time = 0
+        self.thinking_wait_timeout = 20  # 思考等待超时时间（秒） - 从旧 sender 合并
+
+    def count_thinking_messages(self) -> int:
+        """计算当前容器中思考消息的数量"""
+        return sum(1 for msg in self.messages if isinstance(msg, MessageThinking))
+
+    def get_timeout_sending_messages(self) -> List[MessageSending]:
+        """获取所有超时的MessageSending对象（思考时间超过20秒），按thinking_start_time排序 - 从旧 sender 合并"""
+        current_time = time.time()
+        timeout_messages = []
+
+        for msg in self.messages:
+            # 只检查 MessageSending 类型
+            if isinstance(msg, MessageSending):
+                # 确保 thinking_start_time 有效
+                if msg.thinking_start_time and current_time - msg.thinking_start_time > self.thinking_wait_timeout:
+                    timeout_messages.append(msg)
+
+        # 按thinking_start_time排序，时间早的在前面
+        timeout_messages.sort(key=lambda x: x.thinking_start_time)
+        return timeout_messages
+
+    def get_earliest_message(self) -> Optional[Union[MessageThinking, MessageSending]]:
+        """获取thinking_start_time最早的消息对象"""
+        if not self.messages:
+            return None
+        earliest_time = float("inf")
+        earliest_message = None
+        for msg in self.messages:
+            # 确保消息有 thinking_start_time 属性
+            msg_time = getattr(msg, 'thinking_start_time', float('inf'))
+            if msg_time < earliest_time:
+                earliest_time = msg_time
+                earliest_message = msg
+        return earliest_message
+
+    def add_message(self, message: Union[MessageThinking, MessageSending, MessageSet]) -> None:
+        """添加消息到队列"""
+        if isinstance(message, MessageSet):
+            for single_message in message.messages:
+                self.messages.append(single_message)
+        else:
+            self.messages.append(message)
+
+    def remove_message(self, message_to_remove: Union[MessageThinking, MessageSending]) -> bool:
+        """移除指定的消息对象，如果消息存在则返回True，否则返回False"""
+        try:
+            initial_len = len(self.messages)
+            # 使用列表推导式或 filter 创建新列表，排除要删除的元素
+            # self.messages = [msg for msg in self.messages if msg is not message_to_remove]
+            # 或者直接 remove (如果确定对象唯一性)
+            if message_to_remove in self.messages:
+                 self.messages.remove(message_to_remove)
+                 return True
+            # logger.debug(f"Removed message {getattr(message_to_remove, 'message_info', {}).get('message_id', 'UNKNOWN')}. Old len: {initial_len}, New len: {len(self.messages)}")
+            # return len(self.messages) < initial_len
+            return False
+
+        except Exception as e:
+            logger.exception(f"移除消息时发生错误: {e}")
+            return False
+
+    def has_messages(self) -> bool:
+        """检查是否有待发送的消息"""
+        return bool(self.messages)
+
+    def get_all_messages(self) -> List[Union[MessageSending, MessageThinking]]:
+        """获取所有消息"""
+        return list(self.messages) # 返回副本
+
+
+class MessageManager:
+    """管理所有聊天流的消息容器 (不再是单例)"""
+
+    def __init__(self):
+        self.containers: Dict[str, MessageContainer] = {}
+        self.storage = MessageStorage() # 添加 storage 实例
+        self._running = True # 处理器运行状态
+        self._container_lock = asyncio.Lock() # 保护 containers 字典的锁
+        # self.message_sender = MessageSender() # 创建发送器实例 (改为全局实例)
+
+    async def start(self):
+        """启动后台处理器任务。"""
+        # 检查是否已有任务在运行，避免重复启动
+        if hasattr(self, '_processor_task') and not self._processor_task.done():
+             logger.warning("Processor task already running.")
+             return
+        self._processor_task = asyncio.create_task(self._start_processor_loop())
+        logger.info("MessageManager processor task started.")
+
+    def stop(self):
+         """停止后台处理器任务。"""
+         self._running = False
+         if hasattr(self, '_processor_task') and not self._processor_task.done():
+              self._processor_task.cancel()
+              logger.info("MessageManager processor task stopping.")
+         else:
+              logger.info("MessageManager processor task not running or already stopped.")
+
+
+    async def get_container(self, chat_id: str) -> MessageContainer:
+        """获取或创建聊天流的消息容器 (异步，使用锁)"""
+        async with self._container_lock:
+            if chat_id not in self.containers:
+                self.containers[chat_id] = MessageContainer(chat_id)
+            return self.containers[chat_id]
+
+    async def add_message(self, message: Union[MessageThinking, MessageSending, MessageSet]) -> None:
+        """添加消息到对应容器"""
+        chat_stream = message.chat_stream
+        if not chat_stream:
+            logger.error("消息缺少 chat_stream，无法添加到容器")
+            return # 或者抛出异常
+        container = await self.get_container(chat_stream.stream_id)
+        container.add_message(message)
+
+    def check_if_sending_message_exist(self, chat_id, thinking_id):
+        """检查指定聊天流的容器中是否存在具有特定 thinking_id 的 MessageSending 消息 或 emoji 消息"""
+        # 这个方法现在是非异步的，因为它只读取数据
+        container = self.containers.get(chat_id) # 直接 get，因为读取不需要锁
+        if container and container.has_messages():
+            for message in container.get_all_messages():
+                if isinstance(message, MessageSending):
+                    msg_id = getattr(message.message_info, 'message_id', None)
+                    # 检查 message_id 是否匹配 thinking_id 或以 "me" 开头 (emoji)
+                    if msg_id == thinking_id or (msg_id and msg_id.startswith("me")):
+                        # logger.debug(f"检查到存在相同thinking_id或emoji的消息: {msg_id} for {thinking_id}")
+                        return True
+        return False
+
+    async def _handle_sending_message(self, container: MessageContainer, message: MessageSending):
+        """处理单个 MessageSending 消息 (包含 set_reply 逻辑)"""
+        try:
+            _ = message.update_thinking_time() # 更新思考时间
+            thinking_start_time = message.thinking_start_time
+            now_time = time.time()
+            thinking_messages_count, thinking_messages_length = count_messages_between(
+                start_time=thinking_start_time, end_time=now_time, stream_id=message.chat_stream.stream_id
+            )
+
+            # --- 条件应用 set_reply 逻辑 ---
+            if (
+                message.apply_set_reply_logic # 检查标记
+                and message.is_head
+                and (thinking_messages_count > 4 or thinking_messages_length > 250)
+                and not message.is_private_message()
+            ):
+                logger.debug(f"[{message.chat_stream.stream_id}] 应用 set_reply 逻辑: {message.processed_plain_text[:20]}...")
+                message.set_reply()
+            # --- 结束条件 set_reply ---
+
+            await message.process() # 预处理消息内容
+
+            # 使用全局 message_sender 实例
+            await message_sender.send_message(message)
+            await self.storage.store_message(message, message.chat_stream)
+
+            # 移除消息要在发送 *之后*
+            container.remove_message(message)
+            # logger.debug(f"[{message.chat_stream.stream_id}] Sent and removed message: {message.message_info.message_id}")
+
+        except Exception as e:
+            logger.error(f"[{message.chat_stream.stream_id}] 处理发送消息 {getattr(message.message_info, 'message_id', 'N/A')} 时出错: {e}")
+            logger.exception("详细错误信息:")
+            # 考虑是否移除出错的消息，防止无限循环
+            removed = container.remove_message(message)
+            if removed:
+                 logger.warning(f"[{message.chat_stream.stream_id}] 已移除处理出错的消息。")
+
+
+    async def _process_chat_messages(self, chat_id: str):
+        """处理单个聊天流消息 (合并后的逻辑)"""
+        container = await self.get_container(chat_id) # 获取容器是异步的了
+
+        if container.has_messages():
+            message_earliest = container.get_earliest_message()
+
+            if not message_earliest: # 如果最早消息为空，则退出
+                return
+
+            if isinstance(message_earliest, MessageThinking):
+                # --- 处理思考消息 (来自旧 sender) ---
+                message_earliest.update_thinking_time()
+                thinking_time = message_earliest.thinking_time
+                # 减少控制台刷新频率或只在时间显著变化时打印
+                if int(thinking_time) % 5 == 0: # 每5秒打印一次
+                    print(
+                        f"消息 {message_earliest.message_info.message_id} 正在思考中，已思考 {int(thinking_time)} 秒\r",
+                        end="",
+                        flush=True,
+                    )
+
+                # 检查是否超时
+                if thinking_time > global_config.thinking_timeout:
+                    logger.warning(f"[{chat_id}] 消息思考超时 ({thinking_time:.1f}秒)，移除消息 {message_earliest.message_info.message_id}")
+                    container.remove_message(message_earliest)
+                    print() # 超时后换行，避免覆盖下一条日志
+
+            elif isinstance(message_earliest, MessageSending):
+                # --- 处理发送消息 ---
+                await self._handle_sending_message(container, message_earliest)
+
+            # --- 处理超时发送消息 (来自旧 sender) ---
+            # 在处理完最早的消息后，检查是否有超时的发送消息
+            timeout_sending_messages = container.get_timeout_sending_messages()
+            if timeout_sending_messages:
+                logger.debug(f"[{chat_id}] 发现 {len(timeout_sending_messages)} 条超时的发送消息")
+                for msg in timeout_sending_messages:
+                    # 确保不是刚刚处理过的最早消息 (虽然理论上应该已被移除，但以防万一)
+                    if msg is message_earliest:
+                        continue
+                    logger.info(f"[{chat_id}] 处理超时发送消息: {msg.message_info.message_id}")
+                    await self._handle_sending_message(container, msg) # 复用处理逻辑
+
+        # 清理空容器 (可选)
+        # async with self._container_lock:
+        #     if not container.has_messages() and chat_id in self.containers:
+        #         logger.debug(f"[{chat_id}] 容器已空，准备移除。")
+        #         del self.containers[chat_id]
+
+
+    async def _start_processor_loop(self):
+        """消息处理器主循环"""
+        while self._running:
+            tasks = []
+            # 使用异步锁保护迭代器创建过程
+            async with self._container_lock:
+                # 创建 keys 的快照以安全迭代
+                chat_ids = list(self.containers.keys())
+
+            for chat_id in chat_ids:
+                 # 为每个 chat_id 创建一个处理任务
+                 tasks.append(asyncio.create_task(self._process_chat_messages(chat_id)))
+
+            if tasks:
+                try:
+                    # 等待当前批次的所有任务完成
+                    await asyncio.gather(*tasks)
+                except Exception as e:
+                    logger.error(f"消息处理循环 gather 出错: {e}")
+
+            # 等待一小段时间，避免CPU空转
+            try:
+                await asyncio.sleep(0.1) # 稍微降低轮询频率
+            except asyncio.CancelledError:
+                 logger.info("Processor loop sleep cancelled.")
+                 break # 退出循环
+        logger.info("MessageManager processor loop finished.")
+
+# --- 创建全局实例 ---
+message_manager = MessageManager()
+message_sender = MessageSender()
+# --- 结束全局实例 --- 

src/plugins/chat/messagesender.py

@@ -1,291 +0,0 @@
-import asyncio
-import time
-from typing import Dict, List, Optional, Union
-
-from src.common.logger import get_module_logger
-from ...common.database import db
-from ..message.api import global_api
-from .message import MessageSending, MessageThinking, MessageSet
-
-from ..storage.storage import MessageStorage
-from ...config.config import global_config
-from .utils import truncate_message, calculate_typing_time, count_messages_between
-
-from src.common.logger import LogConfig, SENDER_STYLE_CONFIG
-
-# 定义日志配置
-sender_config = LogConfig(
-    # 使用消息发送专用样式
-    console_format=SENDER_STYLE_CONFIG["console_format"],
-    file_format=SENDER_STYLE_CONFIG["file_format"],
-)
-
-logger = get_module_logger("msg_sender", config=sender_config)
-
-
-class MessageSender:
-    """发送器"""
-
-    def __init__(self):
-        self.message_interval = (0.5, 1)  # 消息间隔时间范围(秒)
-        self.last_send_time = 0
-        self._current_bot = None
-
-    def set_bot(self, bot):
-        """设置当前bot实例"""
-        pass
-
-    @staticmethod
-    def get_recalled_messages(stream_id: str) -> list:
-        """获取所有撤回的消息"""
-        recalled_messages = []
-
-        recalled_messages = list(db.recalled_messages.find({"stream_id": stream_id}, {"message_id": 1}))
-        # 按thinking_start_time排序，时间早的在前面
-        return recalled_messages
-
-    @staticmethod
-    async def send_via_ws(message: MessageSending) -> None:
-        try:
-            await global_api.send_message(message)
-        except Exception as e:
-            raise ValueError(f"未找到平台：{message.message_info.platform} 的url配置，请检查配置文件") from e
-
-    async def send_message(
-        self,
-        message: MessageSending,
-    ) -> None:
-        """发送消息"""
-
-        if isinstance(message, MessageSending):
-            recalled_messages = self.get_recalled_messages(message.chat_stream.stream_id)
-            is_recalled = False
-            for recalled_message in recalled_messages:
-                if message.reply_to_message_id == recalled_message["message_id"]:
-                    is_recalled = True
-                    logger.warning(f"消息“{message.processed_plain_text}”已被撤回，不发送")
-                    break
-            if not is_recalled:
-                # print(message.processed_plain_text + str(message.is_emoji))
-                typing_time = calculate_typing_time(
-                    input_string=message.processed_plain_text,
-                    thinking_start_time=message.thinking_start_time,
-                    is_emoji=message.is_emoji,
-                )
-                logger.trace(f"{message.processed_plain_text},{typing_time},计算输入时间结束")
-                await asyncio.sleep(typing_time)
-                logger.trace(f"{message.processed_plain_text},{typing_time},等待输入时间结束")
-
-                message_json = message.to_dict()
-
-                message_preview = truncate_message(message.processed_plain_text)
-                try:
-                    end_point = global_config.api_urls.get(message.message_info.platform, None)
-                    if end_point:
-                        # logger.info(f"发送消息到{end_point}")
-                        # logger.info(message_json)
-                        try:
-                            await global_api.send_message_rest(end_point, message_json)
-                        except Exception as e:
-                            logger.error(f"REST方式发送失败，出现错误: {str(e)}")
-                            logger.info("尝试使用ws发送")
-                            await self.send_via_ws(message)
-                    else:
-                        await self.send_via_ws(message)
-                    logger.success(f"发送消息“{message_preview}”成功")
-                except Exception as e:
-                    logger.error(f"发送消息“{message_preview}”失败: {str(e)}")
-
-
-class MessageContainer:
-    """单个聊天流的发送/思考消息容器"""
-
-    def __init__(self, chat_id: str, max_size: int = 100):
-        self.chat_id = chat_id
-        self.max_size = max_size
-        self.messages = []
-        self.last_send_time = 0
-        self.thinking_wait_timeout = 20  # 思考等待超时时间（秒）
-
-    def get_timeout_messages(self) -> List[MessageSending]:
-        """获取所有超时的Message_Sending对象（思考时间超过20秒），按thinking_start_time排序"""
-        current_time = time.time()
-        timeout_messages = []
-
-        for msg in self.messages:
-            if isinstance(msg, MessageSending):
-                if current_time - msg.thinking_start_time > self.thinking_wait_timeout:
-                    timeout_messages.append(msg)
-
-        # 按thinking_start_time排序，时间早的在前面
-        timeout_messages.sort(key=lambda x: x.thinking_start_time)
-
-        return timeout_messages
-
-    def get_earliest_message(self) -> Optional[Union[MessageThinking, MessageSending]]:
-        """获取thinking_start_time最早的消息对象"""
-        if not self.messages:
-            return None
-        earliest_time = float("inf")
-        earliest_message = None
-        for msg in self.messages:
-            msg_time = msg.thinking_start_time
-            if msg_time < earliest_time:
-                earliest_time = msg_time
-                earliest_message = msg
-        return earliest_message
-
-    def add_message(self, message: Union[MessageThinking, MessageSending]) -> None:
-        """添加消息到队列"""
-        if isinstance(message, MessageSet):
-            for single_message in message.messages:
-                self.messages.append(single_message)
-        else:
-            self.messages.append(message)
-
-    def remove_message(self, message: Union[MessageThinking, MessageSending]) -> bool:
-        """移除消息，如果消息存在则返回True，否则返回False"""
-        try:
-            if message in self.messages:
-                self.messages.remove(message)
-                return True
-            return False
-        except Exception:
-            logger.exception("移除消息时发生错误")
-            return False
-
-    def has_messages(self) -> bool:
-        """检查是否有待发送的消息"""
-        return bool(self.messages)
-
-    def get_all_messages(self) -> List[Union[MessageSending, MessageThinking]]:
-        """获取所有消息"""
-        return list(self.messages)
-
-
-class MessageManager:
-    """管理所有聊天流的消息容器"""
-
-    def __init__(self):
-        self.containers: Dict[str, MessageContainer] = {}  # chat_id -> MessageContainer
-        self.storage = MessageStorage()
-        self._running = True
-
-    def get_container(self, chat_id: str) -> MessageContainer:
-        """获取或创建聊天流的消息容器"""
-        if chat_id not in self.containers:
-            self.containers[chat_id] = MessageContainer(chat_id)
-        return self.containers[chat_id]
-
-    def add_message(self, message: Union[MessageThinking, MessageSending, MessageSet]) -> None:
-        chat_stream = message.chat_stream
-        if not chat_stream:
-            raise ValueError("无法找到对应的聊天流")
-        container = self.get_container(chat_stream.stream_id)
-        container.add_message(message)
-
-    async def process_chat_messages(self, chat_id: str):
-        """处理聊天流消息"""
-        container = self.get_container(chat_id)
-        if container.has_messages():
-            # print(f"处理有message的容器chat_id: {chat_id}")
-            message_earliest = container.get_earliest_message()
-
-            if isinstance(message_earliest, MessageThinking):
-                """取得了思考消息"""
-                message_earliest.update_thinking_time()
-                thinking_time = message_earliest.thinking_time
-                # print(thinking_time)
-                print(
-                    f"消息正在思考中，已思考{int(thinking_time)}秒\r",
-                    end="",
-                    flush=True,
-                )
-
-                # 检查是否超时
-                if thinking_time > global_config.thinking_timeout:
-                    logger.warning(f"消息思考超时({thinking_time}秒)，移除该消息")
-                    container.remove_message(message_earliest)
-
-            else:
-                """取得了发送消息"""
-                thinking_time = message_earliest.update_thinking_time()
-                thinking_start_time = message_earliest.thinking_start_time
-                now_time = time.time()
-                thinking_messages_count, thinking_messages_length = count_messages_between(
-                    start_time=thinking_start_time, end_time=now_time, stream_id=message_earliest.chat_stream.stream_id
-                )
-                # print(thinking_time)
-                # print(thinking_messages_count)
-                # print(thinking_messages_length)
-
-                if (
-                    message_earliest.is_head
-                    and (thinking_messages_count > 4 or thinking_messages_length > 250)
-                    and not message_earliest.is_private_message()  # 避免在私聊时插入reply
-                ):
-                    logger.debug(f"设置回复消息{message_earliest.processed_plain_text}")
-                    message_earliest.set_reply()
-
-                await message_earliest.process()
-
-                # print(f"message_earliest.thinking_start_tim22222e:{message_earliest.thinking_start_time}")
-
-                await message_sender.send_message(message_earliest)
-
-                await self.storage.store_message(message_earliest, message_earliest.chat_stream)
-
-                container.remove_message(message_earliest)
-
-            message_timeout = container.get_timeout_messages()
-            if message_timeout:
-                logger.debug(f"发现{len(message_timeout)}条超时消息")
-                for msg in message_timeout:
-                    if msg == message_earliest:
-                        continue
-
-                    try:
-                        thinking_time = msg.update_thinking_time()
-                        thinking_start_time = msg.thinking_start_time
-                        now_time = time.time()
-                        thinking_messages_count, thinking_messages_length = count_messages_between(
-                            start_time=thinking_start_time, end_time=now_time, stream_id=msg.chat_stream.stream_id
-                        )
-                        # print(thinking_time)
-                        # print(thinking_messages_count)
-                        # print(thinking_messages_length)
-                        if (
-                            msg.is_head
-                            and (thinking_messages_count > 4 or thinking_messages_length > 250)
-                            and not msg.is_private_message()  # 避免在私聊时插入reply
-                        ):
-                            logger.debug(f"设置回复消息{msg.processed_plain_text}")
-                            msg.set_reply()
-
-                        await msg.process()
-
-                        await message_sender.send_message(msg)
-
-                        await self.storage.store_message(msg, msg.chat_stream)
-
-                        if not container.remove_message(msg):
-                            logger.warning("尝试删除不存在的消息")
-                    except Exception:
-                        logger.exception("处理超时消息时发生错误")
-                        continue
-
-    async def start_processor(self):
-        """启动消息处理器"""
-        while self._running:
-            await asyncio.sleep(1)
-            tasks = []
-            for chat_id in self.containers.keys():
-                tasks.append(self.process_chat_messages(chat_id))
-
-            await asyncio.gather(*tasks)
-
-
-# 创建全局消息管理器实例
-message_manager = MessageManager()
-# 创建全局发送器实例
-message_sender = MessageSender()

src/plugins/heartFC_chat/heartFC_chat.py

@@ -7,7 +7,6 @@ from src.plugins.chat.message import MessageRecv, BaseMessageInfo, MessageThinki
 from src.plugins.chat.message import MessageSet, Seg  # Local import needed after move
 from src.plugins.chat.chat_stream import ChatStream
 from src.plugins.chat.message import UserInfo
-from src.heart_flow.heartflow import heartflow, SubHeartflow
 from src.plugins.chat.chat_stream import chat_manager
 from src.common.logger import get_module_logger, LogConfig, PFC_STYLE_CONFIG  # 引入 DEFAULT_CONFIG
 from src.plugins.models.utils_model import LLMRequest
@@ -18,7 +17,7 @@ from src.plugins.utils.timer_calculater import Timer  # <--- Import Timer
 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 .heartflow_message_sender import MessageManager # Assuming this is the type
+from ..chat.message_sender import message_manager # <-- Import the global manager
 # --- End import ---
 
 
@@ -37,7 +36,8 @@ logger = get_module_logger("HeartFCLoop", config=interest_log_config)  # Logger
 if TYPE_CHECKING:
     # Keep this if HeartFCController methods are still needed elsewhere,
     # but the instance variable will be removed from HeartFChatting
-    from .heartFC_controler import HeartFCController
+    # from .heartFC_controler import HeartFCController
+    from src.heart_flow.heartflow import SubHeartflow, heartflow # <-- 同时导入 heartflow 实例用于类型检查
 
 PLANNER_TOOL_DEFINITION = [
     {
@@ -76,54 +76,51 @@ class HeartFChatting:
 
     def __init__(self,
                 chat_id: str,
-                # --- Explicit Dependencies ---
+                # 显式依赖注入
-                gpt_instance: ResponseGenerator,
+                gpt_instance: ResponseGenerator,  # 文本回复生成器
-                tool_user_instance: ToolUser,
+                tool_user_instance: ToolUser,     # 工具使用实例
-                emoji_manager_instance: EmojiManager,
+                emoji_manager_instance: EmojiManager,  # 表情管理实例
-                message_manager_instance: MessageManager
-                # --- End Explicit Dependencies ---
                 ):
         """
-        初始化HeartFChatting实例。
+        HeartFChatting 初始化函数
-
+        
-        Args:
+        参数:
-            chat_id: The identifier for the chat stream (e.g., stream_id).
+            chat_id: 聊天流唯一标识符(如stream_id)
-            gpt_instance: The ResponseGenerator instance for generating text replies.
+            gpt_instance: 文本回复生成器实例
-            tool_user_instance: The ToolUser instance for using tools.
+            tool_user_instance: 工具使用实例
-            emoji_manager_instance: The EmojiManager instance for handling emojis.
+            emoji_manager_instance: 表情管理实例
-            message_manager_instance: The MessageManager instance for sending/managing messages.
         """
-        self.stream_id: str = chat_id
+        # 基础属性
-        self.chat_stream: Optional[ChatStream] = None
+        self.stream_id: str = chat_id  # 聊天流ID
-        self.sub_hf: Optional[SubHeartflow] = None
+        self.chat_stream: Optional[ChatStream] = None  # 关联的聊天流
-        self._initialized = False
+        self.sub_hf: Optional[SubHeartflow] = None  # 关联的子心流
-        self._init_lock = asyncio.Lock()  # Ensure initialization happens only once
+        
-        self._processing_lock = asyncio.Lock()  # 确保只有一个 Plan-Replier-Sender 周期在运行
+        # 初始化状态控制
-        self._timer_lock = asyncio.Lock()  # 用于安全更新计时器
+        self._initialized = False  # 是否已初始化标志
+        self._init_lock = asyncio.Lock()  # 初始化锁(确保只初始化一次)
+        self._processing_lock = asyncio.Lock()  # 处理锁(确保单次Plan-Replier-Sender周期)
+        self._timer_lock = asyncio.Lock()  # 计时器锁(安全更新计时器)
 
-        # --- Store Dependencies ---
+        # 依赖注入存储
-        self.gpt_instance = gpt_instance
+        self.gpt_instance = gpt_instance  # 文本回复生成器
-        self.tool_user = tool_user_instance
+        self.tool_user = tool_user_instance  # 工具使用实例
-        self.emoji_manager = emoji_manager_instance
+        self.emoji_manager = emoji_manager_instance  # 表情管理实例
-        self.message_manager = message_manager_instance
-        # --- End Store Dependencies ---
 
-
+        # LLM规划器配置
-        # Access LLM config through global_config or pass if needed
         self.planner_llm = LLMRequest(
             model=global_config.llm_normal,
             temperature=global_config.llm_normal["temp"],
             max_tokens=1000,
-            request_type="action_planning",
+            request_type="action_planning",  # 用于动作规划
         )
 
-        # Internal state for loop control
+        # 循环控制内部状态
-        self._loop_timer: float = 0.0  # Remaining time for the loop in seconds
+        self._loop_timer: float = 0.0  # 循环剩余时间(秒)
-        self._loop_active: bool = False  # Is the loop currently running?
+        self._loop_active: bool = False  # 循环是否正在运行
-        self._loop_task: Optional[asyncio.Task] = None  # Stores the main loop task
+        self._loop_task: Optional[asyncio.Task] = None  # 主循环任务
-        self._trigger_count_this_activation: int = 0  # Counts triggers within an active period
+        self._trigger_count_this_activation: int = 0  # 当前激活周期内的触发计数
         self._initial_duration: float = INITIAL_DURATION  # 首次触发增加的时间
-        self._last_added_duration: float = self._initial_duration  # <--- 新增：存储上次增加的时间
+        self._last_added_duration: float = self._initial_duration  # 上次增加的时间
 
     def _get_log_prefix(self) -> str:
         """获取日志前缀，包含可读的流名称"""
@@ -146,6 +143,8 @@ class HeartFChatting:
                     logger.error(f"{log_prefix} 获取ChatStream失败。")
                     return False
 
+                # <-- 在这里导入 heartflow 实例
+                from src.heart_flow.heartflow import heartflow
                 self.sub_hf = heartflow.get_subheartflow(self.stream_id)
                 if not self.sub_hf:
                     logger.warning(f"{log_prefix} 获取SubHeartflow失败。一些功能可能受限。")
@@ -245,7 +244,7 @@ class HeartFChatting:
                 cycle_timers = {}  # <--- Initialize timers dict for this cycle
 
                 # Access MessageManager directly
-                if self.message_manager.check_if_sending_message_exist(self.stream_id, thinking_id):
+                if message_manager.check_if_sending_message_exist(self.stream_id, thinking_id):
                     # logger.info(f"{log_prefix} HeartFChatting: 11111111111111111111111111111111麦麦还在发消息，等会再规划")
                     await asyncio.sleep(1)
                     continue
@@ -318,7 +317,7 @@ class HeartFChatting:
                                             )
                                     except Exception as e_replier:
                                         logger.error(f"{log_prefix} 循环: 回复器工作失败: {e_replier}")
-                                        self._cleanup_thinking_message(thinking_id)
+                                        # self._cleanup_thinking_message(thinking_id) <-- Remove cleanup call
 
                                     if replier_result:
                                         # --- Sender Work --- #
@@ -334,10 +333,10 @@ class HeartFChatting:
                                         except Exception as e_sender:
                                             logger.error(f"{log_prefix} 循环: 发送器失败: {e_sender}")
                                             # _sender should handle cleanup, but double check
-                                            # self._cleanup_thinking_message(thinking_id)
+                                            # self._cleanup_thinking_message(thinking_id) <-- Remove cleanup call
                                     else:
                                         logger.warning(f"{log_prefix} 循环: 回复器未产生结果. 跳过发送.")
-                                        self._cleanup_thinking_message(thinking_id)
+                                        # self._cleanup_thinking_message(thinking_id) <-- Remove cleanup call
                         elif action == "emoji_reply":
                             logger.info(
                                 f"{log_prefix} HeartFChatting: 麦麦决定回复表情 ('{emoji_query}'). 理由: {reasoning}"
@@ -652,16 +651,25 @@ class HeartFChatting:
             logger.error(traceback.format_exc())
             return None
 
-    def _cleanup_thinking_message(self, thinking_id: str):
+    # def _cleanup_thinking_message(self, thinking_id: str):
-        """Safely removes the thinking message."""
+    #     """Safely removes the thinking message."""
-        log_prefix = self._get_log_prefix()
+    #     log_prefix = self._get_log_prefix()
-        try:
+    #     try:
-            # Access MessageManager directly
+    #         # Access MessageManager directly
-            container = self.message_manager.get_container(self.stream_id)
+    #         container = await message_manager.get_container(self.stream_id)
-            container.remove_message(thinking_id, msg_type=MessageThinking)
+    #         # container.remove_message(thinking_id, msg_type=MessageThinking) # Need to find the message object first
-            logger.debug(f"{log_prefix} Cleaned up thinking message {thinking_id}.")
+    #         found_msg = None
-        except Exception as e:
+    #         for msg in container.get_all_messages():
-            logger.error(f"{log_prefix} Error cleaning up thinking message {thinking_id}: {e}")
+    #             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(
@@ -774,10 +782,10 @@ class HeartFChatting:
             # 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(
-                self.sub_hf,
+                current_mind_info=self.sub_hf.current_mind,
-                reason,
+                reason=reason,
-                anchor_message,  # Pass anchor_message positionally (matches 'message' parameter)
+                message=anchor_message,  # Pass anchor_message positionally (matches 'message' parameter)
-                thinking_id,  # Pass thinking_id positionally
+                thinking_id=thinking_id,  # Pass thinking_id positionally
             )
 
             if not response_set:
@@ -818,7 +826,7 @@ class HeartFChatting:
             thinking_start_time=thinking_time_point,
         )
         # Access MessageManager directly
-        self.message_manager.add_message(thinking_message)
+        await message_manager.add_message(thinking_message)
         return thinking_id
 
     async def _send_response_messages(
@@ -831,7 +839,7 @@ class HeartFChatting:
 
         chat = anchor_message.chat_stream
         # Access MessageManager directly
-        container = self.message_manager.get_container(chat.stream_id)
+        container = await message_manager.get_container(chat.stream_id)
         thinking_message = None
 
         # 移除思考消息
@@ -875,7 +883,7 @@ class HeartFChatting:
             message_set.add_message(bot_message)
 
         # Access MessageManager directly
-        self.message_manager.add_message(message_set)
+        await message_manager.add_message(message_set)
         return first_bot_msg
 
     async def _handle_emoji(self, anchor_message: Optional[MessageRecv], response_set: List[str], send_emoji: str = ""):
@@ -917,4 +925,4 @@ class HeartFChatting:
                 is_emoji=True,
             )
             # Access MessageManager directly
-            self.message_manager.add_message(bot_message)
+            await message_manager.add_message(bot_message)

src/plugins/heartFC_chat/heartFC_generator.py

@@ -11,7 +11,6 @@ from src.plugins.respon_info_catcher.info_catcher import info_catcher_manager
 from ..utils.timer_calculater import Timer
 
 from src.plugins.moods.moods import MoodManager
-from src.heart_flow.sub_heartflow import SubHeartflow
 # 定义日志配置
 llm_config = LogConfig(
     # 使用消息发送专用样式
@@ -39,7 +38,7 @@ class ResponseGenerator:
 
     async def generate_response(
         self,
-        sub_hf: SubHeartflow,
+        current_mind_info: str,
         reason: str,
         message: MessageRecv,
         thinking_id: str,
@@ -56,7 +55,7 @@ class ResponseGenerator:
             current_model = self.model_normal
             current_model.temperature = global_config.llm_normal["temp"] * arousal_multiplier  # 激活度越高，温度越高
             model_response = await self._generate_response_with_model(
-                sub_hf, reason, message, current_model, thinking_id
+                current_mind_info, reason, message, current_model, thinking_id
             )
 
         if model_response:
@@ -71,7 +70,7 @@ class ResponseGenerator:
             return None
 
     async def _generate_response_with_model(
-        self, sub_hf: SubHeartflow, reason: str, message: MessageRecv, model: LLMRequest, thinking_id: str
+        self, current_mind_info: str, reason: str, message: MessageRecv, model: LLMRequest, thinking_id: str
     ) -> str:
         sender_name = ""
 
@@ -84,9 +83,10 @@ class ResponseGenerator:
             prompt = await prompt_builder.build_prompt(
                 build_mode="focus",
                 reason=reason,
+                current_mind_info=current_mind_info,
                 message_txt=message.processed_plain_text,
                 sender_name=sender_name,
-                subheartflow=sub_hf
+                chat_stream=message.chat_stream
             )
         logger.info(f"构建prompt时间: {t_build_prompt.human_readable}")
 

src/plugins/heartFC_chat/heartflow_message_sender.py

@@ -1,241 +0,0 @@
-import asyncio
-import time
-from typing import Dict, List, Optional, Union
-
-from src.common.logger import get_module_logger
-from ..message.api import global_api
-from ..chat.message import MessageSending, MessageThinking, MessageSet
-from ..storage.storage import MessageStorage
-from ...config.config import global_config
-from ..chat.utils import truncate_message, calculate_typing_time, count_messages_between
-
-from src.common.logger import LogConfig, SENDER_STYLE_CONFIG
-
-# 定义日志配置
-sender_config = LogConfig(
-    # 使用消息发送专用样式
-    console_format=SENDER_STYLE_CONFIG["console_format"],
-    file_format=SENDER_STYLE_CONFIG["file_format"],
-)
-
-logger = get_module_logger("msg_sender", config=sender_config)
-
-
-class MessageSender:
-    """发送器"""
-
-    _instance = None
-
-    def __new__(cls, *args, **kwargs):
-        if cls._instance is None:
-            cls._instance = super(MessageSender, cls).__new__(cls, *args, **kwargs)
-        return cls._instance
-
-    def __init__(self):
-        # 确保 __init__ 只被调用一次
-        if not hasattr(self, "_initialized"):
-            self.message_interval = (0.5, 1)  # 消息间隔时间范围(秒)
-            self.last_send_time = 0
-            self._current_bot = None
-            self._initialized = True
-
-    def set_bot(self, bot):
-        """设置当前bot实例"""
-        pass
-
-    async def send_via_ws(self, message: MessageSending) -> None:
-        try:
-            await global_api.send_message(message)
-        except Exception as e:
-            raise ValueError(f"未找到平台：{message.message_info.platform} 的url配置，请检查配置文件") from e
-
-    async def send_message(
-        self,
-        message: MessageSending,
-    ) -> None:
-        """发送消息"""
-
-        message_json = message.to_dict()
-
-        message_preview = truncate_message(message.processed_plain_text)
-        try:
-            end_point = global_config.api_urls.get(message.message_info.platform, None)
-            if end_point:
-                try:
-                    await global_api.send_message_rest(end_point, message_json)
-                except Exception as e:
-                    logger.error(f"REST方式发送失败，出现错误: {str(e)}")
-                    logger.info("尝试使用ws发送")
-                    await self.send_via_ws(message)
-            else:
-                await self.send_via_ws(message)
-            logger.success(f"发送消息   {message_preview}   成功")
-        except Exception as e:
-            logger.error(f"发送消息   {message_preview}   失败: {str(e)}")
-
-
-class MessageContainer:
-    """单个聊天流的发送/思考消息容器"""
-
-    def __init__(self, chat_id: str, max_size: int = 100):
-        self.chat_id = chat_id
-        self.max_size = max_size
-        self.messages = []
-        self.last_send_time = 0
-
-    def count_thinking_messages(self) -> int:
-        """计算当前容器中思考消息的数量"""
-        return sum(1 for msg in self.messages if isinstance(msg, MessageThinking))
-
-    def get_earliest_message(self) -> Optional[Union[MessageThinking, MessageSending]]:
-        """获取thinking_start_time最早的消息对象"""
-        if not self.messages:
-            return None
-        earliest_time = float("inf")
-        earliest_message = None
-        for msg in self.messages:
-            msg_time = msg.thinking_start_time
-            if msg_time < earliest_time:
-                earliest_time = msg_time
-                earliest_message = msg
-        return earliest_message
-
-    def add_message(self, message: Union[MessageThinking, MessageSending]) -> None:
-        """添加消息到队列"""
-        if isinstance(message, MessageSet):
-            for single_message in message.messages:
-                self.messages.append(single_message)
-        else:
-            self.messages.append(message)
-
-    def remove_message(self, message: Union[MessageThinking, MessageSending]) -> bool:
-        """移除消息，如果消息存在则返回True，否则返回False"""
-        try:
-            if message in self.messages:
-                self.messages.remove(message)
-                return True
-            return False
-        except Exception:
-            logger.exception("移除消息时发生错误")
-            return False
-
-    def has_messages(self) -> bool:
-        """检查是否有待发送的消息"""
-        return bool(self.messages)
-
-    def get_all_messages(self) -> List[Union[MessageSending, MessageThinking]]:
-        """获取所有消息"""
-        return list(self.messages)
-
-
-class MessageManager:
-    """管理所有聊天流的消息容器"""
-
-    _instance = None
-    _lock = asyncio.Lock()
-
-    def __init__(self):
-        if MessageManager._instance is not None:
-            raise Exception("This class is a singleton!")
-        else:
-            self.containers: Dict[str, MessageContainer] = {}
-            self._container_lock = asyncio.Lock()
-            self.running = True
-            MessageManager._instance = self
-
-    async def start(self):
-        """Starts the background processor task."""
-        asyncio.create_task(self.start_processor())
-        logger.info("MessageManager processor task started.")
-
-    def get_container(self, chat_id: str) -> MessageContainer:
-        """获取或创建聊天流的消息容器"""
-        if chat_id not in self.containers:
-            self.containers[chat_id] = MessageContainer(chat_id)
-        return self.containers[chat_id]
-
-    def add_message(self, message: Union[MessageThinking, MessageSending, MessageSet]) -> None:
-        chat_stream = message.chat_stream
-        if not chat_stream:
-            raise ValueError("无法找到对应的聊天流")
-        container = self.get_container(chat_stream.stream_id)
-        container.add_message(message)
-
-    def check_if_sending_message_exist(self, chat_id, thinking_id):
-        """检查指定聊天流的容器中是否存在具有特定 thinking_id 的 MessageSending 消息"""
-        container = self.get_container(chat_id)
-        if container.has_messages():
-            for message in container.get_all_messages():
-                # 首先确保是 MessageSending 类型
-                if isinstance(message, MessageSending):
-                    # 然后再访问 message_info.message_id
-                    # 检查 message_id 是否匹配 thinking_id 或以 "me" 开头
-                    if message.message_info.message_id == thinking_id or message.message_info.message_id[:2] == "me":
-                        # print(f"检查到存在相同thinking_id的消息: {message.message_info.message_id}???{thinking_id}")
-
-                        return True
-        return False
-
-    async def process_chat_messages(self, chat_id: str):
-        """处理聊天流消息"""
-        container = self.get_container(chat_id)
-        if container.has_messages():
-            # print(f"处理有message的容器chat_id: {chat_id}")
-            message_earliest = container.get_earliest_message()
-
-            if isinstance(message_earliest, MessageThinking):
-                """取得了思考消息"""
-                message_earliest.update_thinking_time()
-                thinking_time = message_earliest.thinking_time
-                # print(thinking_time)
-                print(
-                    f"消息正在思考中，已思考{int(thinking_time)}秒\r",
-                    end="",
-                    flush=True,
-                )
-
-                # 检查是否超时
-                if thinking_time > global_config.thinking_timeout:
-                    logger.warning(f"消息思考超时({thinking_time}秒)，移除该消息")
-                    container.remove_message(message_earliest)
-
-            else:
-                """取得了发送消息"""
-                thinking_time = message_earliest.update_thinking_time()
-                thinking_start_time = message_earliest.thinking_start_time
-                now_time = time.time()
-                thinking_messages_count, thinking_messages_length = count_messages_between(
-                    start_time=thinking_start_time, end_time=now_time, stream_id=message_earliest.chat_stream.stream_id
-                )
-
-                await message_earliest.process()
-
-                # 获取 MessageSender 的单例实例并发送消息
-                typing_time = calculate_typing_time(
-                    input_string=message_earliest.processed_plain_text,
-                    thinking_start_time=message_earliest.thinking_start_time,
-                    is_emoji=message_earliest.is_emoji,
-                )
-                await asyncio.sleep(typing_time)
-
-                await MessageSender().send_message(message_earliest)
-                await self.storage.store_message(message_earliest, message_earliest.chat_stream)
-
-                container.remove_message(message_earliest)
-
-    async def start_processor(self):
-        """启动消息处理器"""
-        while self.running:
-            await asyncio.sleep(1)
-            tasks = []
-            for chat_id in list(self.containers.keys()):  # 使用 list 复制 key，防止在迭代时修改字典
-                tasks.append(self.process_chat_messages(chat_id))
-
-            if tasks:  # 仅在有任务时执行 gather
-                await asyncio.gather(*tasks)
-
-
-# # 创建全局消息管理器实例 # 已改为单例模式
-# message_manager = MessageManager()
-# # 创建全局发送器实例 # 已改为单例模式
-# message_sender = MessageSender()

src/plugins/heartFC_chat/heartflow_prompt_builder.py

@@ -1,6 +1,5 @@
 import random
 from typing import Optional
-from src.heart_flow.sub_heartflow import SubHeartflow
 from ...config.config import global_config
 from ..chat.chat_stream import chat_manager
 from src.common.logger import get_module_logger
@@ -82,23 +81,20 @@ class PromptBuilder:
 
     
     async def build_prompt(
-        self, build_mode,reason, message_txt: str, sender_name: str = "某人",subheartflow: SubHeartflow =None
+        self, build_mode,reason,current_mind_info, message_txt: str, sender_name: str = "某人",chat_stream=None
     ) -> tuple[str, str]:
-        
+
-        chat_stream = chat_manager.get_stream(subheartflow.subheartflow_id)
-        
         if build_mode == "normal":
-            return await self._build_prompt_normal(chat_stream, message_txt, sender_name, subheartflow)
+            return await self._build_prompt_normal(chat_stream, message_txt, sender_name)
         
         elif build_mode == "focus":
-            return await self._build_prompt_focus(reason, chat_stream, message_txt, sender_name, subheartflow)
+            return await self._build_prompt_focus(reason, current_mind_info, chat_stream, message_txt, sender_name)
 
 
     
     async def _build_prompt_focus(
-        self, reason, chat_stream, message_txt: str, sender_name: str = "某人", subheartflow: SubHeartflow =None
+        self, reason, current_mind_info, chat_stream, message_txt: str, sender_name: str = "某人"
     ) -> tuple[str, str]:
-        current_mind_info = subheartflow.current_mind
 
         individuality = Individuality.get_instance()
         prompt_personality = individuality.get_prompt(type="personality", x_person=2, level=1)
@@ -107,7 +103,6 @@ class PromptBuilder:
         # 日程构建
         # schedule_prompt = f'''你现在正在做的事情是：{bot_schedule.get_current_num_task(num = 1,time_info = False)}'''
 
-        chat_stream = chat_manager.get_stream(subheartflow.subheartflow_id)
         if chat_stream.group_info:
             chat_in_group = True
         else:
@@ -186,7 +181,7 @@ class PromptBuilder:
 
 
     async def _build_prompt_normal(
-        self, chat_stream, message_txt: str, sender_name: str = "某人", subheartflow=None
+        self, chat_stream, message_txt: str, sender_name: str = "某人"
     ) -> tuple[str, str]:
         # 开始构建prompt
         prompt_personality = "你"
@@ -209,7 +204,7 @@ class PromptBuilder:
             (chat_stream.user_info.platform, chat_stream.user_info.user_id, chat_stream.user_info.user_nickname)
         ]
         who_chat_in_group += get_recent_group_speaker(
-            subheartflow.subheartflow_id,
+            chat_stream.stream_id,
             (chat_stream.user_info.platform, chat_stream.user_info.user_id),
             limit=global_config.MAX_CONTEXT_SIZE,
         )
@@ -249,7 +244,6 @@ class PromptBuilder:
         # schedule_prompt = f"""你现在正在做的事情是：{bot_schedule.get_current_num_task(num=1, time_info=False)}"""
 
         # 获取聊天上下文
-        chat_stream = chat_manager.get_stream(subheartflow.subheartflow_id)
         if chat_stream.group_info:
             chat_in_group = True
         else:

src/plugins/heartFC_chat/normal_chat.py

@@ -9,7 +9,7 @@ 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.messagesender import message_manager
+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
@@ -96,18 +96,18 @@ class NormalChat:
     @staticmethod
     async def _send_response_messages(message, chat, response_set: List[str], thinking_id) -> MessageSending:
         """发送回复消息"""
-        container = message_manager.get_container(chat.stream_id)
+        container = await message_manager.get_container(chat.stream_id)
         thinking_message = None
 
-        for msg in container.messages:
+        for msg in container.messages[:]:
             if isinstance(msg, MessageThinking) and msg.message_info.message_id == thinking_id:
                 thinking_message = msg
                 container.messages.remove(msg)
                 break
 
         if not thinking_message:
-            logger.warning("未找到对应的思考消息，可能已超时被移除")
+            logger.warning(f"[{chat.stream_id}] 未找到对应的思考消息 {thinking_id}，可能已超时被移除")
-            return
+            return None
 
         thinking_start_time = thinking_message.thinking_start_time
         message_set = MessageSet(chat, thinking_id)
@@ -130,12 +130,14 @@ class NormalChat:
                 is_head=not mark_head,
                 is_emoji=False,
                 thinking_start_time=thinking_start_time,
+                apply_set_reply_logic=True
             )
             if not mark_head:
                 mark_head = True
                 first_bot_msg = bot_message
             message_set.add_message(bot_message)
-        message_manager.add_message(message_set)
+
+        await message_manager.add_message(message_set)
 
         return first_bot_msg
 
@@ -164,8 +166,9 @@ class NormalChat:
                     reply=message,
                     is_head=False,
                     is_emoji=True,
+                    apply_set_reply_logic=True
                 )
-                message_manager.add_message(bot_message)
+                await message_manager.add_message(bot_message)
 
     async def _update_relationship(self, message: MessageRecv, response_set):
         """更新关系情绪"""
@@ -328,12 +331,13 @@ class NormalChat:
             if not response_set:
                 logger.info(f"[{chat.stream_id}] 模型未生成回复内容")
                 # 如果模型未生成回复，移除思考消息
-                container = message_manager.get_container(chat.stream_id)
+                container = await message_manager.get_container(chat.stream_id)
                 # thinking_message = None
                 for msg in container.messages[:]:  # Iterate over a copy
                     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}")
                         break
                 return  # 不发送回复

src/plugins/heartFC_chat/normal_chat_generator.py

@@ -86,9 +86,10 @@ class ResponseGenerator:
             prompt = await prompt_builder.build_prompt(
                 build_mode="normal",
                 reason= "",
+                current_mind_info="",
                 message_txt=message.processed_plain_text,
                 sender_name=sender_name,
-                subheartflow=sub_hf,
+                chat_stream=message.chat_stream,
             )
         logger.info(f"构建prompt时间: {t_build_prompt.human_readable}")