Merge branch 'dev' of https://github.com/MoFox-Studio/MoFox_Bot into dev

src/chat/express/expression_selector.py

@@ -1,3 +1,4 @@
+import asyncio
 import hashlib
 import random
 import time
@@ -304,14 +305,6 @@ class ExpressionSelector:
         try:
             # start_time = time.time()
             content, (reasoning_content, model_name, _) = await self.llm_model.generate_response_async(prompt=prompt)
-            # logger.info(f"LLM请求时间: {model_name}  {time.time() - start_time} \n{prompt}")
-
-            # logger.info(f"模型名称: {model_name}")
-            # logger.info(f"LLM返回结果: {content}")
-            # if reasoning_content:
-            #     logger.info(f"LLM推理: {reasoning_content}")
-            # else:
-            #     logger.info(f"LLM推理: 无")
 
             if not content:
                 logger.warning("LLM返回空结果")
@@ -338,7 +331,7 @@ class ExpressionSelector:
 
             # 对选中的所有表达方式，一次性更新count数
             if valid_expressions:
-                await self.update_expressions_count_batch(valid_expressions, 0.006)
+                asyncio.create_task(self.update_expressions_count_batch(valid_expressions, 0.006))
 
             # logger.info(f"LLM从{len(all_expressions)}个情境中选择了{len(valid_expressions)}个")
             return valid_expressions

src/chat/memory_system/memory_system.py

@@ -997,7 +997,6 @@ class MemorySystem:
             from src.chat.message_receive.chat_stream import get_chat_manager
 
             chat_manager = get_chat_manager()
-            # get_stream 为异步方法，需要 await
             chat_stream = await chat_manager.get_stream(stream_id)
 
             if not chat_stream or not hasattr(chat_stream, "context_manager"):

src/chat/message_manager/context_manager.py

@@ -13,6 +13,7 @@ from src.common.data_models.database_data_model import DatabaseMessages
 from src.common.data_models.message_manager_data_model import StreamContext
 from src.common.logger import get_logger
 from src.config.config import global_config
+from src.plugin_system.base.component_types import ChatType
 
 from .distribution_manager import stream_loop_manager
 
@@ -54,7 +55,13 @@ class SingleStreamContextManager:
             bool: 是否成功添加
         """
         try:
-            self.context.add_message(message)
+            # 直接操作上下文的消息列表
+            message.is_read = False
+            self.context.unread_messages.append(message)
+
+            # 自动检测和更新chat type
+            self._detect_chat_type(message)
+
             # 在上下文管理器中计算兴趣值
             await self._calculate_message_interest(message)
             self.total_messages += 1
@@ -78,7 +85,28 @@ class SingleStreamContextManager:
             bool: 是否成功更新
         """
         try:
-            self.context.update_message_info(message_id, **updates)
+            # 直接在未读消息中查找并更新
+            for message in self.context.unread_messages:
+                if message.message_id == message_id:
+                    if "interest_value" in updates:
+                        message.interest_value = updates["interest_value"]
+                    if "actions" in updates:
+                        message.actions = updates["actions"]
+                    if "should_reply" in updates:
+                        message.should_reply = updates["should_reply"]
+                    break
+
+            # 在历史消息中查找并更新
+            for message in self.context.history_messages:
+                if message.message_id == message_id:
+                    if "interest_value" in updates:
+                        message.interest_value = updates["interest_value"]
+                    if "actions" in updates:
+                        message.actions = updates["actions"]
+                    if "should_reply" in updates:
+                        message.should_reply = updates["should_reply"]
+                    break
+
             logger.debug(f"更新单流上下文消息: {self.stream_id}/{message_id}")
             return True
         except Exception as e:
@@ -259,36 +287,17 @@ class SingleStreamContextManager:
             logger.error(f"计算消息兴趣度时发生错误: {e}", exc_info=True)
             return 0.5
 
-    async def add_message_async(self, message: DatabaseMessages, skip_energy_update: bool = False) -> bool:
-        """异步实现的 add_message：将消息添加到 context，并 await 能量更新与分发。"""
-        try:
-            self.context.add_message(message)
-
-            # 在上下文管理器中计算兴趣值
-            await self._calculate_message_interest(message)
-
-            self.total_messages += 1
-            self.last_access_time = time.time()
-
-            # 启动流的循环任务（如果还未启动）
-            asyncio.create_task(stream_loop_manager.start_stream_loop(self.stream_id))
-
-            logger.debug(f"添加消息到单流上下文(异步): {self.stream_id}")
-            return True
-        except Exception as e:
-            logger.error(f"添加消息到单流上下文失败 (async) {self.stream_id}: {e}", exc_info=True)
-            return False
-
-    async def update_message_async(self, message_id: str, updates: dict[str, Any]) -> bool:
-        """异步实现的 update_message：更新消息并在需要时 await 能量更新。"""
-        try:
-            self.context.update_message_info(message_id, **updates)
-
-            logger.debug(f"更新单流上下文消息(异步): {self.stream_id}/{message_id}")
-            return True
-        except Exception as e:
-            logger.error(f"更新单流上下文消息失败 (async) {self.stream_id}/{message_id}: {e}", exc_info=True)
-            return False
+    def _detect_chat_type(self, message: DatabaseMessages):
+        """根据消息内容自动检测聊天类型"""
+        # 只有在第一次添加消息时才检测聊天类型，避免后续消息改变类型
+        if len(self.context.unread_messages) == 1:  # 只有这条消息
+            # 如果消息包含群组信息，则为群聊
+            if hasattr(message, "chat_info_group_id") and message.chat_info_group_id:
+                self.context.chat_type = ChatType.GROUP
+            elif hasattr(message, "chat_info_group_name") and message.chat_info_group_name:
+                self.context.chat_type = ChatType.GROUP
+            else:
+                self.context.chat_type = ChatType.PRIVATE
 
     async def clear_context_async(self) -> bool:
         """异步实现的 clear_context：清空消息并 await 能量重算。"""

src/chat/message_manager/distribution_manager.py

@@ -23,8 +23,6 @@ class StreamLoopManager:
     def __init__(self, max_concurrent_streams: int | None = None):
         # 流循环任务管理
         self.stream_loops: dict[str, asyncio.Task] = {}
-        # 跟踪流使用的管理器类型
-        self.stream_management_type: dict[str, str] = {}  # stream_id -> "adaptive" or "fallback"
 
         # 统计信息
         self.stats: dict[str, Any] = {
@@ -115,7 +113,6 @@ class StreamLoopManager:
             return True
 
         # 使用自适应流管理器获取槽位
-        use_adaptive = False
         try:
             from src.chat.message_manager.adaptive_stream_manager import get_adaptive_stream_manager
             adaptive_manager = get_adaptive_stream_manager()
@@ -132,21 +129,14 @@ class StreamLoopManager:
                 )
 
                 if slot_acquired:
-                    use_adaptive = True
                     logger.debug(f"成功获取流处理槽位: {stream_id} (优先级: {priority.name})")
                 else:
                     logger.debug(f"自适应管理器拒绝槽位请求: {stream_id}，尝试回退方案")
             else:
-                logger.debug("自适应管理器未运行，使用原始方法")
+                logger.debug("自适应管理器未运行")
 
         except Exception as e:
-            logger.debug(f"自适应管理器获取槽位失败，使用原始方法: {e}")
-
-        # 如果自适应管理器失败或未运行，使用回退方案
-        if not use_adaptive:
-            if not await self._fallback_acquire_slot(stream_id, force):
-                logger.debug(f"回退方案也失败: {stream_id}")
-                return False
+            logger.debug(f"自适应管理器获取槽位失败: {e}")
 
         # 创建流循环任务
         try:
@@ -155,68 +145,22 @@ class StreamLoopManager:
                 name=f"stream_loop_{stream_id}"
             )
             self.stream_loops[stream_id] = loop_task
-            # 记录管理器类型
-            self.stream_management_type[stream_id] = "adaptive" if use_adaptive else "fallback"
-
             # 更新统计信息
             self.stats["active_streams"] += 1
             self.stats["total_loops"] += 1
 
-            logger.info(f"启动流循环任务: {stream_id} (管理器: {'adaptive' if use_adaptive else 'fallback'})")
+            logger.info(f"启动流循环任务: {stream_id}")
             return True
 
         except Exception as e:
             logger.error(f"启动流循环任务失败 {stream_id}: {e}")
             # 释放槽位
-            if use_adaptive:
-                try:
-                    from src.chat.message_manager.adaptive_stream_manager import get_adaptive_stream_manager
-                    adaptive_manager = get_adaptive_stream_manager()
-                    adaptive_manager.release_stream_slot(stream_id)
-                except:
-                    pass
+            from src.chat.message_manager.adaptive_stream_manager import get_adaptive_stream_manager
+            adaptive_manager = get_adaptive_stream_manager()
+            adaptive_manager.release_stream_slot(stream_id)
+
             return False
 
-    async def _fallback_acquire_slot(self, stream_id: str, force: bool) -> bool:
-        """回退方案：获取槽位（原始方法）"""
-        # 判断是否需要强制分发
-        should_force = force or await self._should_force_dispatch_for_stream(stream_id)
-
-        # 检查是否超过最大并发限制
-        current_streams = len(self.stream_loops)
-        if current_streams >= self.max_concurrent_streams and not should_force:
-            logger.warning(
-                f"超过最大并发流数限制({current_streams}/{self.max_concurrent_streams})，无法启动流 {stream_id}"
-            )
-            return False
-
-        # 处理强制分发情况
-        if should_force and current_streams >= self.max_concurrent_streams:
-            logger.warning(
-                f"流 {stream_id} 未读消息积压严重(>{self.force_dispatch_unread_threshold})，突破并发限制强制启动分发 (当前: {current_streams}/{self.max_concurrent_streams})"
-            )
-            # 检查是否有现有的分发循环，如果有则先移除
-            if stream_id in self.stream_loops:
-                logger.info(f"发现现有流循环 {stream_id}，将先移除再重新创建")
-                existing_task = self.stream_loops[stream_id]
-                if not existing_task.done():
-                    existing_task.cancel()
-                    # 创建异步任务来等待取消完成，并添加异常处理
-                    cancel_task = asyncio.create_task(
-                        self._wait_for_task_cancel(stream_id, existing_task),
-                        name=f"cancel_existing_loop_{stream_id}"
-                    )
-                    # 为取消任务添加异常处理，避免孤儿任务
-                    cancel_task.add_done_callback(
-                        lambda task: logger.debug(f"取消任务完成: {stream_id}") if not task.exception()
-                        else logger.error(f"取消任务异常: {stream_id} - {task.exception()}")
-                    )
-                # 从字典中移除
-                del self.stream_loops[stream_id]
-                current_streams -= 1  # 更新当前流数量
-
-        return True
-
     def _determine_stream_priority(self, stream_id: str) -> "StreamPriority":
         """确定流优先级"""
         try:
@@ -237,20 +181,6 @@ class StreamLoopManager:
             from src.chat.message_manager.adaptive_stream_manager import StreamPriority
             return StreamPriority.NORMAL
 
-        # 创建流循环任务
-        try:
-            task = asyncio.create_task(
-                self._stream_loop(stream_id),
-                name=f"stream_loop_{stream_id}"  # 为任务添加名称，便于调试
-            )
-            self.stream_loops[stream_id] = task
-            self.stats["total_loops"] += 1
-
-            logger.info(f"启动流循环: {stream_id} (当前总数: {len(self.stream_loops)})")
-            return True
-        except Exception as e:
-            logger.error(f"创建流循环任务失败: {stream_id} - {e}")
-            return False
 
     async def stop_stream_loop(self, stream_id: str) -> bool:
         """停止指定流的循环任务
@@ -342,17 +272,6 @@ class StreamLoopManager:
                     # 4. 计算下次检查间隔
                     interval = await self._calculate_interval(stream_id, has_messages)
 
-                    if has_messages:
-                        updated_unread_count = self._get_unread_count(context)
-                        if self._needs_force_dispatch_for_context(context, updated_unread_count):
-                            interval = min(interval, max(self.force_dispatch_min_interval, 0.0))
-                            logger.debug(
-                                "流 %s 未读消息仍有 %d 条，使用加速分发间隔 %.2fs",
-                                stream_id,
-                                updated_unread_count,
-                                interval,
-                            )
-
                     # 5. sleep等待下次检查
                     logger.info(f"流 {stream_id} 等待 {interval:.2f}s")
                     await asyncio.sleep(interval)
@@ -378,23 +297,14 @@ class StreamLoopManager:
                 del self.stream_loops[stream_id]
                 logger.debug(f"清理流循环标记: {stream_id}")
 
-            # 根据管理器类型释放相应的槽位
-            management_type = self.stream_management_type.get(stream_id, "fallback")
-            if management_type == "adaptive":
-                # 释放自适应管理器的槽位
-                try:
-                    from src.chat.message_manager.adaptive_stream_manager import get_adaptive_stream_manager
-                    adaptive_manager = get_adaptive_stream_manager()
-                    adaptive_manager.release_stream_slot(stream_id)
-                    logger.debug(f"释放自适应流处理槽位: {stream_id}")
-                except Exception as e:
-                    logger.debug(f"释放自适应流处理槽位失败: {e}")
-            else:
-                logger.debug(f"流 {stream_id} 使用回退方案，无需释放自适应槽位")
-
-            # 清理管理器类型记录
-            if stream_id in self.stream_management_type:
-                del self.stream_management_type[stream_id]
+            # 释放自适应管理器的槽位
+            try:
+                from src.chat.message_manager.adaptive_stream_manager import get_adaptive_stream_manager
+                adaptive_manager = get_adaptive_stream_manager()
+                adaptive_manager.release_stream_slot(stream_id)
+                logger.debug(f"释放自适应流处理槽位: {stream_id}")
+            except Exception as e:
+                logger.debug(f"释放自适应流处理槽位失败: {e}")
 
             logger.info(f"流循环结束: {stream_id}")
 
@@ -417,7 +327,7 @@ class StreamLoopManager:
             logger.error(f"获取流上下文失败 {stream_id}: {e}")
             return None
 
-    async def _has_messages_to_process(self, context: Any) -> bool:
+    async def _has_messages_to_process(self, context: StreamContext) -> bool:
         """检查是否有消息需要处理
 
         Args:
@@ -464,7 +374,7 @@ class StreamLoopManager:
             success = results.get("success", False)
 
             if success:
-                await self._refresh_focus_energy(stream_id)
+                asyncio.create_task(self._refresh_focus_energy(stream_id))
                 process_time = time.time() - start_time
                 logger.debug(f"流处理成功: {stream_id} (耗时: {process_time:.2f}s)")
             else:
@@ -553,16 +463,16 @@ class StreamLoopManager:
             logger.debug(f"检查流 {stream_id} 是否需要强制分发失败: {e}")
             return False
 
-    def _get_unread_count(self, context: Any) -> int:
+    def _get_unread_count(self, context: StreamContext) -> int:
         try:
-            unread_messages = getattr(context, "unread_messages", None)
+            unread_messages = context.unread_messages
             if unread_messages is None:
                 return 0
             return len(unread_messages)
         except Exception:
             return 0
 
-    def _needs_force_dispatch_for_context(self, context: Any, unread_count: int | None = None) -> bool:
+    def _needs_force_dispatch_for_context(self, context: StreamContext, unread_count: int | None = None) -> bool:
         if not self.force_dispatch_unread_threshold or self.force_dispatch_unread_threshold <= 0:
             return False
 

src/chat/message_receive/bot.py

@@ -1,5 +1,6 @@
 import os
 import re
+import time
 import traceback
 from typing import Any
 
@@ -468,55 +469,110 @@ class ChatBot:
                 template_group_name = None
 
             async def preprocess():
-                # 存储消息到数据库
-                from .storage import MessageStorage
-
-                try:
-                    await MessageStorage.store_message(message, message.chat_stream)
-                    logger.debug(f"消息已存储到数据库: {message.message_info.message_id}")
-                except Exception as e:
-                    logger.error(f"存储消息到数据库失败: {e}")
-                    traceback.print_exc()
-
-                # 使用消息管理器处理消息（保持原有功能）
                 from src.common.data_models.database_data_model import DatabaseMessages
 
+                message_info = message.message_info
+                msg_user_info = getattr(message_info, "user_info", None)
+                stream_user_info = getattr(message.chat_stream, "user_info", None)
+                group_info = getattr(message.chat_stream, "group_info", None)
+
+                message_id = message_info.message_id or ""
+                message_time = message_info.time if message_info.time is not None else time.time()
+                is_mentioned = None
+                if isinstance(message.is_mentioned, bool):
+                    is_mentioned = message.is_mentioned
+                elif isinstance(message.is_mentioned, (int, float)):
+                    is_mentioned = message.is_mentioned != 0
+
+                user_id = ""
+                user_nickname = ""
+                user_cardname = None
+                user_platform = ""
+                if msg_user_info:
+                    user_id = str(getattr(msg_user_info, "user_id", "") or "")
+                    user_nickname = getattr(msg_user_info, "user_nickname", "") or ""
+                    user_cardname = getattr(msg_user_info, "user_cardname", None)
+                    user_platform = getattr(msg_user_info, "platform", "") or ""
+                elif stream_user_info:
+                    user_id = str(getattr(stream_user_info, "user_id", "") or "")
+                    user_nickname = getattr(stream_user_info, "user_nickname", "") or ""
+                    user_cardname = getattr(stream_user_info, "user_cardname", None)
+                    user_platform = getattr(stream_user_info, "platform", "") or ""
+
+                chat_user_id = str(getattr(stream_user_info, "user_id", "") or "")
+                chat_user_nickname = getattr(stream_user_info, "user_nickname", "") or ""
+                chat_user_cardname = getattr(stream_user_info, "user_cardname", None)
+                chat_user_platform = getattr(stream_user_info, "platform", "") or ""
+
+                group_id = getattr(group_info, "group_id", None)
+                group_name = getattr(group_info, "group_name", None)
+                group_platform = getattr(group_info, "platform", None)
+
                 # 创建数据库消息对象
                 db_message = DatabaseMessages(
-                    message_id=message.message_info.message_id,
-                    time=message.message_info.time,
+                    message_id=message_id,
+                    time=float(message_time),
                     chat_id=message.chat_stream.stream_id,
                     processed_plain_text=message.processed_plain_text,
                     display_message=message.processed_plain_text,
-                    is_mentioned=message.is_mentioned,
-                    is_at=message.is_at,
-                    is_emoji=message.is_emoji,
-                    is_picid=message.is_picid,
-                    is_command=message.is_command,
-                    is_notify=message.is_notify,
-                    user_id=message.message_info.user_info.user_id,
-                    user_nickname=message.message_info.user_info.user_nickname,
-                    user_cardname=message.message_info.user_info.user_cardname,
-                    user_platform=message.message_info.user_info.platform,
+                    is_mentioned=is_mentioned,
+                    is_at=bool(message.is_at) if message.is_at is not None else None,
+                    is_emoji=bool(message.is_emoji),
+                    is_picid=bool(message.is_picid),
+                    is_command=bool(message.is_command),
+                    is_notify=bool(message.is_notify),
+                    user_id=user_id,
+                    user_nickname=user_nickname,
+                    user_cardname=user_cardname,
+                    user_platform=user_platform,
                     chat_info_stream_id=message.chat_stream.stream_id,
                     chat_info_platform=message.chat_stream.platform,
-                    chat_info_create_time=message.chat_stream.create_time,
-                    chat_info_last_active_time=message.chat_stream.last_active_time,
-                    chat_info_user_id=message.chat_stream.user_info.user_id,
-                    chat_info_user_nickname=message.chat_stream.user_info.user_nickname,
-                    chat_info_user_cardname=message.chat_stream.user_info.user_cardname,
-                    chat_info_user_platform=message.chat_stream.user_info.platform,
+                    chat_info_create_time=float(message.chat_stream.create_time),
+                    chat_info_last_active_time=float(message.chat_stream.last_active_time),
+                    chat_info_user_id=chat_user_id,
+                    chat_info_user_nickname=chat_user_nickname,
+                    chat_info_user_cardname=chat_user_cardname,
+                    chat_info_user_platform=chat_user_platform,
+                    chat_info_group_id=group_id,
+                    chat_info_group_name=group_name,
+                    chat_info_group_platform=group_platform,
                 )
 
-                # 如果是群聊，添加群组信息
-                if message.chat_stream.group_info:
-                    db_message.chat_info_group_id = message.chat_stream.group_info.group_id
-                    db_message.chat_info_group_name = message.chat_stream.group_info.group_name
-                    db_message.chat_info_group_platform = message.chat_stream.group_info.platform
+                # 兼容历史逻辑：显式设置群聊相关属性，便于后续逻辑通过 hasattr 判断
+                if group_info:
+                    setattr(db_message, "chat_info_group_id", group_id)
+                    setattr(db_message, "chat_info_group_name", group_name)
+                    setattr(db_message, "chat_info_group_platform", group_platform)
+                else:
+                    setattr(db_message, "chat_info_group_id", None)
+                    setattr(db_message, "chat_info_group_name", None)
+                    setattr(db_message, "chat_info_group_platform", None)
 
-                # 添加消息到消息管理器
-                await message_manager.add_message(message.chat_stream.stream_id, db_message)
-                logger.debug(f"消息已添加到消息管理器: {message.chat_stream.stream_id}")
+                # 先交给消息管理器处理，计算兴趣度等衍生数据
+                try:
+                    await message_manager.add_message(message.chat_stream.stream_id, db_message)
+                    logger.debug(f"消息已添加到消息管理器: {message.chat_stream.stream_id}")
+                except Exception as e:
+                    logger.error(f"消息添加到消息管理器失败: {e}")
+
+                # 将兴趣度结果同步回原始消息，便于后续流程使用
+                message.interest_value = getattr(db_message, "interest_value", getattr(message, "interest_value", 0.0))
+                setattr(message, "should_reply", getattr(db_message, "should_reply", getattr(message, "should_reply", False)))
+                setattr(message, "should_act", getattr(db_message, "should_act", getattr(message, "should_act", False)))
+
+                # 存储消息到数据库，只进行一次写入
+                try:
+                    await MessageStorage.store_message(message, message.chat_stream)
+                    logger.debug(
+                        "消息已存储到数据库: %s (interest=%.3f, should_reply=%s, should_act=%s)",
+                        message.message_info.message_id,
+                        getattr(message, "interest_value", -1.0),
+                        getattr(message, "should_reply", None),
+                        getattr(message, "should_act", None),
+                    )
+                except Exception as e:
+                    logger.error(f"存储消息到数据库失败: {e}")
+                    traceback.print_exc()
 
             if template_group_name:
                 async with global_prompt_manager.async_message_scope(template_group_name):

src/chat/planner_actions/action_manager.py

@@ -33,6 +33,10 @@ class ChatterActionManager:
         self._using_actions = component_registry.get_default_actions()
 
         self.log_prefix: str = "ChatterActionManager"
+        # 批量存储支持
+        self._batch_storage_enabled = False
+        self._pending_actions = []
+        self._current_chat_id = None
 
     # === 执行Action方法 ===
 
@@ -184,19 +188,29 @@ class ChatterActionManager:
                 reason = reasoning or "选择不回复"
                 logger.info(f"{log_prefix} 选择不回复，原因: {reason}")
 
-                # 存储no_reply信息到数据库
-                await database_api.store_action_info(
-                    chat_stream=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",
-                )
+                # 存储no_reply信息到数据库（支持批量存储）
+                if self._batch_storage_enabled:
+                    self.add_action_to_batch(
+                        action_name="no_reply",
+                        action_data={"reason": reason},
+                        thinking_id=thinking_id or "",
+                        action_done=True,
+                        action_build_into_prompt=False,
+                        action_prompt_display=reason
+                    )
+                else:
+                    asyncio.create_task(database_api.store_action_info(
+                        chat_stream=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",
+                    ))
 
                 # 自动清空所有未读消息
-                await self._clear_all_unread_messages(chat_stream.stream_id, "no_reply")
+                asyncio.create_task(self._clear_all_unread_messages(chat_stream.stream_id, "no_reply"))
 
                 return {"action_type": "no_reply", "success": True, "reply_text": "", "command": ""}
 
@@ -214,12 +228,12 @@ class ChatterActionManager:
 
                 # 记录执行的动作到目标消息
                 if success:
-                    await self._record_action_to_message(chat_stream, action_name, target_message, action_data)
+                    asyncio.create_task(self._record_action_to_message(chat_stream, action_name, target_message, action_data))
                     # 自动清空所有未读消息
                     if clear_unread_messages:
-                        await self._clear_all_unread_messages(chat_stream.stream_id, action_name)
+                        asyncio.create_task(self._clear_all_unread_messages(chat_stream.stream_id, action_name))
                     # 重置打断计数
-                    await self._reset_interruption_count_after_action(chat_stream.stream_id)
+                    asyncio.create_task(self._reset_interruption_count_after_action(chat_stream.stream_id))
 
                 return {
                     "action_type": action_name,
@@ -260,13 +274,13 @@ class ChatterActionManager:
                 )
 
                 # 记录回复动作到目标消息
-                await self._record_action_to_message(chat_stream, "reply", target_message, action_data)
+                asyncio.create_task(self._record_action_to_message(chat_stream, "reply", target_message, action_data))
 
                 if clear_unread_messages:
-                    await self._clear_all_unread_messages(chat_stream.stream_id, "reply")
+                    asyncio.create_task(self._clear_all_unread_messages(chat_stream.stream_id, "reply"))
 
                 # 回复成功，重置打断计数
-                await self._reset_interruption_count_after_action(chat_stream.stream_id)
+                asyncio.create_task(self._reset_interruption_count_after_action(chat_stream.stream_id))
 
                 return {"action_type": "reply", "success": True, "reply_text": reply_text, "loop_info": loop_info}
 
@@ -474,16 +488,26 @@ class ChatterActionManager:
         person_name = await person_info_manager.get_value(person_id, "person_name")
         action_prompt_display = f"你对{person_name}进行了回复：{reply_text}"
 
-        # 存储动作信息到数据库
-        await database_api.store_action_info(
-            chat_stream=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",
-        )
+        # 存储动作信息到数据库（支持批量存储）
+        if self._batch_storage_enabled:
+            self.add_action_to_batch(
+                action_name="reply",
+                action_data={"reply_text": reply_text},
+                thinking_id=thinking_id or "",
+                action_done=True,
+                action_build_into_prompt=False,
+                action_prompt_display=action_prompt_display
+            )
+        else:
+            await database_api.store_action_info(
+                chat_stream=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] = {
@@ -579,3 +603,71 @@ class ChatterActionManager:
                 )
 
         return reply_text
+
+    def enable_batch_storage(self, chat_id: str):
+        """启用批量存储模式"""
+        self._batch_storage_enabled = True
+        self._current_chat_id = chat_id
+        self._pending_actions.clear()
+        logger.debug(f"已启用批量存储模式，chat_id: {chat_id}")
+
+    def disable_batch_storage(self):
+        """禁用批量存储模式"""
+        self._batch_storage_enabled = False
+        self._current_chat_id = None
+        logger.debug("已禁用批量存储模式")
+
+    def add_action_to_batch(self, action_name: str, action_data: dict, thinking_id: str = "",
+                           action_done: bool = True, action_build_into_prompt: bool = False,
+                           action_prompt_display: str = ""):
+        """添加动作到批量存储列表"""
+        if not self._batch_storage_enabled:
+            return False
+
+        action_record = {
+            "action_name": action_name,
+            "action_data": action_data,
+            "thinking_id": thinking_id,
+            "action_done": action_done,
+            "action_build_into_prompt": action_build_into_prompt,
+            "action_prompt_display": action_prompt_display,
+            "timestamp": time.time()
+        }
+        self._pending_actions.append(action_record)
+        logger.debug(f"已添加动作到批量存储列表: {action_name} (当前待处理: {len(self._pending_actions)} 个)")
+        return True
+
+    async def flush_batch_storage(self, chat_stream):
+        """批量存储所有待处理的动作记录"""
+        if not self._pending_actions:
+            logger.debug("没有待处理的动作需要批量存储")
+            return
+
+        try:
+            logger.info(f"开始批量存储 {len(self._pending_actions)} 个动作记录")
+
+            # 批量存储所有动作
+            stored_count = 0
+            for action_data in self._pending_actions:
+                try:
+                    result = await database_api.store_action_info(
+                        chat_stream=chat_stream,
+                        action_name=action_data.get("action_name", ""),
+                        action_data=action_data.get("action_data", {}),
+                        action_done=action_data.get("action_done", True),
+                        action_build_into_prompt=action_data.get("action_build_into_prompt", False),
+                        action_prompt_display=action_data.get("action_prompt_display", ""),
+                        thinking_id=action_data.get("thinking_id", "")
+                    )
+                    if result:
+                        stored_count += 1
+                except Exception as e:
+                    logger.error(f"存储单个动作记录失败: {e}")
+
+            logger.info(f"批量存储完成: 成功存储 {stored_count}/{len(self._pending_actions)} 个动作记录")
+
+            # 清空待处理列表
+            self._pending_actions.clear()
+
+        except Exception as e:
+            logger.error(f"批量存储动作记录时发生错误: {e}")

src/chat/replyer/default_generator.py

@@ -287,13 +287,13 @@ class DefaultReplyer:
         try:
             # 构建 Prompt
             with Timer("构建Prompt", {}):  # 内部计时器，可选保留
-                prompt = await asyncio.create_task(self.build_prompt_reply_context(
+                prompt = await self.build_prompt_reply_context(
                     reply_to=reply_to,
                     extra_info=extra_info,
                     available_actions=available_actions,
                     enable_tool=enable_tool,
                     reply_message=reply_message,
-                ))
+                )
 
             if not prompt:
                 logger.warning("构建prompt失败，跳过回复生成")

src/chat/utils/prompt.py

@@ -175,7 +175,6 @@ class PromptManager:
         self._prompts = {}
         self._counter = 0
         self._context = PromptContext()
-        self._lock = asyncio.Lock()
 
     @asynccontextmanager
     async def async_message_scope(self, message_id: str | None = None):
@@ -190,10 +189,9 @@ class PromptManager:
             logger.debug(f"从上下文中获取提示词: {name} {context_prompt}")
             return context_prompt
 
-        async with self._lock:
-            if name not in self._prompts:
-                raise KeyError(f"Prompt '{name}' not found")
-            return self._prompts[name]
+        if name not in self._prompts:
+            raise KeyError(f"Prompt '{name}' not found")
+        return self._prompts[name]
 
     def generate_name(self, template: str) -> str:
         """为未命名的prompt生成名称"""