diff --git a/src/chat/brain_chat/brain_chat.py b/src/chat/brain_chat/brain_chat.py
index b5d2cec7..68790ae0 100644
--- a/src/chat/brain_chat/brain_chat.py
+++ b/src/chat/brain_chat/brain_chat.py
@@ -1,570 +1,572 @@
-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.data_models.message_data_model import ReplyContentType
-from src.chat.message_receive.chat_stream import ChatStream, get_chat_manager
-from src.chat.utils.prompt_builder import global_prompt_manager
-from src.chat.utils.timer_calculator import Timer
-from src.chat.brain_chat.brain_planner import BrainPlanner
-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.express.expression_learner import expression_learner_manager
-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,
-)
-
-if TYPE_CHECKING:
- from src.common.data_models.database_data_model import DatabaseMessages
- from src.common.data_models.message_data_model import ReplySetModel
-
-
-ERROR_LOOP_INFO = {
- "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("bc") # Logger Name Changed
-
-
-class BrainChatting:
- """
- 管理一个连续的私聊Brain Chat循环
- 用于在特定聊天流中生成回复。
- """
-
- def __init__(self, chat_id: str):
- """
- BrainChatting 初始化函数
-
- 参数:
- chat_id: 聊天流唯一标识符(如stream_id)
- on_stop_focus_chat: 当收到stop_focus_chat命令时调用的回调函数
- performance_version: 性能记录版本号，用于区分不同启动版本
- """
- # 基础属性
- self.stream_id: str = chat_id # 聊天流ID
- self.chat_stream: ChatStream = get_chat_manager().get_stream(self.stream_id) # type: ignore
- if not self.chat_stream:
- 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.action_manager = ActionManager()
- self.action_planner = BrainPlanner(chat_id=self.stream_id, action_manager=self.action_manager)
- self.action_modifier = ActionModifier(action_manager=self.action_manager, chat_id=self.stream_id)
-
- # 循环控制内部状态
- self.running: bool = False
- 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.more_plan = False
-
- async def start(self):
- """检查是否需要启动主循环，如果未激活则启动。"""
-
- # 如果循环已经激活，直接返回
- if self.running:
- logger.debug(f"{self.log_prefix} BrainChatting 已激活，无需重复启动")
- return
-
- try:
- # 标记为活动状态，防止重复启动
- self.running = True
-
- self._loop_task = asyncio.create_task(self._main_chat_loop())
- self._loop_task.add_done_callback(self._handle_loop_completion)
- logger.info(f"{self.log_prefix} BrainChatting 启动完成")
-
- except Exception as e:
- # 启动失败时重置状态
- self.running = False
- self._loop_task = None
- logger.error(f"{self.log_prefix} BrainChatting 启动失败: {e}")
- raise
-
- def _handle_loop_completion(self, task: asyncio.Task):
- """当 _hfc_loop 任务完成时执行的回调。"""
- try:
- if exception := task.exception():
- logger.error(f"{self.log_prefix} BrainChatting: 脱离了聊天(异常): {exception}")
- logger.error(traceback.format_exc()) # Log full traceback for exceptions
- else:
- logger.info(f"{self.log_prefix} BrainChatting: 脱离了聊天 (外部停止)")
- except asyncio.CancelledError:
- logger.info(f"{self.log_prefix} BrainChatting: 结束了聊天")
-
- def start_cycle(self) -> Tuple[Dict[str, float], str]:
- self._cycle_counter += 1
- 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():
- formatted_time = f"{elapsed * 1000:.2f}毫秒" if elapsed < 1 else 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"\n详情: {'; '.join(timer_strings)}" if timer_strings else "")
- )
-
- async def _loopbody(self): # sourcery skip: hoist-if-from-if
- 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=True,
- )
-
- if len(recent_messages_list) >= 1:
- self.last_read_time = time.time()
- await self._observe(recent_messages_list=recent_messages_list)
-
- else:
- # Normal模式：消息数量不足，等待
- 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,
- ) -> 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,
- )
-
- # 获取 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,
- ) -> 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
-
- async with global_prompt_manager.async_message_scope(self.chat_stream.context.get_template_name()):
- await self.expression_learner.trigger_learning_for_chat()
-
- cycle_timers, thinking_id = self.start_cycle()
- logger.info(f"{self.log_prefix} 开始第{self._cycle_counter}次思考")
-
- # 第一步：动作检查
- 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),
- )
- 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,
- interest=global_config.personality.interest,
- )
- 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,
- )
-
- # 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 = ""
-
- for result in results:
- 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["reply_text"]
- elif result["action_type"] == "reply":
- if result["success"]:
- reply_loop_info = result["loop_info"]
- reply_text_from_reply = result["reply_text"]
- else:
- logger.warning(f"{self.log_prefix} 回复动作执行失败")
-
- # 构建最终的循环信息
- 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)
-
- 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,
- reasoning: str,
- action_data: dict,
- cycle_timers: Dict[str, float],
- thinking_id: str,
- action_message: Optional["DatabaseMessages"] = None,
- ) -> tuple[bool, str, str]:
- """
- 处理规划动作，使用动作工厂创建相应的动作处理器
-
- 参数:
- action: 动作类型
- reasoning: 决策理由
- action_data: 动作数据，包含不同动作需要的参数
- cycle_timers: 计时器字典
- thinking_id: 思考ID
-
- 返回:
- tuple[bool, str, str]: (是否执行了动作, 思考消息ID, 命令)
- """
- try:
- # 使用工厂创建动作处理器实例
- try:
- action_handler = self.action_manager.create_action(
- action_name=action,
- action_data=action_data,
- action_reasoning=reasoning,
- cycle_timers=cycle_timers,
- thinking_id=thinking_id,
- chat_stream=self.chat_stream,
- log_prefix=self.log_prefix,
- action_message=action_message,
- )
- except Exception as e:
- logger.error(f"{self.log_prefix} 创建动作处理器时出错: {e}")
- traceback.print_exc()
- return False, "", ""
-
- if not action_handler:
- logger.warning(f"{self.log_prefix} 未能创建动作处理器: {action}")
- return False, "", ""
-
+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.data_models.message_data_model import ReplyContentType
+from src.chat.message_receive.chat_stream import ChatStream, get_chat_manager
+from src.chat.utils.prompt_builder import global_prompt_manager
+from src.chat.utils.timer_calculator import Timer
+from src.chat.brain_chat.brain_planner import BrainPlanner
+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.express.expression_learner import expression_learner_manager
+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,
+)
+
+if TYPE_CHECKING:
+ from src.common.data_models.database_data_model import DatabaseMessages
+ from src.common.data_models.message_data_model import ReplySetModel
+
+
+ERROR_LOOP_INFO = {
+ "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("bc") # Logger Name Changed
+
+
+class BrainChatting:
+ """
+ 管理一个连续的私聊Brain Chat循环
+ 用于在特定聊天流中生成回复。
+ """
+
+ def __init__(self, chat_id: str):
+ """
+ BrainChatting 初始化函数
+
+ 参数:
+ chat_id: 聊天流唯一标识符(如stream_id)
+ on_stop_focus_chat: 当收到stop_focus_chat命令时调用的回调函数
+ performance_version: 性能记录版本号，用于区分不同启动版本
+ """
+ # 基础属性
+ self.stream_id: str = chat_id # 聊天流ID
+ self.chat_stream: ChatStream = get_chat_manager().get_stream(self.stream_id) # type: ignore
+ if not self.chat_stream:
+ 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.action_manager = ActionManager()
+ self.action_planner = BrainPlanner(chat_id=self.stream_id, action_manager=self.action_manager)
+ self.action_modifier = ActionModifier(action_manager=self.action_manager, chat_id=self.stream_id)
+
+ # 循环控制内部状态
+ self.running: bool = False
+ 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.more_plan = False
+
+ async def start(self):
+ """检查是否需要启动主循环，如果未激活则启动。"""
+
+ # 如果循环已经激活，直接返回
+ if self.running:
+ logger.debug(f"{self.log_prefix} BrainChatting 已激活，无需重复启动")
+ return
+
+ try:
+ # 标记为活动状态，防止重复启动
+ self.running = True
+
+ self._loop_task = asyncio.create_task(self._main_chat_loop())
+ self._loop_task.add_done_callback(self._handle_loop_completion)
+ logger.info(f"{self.log_prefix} BrainChatting 启动完成")
+
+ except Exception as e:
+ # 启动失败时重置状态
+ self.running = False
+ self._loop_task = None
+ logger.error(f"{self.log_prefix} BrainChatting 启动失败: {e}")
+ raise
+
+ def _handle_loop_completion(self, task: asyncio.Task):
+ """当 _hfc_loop 任务完成时执行的回调。"""
+ try:
+ if exception := task.exception():
+ logger.error(f"{self.log_prefix} BrainChatting: 脱离了聊天(异常): {exception}")
+ logger.error(traceback.format_exc()) # Log full traceback for exceptions
+ else:
+ logger.info(f"{self.log_prefix} BrainChatting: 脱离了聊天 (外部停止)")
+ except asyncio.CancelledError:
+ logger.info(f"{self.log_prefix} BrainChatting: 结束了聊天")
+
+ def start_cycle(self) -> Tuple[Dict[str, float], str]:
+ self._cycle_counter += 1
+ 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():
+ formatted_time = f"{elapsed * 1000:.2f}毫秒" if elapsed < 1 else 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"\n详情: {'; '.join(timer_strings)}" if timer_strings else "")
+ )
+
+ async def _loopbody(self): # sourcery skip: hoist-if-from-if
+ 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=True,
+ )
+
+ if len(recent_messages_list) >= 1:
+ self.last_read_time = time.time()
+ await self._observe(recent_messages_list=recent_messages_list)
+
+ else:
+ # Normal模式：消息数量不足，等待
+ 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,
+ ) -> 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,
+ )
+
+ # 获取 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,
+ ) -> 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
+
+ async with global_prompt_manager.async_message_scope(self.chat_stream.context.get_template_name()):
+ await self.expression_learner.trigger_learning_for_chat()
+
+ cycle_timers, thinking_id = self.start_cycle()
+ logger.info(f"{self.log_prefix} 开始第{self._cycle_counter}次思考")
+
+ # 第一步：动作检查
+ 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),
+ )
+ 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,
+ interest=global_config.personality.interest,
+ )
+ 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,
+ )
+
+ # 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 = ""
+
+ for result in results:
+ 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["reply_text"]
+ elif result["action_type"] == "reply":
+ if result["success"]:
+ reply_loop_info = result["loop_info"]
+ reply_text_from_reply = result["reply_text"]
+ else:
+ logger.warning(f"{self.log_prefix} 回复动作执行失败")
+
+ # 构建最终的循环信息
+ 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)
+
+ 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,
+ reasoning: str,
+ action_data: dict,
+ cycle_timers: Dict[str, float],
+ thinking_id: str,
+ action_message: Optional["DatabaseMessages"] = None,
+ ) -> tuple[bool, str, str]:
+ """
+ 处理规划动作，使用动作工厂创建相应的动作处理器
+
+ 参数:
+ action: 动作类型
+ reasoning: 决策理由
+ action_data: 动作数据，包含不同动作需要的参数
+ cycle_timers: 计时器字典
+ thinking_id: 思考ID
+
+ 返回:
+ tuple[bool, str, str]: (是否执行了动作, 思考消息ID, 命令)
+ """
+ try:
+ # 使用工厂创建动作处理器实例
+ try:
+ action_handler = self.action_manager.create_action(
+ action_name=action,
+ action_data=action_data,
+ action_reasoning=reasoning,
+ cycle_timers=cycle_timers,
+ thinking_id=thinking_id,
+ chat_stream=self.chat_stream,
+ log_prefix=self.log_prefix,
+ action_message=action_message,
+ )
+ except Exception as e:
+ logger.error(f"{self.log_prefix} 创建动作处理器时出错: {e}")
+ traceback.print_exc()
+ return False, "", ""
+
+ if not action_handler:
+ logger.warning(f"{self.log_prefix} 未能创建动作处理器: {action}")
+ return False, "", ""
+
# 处理动作并获取结果（固定记录一次动作信息）
- result = await action_handler.run()
- success, action_text = result
- command = ""
-
- return success, action_text, command
-
- 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,
- ) -> str:
- 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, 4)
-
- if need_reply:
- logger.info(f"{self.log_prefix} 从思考到回复，共有{new_message_count}条新消息，使用引用回复")
-
- 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):
- 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信息到数据库
- 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={"reason": reason},
- action_name="no_reply",
- )
- return {"action_type": "no_reply", "success": True, "reply_text": "", "command": ""}
-
- elif action_planner_info.action_type == "reply":
- try:
- 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=action_planner_info.reasoning or "",
- enable_tool=global_config.tool.enable_tool,
- request_type="replyer",
- from_plugin=False,
- )
-
- 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, "reply_text": "", "loop_info": None}
-
- except asyncio.CancelledError:
- logger.debug(f"{self.log_prefix} 并行执行：回复生成任务已被取消")
- return {"action_type": "reply", "success": False, "reply_text": "", "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,
- )
- return {
- "action_type": "reply",
- "success": True,
- "reply_text": reply_text,
- "loop_info": loop_info,
- }
-
- # 其他动作
- else:
- # 执行普通动作
- with Timer("动作执行", cycle_timers):
- success, reply_text, command = await self._handle_action(
- action_planner_info.action_type,
- action_planner_info.reasoning or "",
- action_planner_info.action_data or {},
- cycle_timers,
- thinking_id,
- action_planner_info.action_message,
- )
- return {
- "action_type": action_planner_info.action_type,
- "success": success,
- "reply_text": reply_text,
- "command": command,
- }
-
- 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,
- "reply_text": "",
- "loop_info": None,
- "error": str(e),
- }
+ # BaseAction 定义了异步方法 execute() 作为统一执行入口
+ # 这里调用 execute() 以兼容所有 Action 实现
+ result = await action_handler.execute()
+ success, action_text = result
+ command = ""
+
+ return success, action_text, command
+
+ 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,
+ ) -> str:
+ 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, 4)
+
+ if need_reply:
+ logger.info(f"{self.log_prefix} 从思考到回复，共有{new_message_count}条新消息，使用引用回复")
+
+ 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):
+ 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信息到数据库
+ 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={"reason": reason},
+ action_name="no_reply",
+ )
+ return {"action_type": "no_reply", "success": True, "reply_text": "", "command": ""}
+
+ elif action_planner_info.action_type == "reply":
+ try:
+ 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=action_planner_info.reasoning or "",
+ enable_tool=global_config.tool.enable_tool,
+ request_type="replyer",
+ from_plugin=False,
+ )
+
+ 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, "reply_text": "", "loop_info": None}
+
+ except asyncio.CancelledError:
+ logger.debug(f"{self.log_prefix} 并行执行：回复生成任务已被取消")
+ return {"action_type": "reply", "success": False, "reply_text": "", "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,
+ )
+ return {
+ "action_type": "reply",
+ "success": True,
+ "reply_text": reply_text,
+ "loop_info": loop_info,
+ }
+
+ # 其他动作
+ else:
+ # 执行普通动作
+ with Timer("动作执行", cycle_timers):
+ success, reply_text, command = await self._handle_action(
+ action_planner_info.action_type,
+ action_planner_info.reasoning or "",
+ action_planner_info.action_data or {},
+ cycle_timers,
+ thinking_id,
+ action_planner_info.action_message,
+ )
+ return {
+ "action_type": action_planner_info.action_type,
+ "success": success,
+ "reply_text": reply_text,
+ "command": command,
+ }
+
+ 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,
+ "reply_text": "",
+ "loop_info": None,
+ "error": str(e),
+ }
diff --git a/src/chat/brain_chat/brain_planner.py b/src/chat/brain_chat/brain_planner.py
index 35cdf172..0cbaab36 100644
--- a/src/chat/brain_chat/brain_planner.py
+++ b/src/chat/brain_chat/brain_planner.py
@@ -249,6 +249,8 @@ class BrainPlanner:
# 获取必要信息
is_group_chat, chat_target_info, current_available_actions = self.get_necessary_info()
+ # 提及/被@ 的处理由心流或统一判定模块驱动；Planner 不再做硬编码强制回复
+
# 应用激活类型过滤
filtered_actions = self._filter_actions_by_activation_type(available_actions, chat_content_block_short)
diff --git a/src/chat/heart_flow/heartflow_message_processor.py b/src/chat/heart_flow/heartflow_message_processor.py
index fdc13e2f..822d05de 100644
--- a/src/chat/heart_flow/heartflow_message_processor.py
+++ b/src/chat/heart_flow/heartflow_message_processor.py
@@ -1,7 +1,7 @@
import re
import traceback
-from typing import Tuple, TYPE_CHECKING
+from typing import TYPE_CHECKING
from src.chat.message_receive.message import MessageRecv
from src.chat.message_receive.storage import MessageStorage
@@ -17,31 +17,6 @@ if TYPE_CHECKING:
logger = get_logger("chat")
-
-async def _calculate_interest(message: MessageRecv) -> Tuple[float, list[str]]:
- """计算消息的兴趣度
-
- Args:
- message: 待处理的消息对象
-
- Returns:
- Tuple[float, bool, list[str]]: (兴趣度, 是否被提及, 关键词)
- """
- if message.is_picid or message.is_emoji:
- return 0.0, []
-
- is_mentioned, is_at, reply_probability_boost = is_mentioned_bot_in_message(message)
- # interested_rate = 0.0
- keywords = []
-
- message.interest_value = 1
- message.is_mentioned = is_mentioned
- message.is_at = is_at
- message.reply_probability_boost = reply_probability_boost
-
- return 1, keywords
-
-
class HeartFCMessageReceiver:
"""心流处理器，负责处理接收到的消息并计算兴趣度"""
@@ -67,12 +42,16 @@ class HeartFCMessageReceiver:
userinfo = message.message_info.user_info
chat = message.chat_stream
- # 2. 兴趣度计算与更新
- _, keywords = await _calculate_interest(message)
+ # 2. 计算at信息
+ is_mentioned, is_at, reply_probability_boost = is_mentioned_bot_in_message(message)
+ print(f"is_mentioned: {is_mentioned}, is_at: {is_at}, reply_probability_boost: {reply_probability_boost}")
+ message.is_mentioned = is_mentioned
+ message.is_at = is_at
+ message.reply_probability_boost = reply_probability_boost
await self.storage.store_message(message, chat)
- _heartflow_chat: HeartFChatting = await heartflow.get_or_create_heartflow_chat(chat.stream_id) # type: ignore
+ await heartflow.get_or_create_heartflow_chat(chat.stream_id) # type: ignore
# 3. 日志记录
mes_name = chat.group_info.group_name if chat.group_info else "私聊"
diff --git a/src/chat/message_receive/bot.py b/src/chat/message_receive/bot.py
index 2eb0b4e3..43d2754a 100644
--- a/src/chat/message_receive/bot.py
+++ b/src/chat/message_receive/bot.py
@@ -221,6 +221,8 @@ class ChatBot:
# 处理消息内容，生成纯文本
await message.process()
+ # 平台层的 @ 检测由底层 is_mentioned_bot_in_message 统一处理；此处不做用户名硬编码匹配
+
# 过滤检查
if _check_ban_words(
message.processed_plain_text,
diff --git a/src/chat/message_receive/message.py b/src/chat/message_receive/message.py
index aa992552..6c1ec1a7 100644
--- a/src/chat/message_receive/message.py
+++ b/src/chat/message_receive/message.py
@@ -130,6 +130,16 @@ class MessageRecv(Message):
self.key_words = []
self.key_words_lite = []
+ # 兼容适配器通过 additional_config 传入的 @ 标记
+ try:
+ msg_info_dict = message_dict.get("message_info", {})
+ add_cfg = msg_info_dict.get("additional_config") or {}
+ if isinstance(add_cfg, dict) and add_cfg.get("at_bot"):
+ # 标记为被提及，提高后续回复优先级
+ self.is_mentioned = True # type: ignore
+ except Exception:
+ pass
+
def update_chat_stream(self, chat_stream: "ChatStream"):
self.chat_stream = chat_stream
diff --git a/src/chat/replyer/group_generator.py b/src/chat/replyer/group_generator.py
index 88935da7..3a24f04f 100644
--- a/src/chat/replyer/group_generator.py
+++ b/src/chat/replyer/group_generator.py
@@ -529,7 +529,7 @@ class DefaultReplyer:
show_actions=True,
)
core_dialogue_prompt = f"""--------------------------------
-这是你和{sender}的对话，你们正在交流中：
+这是上述中你和{sender}的对话摘要，内容从上面的对话中截取，便于你理解：
{core_dialogue_prompt_str}
--------------------------------
"""
diff --git a/src/chat/utils/chat_message_builder.py b/src/chat/utils/chat_message_builder.py
index 52559ecb..8915e810 100644
--- a/src/chat/utils/chat_message_builder.py
+++ b/src/chat/utils/chat_message_builder.py
@@ -43,9 +43,12 @@ def replace_user_references(
if name_resolver is None:
def default_resolver(platform: str, user_id: str) -> str:
- # 检查是否是机器人自己
- if replace_bot_name and user_id == global_config.bot.qq_account:
- return f"{global_config.bot.nickname}(你)"
+ # 检查是否是机器人自己（支持多平台）
+ if replace_bot_name:
+ if platform == "qq" and user_id == global_config.bot.qq_account:
+ return f"{global_config.bot.nickname}(你)"
+ if platform == "telegram" and user_id == getattr(global_config.bot, "telegram_account", ""):
+ return f"{global_config.bot.nickname}(你)"
person = Person(platform=platform, user_id=user_id)
return person.person_name or user_id # type: ignore
@@ -92,6 +95,8 @@ def replace_user_references(
new_content += content[last_end:]
content = new_content
+ # Telegram 文本 @username 的显示映射交由适配器或平台层处理；此处不做硬编码替换
+
return content
@@ -432,7 +437,10 @@ def _build_readable_messages_internal(
person_name = (
person.person_name or f"{user_nickname}" or (f"昵称：{user_cardname}" if user_cardname else "某人")
)
- if replace_bot_name and user_id == global_config.bot.qq_account:
+ if replace_bot_name and (
+ (platform == global_config.bot.platform and user_id == global_config.bot.qq_account)
+ or (platform == "telegram" and user_id == getattr(global_config.bot, "telegram_account", ""))
+ ):
person_name = f"{global_config.bot.nickname}(你)"
# 使用独立函数处理用户引用格式
@@ -866,7 +874,9 @@ async def build_anonymous_messages(messages: List[DatabaseMessages]) -> str:
# print(f"get_anon_name: platform:{platform}, user_id:{user_id}")
# print(f"global_config.bot.qq_account:{global_config.bot.qq_account}")
- if user_id == global_config.bot.qq_account:
+ if (platform == "qq" and user_id == global_config.bot.qq_account) or (
+ platform == "telegram" and user_id == getattr(global_config.bot, "telegram_account", "")
+ ):
# print("SELF11111111111111")
return "SELF"
try:
diff --git a/src/chat/utils/statistic.py b/src/chat/utils/statistic.py
index 97ef1cc0..2c4dd619 100644
--- a/src/chat/utils/statistic.py
+++ b/src/chat/utils/statistic.py
@@ -492,10 +492,15 @@ class StatisticOutputTask(AsyncTask):
continue
# Update name_mapping
- if chat_id in self.name_mapping:
- if chat_name != self.name_mapping[chat_id][0] and message_time_ts > self.name_mapping[chat_id][1]:
+ try:
+ if chat_id in self.name_mapping:
+ if chat_name != self.name_mapping[chat_id][0] and message_time_ts > self.name_mapping[chat_id][1]:
+ self.name_mapping[chat_id] = (chat_name, message_time_ts)
+ else:
self.name_mapping[chat_id] = (chat_name, message_time_ts)
- else:
+ except (IndexError, TypeError) as e:
+ logger.warning(f"更新 name_mapping 时发生错误，chat_id: {chat_id}, 错误: {e}")
+ # 重置为正确的格式
self.name_mapping[chat_id] = (chat_name, message_time_ts)
for idx, (_, period_start_dt) in enumerate(collect_period):
@@ -518,7 +523,21 @@ class StatisticOutputTask(AsyncTask):
# 如果存在上次完整统计数据，则使用该数据进行增量统计
last_stat: Dict[str, Any] = local_storage["last_full_statistics"] # 上次完整统计数据 # type: ignore
- self.name_mapping = last_stat["name_mapping"] # 上次完整统计数据的名称映射
+ # 修复 name_mapping 数据类型不匹配问题
+ # JSON 中存储为列表，但代码期望为元组
+ raw_name_mapping = last_stat["name_mapping"]
+ self.name_mapping = {}
+ for chat_id, value in raw_name_mapping.items():
+ if isinstance(value, list) and len(value) == 2:
+ # 将列表转换为元组
+ self.name_mapping[chat_id] = (value[0], value[1])
+ elif isinstance(value, tuple) and len(value) == 2:
+ # 已经是元组，直接使用
+ self.name_mapping[chat_id] = value
+ else:
+ # 数据格式不正确，跳过或使用默认值
+ logger.warning(f"name_mapping 中 chat_id {chat_id} 的数据格式不正确: {value}")
+ continue
last_all_time_stat = last_stat["stat_data"] # 上次完整统计的统计数据
last_stat_timestamp = datetime.fromtimestamp(last_stat["timestamp"]) # 上次完整统计数据的时间戳
self.stat_period = [item for item in self.stat_period if item[0] != "all_time"] # 删除"所有时间"的统计时段
@@ -571,8 +590,14 @@ class StatisticOutputTask(AsyncTask):
# 更新上次完整统计数据的时间戳
# 将所有defaultdict转换为普通dict以避免类型冲突
clean_stat_data = self._convert_defaultdict_to_dict(stat["all_time"])
+
+ # 将 name_mapping 中的元组转换为列表，因为JSON不支持元组
+ json_safe_name_mapping = {}
+ for chat_id, (chat_name, timestamp) in self.name_mapping.items():
+ json_safe_name_mapping[chat_id] = [chat_name, timestamp]
+
local_storage["last_full_statistics"] = {
- "name_mapping": self.name_mapping,
+ "name_mapping": json_safe_name_mapping,
"stat_data": clean_stat_data,
"timestamp": now.timestamp(),
}
@@ -651,10 +676,13 @@ class StatisticOutputTask(AsyncTask):
if stats[TOTAL_MSG_CNT] <= 0:
return ""
output = ["聊天消息统计:", " 联系人/群组名称 消息数量"]
- output.extend(
- f"{self.name_mapping[chat_id][0][:32]:<32} {count:>10}"
- for chat_id, count in sorted(stats[MSG_CNT_BY_CHAT].items())
- )
+ for chat_id, count in sorted(stats[MSG_CNT_BY_CHAT].items()):
+ try:
+ chat_name = self.name_mapping.get(chat_id, ("未知聊天", 0))[0]
+ output.append(f"{chat_name[:32]:<32} {count:>10}")
+ except (IndexError, TypeError) as e:
+ logger.warning(f"格式化聊天统计时发生错误，chat_id: {chat_id}, 错误: {e}")
+ output.append(f"{'未知聊天':<32} {count:>10}")
output.append("")
return "\n".join(output)
@@ -770,14 +798,16 @@ class StatisticOutputTask(AsyncTask):
)
# 聊天消息统计
- chat_rows = "\n".join(
- [
- f"| {self.name_mapping[chat_id][0]} | {count} |
"
- for chat_id, count in sorted(stat_data[MSG_CNT_BY_CHAT].items())
- ]
- if stat_data[MSG_CNT_BY_CHAT]
- else ["| 暂无数据 |
"]
- )
+ chat_rows = []
+ for chat_id, count in sorted(stat_data[MSG_CNT_BY_CHAT].items()):
+ try:
+ chat_name = self.name_mapping.get(chat_id, ("未知聊天", 0))[0]
+ chat_rows.append(f"| {chat_name} | {count} |
")
+ except (IndexError, TypeError) as e:
+ logger.warning(f"生成HTML聊天统计时发生错误，chat_id: {chat_id}, 错误: {e}")
+ chat_rows.append(f"| 未知聊天 | {count} |
")
+
+ chat_rows_html = "\n".join(chat_rows) if chat_rows else "| 暂无数据 |
"
# 生成HTML
return f"""
@@ -824,7 +854,7 @@ class StatisticOutputTask(AsyncTask):
| 联系人/群组名称 | 消息数量 |
|---|
- {chat_rows}
+ {chat_rows_html}
@@ -975,7 +1005,7 @@ class StatisticOutputTask(AsyncTask):
}}
// 聊天消息分布饼图
- const chatLabels = {[self.name_mapping[chat_id][0] for chat_id in sorted(stat_data[MSG_CNT_BY_CHAT].keys())] if stat_data[MSG_CNT_BY_CHAT] else []};
+ const chatLabels = {[self.name_mapping.get(chat_id, ("未知聊天", 0))[0] for chat_id in sorted(stat_data[MSG_CNT_BY_CHAT].keys())] if stat_data[MSG_CNT_BY_CHAT] else []};
if (chatLabels.length > 0) {{
const chatData = {{
labels: chatLabels,
diff --git a/src/chat/utils/utils.py b/src/chat/utils/utils.py
index 240ce609..ce3eab08 100644
--- a/src/chat/utils/utils.py
+++ b/src/chat/utils/utils.py
@@ -30,76 +30,146 @@ def is_english_letter(char: str) -> bool:
return "a" <= char.lower() <= "z"
-def db_message_to_str(message_dict: dict) -> str:
- logger.debug(f"message_dict: {message_dict}")
- time_str = time.strftime("%m-%d %H:%M:%S", time.localtime(message_dict["time"]))
- try:
- name = f"[({message_dict['user_id']}){message_dict.get('user_nickname', '')}]{message_dict.get('user_cardname', '')}"
- except Exception:
- name = message_dict.get("user_nickname", "") or f"用户{message_dict['user_id']}"
- content = message_dict.get("processed_plain_text", "")
- result = f"[{time_str}] {name}: {content}\n"
- logger.debug(f"result: {result}")
+def parse_platform_accounts(platforms: list[str]) -> dict[str, str]:
+ """解析 platforms 列表，返回平台到账号的映射
+
+ Args:
+ platforms: 格式为 ["platform:account"] 的列表，如 ["tg:123456789", "wx:wxid123"]
+
+ Returns:
+ 字典，键为平台名，值为账号
+ """
+ result = {}
+ for platform_entry in platforms:
+ if ":" in platform_entry:
+ platform_name, account = platform_entry.split(":", 1)
+ result[platform_name.strip()] = account.strip()
return result
+def get_current_platform_account(platform: str, platform_accounts: dict[str, str], qq_account: str) -> str:
+ """根据当前平台获取对应的账号
+
+ Args:
+ platform: 当前消息的平台
+ platform_accounts: 从 platforms 列表解析的平台账号映射
+ qq_account: QQ 账号（兼容旧配置）
+
+ Returns:
+ 当前平台对应的账号
+ """
+ if platform == "qq":
+ return qq_account
+ elif platform == "telegram":
+ # 优先使用 tg，其次使用 telegram
+ return platform_accounts.get("tg", "") or platform_accounts.get("telegram", "")
+ else:
+ # 其他平台直接使用平台名作为键
+ return platform_accounts.get(platform, "")
+
+
def is_mentioned_bot_in_message(message: MessageRecv) -> tuple[bool, bool, float]:
- """检查消息是否提到了机器人"""
- keywords = [global_config.bot.nickname] + list(global_config.bot.alias_names)
+ """检查消息是否提到了机器人（统一多平台实现）"""
+ text = message.processed_plain_text or ""
+ platform = getattr(message.message_info, "platform", "") or ""
+
+ # 获取各平台账号
+ platforms_list = getattr(global_config.bot, "platforms", []) or []
+ platform_accounts = parse_platform_accounts(platforms_list)
+ qq_account = str(getattr(global_config.bot, "qq_account", "") or "")
+
+ # 获取当前平台对应的账号
+ current_account = get_current_platform_account(platform, platform_accounts, qq_account)
+
+ nickname = str(global_config.bot.nickname or "")
+ alias_names = list(getattr(global_config.bot, "alias_names", []) or [])
+ keywords = [nickname] + alias_names
+
reply_probability = 0.0
is_at = False
is_mentioned = False
- # 这部分怎么处理啊啊啊啊
- # 我觉得可以给消息加一个 reply_probability_boost字段
- if (
- message.message_info.additional_config is not None
- and message.message_info.additional_config.get("is_mentioned") is not None
- ):
+ # 1) 直接的 additional_config 标记
+ add_cfg = getattr(message.message_info, "additional_config", None) or {}
+ if isinstance(add_cfg, dict):
+ if add_cfg.get("at_bot") or add_cfg.get("is_mentioned"):
+ is_mentioned = True
+ # 当提供数值型 is_mentioned 时，当作概率提升
+ try:
+ if add_cfg.get("is_mentioned") not in (None, ""):
+ reply_probability = float(add_cfg.get("is_mentioned")) # type: ignore
+ except Exception:
+ pass
+
+ # 2) 已经在上游设置过的 message.is_mentioned
+ if getattr(message, "is_mentioned", False):
+ is_mentioned = True
+
+ # 3) 扫描分段：是否包含 mention_bot（适配器插入）
+ def _has_mention_bot(seg) -> bool:
try:
- reply_probability = float(message.message_info.additional_config.get("is_mentioned")) # type: ignore
- is_mentioned = True
- return is_mentioned, is_at, reply_probability
- except Exception as e:
- logger.warning(str(e))
- logger.warning(
- f"消息中包含不合理的设置 is_mentioned: {message.message_info.additional_config.get('is_mentioned')}"
- )
+ if seg is None:
+ return False
+ if getattr(seg, "type", None) == "mention_bot":
+ return True
+ if getattr(seg, "type", None) == "seglist":
+ for s in getattr(seg, "data", []) or []:
+ if _has_mention_bot(s):
+ return True
+ return False
+ except Exception:
+ return False
- for keyword in keywords:
- if keyword in message.processed_plain_text:
- is_mentioned = True
-
- # 判断是否被@
- if re.search(rf"@<(.+?):{global_config.bot.qq_account}>", message.processed_plain_text):
+ if _has_mention_bot(getattr(message, "message_segment", None)):
is_at = True
is_mentioned = True
- if is_at and global_config.chat.at_bot_inevitable_reply:
+ # 4) 统一的 @ 检测逻辑
+ if current_account and not is_at and not is_mentioned:
+ if platform == "qq":
+ # QQ 格式: @
+ if re.search(rf"@<(.+?):{re.escape(current_account)}>", text):
+ is_at = True
+ is_mentioned = True
+ else:
+ # 其他平台格式: @username 或 @account
+ if re.search(rf"@{re.escape(current_account)}(\b|$)", text, flags=re.IGNORECASE):
+ is_at = True
+ is_mentioned = True
+
+ # 5) 统一的回复检测逻辑
+ if not is_mentioned:
+ # 通用回复格式：包含 "(你)" 或 "（你）"
+ if re.search(r"\[回复 .*?\(你\)：", text) or re.search(r"\[回复 .*?（你）：", text):
+ is_mentioned = True
+ # ID 形式的回复检测
+ elif current_account:
+ if re.search(rf"\[回复 (.+?)\({re.escape(current_account)}\)：(.+?)\]，说：", text):
+ is_mentioned = True
+ elif re.search(rf"\[回复<(.+?)(?=:{re.escape(current_account)}>)\:{re.escape(current_account)}>：(.+?)\]，说：", text):
+ is_mentioned = True
+
+ # 6) 名称/别名 提及（去除 @/回复标记后再匹配）
+ if not is_mentioned and keywords:
+ msg_content = text
+ # 去除各种 @ 与 回复标记，避免误判
+ msg_content = re.sub(r"@(.+?)（(\d+)）", "", msg_content)
+ msg_content = re.sub(r"@<(.+?)(?=:(\d+))\:(\d+)>", "", msg_content)
+ msg_content = re.sub(r"\[回复 (.+?)\(((\d+)|未知id|你)\)：(.+?)\]，说：", "", msg_content)
+ msg_content = re.sub(r"\[回复<(.+?)(?=:(\d+))\:(\d+)>：(.+?)\]，说：", "", msg_content)
+ for kw in keywords:
+ if kw and kw in msg_content:
+ is_mentioned = True
+ break
+
+ # 7) 概率设置
+ if is_at and getattr(global_config.chat, "at_bot_inevitable_reply", 1):
reply_probability = 1.0
logger.debug("被@，回复概率设置为100%")
- else:
- if not is_mentioned:
- # 判断是否被回复
- if re.match(
- rf"\[回复 (.+?)\({str(global_config.bot.qq_account)}\)：(.+?)\]，说：", message.processed_plain_text
- ) or re.match(
- rf"\[回复<(.+?)(?=:{str(global_config.bot.qq_account)}>)\:{str(global_config.bot.qq_account)}>：(.+?)\]，说：",
- message.processed_plain_text,
- ):
- is_mentioned = True
- else:
- # 判断内容中是否被提及
- message_content = re.sub(r"@(.+?)（(\d+)）", "", message.processed_plain_text)
- message_content = re.sub(r"@<(.+?)(?=:(\d+))\:(\d+)>", "", message_content)
- message_content = re.sub(r"\[回复 (.+?)\(((\d+)|未知id)\)：(.+?)\]，说：", "", message_content)
- message_content = re.sub(r"\[回复<(.+?)(?=:(\d+))\:(\d+)>：(.+?)\]，说：", "", message_content)
- for keyword in keywords:
- if keyword in message_content:
- is_mentioned = True
- if is_mentioned and global_config.chat.mentioned_bot_reply:
- reply_probability = 1.0
- logger.debug("被提及，回复概率设置为100%")
+ elif is_mentioned and getattr(global_config.chat, "mentioned_bot_reply", 1):
+ reply_probability = max(reply_probability, 1.0)
+ logger.debug("被提及，回复概率设置为100%")
+
return is_mentioned, is_at, reply_probability
@@ -115,45 +185,6 @@ async def get_embedding(text, request_type="embedding") -> Optional[List[float]]
return embedding
-def get_recent_group_speaker(chat_stream_id: str, sender, limit: int = 12) -> list:
- # 获取当前群聊记录内发言的人
- filter_query = {"chat_id": chat_stream_id}
- sort_order = [("time", -1)]
- recent_messages = find_messages(message_filter=filter_query, sort=sort_order, limit=limit)
-
- if not recent_messages:
- return []
-
- who_chat_in_group = []
- for db_msg in recent_messages:
- # user_info = UserInfo.from_dict(
- # {
- # "platform": msg_db_data["user_platform"],
- # "user_id": msg_db_data["user_id"],
- # "user_nickname": msg_db_data["user_nickname"],
- # "user_cardname": msg_db_data.get("user_cardname", ""),
- # }
- # )
- # if (
- # (user_info.platform, user_info.user_id) != sender
- # and user_info.user_id != global_config.bot.qq_account
- # and (user_info.platform, user_info.user_id, user_info.user_nickname) not in who_chat_in_group
- # and len(who_chat_in_group) < 5
- # ): # 排除重复，排除消息发送者，排除bot，限制加载的关系数目
- # who_chat_in_group.append((user_info.platform, user_info.user_id, user_info.user_nickname))
- if (
- (db_msg.user_info.platform, db_msg.user_info.user_id) != sender
- and db_msg.user_info.user_id != global_config.bot.qq_account
- and (db_msg.user_info.platform, db_msg.user_info.user_id, db_msg.user_info.user_nickname)
- not in who_chat_in_group
- and len(who_chat_in_group) < 5
- ): # 排除重复，排除消息发送者，排除bot，限制加载的关系数目
- who_chat_in_group.append(
- (db_msg.user_info.platform, db_msg.user_info.user_id, db_msg.user_info.user_nickname)
- )
-
- return who_chat_in_group
-
def split_into_sentences_w_remove_punctuation(text: str) -> list[str]:
"""将文本分割成句子，并根据概率合并
@@ -410,42 +441,6 @@ def calculate_typing_time(
return total_time # 加上回车时间
-def cosine_similarity(v1, v2):
- """计算余弦相似度"""
- dot_product = np.dot(v1, v2)
- norm1 = np.linalg.norm(v1)
- norm2 = np.linalg.norm(v2)
- return 0 if norm1 == 0 or norm2 == 0 else dot_product / (norm1 * norm2)
-
-
-def text_to_vector(text):
- """将文本转换为词频向量"""
- # 分词
- words = jieba.lcut(text)
- return Counter(words)
-
-
-def find_similar_topics_simple(text: str, topics: list, top_k: int = 5) -> list:
- """使用简单的余弦相似度计算文本相似度"""
- # 将输入文本转换为词频向量
- text_vector = text_to_vector(text)
-
- # 计算每个主题的相似度
- similarities = []
- for topic in topics:
- topic_vector = text_to_vector(topic)
- # 获取所有唯一词
- all_words = set(text_vector.keys()) | set(topic_vector.keys())
- # 构建向量
- v1 = [text_vector.get(word, 0) for word in all_words]
- v2 = [topic_vector.get(word, 0) for word in all_words]
- # 计算相似度
- similarity = cosine_similarity(v1, v2)
- similarities.append((topic, similarity))
-
- # 按相似度降序排序并返回前k个
- return sorted(similarities, key=lambda x: x[1], reverse=True)[:top_k]
-
def truncate_message(message: str, max_length=20) -> str:
"""截断消息，使其不超过指定长度"""
@@ -523,47 +518,6 @@ def get_western_ratio(paragraph):
return western_count / len(alnum_chars)
-def count_messages_between(start_time: float, end_time: float, stream_id: str) -> tuple[int, int]:
- """计算两个时间点之间的消息数量和文本总长度
-
- Args:
- start_time (float): 起始时间戳 (不包含)
- end_time (float): 结束时间戳 (包含)
- stream_id (str): 聊天流ID
-
- Returns:
- tuple[int, int]: (消息数量, 文本总长度)
- """
- count = 0
- total_length = 0
-
- # 参数校验 (可选但推荐)
- if start_time >= end_time:
- # logger.debug(f"开始时间 {start_time} 大于或等于结束时间 {end_time}，返回 0, 0")
- return 0, 0
- if not stream_id:
- logger.error("stream_id 不能为空")
- return 0, 0
-
- # 使用message_repository中的count_messages和find_messages函数
-
- # 构建查询条件
- filter_query = {"chat_id": stream_id, "time": {"$gt": start_time, "$lte": end_time}}
-
- try:
- # 先获取消息数量
- count = count_messages(filter_query)
-
- # 获取消息内容计算总长度
- messages = find_messages(message_filter=filter_query)
- total_length = sum(len(msg.processed_plain_text or "") for msg in messages)
-
- return count, total_length
-
- except Exception as e:
- logger.error(f"计算消息数量时发生意外错误: {e}")
- return 0, 0
-
def translate_timestamp_to_human_readable(timestamp: float, mode: str = "normal") -> str:
# sourcery skip: merge-comparisons, merge-duplicate-blocks, switch
@@ -698,65 +652,6 @@ def assign_message_ids(messages: List[DatabaseMessages]) -> List[Tuple[str, Data
return result
-# def assign_message_ids_flexible(
-# messages: list, prefix: str = "msg", id_length: int = 6, use_timestamp: bool = False
-# ) -> list:
-# """
-# 为消息列表中的每个消息分配唯一的简短随机ID（增强版）
-
-# Args:
-# messages: 消息列表
-# prefix: ID前缀，默认为"msg"
-# id_length: ID的总长度（不包括前缀），默认为6
-# use_timestamp: 是否在ID中包含时间戳，默认为False
-
-# Returns:
-# 包含 {'id': str, 'message': any} 格式的字典列表
-# """
-# result = []
-# used_ids = set()
-
-# for i, message in enumerate(messages):
-# # 生成唯一的ID
-# while True:
-# if use_timestamp:
-# # 使用时间戳的后几位 + 随机字符
-# timestamp_suffix = str(int(time.time() * 1000))[-3:]
-# remaining_length = id_length - 3
-# random_chars = "".join(random.choices(string.ascii_lowercase + string.digits, k=remaining_length))
-# message_id = f"{prefix}{timestamp_suffix}{random_chars}"
-# else:
-# # 使用索引 + 随机字符
-# index_str = str(i + 1)
-# remaining_length = max(1, id_length - len(index_str))
-# random_chars = "".join(random.choices(string.ascii_lowercase + string.digits, k=remaining_length))
-# message_id = f"{prefix}{index_str}{random_chars}"
-
-# if message_id not in used_ids:
-# used_ids.add(message_id)
-# break
-
-# result.append({"id": message_id, "message": message})
-
-# return result
-
-
-# 使用示例:
-# messages = ["Hello", "World", "Test message"]
-#
-# # 基础版本
-# result1 = assign_message_ids(messages)
-# # 结果: [{'id': 'm1123', 'message': 'Hello'}, {'id': 'm2456', 'message': 'World'}, {'id': 'm3789', 'message': 'Test message'}]
-#
-# # 增强版本 - 自定义前缀和长度
-# result2 = assign_message_ids_flexible(messages, prefix="chat", id_length=8)
-# # 结果: [{'id': 'chat1abc2', 'message': 'Hello'}, {'id': 'chat2def3', 'message': 'World'}, {'id': 'chat3ghi4', 'message': 'Test message'}]
-#
-# # 增强版本 - 使用时间戳
-# result3 = assign_message_ids_flexible(messages, prefix="ts", use_timestamp=True)
-# # 结果: [{'id': 'ts123a1b', 'message': 'Hello'}, {'id': 'ts123c2d', 'message': 'World'}, {'id': 'ts123e3f', 'message': 'Test message'}]
-
-
def parse_keywords_string(keywords_input) -> list[str]:
# sourcery skip: use-contextlib-suppress
"""
diff --git a/src/config/official_configs.py b/src/config/official_configs.py
index df616a64..641b287a 100644
--- a/src/config/official_configs.py
+++ b/src/config/official_configs.py
@@ -27,6 +27,9 @@ class BotConfig(ConfigBase):
nickname: str
"""昵称"""
+
+ platforms: list[str] = field(default_factory=lambda: [])
+ """其他平台列表"""
alias_names: list[str] = field(default_factory=lambda: [])
"""别名列表"""
diff --git a/src/memory_system/question_maker.py b/src/memory_system/question_maker.py
index 316afee5..bd9c6929 100644
--- a/src/memory_system/question_maker.py
+++ b/src/memory_system/question_maker.py
@@ -50,15 +50,14 @@ class QuestionMaker:
"""按权重随机选取一个未回答的冲突并自增 raise_time。
选择规则：
- - 若存在 `raise_time == 0` 的项：按权重抽样（0 次权重 1.0，≥1 次权重 0.05）。
- - 若不存在 `raise_time == 0` 的项：仅 5% 概率返回其中任意一条，否则返回 None。
+ - 若存在 `raise_time == 0` 的项：按权重抽样（0 次权重 1.0，≥1 次权重 0.01）。
+ - 若不存在，返回 None。
- 每次成功选中后，将该条目的 `raise_time` 自增 1 并保存。
"""
conflicts = await self.get_un_answered_conflict()
if not conflicts:
return None
- # 如果没有 raise_time==0 的项，则仅有 5% 概率抽样一个
conflicts_with_zero = [c for c in conflicts if (getattr(c, "raise_time", 0) or 0) == 0]
if conflicts_with_zero:
# 权重规则：raise_time == 0 -> 1.0；raise_time >= 1 -> 0.01
@@ -71,12 +70,14 @@ class QuestionMaker:
# 按权重随机选择
chosen_conflict = random.choices(conflicts, weights=weights, k=1)[0]
- # 选中后，自增 raise_time 并保存
+ # 选中后，自增 raise_time 并保存
chosen_conflict.raise_time = (getattr(chosen_conflict, "raise_time", 0) or 0) + 1
chosen_conflict.save()
-
- return chosen_conflict
+ return chosen_conflict
+ else:
+ # 如果没有 raise_time == 0 的冲突，返回 None
+ return None
async def make_question(self) -> Tuple[Optional[str], Optional[str], Optional[str]]:
"""生成一条用于询问用户的冲突问题与上下文。
diff --git a/template/bot_config_template.toml b/template/bot_config_template.toml
index b88077ca..f51c5203 100644
--- a/template/bot_config_template.toml
+++ b/template/bot_config_template.toml
@@ -1,5 +1,5 @@
[inner]
-version = "6.18.3"
+version = "6.18.4"
#----以下是给开发人员阅读的，如果你只是部署了麦麦，不需要阅读----
#如果你想要修改配置文件，请递增version的值
@@ -14,6 +14,9 @@ version = "6.18.3"
[bot]
platform = "qq"
qq_account = "1145141919810" # 麦麦的QQ账号
+
+platforms = ["wx:114514","xx:1919810"] # 麦麦的其他平台账号
+
nickname = "麦麦" # 麦麦的昵称
alias_names = ["麦叠", "牢麦"] # 麦麦的别名