应要求提交上未完成的HFC, expression部分

src/bw_learner/expression_reflect_tracker.py

@@ -0,0 +1,60 @@
+from typing import TYPE_CHECKING, Optional
+
+import time
+
+from src.common.logger import get_logger
+from src.common.database.database import get_db_session
+from src.llm_models.utils_model import LLMRequest
+from src.config.config import model_config
+
+if TYPE_CHECKING:
+    from src.common.data_models.expression_data_model import MaiExpression
+
+# TODO: 这个LLMRequest实例被更优雅的方式替换掉
+judge_model = LLMRequest(model_set=model_config.model_task_config.tool_use, request_type="reflect.tracker")
+
+logger = get_logger("reflect_tracker")
+
+class ReflectTracker:
+    def __init__(self, session_id: str):
+        self.session_id = session_id
+        self.last_check_msg_count = 0
+        self.max_msg_count = 30
+        self.max_duration = 15 * 60  # 15 分钟
+        self.expression: Optional["MaiExpression"] = None  # 当前正在追踪的表达，由外部设置
+
+        # 运行状态
+        self.tracking = False
+        self.tracking_start_time: float = 0.0
+
+    def register_expression_and_track(self, expression: "MaiExpression"):
+        """注册需要追踪的表达"""
+        if self.tracking:
+            raise RuntimeError("ReflectTracker is already tracking an expression.")
+        self.expression = expression
+        self.tracking = True
+        self.tracking_start_time = time.time()
+    
+    def _reset_tracker(self):
+        """重置追踪状态"""
+        self.expression = None
+        self.tracking = False
+        self.last_check_msg_count = 0
+
+    async def trigger_tracker(self) -> bool:
+        """
+        触发追踪检查
+
+        Returns:
+            return (bool): 如果返回True，表示追踪完成，Tracker运行结束（运行状态置为`False`）；如果返回False，表示继续追踪
+        """
+        # 对于没有正在追踪的表达，直接返回False
+        if not self.tracking:
+            return False
+
+        # 检查是否超时（无论是消息数量还是时间）
+        if time.time() - self.tracking_start_time > self.max_duration:
+            self._reset_tracker()
+            return True
+        
+        # TODO: 完成追踪检查逻辑

src/bw_learner/expression_reflector.py

@@ -1,22 +1,35 @@
+from rich.traceback import install
+from sqlmodel import select
+from typing import TYPE_CHECKING
+
 import random
 import time
-from typing import Optional, Dict
 
 from src.common.logger import get_logger
-from src.common.database.database_model import Expression
 from src.config.config import global_config
-from src.chat.message_receive.chat_stream import get_chat_manager
+from src.common.database.database_model import Expression
-from src.plugin_system.apis import send_api
+from src.common.database.database import get_db_session
+from src.common.data_models.expression_data_model import MaiExpression
+from src.common.utils.utils_session import SessionUtils
+from .expression_reflect_tracker import ReflectTracker
+
+if TYPE_CHECKING:
+    from src.config.official_configs import TargetItem
 
 logger = get_logger("expression_reflector")
 
+install(extra_lines=3)
+
+LOG_PREFIX = "[Expression Reflector]"
+
 
 class ExpressionReflector:
-    """表达反思器，管理单个聊天流的表达反思提问"""
+    """表达反思器，管理单个聊天流的表达反思提问，使用每个session_id独立的实例"""
 
-    def __init__(self, chat_id: str):
+    def __init__(self, session_id: str):
-        self.chat_id = chat_id
+        self.session_id = session_id
-        self.last_ask_time: float = 0.0
+        self.last_ask_time: float = time.time()
+        self.reflect_tracker: ReflectTracker = ReflectTracker(session_id)
 
     async def check_and_ask(self) -> bool:
         """
@@ -25,226 +38,99 @@ class ExpressionReflector:
         Returns:
             bool: 是否执行了提问
         """
-        try:
+        if not await self.check_need_ask():
-            logger.debug(f"[Expression Reflection] 开始检查是否需要提问 (stream_id: {self.chat_id})")
-
-            if not global_config.expression.expression_manual_reflect:
-                logger.debug("[Expression Reflection] 表达反思功能未启用，跳过")
-                return False
-
-            operator_config = global_config.expression.manual_reflect_operator_id
-            if not operator_config:
-                logger.debug("[Expression Reflection] Operator ID 未配置，跳过")
-                return False
-
-            # 检查是否在允许列表中
-            allow_reflect = global_config.expression.allow_reflect
-            if allow_reflect:
-                # 将 allow_reflect 中的 platform:id:type 格式转换为 chat_id 列表
-                allow_reflect_chat_ids = []
-                for stream_config in allow_reflect:
-                    parsed_chat_id = global_config.expression._parse_stream_config_to_chat_id(stream_config)
-                    if parsed_chat_id:
-                        allow_reflect_chat_ids.append(parsed_chat_id)
-                    else:
-                        logger.warning(f"[Expression Reflection] 无法解析 allow_reflect 配置项: {stream_config}")
-
-                if self.chat_id not in allow_reflect_chat_ids:
-                    logger.info(f"[Expression Reflection] 当前聊天流 {self.chat_id} 不在允许列表中，跳过")
-                    return False
-
-            # 检查上一次提问时间
-            current_time = time.time()
-            time_since_last_ask = current_time - self.last_ask_time
-
-            # 5-10分钟间隔，随机选择
-            min_interval = 10 * 60  # 5分钟
-            max_interval = 15 * 60  # 10分钟
-            interval = random.uniform(min_interval, max_interval)
-
-            logger.info(
-                f"[Expression Reflection] 上次提问时间: {self.last_ask_time:.2f}, 当前时间: {current_time:.2f}, 已过时间: {time_since_last_ask:.2f}秒 ({time_since_last_ask / 60:.2f}分钟), 需要间隔: {interval:.2f}秒 ({interval / 60:.2f}分钟)"
-            )
-
-            if time_since_last_ask < interval:
-                remaining_time = interval - time_since_last_ask
-                logger.info(
-                    f"[Expression Reflection] 距离上次提问时间不足，还需等待 {remaining_time:.2f}秒 ({remaining_time / 60:.2f}分钟)，跳过"
-                )
-                return False
-
-            # 检查是否已经有针对该 Operator 的 Tracker 在运行
-            logger.info(f"[Expression Reflection] 检查 Operator {operator_config} 是否已有活跃的 Tracker")
-            if await _check_tracker_exists(operator_config):
-                logger.info(f"[Expression Reflection] Operator {operator_config} 已有活跃的 Tracker，跳过本次提问")
-                return False
-
-            # 获取未检查的表达
-            try:
-                logger.info("[Expression Reflection] 查询未检查且未拒绝的表达")
-                expressions = Expression.select().where((~Expression.checked) & (~Expression.rejected)).limit(50)
-
-                expr_list = list(expressions)
-                logger.info(f"[Expression Reflection] 找到 {len(expr_list)} 个候选表达")
-
-                if not expr_list:
-                    logger.info("[Expression Reflection] 没有可用的表达，跳过")
-                    return False
-
-                target_expr: Expression = random.choice(expr_list)
-                logger.info(
-                    f"[Expression Reflection] 随机选择了表达 ID: {target_expr.id}, Situation: {target_expr.situation}, Style: {target_expr.style}"
-                )
-
-                # 生成询问文本
-                ask_text = _generate_ask_text(target_expr)
-                if not ask_text:
-                    logger.warning("[Expression Reflection] 生成询问文本失败，跳过")
-                    return False
-
-                logger.info(f"[Expression Reflection] 准备向 Operator {operator_config} 发送提问")
-                # 发送给 Operator
-                await _send_to_operator(operator_config, ask_text, target_expr)
-
-                # 更新上一次提问时间
-                self.last_ask_time = current_time
-                logger.info(f"[Expression Reflection] 提问成功，已更新上次提问时间为 {current_time:.2f}")
-
-                return True
-
-            except Exception as e:
-                logger.error(f"[Expression Reflection] 检查或提问过程中出错: {e}")
-                import traceback
-
-                logger.error(traceback.format_exc())
-                return False
-        except Exception as e:
-            logger.error(f"[Expression Reflection] 检查或提问过程中出错: {e}")
-            import traceback
-
-            logger.error(traceback.format_exc())
             return False
 
-
+        operator_config = global_config.expression.manual_reflect_operator_id
-class ExpressionReflectorManager:
+        if not operator_config:
-    """表达反思管理器，管理多个聊天流的表达反思实例"""
+            logger.debug(f"{LOG_PREFIX} Operator ID 未配置，跳过")
-
-    def __init__(self):
-        self.reflectors: Dict[str, ExpressionReflector] = {}
-
-    def get_or_create_reflector(self, chat_id: str) -> ExpressionReflector:
-        """获取或创建指定聊天流的表达反思实例"""
-        if chat_id not in self.reflectors:
-            self.reflectors[chat_id] = ExpressionReflector(chat_id)
-        return self.reflectors[chat_id]
-
-
-# 创建全局实例
-expression_reflector_manager = ExpressionReflectorManager()
-
-
-async def _check_tracker_exists(operator_config: str) -> bool:
-    """检查指定 Operator 是否已有活跃的 Tracker"""
-    from src.bw_learner.reflect_tracker import reflect_tracker_manager
-
-    chat_manager = get_chat_manager()
-    chat_stream = None
-
-    # 尝试解析配置字符串 "platform:id:type"
-    parts = operator_config.split(":")
-    if len(parts) == 3:
-        platform = parts[0]
-        id_str = parts[1]
-        stream_type = parts[2]
-
-        user_info = None
-        group_info = None
-
-        from maim_message import UserInfo, GroupInfo
-
-        if stream_type == "group":
-            group_info = GroupInfo(group_id=id_str, platform=platform)
-            user_info = UserInfo(user_id="system", user_nickname="System", platform=platform)
-        elif stream_type == "private":
-            user_info = UserInfo(user_id=id_str, platform=platform, user_nickname="Operator")
-        else:
             return False
 
-        if user_info:
+        if await self.ask_reflection(operator_config):
-            try:
+            self.last_ask_time = time.time()
-                chat_stream = await chat_manager.get_or_create_stream(platform, user_info, group_info)
+            return True
-            except Exception as e:
-                logger.error(f"Failed to get or create chat stream for checking tracker: {e}")
-                return False
-    else:
-        chat_stream = chat_manager.get_stream(operator_config)
-
-    if not chat_stream:
         return False
 
-    return reflect_tracker_manager.get_tracker(chat_stream.stream_id) is not None
+    async def check_need_ask(self) -> bool:
+        """
+        检查是否需要提问表达反思
 
+        Returns:
+            bool: 是否执行了提问
+        """
+        if not global_config.expression.expression_manual_reflect:
+            logger.debug(f"{LOG_PREFIX} 表达反思功能未启用，跳过")
+            return False
+        logger.debug(f"{LOG_PREFIX} 开始检查是否需要提问 (session_id: {self.session_id})")
+        operator_config = global_config.expression.manual_reflect_operator_id
+        if not operator_config:
+            logger.debug(f"{LOG_PREFIX} Operator ID 未配置，跳过")
+            return False
+
+        if allow_reflect_list := global_config.expression.allow_reflect:
+            # 转换配置项为session_id列表
+            allow_reflect_session_ids = [
+                self._parse_config_item_2_session_id(stream_config) for stream_config in allow_reflect_list
+            ]
+            if self.session_id not in allow_reflect_session_ids:
+                logger.info(f"{LOG_PREFIX} 当前聊天流 {self.session_id} 不在允许列表中，跳过")
+                return False
+
+        # 检查上一次提问时间
+        current_time = time.time()
+        time_since_last_ask = current_time - self.last_ask_time
+
+        # 随机选择10-15分钟间隔
+        ask_interval = random.uniform(10 * 60, 15 * 60)
+        if time_since_last_ask < ask_interval:
+            logger.info(
+                f"{LOG_PREFIX} 距离上次提问时间 {time_since_last_ask:.2f} 秒，未达到随机间隔 {ask_interval:.2f} 秒，跳过"
+            )
+            return False
+
+        if self.reflect_tracker.tracking:
+            logger.info(f"{LOG_PREFIX} Operator {operator_config} 已有活跃的 Tracker，跳过本次提问")
+            return False
+        return True
+
+    async def ask_reflection(self, operator_config: "TargetItem") -> bool:
+        """执行提问表达反思的操作"""
+        # 选取未检查过的表达
+        logger.info(f"{LOG_PREFIX} 查询未检查且未拒绝的表达")
+        try:
+            with get_db_session() as session:
+                statement = select(Expression).filter_by(checked=False, rejected=False).limit(50)
+                results = session.exec(statement).all()
+                if not results:
+                    logger.info(f"{LOG_PREFIX} 未找到未检查且未拒绝的表达")
+                    return False
+                logger.info(f"{LOG_PREFIX} 找到 {len(results)} 个未检查且未拒绝的表达")
+
+        except Exception as selected_expression:
+            logger.error(f"{LOG_PREFIX} 查询表达时发生错误: {selected_expression}")
+            return False
+
+        # 随机选取一个表达进行提问
+        selected_expression = MaiExpression.from_db_instance(random.choice(results))
+        item_id = selected_expression.item_id
+        situation = selected_expression.situation
+        style = selected_expression.style
+        logger.info(f"{LOG_PREFIX} 随机选择了表达 ID: {item_id}, Situation: {situation}, Style: {style}")
 
-def _generate_ask_text(expr: Expression) -> Optional[str]:
-    try:
         ask_text = (
             f"我正在学习新的表达方式，请帮我看看这个是否合适？\n\n"
             f"**学习到的表达信息**\n"
-            f"- 情景 (Situation): {expr.situation}\n"
+            f"- 情景 (Situation): {situation}\n"
-            f"- 风格 (Style): {expr.style}\n"
+            f"- 风格 (Style): {style}\n"
         )
-        return ask_text
-    except Exception as e:
-        logger.error(f"Failed to generate ask text: {e}")
-        return None
 
+        # TODO: 在发送相关API重构完成后完成发送给operator的逻辑
 
-async def _send_to_operator(operator_config: str, text: str, expr: Expression):
+        self.reflect_tracker.register_expression_and_track(selected_expression)
-    chat_manager = get_chat_manager()
+        return True
-    chat_stream = None
 
-    # 尝试解析配置字符串 "platform:id:type"
+    def _parse_config_item_2_session_id(self, config_item: "TargetItem") -> str:
-    parts = operator_config.split(":")
+        if config_item.rule_type == "group":
-    if len(parts) == 3:
+            return SessionUtils.calculate_session_id(config_item.platform, group_id=str(config_item.item_id))
-        platform = parts[0]
-        id_str = parts[1]
-        stream_type = parts[2]
-
-        user_info = None
-        group_info = None
-
-        from maim_message import UserInfo, GroupInfo
-
-        if stream_type == "group":
-            group_info = GroupInfo(group_id=id_str, platform=platform)
-            user_info = UserInfo(user_id="system", user_nickname="System", platform=platform)
-        elif stream_type == "private":
-            user_info = UserInfo(user_id=id_str, platform=platform, user_nickname="Operator")
         else:
-            logger.warning(f"Unknown stream type in operator config: {stream_type}")
+            return SessionUtils.calculate_session_id(config_item.platform, user_id=str(config_item.item_id))
-            return
-
-        if user_info:
-            try:
-                chat_stream = await chat_manager.get_or_create_stream(platform, user_info, group_info)
-            except Exception as e:
-                logger.error(f"Failed to get or create chat stream for operator {operator_config}: {e}")
-                return
-    else:
-        chat_stream = chat_manager.get_stream(operator_config)
-
-    if not chat_stream:
-        logger.warning(f"Could not find or create chat stream for operator: {operator_config}")
-        return
-
-    stream_id = chat_stream.stream_id
-
-    # 注册 Tracker
-    from src.bw_learner.reflect_tracker import ReflectTracker, reflect_tracker_manager
-
-    tracker = ReflectTracker(chat_stream=chat_stream, expression=expr, created_time=time.time())
-    reflect_tracker_manager.add_tracker(stream_id, tracker)
-
-    # 发送消息
-    await send_api.text_to_stream(text=text, stream_id=stream_id, typing=True)
-    logger.info(f"Sent expression reflect query to operator {operator_config} for expr {expr.id}")

src/chat/heart_flow/heartFC_chat.py

@@ -1,148 +1,169 @@
+from rich.traceback import install
+from typing import Optional, List, TYPE_CHECKING
+
 import asyncio
 import time
 import traceback
 import random
-from typing import List, Optional, Dict, Any, Tuple, TYPE_CHECKING
-from rich.traceback import install
 
-from src.config.config import global_config
 from src.common.logger import get_logger
-from src.common.data_models.info_data_model import ActionPlannerInfo
+from src.common.utils.utils_session import SessionUtils
-from src.common.data_models.message_data_model import ReplyContentType
+from src.config.config import global_config
-from src.chat.message_receive.chat_stream import ChatStream, get_chat_manager
+from src.chat.message_receive.chat_manager import chat_manager
-from src.chat.utils.prompt_builder import global_prompt_manager
-from src.chat.utils.timer_calculator import Timer
-from src.chat.planner_actions.planner import ActionPlanner
-from src.chat.planner_actions.action_modifier import ActionModifier
-from src.chat.planner_actions.action_manager import ActionManager
-from src.chat.heart_flow.hfc_utils import CycleDetail
-from src.bw_learner.expression_learner import expression_learner_manager
-from src.chat.heart_flow.frequency_control import frequency_control_manager
-from src.bw_learner.reflect_tracker import reflect_tracker_manager
-from src.bw_learner.expression_reflector import expression_reflector_manager
-from src.bw_learner.message_recorder import extract_and_distribute_messages
-from src.person_info.person_info import Person
-from src.plugin_system.base.component_types import EventType, ActionInfo
-from src.plugin_system.core import events_manager
-from src.plugin_system.apis import generator_api, send_api, message_api, database_api
-from src.chat.utils.chat_message_builder import (
-    build_readable_messages_with_id,
-    get_raw_msg_before_timestamp_with_chat,
-)
-from src.chat.utils.utils import record_replyer_action_temp
-from src.memory_system.chat_history_summarizer import ChatHistorySummarizer
 
 if TYPE_CHECKING:
-    from src.common.data_models.database_data_model import DatabaseMessages
+    from src.chat.message_receive.message import SessionMessage
-    from src.common.data_models.message_data_model import ReplySetModel
 
+install(extra_lines=5)
 
-ERROR_LOOP_INFO = {
+logger = get_logger("heartFC_chat")
-    "loop_plan_info": {
-        "action_result": {
-            "action_type": "error",
-            "action_data": {},
-            "reasoning": "循环处理失败",
-        },
-    },
-    "loop_action_info": {
-        "action_taken": False,
-        "reply_text": "",
-        "command": "",
-        "taken_time": time.time(),
-    },
-}
-
-
-install(extra_lines=3)
-
-# 注释：原来的动作修改超时常量已移除，因为改为顺序执行
-
-logger = get_logger("hfc")  # Logger Name Changed
 
 
 class HeartFChatting:
     """
-    管理一个连续的Focus Chat循环
+    管理一个连续的Focus Chat聊天会话
-    用于在特定聊天流中生成回复。
+    用于在特定的聊天会话里面生成回复
-    其生命周期现在由其关联的 SubHeartflow 的 FOCUSED 状态控制。
     """
 
-    def __init__(self, chat_id: str):
+    def __init__(self, session_id: str):
         """
-        HeartFChatting 初始化函数
+        初始化 HeartFChatting 实例
 
-        参数:
+        Args:
-            chat_id: 聊天流唯一标识符(如stream_id)
+            session_id: 聊天会话ID
-            on_stop_focus_chat: 当收到stop_focus_chat命令时调用的回调函数
-            performance_version: 性能记录版本号，用于区分不同启动版本
         """
         # 基础属性
-        self.stream_id: str = chat_id  # 聊天流ID
+        self.session_id = session_id
-        self.chat_stream: ChatStream = get_chat_manager().get_stream(self.stream_id)  # type: ignore
+        session_name = chat_manager.get_session_name(session_id) or session_id
-        if not self.chat_stream:
+        self.log_prefix = f"[{session_name}]"
-            raise ValueError(f"无法找到聊天流: {self.stream_id}")
-        self.log_prefix = f"[{get_chat_manager().get_stream_name(self.stream_id) or self.stream_id}]"
 
-        self.expression_learner = expression_learner_manager.get_expression_learner(self.stream_id)
+        # 系统运行状态
+        self._running: bool = False
+        self._loop_task: Optional[asyncio.Task] = None
+        self._cycle_counter: int = 0
+        self._hfc_lock: asyncio.Lock = asyncio.Lock()  # 用于保护 _hfc_func 的并发访问
+        # 聊天频率相关
+        self._consecutive_no_reply_count = 0  # 跟踪连续 no_reply 次数，用于动态调整阈值
+        self._talk_frequency_adjust: float = 1.0  # 发言频率修正值，默认为1.0，可以根据需要调整
 
-        self.action_manager = ActionManager()
+        # HFC内消息缓存
-        self.action_planner = ActionPlanner(chat_id=self.stream_id, action_manager=self.action_manager)
+        self.message_cache: List[SessionMessage] = []
-        self.action_modifier = ActionModifier(action_manager=self.action_manager, chat_id=self.stream_id)
 
-        # 循环控制内部状态
+        # Asyncio Event 用于控制循环的开始和结束
-        self.running: bool = False
+        self._cycle_event = asyncio.Event()
-        self._loop_task: Optional[asyncio.Task] = None  # 主循环任务
-
-        # 添加循环信息管理相关的属性
-        self.history_loop: List[CycleDetail] = []
-        self._cycle_counter = 0
-        self._current_cycle_detail: CycleDetail = None  # type: ignore
-
-        self.last_read_time = time.time() - 2
-
-        self.is_mute = False
-
-        self.last_active_time = time.time()  # 记录上一次非noreply时间
-
-        self.question_probability_multiplier = 1
-        self.questioned = False
-
-        # 跟踪连续 no_reply 次数，用于动态调整阈值
-        self.consecutive_no_reply_count = 0
-
-        # 聊天内容概括器
-        self.chat_history_summarizer = ChatHistorySummarizer(chat_id=self.stream_id)
 
     async def start(self):
-        """检查是否需要启动主循环，如果未激活则启动。"""
+        """启动 HeartFChatting 的主循环"""
-
+        # 先检查是否已经启动运行
-        # 如果循环已经激活，直接返回
+        if self._running:
-        if self.running:
+            logger.debug(f"{self.log_prefix} 已经在运行中，无需重复启动")
-            logger.debug(f"{self.log_prefix} HeartFChatting 已激活，无需重复启动")
             return
 
         try:
-            # 标记为活动状态，防止重复启动
+            self._running = True
-            self.running = True
+            self._cycle_event.clear()  # 确保事件初始状态为未设置
 
-            self._loop_task = asyncio.create_task(self._main_chat_loop())
+            self._loop_task = asyncio.create_task(self.main_loop())
             self._loop_task.add_done_callback(self._handle_loop_completion)
 
-            # 启动聊天内容概括器的后台定期检查循环
-            await self.chat_history_summarizer.start()
-
             logger.info(f"{self.log_prefix} HeartFChatting 启动完成")
-
         except Exception as e:
-            # 启动失败时重置状态
+            logger.error(f"{self.log_prefix} 启动 HeartFChatting 失败: {e}", exc_info=True)
-            self.running = False
+            self._running = False  # 确保状态正确
-            self._loop_task = None
+            self._cycle_event.set()  # 确保事件被设置，避免死锁
-            logger.error(f"{self.log_prefix} HeartFChatting 启动失败: {e}")
+            self._loop_task = None  # 确保任务引用被清理
             raise
 
+    async def stop(self):
+        """停止 HeartFChatting 的主循环"""
+        if not self._running:
+            logger.debug(f"{self.log_prefix} HeartFChatting 已经停止，无需重复停止")
+            return
+
+        self._running = False
+        self._cycle_event.set()  # 触发事件，通知循环结束
+
+        if self._loop_task:
+            self._loop_task.cancel()  # 取消主循环任务
+            try:
+                await self._loop_task  # 等待任务完成
+            except asyncio.CancelledError:
+                logger.info(f"{self.log_prefix} HeartFChatting 主循环已成功取消")
+            except Exception as e:
+                logger.error(f"{self.log_prefix} 停止 HeartFChatting 时发生错误: {e}", exc_info=True)
+            finally:
+                self._loop_task = None  # 确保任务引用被清理
+
+        logger.info(f"{self.log_prefix} HeartFChatting 已停止")
+
+    async def adjust_talk_frequency(self, new_value: float):
+        """调整发言频率的调整值
+
+        Args:
+            new_value: 新的修正值，必须为非负数。值越大，修正发言频率越高；值越小，修正发言频率越低。
+        """
+        self._talk_frequency_adjust = max(0.0, new_value)
+
+    async def register_message(self, message: "SessionMessage"):
+        """注册一条消息到 HeartFChatting 的缓存中，并检测其是否产生提及，决定是否唤醒聊天
+
+        Args:
+            message: 待注册的消息对象
+        """
+        self.message_cache.append(message)
+        # 先检查at必回复
+        if global_config.chat.inevitable_at_reply and message.is_at:
+            async with self._hfc_lock:  # 确保与主循环逻辑的互斥访问
+                await self._judge_and_response(message)
+            return  # 直接返回，避免同一条消息被主循环再次处理
+        # 再检查提及必回复
+        if global_config.chat.mentioned_bot_reply and message.is_mentioned:
+            # 直接获取锁，确保一定一定触发回复逻辑，不受当前是否正在执行主循环的影响
+            async with self._hfc_lock:  # 确保与主循环逻辑的互斥访问
+                await self._judge_and_response(message)
+            return
+
+    async def main_loop(self):
+        try:
+            while self._running and not self._cycle_event.is_set():
+                if not self._hfc_lock.locked():
+                    async with self._hfc_lock:  # 确保主循环逻辑的互斥访问
+                        await self._hfc_func()
+                await asyncio.sleep(5)
+        except asyncio.CancelledError:
+            logger.info(f"{self.log_prefix} HeartFChatting: 主循环被取消，正在关闭")
+        except Exception as e:
+            logger.error(f"{self.log_prefix} 麦麦聊天意外错误: {e}，将于3s后尝试重新启动")
+            await self.stop()  # 确保状态正确
+            await asyncio.sleep(3)
+            await self.start()  # 尝试重新启动
+
+    async def _hfc_func(self, mentioned_message: Optional["SessionMessage"] = None):
+        """心流聊天的主循环逻辑"""
+        if self._consecutive_no_reply_count >= 5:
+            threshold = 2
+        elif self._consecutive_no_reply_count >= 3:
+            threshold = 2 if random.random() < 0.5 else 1
+        else:
+            threshold = 1
+
+        if len(self.message_cache) < threshold:
+            await asyncio.sleep(0.2)
+            return True
+
+        talk_value_threshold = random.random() * self._get_talk_value(self.session_id) * self._talk_frequency_adjust
+        if mentioned_message and global_config.chat.mentioned_bot_reply:
+            await self._judge_and_response(mentioned_message)
+        elif random.random() < talk_value_threshold:
+            await self._judge_and_response()
+        return True
+
+    async def _judge_and_response(self, mentioned_message: Optional["SessionMessage"] = None):
+        """判定和生成回复"""
+        # TODO: 在expression和reflector重构完成后完成这里的逻辑
+
     def _handle_loop_completion(self, task: asyncio.Task):
-        """当 _hfc_loop 任务完成时执行的回调。"""
+        """当 _hfc_func 任务完成时执行的回调。"""
         try:
             if exception := task.exception():
                 logger.error(f"{self.log_prefix} HeartFChatting: 脱离了聊天(异常): {exception}")
@@ -152,657 +173,54 @@ class HeartFChatting:
         except asyncio.CancelledError:
             logger.info(f"{self.log_prefix} HeartFChatting: 结束了聊天")
 
-    def start_cycle(self) -> Tuple[Dict[str, float], str]:
+    def _get_talk_value(self, session_id: Optional[str]) -> float:
-        self._cycle_counter += 1
+        result = global_config.chat.talk_value or 0.0
-        self._current_cycle_detail = CycleDetail(self._cycle_counter)
-        self._current_cycle_detail.thinking_id = f"tid{str(round(time.time(), 2))}"
-        cycle_timers = {}
-        return cycle_timers, self._current_cycle_detail.thinking_id
-
-    def end_cycle(self, loop_info, cycle_timers):
-        self._current_cycle_detail.set_loop_info(loop_info)
-        self.history_loop.append(self._current_cycle_detail)
-        self._current_cycle_detail.timers = cycle_timers
-        self._current_cycle_detail.end_time = time.time()
-
-    def print_cycle_info(self, cycle_timers):
-        # 记录循环信息和计时器结果
-        timer_strings = []
-        for name, elapsed in cycle_timers.items():
-            if elapsed < 0.1:
-                # 不显示小于0.1秒的计时器
-                continue
-            formatted_time = f"{elapsed:.2f}秒"
-            timer_strings.append(f"{name}: {formatted_time}")
-
-        logger.info(
-            f"{self.log_prefix} 第{self._current_cycle_detail.cycle_id}次思考,"
-            f"耗时: {self._current_cycle_detail.end_time - self._current_cycle_detail.start_time:.1f}秒;"  # type: ignore
-            + (f"详情: {'; '.join(timer_strings)}" if timer_strings else "")
-        )
-
-    async def _loopbody(self):
-        recent_messages_list = message_api.get_messages_by_time_in_chat(
-            chat_id=self.stream_id,
-            start_time=self.last_read_time,
-            end_time=time.time(),
-            limit=20,
-            limit_mode="latest",
-            filter_mai=True,
-            filter_command=False,
-            filter_intercept_message_level=0,
-        )
-
-        # 根据连续 no_reply 次数动态调整阈值
-        # 3次 no_reply 时，阈值调高到 1.5（50%概率为1，50%概率为2）
-        # 5次 no_reply 时，提高到 2（大于等于两条消息的阈值）
-        if self.consecutive_no_reply_count >= 5:
-            threshold = 2
-        elif self.consecutive_no_reply_count >= 3:
-            # 1.5 的含义：50%概率为1，50%概率为2
-            threshold = 2 if random.random() < 0.5 else 1
-        else:
-            threshold = 1
-
-        if len(recent_messages_list) >= threshold:
-            # for message in recent_messages_list:
-            # print(message.processed_plain_text)
-
-            self.last_read_time = time.time()
-
-            # !此处使at或者提及必定回复
-            mentioned_message = None
-            for message in recent_messages_list:
-                if (message.is_mentioned or message.is_at) and global_config.chat.mentioned_bot_reply:
-                    mentioned_message = message
-
-            # logger.info(f"{self.log_prefix} 当前talk_value: {TempMethods.get_talk_value(self.stream_id)}")
-
-            # *控制频率用
-            if mentioned_message:
-                await self._observe(recent_messages_list=recent_messages_list, force_reply_message=mentioned_message)
-            elif (
-                random.random()
-                < TempMethodsHFC.get_talk_value(self.stream_id)
-                * frequency_control_manager.get_or_create_frequency_control(self.stream_id).get_talk_frequency_adjust()
-            ):
-                await self._observe(recent_messages_list=recent_messages_list)
-            else:
-                # 没有提到，继续保持沉默，等待5秒防止频繁触发
-                await asyncio.sleep(10)
-                return True
-        else:
-            await asyncio.sleep(0.2)
-            return True
-        return True
-
-    async def _send_and_store_reply(
-        self,
-        response_set: "ReplySetModel",
-        action_message: "DatabaseMessages",
-        cycle_timers: Dict[str, float],
-        thinking_id,
-        actions,
-        selected_expressions: Optional[List[int]] = None,
-        quote_message: Optional[bool] = None,
-    ) -> Tuple[Dict[str, Any], str, Dict[str, float]]:
-        with Timer("回复发送", cycle_timers):
-            reply_text = await self._send_response(
-                reply_set=response_set,
-                message_data=action_message,
-                selected_expressions=selected_expressions,
-                quote_message=quote_message,
-            )
-
-        # 获取 platform，如果不存在则从 chat_stream 获取，如果还是 None 则使用默认值
-        platform = action_message.chat_info.platform
-        if platform is None:
-            platform = getattr(self.chat_stream, "platform", "unknown")
-
-        person = Person(platform=platform, user_id=action_message.user_info.user_id)
-        person_name = person.person_name
-        action_prompt_display = f"你对{person_name}进行了回复：{reply_text}"
-
-        await database_api.store_action_info(
-            chat_stream=self.chat_stream,
-            action_build_into_prompt=False,
-            action_prompt_display=action_prompt_display,
-            action_done=True,
-            thinking_id=thinking_id,
-            action_data={"reply_text": reply_text},
-            action_name="reply",
-        )
-
-        # 构建循环信息
-        loop_info: Dict[str, Any] = {
-            "loop_plan_info": {
-                "action_result": actions,
-            },
-            "loop_action_info": {
-                "action_taken": True,
-                "reply_text": reply_text,
-                "command": "",
-                "taken_time": time.time(),
-            },
-        }
-
-        return loop_info, reply_text, cycle_timers
-
-    async def _observe(
-        self,  # interest_value: float = 0.0,
-        recent_messages_list: Optional[List["DatabaseMessages"]] = None,
-        force_reply_message: Optional["DatabaseMessages"] = None,
-    ) -> bool:  # sourcery skip: merge-else-if-into-elif, remove-redundant-if
-        if recent_messages_list is None:
-            recent_messages_list = []
-        _reply_text = ""  # 初始化reply_text变量，避免UnboundLocalError
-
-        # -------------------------------------------------------------------------
-        # ReflectTracker Check
-        # 在每次回复前检查一次上下文，看是否有反思问题得到了解答
-        # -------------------------------------------------------------------------
-
-        reflector = expression_reflector_manager.get_or_create_reflector(self.stream_id)
-        await reflector.check_and_ask()
-        tracker = reflect_tracker_manager.get_tracker(self.stream_id)
-        if tracker:
-            resolved = await tracker.trigger_tracker()
-            if resolved:
-                reflect_tracker_manager.remove_tracker(self.stream_id)
-                logger.info(f"{self.log_prefix} ReflectTracker resolved and removed.")
-
-        start_time = time.time()
-        async with global_prompt_manager.async_message_scope(self.chat_stream.context.get_template_name()):
-            # 通过 MessageRecorder 统一提取消息并分发给 expression_learner 和 jargon_miner
-            # 在 replyer 执行时触发，统一管理时间窗口，避免重复获取消息
-            asyncio.create_task(extract_and_distribute_messages(self.stream_id))
-
-            # 添加curious检测任务 - 检测聊天记录中的矛盾、冲突或需要提问的内容
-            # asyncio.create_task(check_and_make_question(self.stream_id))
-            # 添加聊天内容概括任务 - 累积、打包和压缩聊天记录
-            # 注意：后台循环已在start()中启动，这里作为额外触发点，在有思考时立即处理
-            # asyncio.create_task(self.chat_history_summarizer.process())
-
-            cycle_timers, thinking_id = self.start_cycle()
-            logger.info(
-                f"{self.log_prefix} 开始第{self._cycle_counter}次思考(频率: {TempMethodsHFC.get_talk_value(self.stream_id)})"
-            )
-
-            # 第一步：动作检查
-            available_actions: Dict[str, ActionInfo] = {}
-            try:
-                await self.action_modifier.modify_actions()
-                available_actions = self.action_manager.get_using_actions()
-            except Exception as e:
-                logger.error(f"{self.log_prefix} 动作修改失败: {e}")
-
-            # 执行planner
-            is_group_chat, chat_target_info, _ = self.action_planner.get_necessary_info()
-
-            message_list_before_now = get_raw_msg_before_timestamp_with_chat(
-                chat_id=self.stream_id,
-                timestamp=time.time(),
-                limit=int(global_config.chat.max_context_size * 0.6),
-                filter_intercept_message_level=1,
-            )
-            chat_content_block, message_id_list = build_readable_messages_with_id(
-                messages=message_list_before_now,
-                timestamp_mode="normal_no_YMD",
-                read_mark=self.action_planner.last_obs_time_mark,
-                truncate=True,
-                show_actions=True,
-            )
-
-            prompt_info = await self.action_planner.build_planner_prompt(
-                is_group_chat=is_group_chat,
-                chat_target_info=chat_target_info,
-                current_available_actions=available_actions,
-                chat_content_block=chat_content_block,
-                message_id_list=message_id_list,
-            )
-            continue_flag, modified_message = await events_manager.handle_mai_events(
-                EventType.ON_PLAN, None, prompt_info[0], None, self.chat_stream.stream_id
-            )
-            if not continue_flag:
-                return False
-            if modified_message and modified_message._modify_flags.modify_llm_prompt:
-                prompt_info = (modified_message.llm_prompt, prompt_info[1])
-
-            with Timer("规划器", cycle_timers):
-                action_to_use_info = await self.action_planner.plan(
-                    loop_start_time=self.last_read_time,
-                    available_actions=available_actions,
-                    force_reply_message=force_reply_message,
-                )
-
-            logger.info(
-                f"{self.log_prefix} 决定执行{len(action_to_use_info)}个动作: {' '.join([a.action_type for a in action_to_use_info])}"
-            )
-
-            # 3. 并行执行所有动作
-            action_tasks = [
-                asyncio.create_task(
-                    self._execute_action(action, action_to_use_info, thinking_id, available_actions, cycle_timers)
-                )
-                for action in action_to_use_info
-            ]
-
-            # 并行执行所有任务
-            results = await asyncio.gather(*action_tasks, return_exceptions=True)
-
-            # 处理执行结果
-            reply_loop_info = None
-            reply_text_from_reply = ""
-            action_success = False
-            action_reply_text = ""
-
-            excute_result_str = ""
-            for result in results:
-                excute_result_str += f"{result['action_type']} 执行结果:{result['result']}\n"
-
-                if isinstance(result, BaseException):
-                    logger.error(f"{self.log_prefix} 动作执行异常: {result}")
-                    continue
-
-                if result["action_type"] != "reply":
-                    action_success = result["success"]
-                    action_reply_text = result["result"]
-                elif result["action_type"] == "reply":
-                    if result["success"]:
-                        reply_loop_info = result["loop_info"]
-                        reply_text_from_reply = result["result"]
-                    else:
-                        logger.warning(f"{self.log_prefix} 回复动作执行失败")
-
-            self.action_planner.add_plan_excute_log(result=excute_result_str)
-
-            # 构建最终的循环信息
-            if reply_loop_info:
-                # 如果有回复信息，使用回复的loop_info作为基础
-                loop_info = reply_loop_info
-                # 更新动作执行信息
-                loop_info["loop_action_info"].update(
-                    {
-                        "action_taken": action_success,
-                        "taken_time": time.time(),
-                    }
-                )
-                _reply_text = reply_text_from_reply
-            else:
-                # 没有回复信息，构建纯动作的loop_info
-                loop_info = {
-                    "loop_plan_info": {
-                        "action_result": action_to_use_info,
-                    },
-                    "loop_action_info": {
-                        "action_taken": action_success,
-                        "reply_text": action_reply_text,
-                        "taken_time": time.time(),
-                    },
-                }
-                _reply_text = action_reply_text
-
-            self.end_cycle(loop_info, cycle_timers)
-            self.print_cycle_info(cycle_timers)
-
-            end_time = time.time()
-            if end_time - start_time < global_config.chat.planner_smooth:
-                wait_time = global_config.chat.planner_smooth - (end_time - start_time)
-                await asyncio.sleep(wait_time)
-            else:
-                await asyncio.sleep(0.1)
-            return True
-
-    async def _main_chat_loop(self):
-        """主循环，持续进行计划并可能回复消息，直到被外部取消。"""
-        try:
-            while self.running:
-                # 主循环
-                success = await self._loopbody()
-                await asyncio.sleep(0.1)
-                if not success:
-                    break
-        except asyncio.CancelledError:
-            # 设置了关闭标志位后被取消是正常流程
-            logger.info(f"{self.log_prefix} 麦麦已关闭聊天")
-        except Exception:
-            logger.error(f"{self.log_prefix} 麦麦聊天意外错误，将于3s后尝试重新启动")
-            print(traceback.format_exc())
-            await asyncio.sleep(3)
-            self._loop_task = asyncio.create_task(self._main_chat_loop())
-        logger.error(f"{self.log_prefix} 结束了当前聊天循环")
-
-    async def _handle_action(
-        self,
-        action: str,
-        action_reasoning: str,
-        action_data: dict,
-        cycle_timers: Dict[str, float],
-        thinking_id: str,
-        action_message: Optional["DatabaseMessages"] = None,
-    ) -> tuple[bool, str, str]:
-        """
-        处理规划动作，使用动作工厂创建相应的动作处理器
-
-        参数:
-            action: 动作类型
-            action_reasoning: 决策理由
-            action_data: 动作数据，包含不同动作需要的参数
-            cycle_timers: 计时器字典
-            thinking_id: 思考ID
-            action_message: 消息数据
-        返回:
-            tuple[bool, str, str]: (是否执行了动作, 思考消息ID, 命令)
-        """
-        try:
-            # 使用工厂创建动作处理器实例
-            try:
-                action_handler = self.action_manager.create_action(
-                    action_name=action,
-                    action_data=action_data,
-                    cycle_timers=cycle_timers,
-                    thinking_id=thinking_id,
-                    chat_stream=self.chat_stream,
-                    log_prefix=self.log_prefix,
-                    action_reasoning=action_reasoning,
-                    action_message=action_message,
-                )
-            except Exception as e:
-                logger.error(f"{self.log_prefix} 创建动作处理器时出错: {e}")
-                traceback.print_exc()
-                return False, ""
-
-            # 处理动作并获取结果（固定记录一次动作信息）
-            result = await action_handler.execute()
-            success, action_text = result
-
-            return success, action_text
-
-        except Exception as e:
-            logger.error(f"{self.log_prefix} 处理{action}时出错: {e}")
-            traceback.print_exc()
-            return False, ""
-
-    async def _send_response(
-        self,
-        reply_set: "ReplySetModel",
-        message_data: "DatabaseMessages",
-        selected_expressions: Optional[List[int]] = None,
-        quote_message: Optional[bool] = None,
-    ) -> str:
-        # 根据 llm_quote 配置决定是否使用 quote_message 参数
-        if global_config.chat.llm_quote:
-            # 如果配置为 true，使用 llm_quote 参数决定是否引用回复
-            if quote_message is None:
-                logger.warning(f"{self.log_prefix} quote_message 参数为空，不引用")
-                need_reply = False
-            else:
-                need_reply = quote_message
-                if need_reply:
-                    logger.info(f"{self.log_prefix} LLM 决定使用引用回复")
-        else:
-            # 如果配置为 false，使用原来的模式
-            new_message_count = message_api.count_new_messages(
-                chat_id=self.chat_stream.stream_id, start_time=self.last_read_time, end_time=time.time()
-            )
-            need_reply = new_message_count >= random.randint(2, 3) or time.time() - self.last_read_time > 90
-            if need_reply:
-                logger.info(
-                    f"{self.log_prefix} 从思考到回复，共有{new_message_count}条新消息，使用引用回复，或者上次回复时间超过90秒"
-                )
-
-        reply_text = ""
-        first_replied = False
-        for reply_content in reply_set.reply_data:
-            if reply_content.content_type != ReplyContentType.TEXT:
-                continue
-            data: str = reply_content.content  # type: ignore
-            if not first_replied:
-                await send_api.text_to_stream(
-                    text=data,
-                    stream_id=self.chat_stream.stream_id,
-                    reply_message=message_data,
-                    set_reply=need_reply,
-                    typing=False,
-                    selected_expressions=selected_expressions,
-                )
-                first_replied = True
-            else:
-                await send_api.text_to_stream(
-                    text=data,
-                    stream_id=self.chat_stream.stream_id,
-                    reply_message=message_data,
-                    set_reply=False,
-                    typing=True,
-                    selected_expressions=selected_expressions,
-                )
-            reply_text += data
-
-        return reply_text
-
-    async def _execute_action(
-        self,
-        action_planner_info: ActionPlannerInfo,
-        chosen_action_plan_infos: List[ActionPlannerInfo],
-        thinking_id: str,
-        available_actions: Dict[str, ActionInfo],
-        cycle_timers: Dict[str, float],
-    ):
-        """执行单个动作的通用函数"""
-        try:
-            with Timer(f"动作{action_planner_info.action_type}", cycle_timers):
-                # 直接当场执行no_reply逻辑
-                if action_planner_info.action_type == "no_reply":
-                    # 直接处理no_reply逻辑，不再通过动作系统
-                    reason = action_planner_info.reasoning or "选择不回复"
-                    # logger.info(f"{self.log_prefix} 选择不回复，原因: {reason}")
-
-                    # 增加连续 no_reply 计数
-                    self.consecutive_no_reply_count += 1
-
-                    await database_api.store_action_info(
-                        chat_stream=self.chat_stream,
-                        action_build_into_prompt=False,
-                        action_prompt_display=reason,
-                        action_done=True,
-                        thinking_id=thinking_id,
-                        action_data={},
-                        action_name="no_reply",
-                        action_reasoning=reason,
-                    )
-
-                    return {"action_type": "no_reply", "success": True, "result": "选择不回复", "command": ""}
-
-                elif action_planner_info.action_type == "reply":
-                    # 直接当场执行reply逻辑
-                    self.questioned = False
-                    # 刷新主动发言状态
-                    # 重置连续 no_reply 计数
-                    self.consecutive_no_reply_count = 0
-
-                    reason = action_planner_info.reasoning or ""
-                    # 根据 think_mode 配置决定 think_level 的值
-                    think_mode = global_config.chat.think_mode
-                    if think_mode == "default":
-                        think_level = 0
-                    elif think_mode == "deep":
-                        think_level = 1
-                    elif think_mode == "dynamic":
-                        # dynamic 模式：从 planner 返回的 action_data 中获取
-                        think_level = action_planner_info.action_data.get("think_level", 1)
-                    else:
-                        # 默认使用 default 模式
-                        think_level = 0
-                    # 使用 action_reasoning（planner 的整体思考理由）作为 reply_reason
-                    planner_reasoning = action_planner_info.action_reasoning or reason
-
-                    record_replyer_action_temp(
-                        chat_id=self.stream_id,
-                        reason=reason,
-                        think_level=think_level,
-                    )
-
-                    await database_api.store_action_info(
-                        chat_stream=self.chat_stream,
-                        action_build_into_prompt=False,
-                        action_prompt_display=reason,
-                        action_done=True,
-                        thinking_id=thinking_id,
-                        action_data={},
-                        action_name="reply",
-                        action_reasoning=reason,
-                    )
-
-                    # 从 Planner 的 action_data 中提取未知词语列表（仅在 reply 时使用）
-                    unknown_words = None
-                    quote_message = None
-                    if isinstance(action_planner_info.action_data, dict):
-                        uw = action_planner_info.action_data.get("unknown_words")
-                        if isinstance(uw, list):
-                            cleaned_uw: List[str] = []
-                            for item in uw:
-                                if isinstance(item, str):
-                                    s = item.strip()
-                                    if s:
-                                        cleaned_uw.append(s)
-                            if cleaned_uw:
-                                unknown_words = cleaned_uw
-
-                        # 从 Planner 的 action_data 中提取 quote_message 参数
-                        qm = action_planner_info.action_data.get("quote")
-                        if qm is not None:
-                            # 支持多种格式：true/false, "true"/"false", 1/0
-                            if isinstance(qm, bool):
-                                quote_message = qm
-                            elif isinstance(qm, str):
-                                quote_message = qm.lower() in ("true", "1", "yes")
-                            elif isinstance(qm, (int, float)):
-                                quote_message = bool(qm)
-
-                        logger.info(f"{self.log_prefix} {qm}引用回复设置: {quote_message}")
-
-                    success, llm_response = await generator_api.generate_reply(
-                        chat_stream=self.chat_stream,
-                        reply_message=action_planner_info.action_message,
-                        available_actions=available_actions,
-                        chosen_actions=chosen_action_plan_infos,
-                        reply_reason=planner_reasoning,
-                        unknown_words=unknown_words,
-                        enable_tool=global_config.tool.enable_tool,
-                        request_type="replyer",
-                        from_plugin=False,
-                        reply_time_point=action_planner_info.action_data.get("loop_start_time", time.time()),
-                        think_level=think_level,
-                    )
-
-                    if not success or not llm_response or not llm_response.reply_set:
-                        if action_planner_info.action_message:
-                            logger.info(f"对 {action_planner_info.action_message.processed_plain_text} 的回复生成失败")
-                        else:
-                            logger.info("回复生成失败")
-                        return {"action_type": "reply", "success": False, "result": "回复生成失败", "loop_info": None}
-
-                    response_set = llm_response.reply_set
-                    selected_expressions = llm_response.selected_expressions
-                    loop_info, reply_text, _ = await self._send_and_store_reply(
-                        response_set=response_set,
-                        action_message=action_planner_info.action_message,  # type: ignore
-                        cycle_timers=cycle_timers,
-                        thinking_id=thinking_id,
-                        actions=chosen_action_plan_infos,
-                        selected_expressions=selected_expressions,
-                        quote_message=quote_message,
-                    )
-                    self.last_active_time = time.time()
-                    return {
-                        "action_type": "reply",
-                        "success": True,
-                        "result": f"你使用reply动作，对' {action_planner_info.action_message.processed_plain_text} '这句话进行了回复，回复内容为: '{reply_text}'",
-                        "loop_info": loop_info,
-                    }
-
-                else:
-                    # 执行普通动作
-                    with Timer("动作执行", cycle_timers):
-                        success, result = await self._handle_action(
-                            action=action_planner_info.action_type,
-                            action_reasoning=action_planner_info.action_reasoning or "",
-                            action_data=action_planner_info.action_data or {},
-                            cycle_timers=cycle_timers,
-                            thinking_id=thinking_id,
-                            action_message=action_planner_info.action_message,
-                        )
-
-                    self.last_active_time = time.time()
-                    return {
-                        "action_type": action_planner_info.action_type,
-                        "success": success,
-                        "result": result,
-                    }
-
-        except Exception as e:
-            logger.error(f"{self.log_prefix} 执行动作时出错: {e}")
-            logger.error(f"{self.log_prefix} 错误信息: {traceback.format_exc()}")
-            return {
-                "action_type": action_planner_info.action_type,
-                "success": False,
-                "result": "",
-                "loop_info": None,
-                "error": str(e),
-            }
-
-
-class TempMethodsHFC:
-    @staticmethod
-    def get_talk_value(chat_id: Optional[str]) -> float:
-        result = global_config.chat.talk_value or 0.0000001
         if not global_config.chat.enable_talk_value_rules or not global_config.chat.talk_value_rules:
             return result
-        import time
-
         local_time = time.localtime()
         now_min = local_time.tm_hour * 60 + local_time.tm_min
-        # 先处理特定规则
+
-        if chat_id:
+        # 优先匹配会话相关的规则
+        if session_id:
             for rule in global_config.chat.talk_value_rules:
                 if not rule.platform and not rule.item_id:
-                    continue  # 一起留空表示全局，跳过
+                    continue  # 一起留空表示全局
-                is_group = rule.rule_type == "group"
+                if rule.rule_type == "group":
-                from src.chat.message_receive.chat_stream import get_chat_manager
+                    rule_session_id = SessionUtils.calculate_session_id(rule.platform, group_id=str(rule.item_id))
-
+                else:
-                stream_id = get_chat_manager().get_stream_id(rule.platform, str(rule.item_id), is_group)
+                    rule_session_id = SessionUtils.calculate_session_id(rule.platform, user_id=str(rule.item_id))
-                if stream_id != chat_id:
+                if rule_session_id != session_id:
-                    continue
+                    continue  # 不匹配的会话ID，跳过
-                parsed_range = TempMethodsHFC._parse_range(rule.time)
+                parsed_range = self._parse_range(rule.time)
                 if not parsed_range:
-                    continue
+                    continue  # 无法解析的时间范围，跳过
                 start_min, end_min = parsed_range
                 in_range: bool = False
                 if start_min <= end_min:
                     in_range = start_min <= now_min <= end_min
-                else:
+                else:  # 跨天的时间范围
                     in_range = now_min >= start_min or now_min <= end_min
                 if in_range:
-                    return rule.value or 0.0
+                    return rule.value or 0.0  # 如果规则生效但没有设置值，返回0.0
-        # 再处理全局规则
+
+        # 没有匹配到会话相关的规则，继续匹配全局规则
         for rule in global_config.chat.talk_value_rules:
             if rule.platform or rule.item_id:
-                continue  # 有指定表示特定，跳过
+                continue  # 只匹配全局规则
-            parsed_range = TempMethodsHFC._parse_range(rule.time)
+            parsed_range = self._parse_range(rule.time)
             if not parsed_range:
-                continue
+                continue  # 无法解析的时间范围，跳过
             start_min, end_min = parsed_range
             in_range: bool = False
             if start_min <= end_min:
                 in_range = start_min <= now_min <= end_min
-            else:
+            else:  # 跨天的时间范围
                 in_range = now_min >= start_min or now_min <= end_min
             if in_range:
-                return rule.value or 0.0000001
+                return rule.value or 0.0  # 如果规则生效但没有设置值，返回0.0
-        return result
+        return result  # 如果没有任何规则生效，返回默认值
 
-    @staticmethod
+    def _parse_range(self, range_str: str) -> Optional[tuple[int, int]]:
-    def _parse_range(range_str: str) -> Optional[tuple[int, int]]:
         """解析 "HH:MM-HH:MM" 到 (start_min, end_min)。"""
         try:
             start_str, end_str = [s.strip() for s in range_str.split("-")]

src/chat/heart_flow/hfc_utils.py

@@ -1,6 +1,8 @@
-import time
+from dataclasses import dataclass
 from typing import Optional, Dict, Any
 
+import time
+
 from src.config.config import global_config
 from src.common.logger import get_logger
 from src.chat.message_receive.chat_stream import get_chat_manager
@@ -11,6 +13,13 @@ from src.common.message_repository import count_messages
 
 logger = get_logger(__name__)
 
+@dataclass
+class CyclePlanInfo:
+    ...
+    
+@dataclass
+class CycleActionInfo:
+    ...
 
 class CycleDetail:
     """循环信息记录类"""
@@ -22,8 +31,8 @@ class CycleDetail:
         self.end_time: Optional[float] = None
         self.timers: Dict[str, float] = {}
 
-        self.loop_plan_info: Dict[str, Any] = {}
+        self.loop_plan_info: CyclePlanInfo = CyclePlanInfo()
-        self.loop_action_info: Dict[str, Any] = {}
+        self.loop_action_info: CycleActionInfo = CycleActionInfo()
 
     def to_dict(self) -> Dict[str, Any]:
         """将循环信息转换为字典格式"""

src/chat/message_receive/__init__.py

@@ -1,10 +1,5 @@
 from src.chat.emoji_system.emoji_manager import emoji_manager
-from src.chat.message_receive.chat_stream import get_chat_manager
+from src.chat.message_receive.chat_manager import chat_manager
-from src.chat.message_receive.storage import MessageStorage
 
 
-__all__ = [
+__all__ = ["chat_manager", "emoji_manager"]
-    "get_chat_manager",
-    "MessageStorage",
-    "emoji_manager",
-]

src/chat/message_receive/bot.py

@@ -1,21 +1,23 @@
 import traceback
 import os
-import re
 
-from typing import Dict, Any, Optional
+from maim_message import MessageBase
-from maim_message import UserInfo, Seg, GroupInfo
+from typing import Dict, Any
+
 
 from src.common.logger import get_logger
-from src.config.config import global_config
+from src.common.utils.utils_message import MessageUtils
-from src.chat.message_receive.chat_stream import get_chat_manager
+from src.common.utils.utils_session import SessionUtils
-from src.chat.message_receive.message import MessageRecv
+from src.chat.message_receive.message_old import MessageRecv
-from src.chat.message_receive.storage import MessageStorage
 from src.chat.heart_flow.heartflow_message_processor import HeartFCMessageReceiver
 from src.chat.brain_chat.PFC.pfc_manager import PFCManager
 from src.chat.utils.prompt_builder import Prompt, global_prompt_manager
 from src.plugin_system.core import component_registry, events_manager, global_announcement_manager
 from src.plugin_system.base import BaseCommand, EventType
 
+from .message import SessionMessage
+from .chat_manager import chat_manager
+
 # 定义日志配置
 
 # 获取项目根目录（假设本文件在src/chat/message_receive/下，根目录为上上上级目录）
@@ -25,50 +27,6 @@ PROJECT_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "../../..
 logger = get_logger("chat")
 
 
-def _check_ban_words(text: str, userinfo: UserInfo, group_info: Optional[GroupInfo] = None) -> bool:
-    """检查消息是否包含过滤词
-
-    Args:
-        text: 待检查的文本
-        chat: 聊天对象
-        userinfo: 用户信息
-
-    Returns:
-        bool: 是否包含过滤词
-    """
-    for word in global_config.message_receive.ban_words:
-        if word in text:
-            chat_name = group_info.group_name if group_info else "私聊"
-            logger.info(f"[{chat_name}]{userinfo.user_nickname}:{text}")
-            logger.info(f"[过滤词识别]消息中含有{word}，filtered")
-            return True
-    return False
-
-
-def _check_ban_regex(text: str, userinfo: UserInfo, group_info: Optional[GroupInfo] = None) -> bool:
-    """检查消息是否匹配过滤正则表达式
-
-    Args:
-        text: 待检查的文本
-        chat: 聊天对象
-        userinfo: 用户信息
-
-    Returns:
-        bool: 是否匹配过滤正则
-    """
-    # 检查text是否为None或空字符串
-    if text is None or not text:
-        return False
-
-    for pattern in global_config.message_receive.ban_msgs_regex:
-        if re.search(pattern, text):
-            chat_name = group_info.group_name if group_info else "私聊"
-            logger.info(f"[{chat_name}]{userinfo.user_nickname}:{text}")
-            logger.info(f"[正则表达式过滤]消息匹配到{pattern}，filtered")
-            return True
-    return False
-
-
 class ChatBot:
     def __init__(self):
         self.bot = None  # bot 实例引用
@@ -100,9 +58,11 @@ class ChatBot:
             logger.error(f"创建PFC聊天失败: {e}")
             logger.error(traceback.format_exc())
 
-    async def _process_commands(self, message: MessageRecv):
+    async def _process_commands(self, message: SessionMessage):
         # sourcery skip: use-named-expression
         """使用新插件系统处理命令"""
+        if not message.processed_plain_text:
+            return False, None, True  # 没有文本内容，继续处理消息
         try:
             text = message.processed_plain_text
 
@@ -112,11 +72,8 @@ class ChatBot:
                 command_class, matched_groups, command_info = command_result
                 plugin_name = command_info.plugin_name
                 command_name = command_info.name
-                if (
+                if message.session_id and command_name in global_announcement_manager.get_disabled_chat_commands(
-                    message.chat_stream
+                    message.session_id
-                    and message.chat_stream.stream_id
-                    and command_name
-                    in global_announcement_manager.get_disabled_chat_commands(message.chat_stream.stream_id)
                 ):
                     logger.info("用户禁用的命令，跳过处理")
                     return False, None, True
@@ -269,97 +226,115 @@ class ChatBot:
                 )
             # print(message_data)
             # logger.debug(str(message_data))
-            message = MessageRecv(message_data)
+            maim_raw_message = MessageBase.from_dict(message_data)
+            message = SessionMessage.from_maim_message(maim_raw_message)
             group_info = message.message_info.group_info
             user_info = message.message_info.user_info
 
-            continue_flag, modified_message = await events_manager.handle_mai_events(
+            session_id = SessionUtils.calculate_session_id(
-                EventType.ON_MESSAGE_PRE_PROCESS, message
+                message.platform,
+                user_id=message.message_info.user_info.user_id,
+                group_id=group_info.group_id if group_info else None,
             )
-            if not continue_flag:
-                return
-            if modified_message and modified_message._modify_flags.modify_message_segments:
-                message.message_segment = Seg(type="seglist", data=modified_message.message_segments)
 
-            if await self.handle_notice_message(message):
+            message.session_id = session_id  # 正确初始化session_id
-                pass
 
-            # 处理消息内容，生成纯文本
+            # TODO: 修复事件预处理部分
+            # continue_flag, modified_message = await events_manager.handle_mai_events(
+            #     EventType.ON_MESSAGE_PRE_PROCESS, message
+            # )
+            # if not continue_flag:
+            #     return
+            # if modified_message and modified_message._modify_flags.modify_message_segments:
+            #     message.message_segment = Seg(type="seglist", data=modified_message.message_segments)
+
+            # TODO: notice消息处理
+            # if await self.handle_notice_message(message):
+            #     pass
+
+            # 处理消息内容，识别表情包等二进制数据并转化为文本描述
             await message.process()
 
             # 平台层的 @ 检测由底层 is_mentioned_bot_in_message 统一处理；此处不做用户名硬编码匹配
 
             # 过滤检查
-            if _check_ban_words(
+            text = message.processed_plain_text or ""
-                message.processed_plain_text,
+            is_banned, word = MessageUtils.check_ban_words(text)
-                user_info,  # type: ignore
+            if is_banned:
-                group_info,
+                chat_name = group_info.group_name if group_info else "私聊"
-            ) or _check_ban_regex(
+                logger.info(f"[{chat_name}]{user_info.user_nickname}:{text}")
-                message.raw_message,  # type: ignore
+                logger.info(f"[过滤词识别]消息中含有{word}，filtered")
-                user_info,  # type: ignore
+                return
-                group_info,
+            is_banned_regex, pattern = MessageUtils.check_ban_regex(text)
-            ):
+            if is_banned_regex:
+                chat_name = group_info.group_name if group_info else "私聊"
+                logger.info(f"[{chat_name}]{user_info.user_nickname}:{text}")
+                logger.info(f"[正则表达式过滤]消息匹配到{pattern}，filtered")
                 return
 
-            get_chat_manager().register_message(message)
+            chat_manager.register_message(message)
 
-            chat = await get_chat_manager().get_or_create_stream(
+            platform = message.platform
-                platform=message.message_info.platform,  # type: ignore
+            user_id = user_info.user_id
-                user_info=user_info,  # type: ignore
+            group_id = group_info.group_id if group_info else None
-                group_info=group_info,
+            _ = await chat_manager.get_or_create_session(platform, user_id, group_id)  # 确保会话存在
-            )
 
-            message.update_chat_stream(chat)
+            # message.update_chat_stream(chat)
-
-            # if await self.check_ban_content(message):
-            #     logger.warning(f"检测到消息中含有违法，色情，暴力，反动，敏感内容，消息内容：{message.processed_plain_text}，发送者：{message.message_info.user_info.user_nickname}")
-            #     return
 
+            # TODO: 在新命令系统完成后恢复这里
             # 命令处理 - 使用新插件系统检查并处理命令
-            is_command, cmd_result, continue_process = await self._process_commands(message)
+            # is_command, cmd_result, continue_process = await self._process_commands(message)
 
             # 如果是命令且不需要继续处理，则直接返回
-            if is_command and not continue_process:
+            # if is_command and not continue_process:
-                await MessageStorage.store_message(message, chat)
+            #     await MessageStorage.store_message(message, chat)
-                logger.info(f"命令处理完成，跳过后续消息处理: {cmd_result}")
+            #     logger.info(f"命令处理完成，跳过后续消息处理: {cmd_result}")
-                return
+            #     return
 
-            continue_flag, modified_message = await events_manager.handle_mai_events(EventType.ON_MESSAGE, message)
+            # continue_flag, modified_message = await events_manager.handle_mai_events(EventType.ON_MESSAGE, message)
-            if not continue_flag:
+            # if not continue_flag:
-                return
+            #     return
-            if modified_message and modified_message._modify_flags.modify_plain_text:
+            # if modified_message and modified_message._modify_flags.modify_plain_text:
-                message.processed_plain_text = modified_message.plain_text
+            #     message.processed_plain_text = modified_message.plain_text
 
-            # 确认从接口发来的message是否有自定义的prompt模板信息
+            # # 确认从接口发来的message是否有自定义的prompt模板信息
-            if message.message_info.template_info and not message.message_info.template_info.template_default:
+            # if message.message_info.template_info and not message.message_info.template_info.template_default:
-                template_group_name: Optional[str] = message.message_info.template_info.template_name  # type: ignore
+            #     template_group_name: Optional[str] = message.message_info.template_info.template_name  # type: ignore
-                template_items = message.message_info.template_info.template_items
+            #     template_items = message.message_info.template_info.template_items
-                async with global_prompt_manager.async_message_scope(template_group_name):
+            #     async with global_prompt_manager.async_message_scope(template_group_name):
-                    if isinstance(template_items, dict):
+            #         if isinstance(template_items, dict):
-                        for k in template_items.keys():
+            #             for k in template_items.keys():
-                            await Prompt.create_async(template_items[k], k)
+            #                 await Prompt.create_async(template_items[k], k)
-                            logger.debug(f"注册{template_items[k]},{k}")
+            #                 logger.debug(f"注册{template_items[k]},{k}")
-            else:
+            # else:
-                template_group_name = None
+            #     template_group_name = None
 
+            # async def preprocess():
+            #     # 根据聊天类型路由消息
+            #     if group_info is None:
+            #         # 私聊消息 -> PFC系统
+            #         logger.debug("[私聊]检测到私聊消息，路由到PFC系统")
+            #         await MessageStorage.store_message(message, chat)
+            #         await self._create_pfc_chat(message)
+            #     else:
+            #         # 群聊消息 -> HeartFlow系统
+            #         logger.debug("[群聊]检测到群聊消息，路由到HeartFlow系统")
+            #         await self.heartflow_message_receiver.process_message(message)
+
+            # if template_group_name:
+            #     async with global_prompt_manager.async_message_scope(template_group_name):
+            #         await preprocess()
+            # else:
+            #     await preprocess()
             async def preprocess():
-                # 根据聊天类型路由消息
                 if group_info is None:
-                    # 私聊消息 -> PFC系统
                     logger.debug("[私聊]检测到私聊消息，路由到PFC系统")
-                    await MessageStorage.store_message(message, chat)
+                    MessageUtils.store_message_to_db(message)  # 存储消息到数据库
                     await self._create_pfc_chat(message)
                 else:
-                    # 群聊消息 -> HeartFlow系统
                     logger.debug("[群聊]检测到群聊消息，路由到HeartFlow系统")
                     await self.heartflow_message_receiver.process_message(message)
 
-            if template_group_name:
-                async with global_prompt_manager.async_message_scope(template_group_name):
-                    await preprocess()
-            else:
-                await preprocess()
-
         except Exception as e:
             logger.error(f"预处理消息失败: {e}")
             traceback.print_exc()