diff --git a/src/config/config.py b/src/config/config.py index 1e18e58e..8c41e055 100644 --- a/src/config/config.py +++ b/src/config/config.py @@ -274,6 +274,9 @@ class BotConfig: talk_allowed_private = set() enable_pfc_chatting: bool = False # 是否启用PFC聊天 enable_pfc_reply_checker: bool = True # 是否开启PFC回复检查 + pfc_message_buffer_size: int = ( + 2 # PFC 聊天消息缓冲数量，有利于使聊天节奏更加紧凑流畅，请根据实际 LLM 响应速度进行调整，默认2条 + ) # idle_chat enable_idle_chat: bool = False # 是否启用 pfc 主动发言 @@ -670,6 +673,9 @@ class BotConfig: "enable_pfc_reply_checker", config.enable_pfc_reply_checker ) logger.info(f"PFC Reply Checker 状态: {'启用' if config.enable_pfc_reply_checker else '关闭'}") + config.pfc_message_buffer_size = experimental_config.get( + "pfc_message_buffer_size", config.pfc_message_buffer_size + ) def idle_chat(parent: dict): idle_chat_config = parent["idle_chat"] diff --git a/src/plugins/PFC/action_planner.py b/src/plugins/PFC/action_planner.py index 212c2bf0..cde14ea7 100644 --- a/src/plugins/PFC/action_planner.py +++ b/src/plugins/PFC/action_planner.py @@ -3,14 +3,11 @@ import traceback from typing import Tuple, Optional, Dict, Any, List from src.common.logger_manager import get_logger - -# from src.individuality.individuality import Individuality -from src.plugins.utils.chat_message_builder import build_readable_messages from ..models.utils_model import LLMRequest from src.config.config import global_config # 确保导入路径正确 -from .pfc_utils import get_items_from_json +from .pfc_utils import get_items_from_json, build_chat_history_text from .chat_observer import ChatObserver from .observation_info import ObservationInfo from .conversation_info import ConversationInfo @@ -210,9 +207,9 @@ class ActionPlanner: time_since_last_bot_message_info = self._get_bot_last_speak_time_info(observation_info) timeout_context = self._get_timeout_context(conversation_info) goals_str = self._build_goals_string(conversation_info) - chat_history_text = await self._build_chat_history_text(observation_info) + chat_history_text = await build_chat_history_text(observation_info, self.private_name) # 获取 sender_name, relationship_text, current_emotion_text - sender_name_str = getattr(observation_info, "sender_name", "对方") # 从 observation_info 获取 + sender_name_str = self.private_name if not sender_name_str: sender_name_str = "对方" # 再次确保有默认值 @@ -467,50 +464,6 @@ class ActionPlanner: goals_str = "- 构建对话目标时出错。\n" return goals_str - async def _build_chat_history_text(self, observation_info: ObservationInfo) -> str: - """构建聊天历史记录文本 (包含未处理消息)""" - - chat_history_text = "" - try: - if hasattr(observation_info, "chat_history_str") and observation_info.chat_history_str: - chat_history_text = observation_info.chat_history_str - elif hasattr(observation_info, "chat_history") and observation_info.chat_history: - history_slice = observation_info.chat_history[-20:] - chat_history_text = await build_readable_messages( - history_slice, replace_bot_name=True, merge_messages=False, timestamp_mode="relative", read_mark=0.0 - ) - else: - chat_history_text = "还没有聊天记录。\n" - unread_count = getattr(observation_info, "new_messages_count", 0) - unread_messages = getattr(observation_info, "unprocessed_messages", []) - if unread_count > 0 and unread_messages: - bot_qq_str = str(global_config.BOT_QQ) - other_unread_messages = [ - msg for msg in unread_messages if msg.get("user_info", {}).get("user_id") != bot_qq_str - ] - other_unread_count = len(other_unread_messages) - if other_unread_count > 0: - new_messages_str = await build_readable_messages( - other_unread_messages, - replace_bot_name=True, - merge_messages=False, - timestamp_mode="relative", - read_mark=0.0, - ) - chat_history_text += ( - f"\n--- 以下是 {other_unread_count} 条你需要处理的新消息 ---\n{new_messages_str}\n------\n" - ) - logger.debug(f"[私聊][{self.private_name}] 向 LLM 追加了 {other_unread_count} 条未读消息。") - else: - chat_history_text += "\n--- 以上均为已读消息，未读消息均已处理完毕 ---\n" - except AttributeError as e: - logger.warning(f"[私聊][{self.private_name}] 构建聊天记录文本时属性错误: {e}") - chat_history_text = "[获取聊天记录时出错]\n" - except Exception as e: - logger.error(f"[私聊][{self.private_name}] 处理聊天记录时发生未知错误: {e}") - chat_history_text = "[处理聊天记录时出错]\n" - return chat_history_text - def _build_action_history_context(self, conversation_info: ConversationInfo) -> Tuple[str, str]: """构建行动历史概要和上一次行动详细情况""" diff --git a/src/plugins/PFC/conversation.py b/src/plugins/PFC/conversation.py index da613317..6132ae13 100644 --- a/src/plugins/PFC/conversation.py +++ b/src/plugins/PFC/conversation.py @@ -25,12 +25,10 @@ from .observation_info import ObservationInfo from .conversation_info import ConversationInfo from .reply_generator import ReplyGenerator from .PFC_idle.idle_chat import IdleChat -from .pfc_KnowledgeFetcher import KnowledgeFetcher from .waiter import Waiter from .reply_checker import ReplyChecker -# >>> 新增导入 <<< -from .conversation_loop import run_conversation_loop # 导入新的循环函数 +from .conversation_loop import run_conversation_loop from rich.traceback import install @@ -38,13 +36,6 @@ install(extra_lines=3) logger = get_logger("pfc_conversation") -# 时区配置移到 loop 文件或更全局的位置，这里不再需要 -# configured_tz = getattr(global_config, 'TIME_ZONE', 'Asia/Shanghai') -# TIME_ZONE = tz.gettz(configured_tz) -# if TIME_ZONE is None: -# logger.error(f"配置的时区 '{configured_tz}' 无效，将使用默认时区 'Asia/Shanghai'") -# TIME_ZONE = tz.gettz('Asia/Shanghai') - class Conversation: """ @@ -59,7 +50,7 @@ class Conversation: self.stream_id: str = stream_id self.private_name: str = private_name self.state: ConversationState = ConversationState.INIT - self.should_continue: bool = False # Manager 会在初始化后设置 + self.should_continue: bool = False self.ignore_until_timestamp: Optional[float] = None self.generated_reply: str = "" self.chat_stream: Optional[ChatStream] = None @@ -74,7 +65,6 @@ class Conversation: self.action_planner: Optional[ActionPlanner] = None self.goal_analyzer: Optional[GoalAnalyzer] = None self.reply_generator: Optional[ReplyGenerator] = None - self.knowledge_fetcher: Optional[KnowledgeFetcher] = None self.waiter: Optional[Waiter] = None self.direct_sender: Optional[DirectMessageSender] = None self.idle_chat: Optional[IdleChat] = None @@ -83,13 +73,14 @@ class Conversation: self.conversation_info: Optional[ConversationInfo] = None self.reply_checker: Optional[ReplyChecker] = None - self._initialized: bool = False # Manager 会在初始化成功后设为 True + self._initialized: bool = False self.bot_qq_str: Optional[str] = str(global_config.BOT_QQ) if global_config.BOT_QQ else None if not self.bot_qq_str: logger.error(f"[私聊][{self.private_name}] 严重错误：未能从配置中获取 BOT_QQ ID！") - # _initialize 和 _load_initial_history 方法已被移除 + # 确保这个属性被正确初始化 + self.consecutive_llm_action_failures: int = 0 # LLM相关动作连续失败的计数器 async def start(self): """ @@ -105,27 +96,27 @@ class Conversation: logger.info(f"[私聊][{self.private_name}] 对话系统启动，准备创建规划循环任务...") try: - # >>> 修改后的调用 <<< + # 创建PFC主循环任务 _loop_task = asyncio.create_task(run_conversation_loop(self)) logger.info(f"[私聊][{self.private_name}] 规划循环任务已创建。") except Exception as task_err: logger.error(f"[私聊][{self.private_name}] 创建规划循环任务时出错: {task_err}") - await self.stop() + await self.stop() # 发生错误时尝试停止 async def stop(self): """ 停止对话实例并清理相关资源。 """ logger.info(f"[私聊][{self.private_name}] 正在停止对话实例: {self.stream_id}") - self.should_continue = False + self.should_continue = False # 设置标志以退出循环 # 最终关系评估 if ( - self._initialized + self._initialized # 确保已初始化 and self.relationship_updater and self.conversation_info and self.observation_info - and self.chat_observer + and self.chat_observer # 确保所有需要的组件都存在 ): try: logger.info(f"[私聊][{self.private_name}] 准备执行最终关系评估...") @@ -143,11 +134,10 @@ class Conversation: # 停止其他组件 if self.idle_chat: - # 减少活跃实例计数，而不是停止IdleChat - await self.idle_chat.decrement_active_instances() + await self.idle_chat.decrement_active_instances() # 减少活跃实例计数 logger.debug(f"[私聊][{self.private_name}] 已减少IdleChat活跃实例计数") - if self.observation_info and self.chat_observer: - self.observation_info.unbind_from_chat_observer() + if self.observation_info and self.chat_observer: # 确保二者都存在 + self.observation_info.unbind_from_chat_observer() # 解绑 if self.mood_mng and hasattr(self.mood_mng, "stop_mood_update") and self.mood_mng._running: # type: ignore self.mood_mng.stop_mood_update() # type: ignore logger.debug(f"[私聊][{self.private_name}] MoodManager 后台更新已停止。") @@ -155,12 +145,11 @@ class Conversation: self._initialized = False # 标记为未初始化 logger.info(f"[私聊][{self.private_name}] 对话实例 {self.stream_id} 已停止。") - # _plan_and_action_loop 方法已被移除 - def _convert_to_message(self, msg_dict: Dict[str, Any]) -> Optional[Message]: """将从数据库或其他来源获取的消息字典转换为内部使用的 Message 对象""" # 这个方法似乎没有被其他内部方法调用，但为了完整性暂时保留 try: + # 尝试获取与此对话实例关联的 ChatStream chat_stream_to_use = self.chat_stream or chat_manager.get_stream(self.stream_id) if not chat_stream_to_use: logger.error( @@ -168,6 +157,7 @@ class Conversation: ) return None + # 解析 UserInfo user_info_dict = msg_dict.get("user_info", {}) user_info: Optional[UserInfo] = None if isinstance(user_info_dict, dict): @@ -177,21 +167,22 @@ class Conversation: logger.warning( f"[私聊][{self.private_name}] 从字典创建 UserInfo 时出错: {e}, dict: {user_info_dict}" ) - if not user_info: + if not user_info: # 如果没有有效的 UserInfo，则无法创建 Message 对象 logger.warning( f"[私聊][{self.private_name}] 消息缺少有效的 UserInfo，无法转换。 msg_id: {msg_dict.get('message_id')}" ) return None + # 创建并返回 Message 对象 return Message( - message_id=msg_dict.get("message_id", f"gen_{time.time()}"), + message_id=msg_dict.get("message_id", f"gen_{time.time()}"), # 提供默认 message_id chat_stream=chat_stream_to_use, - time=msg_dict.get("time", time.time()), + time=msg_dict.get("time", time.time()), # 提供默认时间 user_info=user_info, - processed_plain_text=msg_dict.get("processed_plain_text", ""), - detailed_plain_text=msg_dict.get("detailed_plain_text", ""), + processed_plain_text=msg_dict.get("processed_plain_text", ""), # 提供默认文本 + detailed_plain_text=msg_dict.get("detailed_plain_text", ""), # 提供默认详细文本 ) except Exception as e: logger.error(f"[私聊][{self.private_name}] 转换消息时出错: {e}") logger.error(f"[私聊][{self.private_name}] {traceback.format_exc()}") - return None + return None # 出错时返回 None diff --git a/src/plugins/PFC/conversation_info.py b/src/plugins/PFC/conversation_info.py index 0e7a5138..8bbf5ef5 100644 --- a/src/plugins/PFC/conversation_info.py +++ b/src/plugins/PFC/conversation_info.py @@ -17,4 +17,5 @@ class ConversationInfo: self.relationship_text: Optional[str] = "你们还不熟悉。" # 与当前对话者的关系描述文本 self.current_emotion_text: Optional[str] = "心情平静。" # 机器人当前的情绪描述文本 self.current_instance_message_count: int = 0 # 当前私聊实例中的消息计数 + self.other_new_messages_during_planning_count: int = 0 # 在计划阶段期间收到的其他新消息计数 # --- 新增字段结束 --- diff --git a/src/plugins/PFC/conversation_initializer.py b/src/plugins/PFC/conversation_initializer.py index 932f74f5..118bc171 100644 --- a/src/plugins/PFC/conversation_initializer.py +++ b/src/plugins/PFC/conversation_initializer.py @@ -18,7 +18,6 @@ from .observation_info import ObservationInfo from .conversation_info import ConversationInfo from .reply_generator import ReplyGenerator from .PFC_idle.idle_chat import IdleChat -from .pfc_KnowledgeFetcher import KnowledgeFetcher # 修正大小写 from .waiter import Waiter from .pfc_utils import get_person_id from .reply_checker import ReplyChecker @@ -166,9 +165,6 @@ async def initialize_core_components(conversation_instance: "Conversation"): conversation_instance.stream_id, conversation_instance.private_name ) - logger.debug(f"[私聊][{conversation_instance.private_name}] (Initializer) 初始化 KnowledgeFetcher...") - conversation_instance.knowledge_fetcher = KnowledgeFetcher(conversation_instance.private_name) - logger.debug(f"[私聊][{conversation_instance.private_name}] (Initializer) 初始化 Waiter...") conversation_instance.waiter = Waiter(conversation_instance.stream_id, conversation_instance.private_name) diff --git a/src/plugins/PFC/conversation_loop.py b/src/plugins/PFC/conversation_loop.py index d0cd287d..aab031cc 100644 --- a/src/plugins/PFC/conversation_loop.py +++ b/src/plugins/PFC/conversation_loop.py @@ -15,18 +15,19 @@ if TYPE_CHECKING: logger = get_logger("pfc_loop") -# 时区配置 (从 conversation.py 移过来，或者考虑放到更全局的配置模块) +# 时区配置 configured_tz = getattr(global_config, "TIME_ZONE", "Asia/Shanghai") TIME_ZONE = tz.gettz(configured_tz) if TIME_ZONE is None: logger.error(f"配置的时区 '{configured_tz}' 无效，将使用默认时区 'Asia/Shanghai'") TIME_ZONE = tz.gettz("Asia/Shanghai") +MAX_CONSECUTIVE_LLM_ACTION_FAILURES = 3 # 可配置的最大LLM连续失败次数 + async def run_conversation_loop(conversation_instance: "Conversation"): """ 核心的规划与行动循环 (PFC Loop)。 - 之前是 Conversation 类中的 _plan_and_action_loop 方法。 """ logger.debug(f"[私聊][{conversation_instance.private_name}] 进入 run_conversation_loop 循环。") @@ -34,16 +35,20 @@ async def run_conversation_loop(conversation_instance: "Conversation"): logger.error(f"[私聊][{conversation_instance.private_name}] 尝试在未初始化状态下运行规划循环，退出。") return - force_reflect_and_act = False # 用于强制使用反思 prompt 的标志 + _force_reflect_and_act_next_iter = False while conversation_instance.should_continue: loop_iter_start_time = time.time() - logger.debug(f"[私聊][{conversation_instance.private_name}] 开始新一轮循环迭代 ({loop_iter_start_time:.2f})") + current_force_reflect_and_act = _force_reflect_and_act_next_iter + _force_reflect_and_act_next_iter = False + + logger.debug( + f"[私聊][{conversation_instance.private_name}] 开始新一轮循环迭代 ({loop_iter_start_time:.2f}), force_reflect_next_iter: {current_force_reflect_and_act}, consecutive_llm_failures: {conversation_instance.consecutive_llm_action_failures}" + ) - # 更新当前时间 try: - global TIME_ZONE # 引用全局 TIME_ZONE - if TIME_ZONE is None: # 如果还未加载成功 + global TIME_ZONE + if TIME_ZONE is None: configured_tz_loop = getattr(global_config, "TIME_ZONE", "Asia/Shanghai") TIME_ZONE = tz.gettz(configured_tz_loop) if TIME_ZONE is None: @@ -54,7 +59,6 @@ async def run_conversation_loop(conversation_instance: "Conversation"): if conversation_instance.observation_info: time_str = current_time_dt.strftime("%Y-%m-%d %H:%M:%S %Z%z") conversation_instance.observation_info.current_time_str = time_str - logger.debug(f"[私聊][{conversation_instance.private_name}] 更新 ObservationInfo 当前时间: {time_str}") else: logger.warning( f"[私聊][{conversation_instance.private_name}] ObservationInfo 未初始化，无法更新当前时间。" @@ -64,15 +68,11 @@ async def run_conversation_loop(conversation_instance: "Conversation"): f"[私聊][{conversation_instance.private_name}] 更新 ObservationInfo 当前时间时出错: {time_update_err}" ) - # 处理忽略状态 if ( conversation_instance.ignore_until_timestamp and loop_iter_start_time < conversation_instance.ignore_until_timestamp ): if conversation_instance.idle_chat and conversation_instance.idle_chat._running: - # 不直接停止服务，改为暂时忽略此用户 - # 虽然我们仍然可以通过active_instances_count来决定是否触发主动聊天 - # 但为了安全起见，我们只记录一个日志 logger.debug(f"[私聊][{conversation_instance.private_name}] 对话被暂时忽略，暂停对该用户的主动聊天") sleep_duration = min(30, conversation_instance.ignore_until_timestamp - loop_iter_start_time) await asyncio.sleep(sleep_duration) @@ -85,18 +85,13 @@ async def run_conversation_loop(conversation_instance: "Conversation"): f"[私聊][{conversation_instance.private_name}] 忽略时间已到 {conversation_instance.stream_id}，准备结束对话。" ) conversation_instance.ignore_until_timestamp = None - await conversation_instance.stop() # 调用 Conversation 实例的 stop 方法 + await conversation_instance.stop() continue else: - # 忽略状态结束，这里不需要任何特殊处理 - # IdleChat会通过active_instances_count自动决定是否触发 pass - # 核心规划与行动逻辑 try: - # 更新关系和情绪文本 (在每次循环开始时进行) if conversation_instance.conversation_info and conversation_instance._initialized: - # 更新关系 if ( conversation_instance.conversation_info.person_id and conversation_instance.relationship_translator @@ -121,13 +116,11 @@ async def run_conversation_loop(conversation_instance: "Conversation"): except Exception as e_rel: logger.error(f"[私聊][{conversation_instance.private_name}] (Loop) 更新关系文本时出错: {e_rel}") conversation_instance.conversation_info.relationship_text = "你们的关系是：普通。" - # 更新情绪 if conversation_instance.mood_mng: conversation_instance.conversation_info.current_emotion_text = ( conversation_instance.mood_mng.get_prompt() - ) # type: ignore + ) - # 检查核心组件 if not all( [ conversation_instance.action_planner, @@ -141,7 +134,6 @@ async def run_conversation_loop(conversation_instance: "Conversation"): await asyncio.sleep(5) continue - # 规划 planning_start_time = time.time() logger.debug( f"[私聊][{conversation_instance.private_name}] --- (Loop) 开始规划 ({planning_start_time:.2f}) ---" @@ -155,79 +147,55 @@ async def run_conversation_loop(conversation_instance: "Conversation"): conversation_instance.conversation_info.last_successful_reply_action if conversation_instance.conversation_info else None, - use_reflect_prompt=force_reflect_and_act, + use_reflect_prompt=current_force_reflect_and_act, ) - force_reflect_and_act = False + logger.debug( f"[私聊][{conversation_instance.private_name}] (Loop) ActionPlanner.plan 完成，初步规划动作: {action}" ) - # 检查中断 - current_unprocessed_messages = getattr(conversation_instance.observation_info, "unprocessed_messages", []) - new_messages_during_planning: List[Dict[str, Any]] = [] - other_new_messages_during_planning: List[Dict[str, Any]] = [] + current_unprocessed_messages_after_plan = getattr( + conversation_instance.observation_info, "unprocessed_messages", [] + ) + new_messages_during_action_planning: List[Dict[str, Any]] = [] + other_new_messages_during_action_planning: List[Dict[str, Any]] = [] - for msg in current_unprocessed_messages: - msg_time = msg.get("time") - sender_id_info = msg.get("user_info", {}) - sender_id = str(sender_id_info.get("user_id")) if sender_id_info else None - if msg_time and msg_time >= planning_start_time: - new_messages_during_planning.append(msg) - if sender_id != conversation_instance.bot_qq_str: - other_new_messages_during_planning.append(msg) + for msg_ap in current_unprocessed_messages_after_plan: + msg_time_ap = msg_ap.get("time") + sender_id_info_ap = msg_ap.get("user_info", {}) + sender_id_ap = str(sender_id_info_ap.get("user_id")) if sender_id_info_ap else None + if msg_time_ap and msg_time_ap >= planning_start_time: + new_messages_during_action_planning.append(msg_ap) + if sender_id_ap != conversation_instance.bot_qq_str: + other_new_messages_during_action_planning.append(msg_ap) - new_msg_count = len(new_messages_during_planning) - other_new_msg_count = len(other_new_messages_during_planning) + new_msg_count_action_planning = len(new_messages_during_action_planning) + other_new_msg_count_action_planning = len(other_new_messages_during_action_planning) - if conversation_instance.conversation_info and other_new_msg_count > 0: - conversation_instance.conversation_info.current_instance_message_count += other_new_msg_count - # 触发关系和情绪更新（如果需要） - if ( - conversation_instance.relationship_updater - and conversation_instance.observation_info - and conversation_instance.chat_observer - ): - await conversation_instance.relationship_updater.update_relationship_incremental( - conversation_info=conversation_instance.conversation_info, - observation_info=conversation_instance.observation_info, - chat_observer_for_history=conversation_instance.chat_observer, - ) - if ( - conversation_instance.emotion_updater - and other_new_messages_during_planning - and conversation_instance.observation_info - and conversation_instance.chat_observer - ): - last_user_msg = other_new_messages_during_planning[-1] - last_user_msg_text = last_user_msg.get("processed_plain_text", "用户发了新消息") - sender_name_for_event = getattr(conversation_instance.observation_info, "sender_name", "对方") - event_desc = f"用户【{sender_name_for_event}】发送了新消息: '{last_user_msg_text[:30]}...'" - await conversation_instance.emotion_updater.update_emotion_based_on_context( - conversation_info=conversation_instance.conversation_info, - observation_info=conversation_instance.observation_info, - chat_observer_for_history=conversation_instance.chat_observer, - event_description=event_desc, - ) + if conversation_instance.conversation_info and other_new_msg_count_action_planning > 0: + pass - should_interrupt: bool = False - interrupt_reason: str = "" - if action in ["wait", "listening"] and new_msg_count > 0: - should_interrupt = True - interrupt_reason = f"规划 {action} 期间收到 {new_msg_count} 条新消息" - elif other_new_msg_count > 2: # Threshold for other actions - should_interrupt = True - interrupt_reason = f"规划 {action} 期间收到 {other_new_msg_count} 条来自他人的新消息" - - if should_interrupt: - logger.info( - f"[私聊][{conversation_instance.private_name}] (Loop) 中断 '{action}'，原因: {interrupt_reason}。重新规划..." + should_interrupt_action_planning: bool = False + interrupt_reason_action_planning: str = "" + if action in ["wait", "listening"] and new_msg_count_action_planning > 0: + should_interrupt_action_planning = True + interrupt_reason_action_planning = f"规划 {action} 期间收到 {new_msg_count_action_planning} 条新消息" + elif other_new_msg_count_action_planning > 2: + should_interrupt_action_planning = True + interrupt_reason_action_planning = ( + f"规划 {action} 期间收到 {other_new_msg_count_action_planning} 条来自他人的新消息" ) - cancel_record = { + + if should_interrupt_action_planning: + logger.info( + f"[私聊][{conversation_instance.private_name}] (Loop) 中断 '{action}' (在ActionPlanner.plan后)，原因: {interrupt_reason_action_planning}。重新规划..." + ) + cancel_record_ap = { "action": action, "plan_reason": reason, - "status": "cancelled_due_to_new_messages", + "status": "cancelled_due_to_new_messages_during_action_plan", "time": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"), - "final_reason": interrupt_reason, + "final_reason": interrupt_reason_action_planning, } if conversation_instance.conversation_info: if ( @@ -235,40 +203,208 @@ async def run_conversation_loop(conversation_instance: "Conversation"): or conversation_instance.conversation_info.done_action is None ): conversation_instance.conversation_info.done_action = [] - conversation_instance.conversation_info.done_action.append(cancel_record) + conversation_instance.conversation_info.done_action.append(cancel_record_ap) conversation_instance.conversation_info.last_successful_reply_action = None conversation_instance.state = ConversationState.ANALYZING await asyncio.sleep(0.1) continue - # 执行动作 (调用 actions 模块的函数) - logger.debug( - f"[私聊][{conversation_instance.private_name}] (Loop) 未中断，调用 actions.handle_action 执行动作 '{action}'..." - ) - if conversation_instance.conversation_info: - conversation_instance.conversation_info.other_new_messages_during_planning_count = other_new_msg_count + # --- LLM Action Handling with Shield and Failure Count --- + if action in ["direct_reply", "send_new_message"]: + logger.debug( + f"[私聊][{conversation_instance.private_name}] (Loop) 动作 '{action}' 需要LLM生成，进入监控执行模式..." + ) + llm_call_start_time = time.time() - await actions.handle_action( - conversation_instance, - action, - reason, - conversation_instance.observation_info, - conversation_instance.conversation_info, - ) - logger.debug(f"[私聊][{conversation_instance.private_name}] (Loop) actions.handle_action 完成。") + if conversation_instance.conversation_info: + conversation_instance.conversation_info.other_new_messages_during_planning_count = ( + other_new_msg_count_action_planning + ) + + llm_action_task = asyncio.create_task( + actions.handle_action( + conversation_instance, + action, + reason, + conversation_instance.observation_info, + conversation_instance.conversation_info, + ) + ) + + interrupted_by_new_messages = False + llm_action_completed_successfully = False + action_outcome_processed = False # Flag to ensure we process outcome only once + + while not llm_action_task.done() and not action_outcome_processed: + try: + # Shield the task so wait_for timeout doesn't cancel it directly + await asyncio.wait_for(asyncio.shield(llm_action_task), timeout=1.5) + # If wait_for completes without timeout, the shielded task is done (or errored/cancelled by itself) + action_outcome_processed = True # Outcome will be processed outside this loop + except asyncio.TimeoutError: + # Shielded task didn't finish in 1.5s. This is our chance to check messages. + current_time_for_check = time.time() + logger.debug( + f"[私聊][{conversation_instance.private_name}] (Loop) LLM Monitor polling. llm_call_start_time: {llm_call_start_time:.2f}, current_check_time: {current_time_for_check:.2f}. Task still running, checking for new messages." + ) + + current_unprocessed_messages_during_llm = getattr( + conversation_instance.observation_info, "unprocessed_messages", [] + ) + other_new_messages_this_check: List[Dict[str, Any]] = [] + + logger.debug( + f"[私聊][{conversation_instance.private_name}] (Loop) Checking unprocessed_messages (count: {len(current_unprocessed_messages_during_llm)}):" + ) + for msg_llm in current_unprocessed_messages_during_llm: + msg_time_llm = msg_llm.get("time") + sender_id_info_llm = msg_llm.get("user_info", {}) + sender_id_llm = str(sender_id_info_llm.get("user_id")) if sender_id_info_llm else None + is_new_enough = msg_time_llm and msg_time_llm >= llm_call_start_time + is_other_sender = sender_id_llm != conversation_instance.bot_qq_str + + time_str_for_log = f"{msg_time_llm:.2f}" if msg_time_llm is not None else "N/A" + logger.debug( + f" - Msg ID: {msg_llm.get('message_id')}, Time: {time_str_for_log}, Sender: {sender_id_llm}. New enough? {is_new_enough}. Other sender? {is_other_sender}." + ) + + if is_new_enough and is_other_sender: + other_new_messages_this_check.append(msg_llm) + + logger.debug( + f"[私聊][{conversation_instance.private_name}] (Loop) Found {len(other_new_messages_this_check)} 'other_new_messages_this_check'." + ) + + if len(other_new_messages_this_check) > global_config.pfc_message_buffer_size: + logger.info( + f"[私聊][{conversation_instance.private_name}] (Loop) LLM动作 '{action}' 执行期间收到 {len(other_new_messages_this_check)} 条来自他人的新消息，将取消LLM任务。" + ) + if not llm_action_task.done(): # Check again before cancelling + llm_action_task.cancel() # Now we explicitly cancel the original task + interrupted_by_new_messages = True + action_outcome_processed = True # We've made a decision, exit monitoring + # else: continue polling if not enough new messages + # Shield ensures CancelledError from llm_action_task itself is caught by the outer try/except + + # After the monitoring loop (either task finished, or we decided to cancel it) + # Await the task properly to get its result or handle its exception/cancellation + action_final_status_in_history = "unknown" + try: + await llm_action_task # This will re-raise CancelledError if we cancelled it, or other exceptions + + # If no exception, it means the task completed. + # actions.handle_action updates done_action, so we check its status. + if conversation_instance.conversation_info and conversation_instance.conversation_info.done_action: + # Check if done_action is not empty + if conversation_instance.conversation_info.done_action: + action_final_status_in_history = conversation_instance.conversation_info.done_action[ + -1 + ].get("status", "unknown") + + if action_final_status_in_history in ["done", "done_no_reply"]: + logger.debug( + f"[私聊][{conversation_instance.private_name}] (Loop) LLM动作 '{action}' 任务最终成功完成 (status: {action_final_status_in_history})。" + ) + conversation_instance.consecutive_llm_action_failures = 0 + llm_action_completed_successfully = True + else: + logger.warning( + f"[私聊][{conversation_instance.private_name}] (Loop) LLM动作 '{action}' 任务完成但未成功 (status: {action_final_status_in_history})。" + ) + if not interrupted_by_new_messages: + conversation_instance.consecutive_llm_action_failures += 1 + + except asyncio.CancelledError: + logger.info( + f"[私聊][{conversation_instance.private_name}] (Loop) LLM动作 '{action}' 任务最终确认被取消。" + ) + if not interrupted_by_new_messages: + conversation_instance.consecutive_llm_action_failures += 1 + logger.warning( + f"[私聊][{conversation_instance.private_name}] (Loop) LLM任务因外部原因取消，连续失败次数: {conversation_instance.consecutive_llm_action_failures}" + ) + else: # interrupted_by_new_messages is True + logger.info( + f"[私聊][{conversation_instance.private_name}] (Loop) LLM任务因新消息被内部逻辑取消，不计为LLM失败。" + ) + + except Exception as e_llm_final: + logger.error( + f"[私聊][{conversation_instance.private_name}] (Loop) LLM动作 '{action}' 任务执行时发生最终错误: {e_llm_final}" + ) + logger.error(traceback.format_exc()) + conversation_instance.state = ConversationState.ERROR + if not interrupted_by_new_messages: + conversation_instance.consecutive_llm_action_failures += 1 + + # --- Post LLM Action Task Handling --- + if not llm_action_completed_successfully: + if conversation_instance.consecutive_llm_action_failures >= MAX_CONSECUTIVE_LLM_ACTION_FAILURES: + logger.error( + f"[私聊][{conversation_instance.private_name}] (Loop) LLM相关动作连续失败或被取消 {conversation_instance.consecutive_llm_action_failures} 次。将强制等待并重置计数器。" + ) + + action = "wait" # Force action to wait + reason = f"LLM连续失败{conversation_instance.consecutive_llm_action_failures}次，强制等待" + conversation_instance.consecutive_llm_action_failures = 0 + + if conversation_instance.conversation_info: + conversation_instance.conversation_info.last_successful_reply_action = None + + logger.info(f"[私聊][{conversation_instance.private_name}] (Loop) 执行强制等待动作...") + await actions.handle_action( + conversation_instance, + action, + reason, + conversation_instance.observation_info, + conversation_instance.conversation_info, + ) + _force_reflect_and_act_next_iter = False + conversation_instance.state = ConversationState.ANALYZING + await asyncio.sleep(1) + continue + else: + conversation_instance.state = ConversationState.ANALYZING + logger.info( + f"[私聊][{conversation_instance.private_name}] (Loop) LLM动作中断/失败，准备重新规划。Interrupted by new msgs: {interrupted_by_new_messages}, Consecutive LLM Failures: {conversation_instance.consecutive_llm_action_failures}" + ) + await asyncio.sleep(0.1) + continue + else: + logger.debug(f"[私聊][{conversation_instance.private_name}] (Loop) 执行非LLM类动作 '{action}'...") + conversation_instance.consecutive_llm_action_failures = 0 + logger.debug( + f"[私聊][{conversation_instance.private_name}] (Loop) 重置 consecutive_llm_action_failures due to non-LLM action." + ) + + if conversation_instance.conversation_info: + conversation_instance.conversation_info.other_new_messages_during_planning_count = ( + other_new_msg_count_action_planning + ) + + await actions.handle_action( + conversation_instance, + action, + reason, + conversation_instance.observation_info, + conversation_instance.conversation_info, + ) + logger.debug(f"[私聊][{conversation_instance.private_name}] (Loop) 非LLM类动作 '{action}' 完成。") - # 检查是否需要反思 last_action_record = {} if conversation_instance.conversation_info and conversation_instance.conversation_info.done_action: - last_action_record = conversation_instance.conversation_info.done_action[-1] + if conversation_instance.conversation_info.done_action: + last_action_record = conversation_instance.conversation_info.done_action[-1] + if ( last_action_record.get("action") == "send_new_message" and last_action_record.get("status") == "done_no_reply" ): - logger.info(f"[私聊][{conversation_instance.private_name}] (Loop) 检测到需反思，设置标志。") - force_reflect_and_act = True + logger.info( + f"[私聊][{conversation_instance.private_name}] (Loop) 检测到 ReplyGenerator 决定不发送消息，下一轮将强制反思。" + ) + _force_reflect_and_act_next_iter = True - # 检查结束条件 goal_ended: bool = False if ( conversation_instance.conversation_info @@ -286,7 +422,9 @@ async def run_conversation_loop(conversation_instance: "Conversation"): last_action_record_for_end_check = {} if conversation_instance.conversation_info and conversation_instance.conversation_info.done_action: - last_action_record_for_end_check = conversation_instance.conversation_info.done_action[-1] + if conversation_instance.conversation_info.done_action: + last_action_record_for_end_check = conversation_instance.conversation_info.done_action[-1] + action_ended: bool = ( last_action_record_for_end_check.get("action") in ["end_conversation", "say_goodbye"] and last_action_record_for_end_check.get("status") == "done" @@ -294,12 +432,12 @@ async def run_conversation_loop(conversation_instance: "Conversation"): if goal_ended or action_ended: logger.info(f"[私聊][{conversation_instance.private_name}] (Loop) 检测到结束条件，停止循环。") - await conversation_instance.stop() # 调用 Conversation 的 stop - continue # 虽然会 break，但 continue 更明确 + await conversation_instance.stop() + continue except asyncio.CancelledError: logger.info(f"[私聊][{conversation_instance.private_name}] (Loop) PFC 主循环任务被取消。") - await conversation_instance.stop() # 调用 Conversation 的 stop + await conversation_instance.stop() break except Exception as loop_err: logger.error(f"[私聊][{conversation_instance.private_name}] (Loop) PFC 主循环出错: {loop_err}") @@ -307,7 +445,6 @@ async def run_conversation_loop(conversation_instance: "Conversation"): conversation_instance.state = ConversationState.ERROR await asyncio.sleep(5) - # 控制循环频率 loop_duration = time.time() - loop_iter_start_time min_loop_interval = 0.1 logger.debug(f"[私聊][{conversation_instance.private_name}] (Loop) 循环迭代耗时: {loop_duration:.3f} 秒。") diff --git a/src/plugins/PFC/observation_info.py b/src/plugins/PFC/observation_info.py index 7b3d6279..a372b3dc 100644 --- a/src/plugins/PFC/observation_info.py +++ b/src/plugins/PFC/observation_info.py @@ -114,8 +114,6 @@ class ObservationInfo: """初始化 ObservationInfo""" self.private_name: str = private_name - # 新增：发信人信息 - self.sender_name: Optional[str] = None self.sender_user_id: Optional[str] = None # 存储为字符串 self.sender_platform: Optional[str] = None @@ -232,23 +230,20 @@ class ObservationInfo: if user_info: try: self.sender_user_id = str(user_info.user_id) # 确保是字符串 - self.sender_name = user_info.user_nickname # 或者 user_info.card 如果私聊时card更准 self.sender_platform = user_info.platform current_message_sender_id = self.sender_user_id # 用于后续逻辑 logger.debug( - f"[私聊][{self.private_name}] 更新发信人信息: ID={self.sender_user_id}, Name={self.sender_name}, Platform={self.sender_platform}" + f"[私聊][{self.private_name}] 更新发信人信息: ID={self.sender_user_id}, Name={self.private_name}, Platform={self.sender_platform}" ) except AttributeError as e: logger.error(f"[私聊][{self.private_name}] 从 UserInfo 对象提取信息时出错: {e}, UserInfo: {user_info}") # 如果提取失败，将这些新字段设为 None，避免使用旧数据 self.sender_user_id = None - self.sender_name = None self.sender_platform = None else: logger.warning(f"[私聊][{self.private_name}] 处理消息更新时缺少有效的 UserInfo, message_id: {message_id}") # 如果没有 UserInfo，也将这些新字段设为 None self.sender_user_id = None - self.sender_name = None self.sender_platform = None # --- 新增/修改结束 --- diff --git a/src/plugins/PFC/pfc.py b/src/plugins/PFC/pfc.py index 77a8f4b7..741f20ea 100644 --- a/src/plugins/PFC/pfc.py +++ b/src/plugins/PFC/pfc.py @@ -3,7 +3,7 @@ from src.common.logger import get_module_logger from ..models.utils_model import LLMRequest from ...config.config import global_config from .chat_observer import ChatObserver -from .pfc_utils import get_items_from_json +from .pfc_utils import get_items_from_json, build_chat_history_text from src.individuality.individuality import Individuality from .conversation_info import ConversationInfo from .observation_info import ObservationInfo @@ -86,21 +86,7 @@ class GoalAnalyzer: goals_str = f"目标：{goal}，产生该对话目标的原因：{reasoning}\n" # 获取聊天历史记录 - chat_history_text = observation_info.chat_history_str - - if observation_info.new_messages_count > 0: - new_messages_list = observation_info.unprocessed_messages - new_messages_str = await build_readable_messages( - new_messages_list, - replace_bot_name=True, - merge_messages=False, - timestamp_mode="relative", - read_mark=0.0, - ) - chat_history_text += f"\n--- 以下是 {observation_info.new_messages_count} 条新消息 ---\n{new_messages_str}" - else: - chat_history_text += "\n--- 以上均为已读消息，未读消息均已处理完毕 ---\n" - # await observation_info.clear_unprocessed_messages() + chat_history_text = await build_chat_history_text(observation_info, self.private_name) persona_text = f"你的名字是{self.name}，{self.personality_info}。" # 构建action历史文本 diff --git a/src/plugins/PFC/pfc_KnowledgeFetcher.py b/src/plugins/PFC/pfc_KnowledgeFetcher.py deleted file mode 100644 index 0989339d..00000000 --- a/src/plugins/PFC/pfc_KnowledgeFetcher.py +++ /dev/null @@ -1,85 +0,0 @@ -from typing import List, Tuple -from src.common.logger import get_module_logger -from src.plugins.memory_system.Hippocampus import HippocampusManager -from ..models.utils_model import LLMRequest -from ...config.config import global_config -from ..chat.message import Message -from ..knowledge.knowledge_lib import qa_manager -from ..utils.chat_message_builder import build_readable_messages - -logger = get_module_logger("knowledge_fetcher") - - -class KnowledgeFetcher: - """知识调取器""" - - def __init__(self, private_name: str): - self.llm = LLMRequest( - model=global_config.llm_normal, - temperature=global_config.llm_normal["temp"], - max_tokens=1000, - request_type="knowledge_fetch", - ) - self.private_name = private_name - - def _lpmm_get_knowledge(self, query: str) -> str: - """获取相关知识 - - Args: - query: 查询内容 - - Returns: - str: 构造好的,带相关度的知识 - """ - - logger.debug(f"[私聊][{self.private_name}]正在从LPMM知识库中获取知识") - try: - knowledge_info = qa_manager.get_knowledge(query) - logger.debug(f"[私聊][{self.private_name}]LPMM知识库查询结果: {knowledge_info:150}") - return knowledge_info - except Exception as e: - logger.error(f"[私聊][{self.private_name}]LPMM知识库搜索工具执行失败: {str(e)}") - return "未找到匹配的知识" - - async def fetch(self, query: str, chat_history: List[Message]) -> Tuple[str, str]: - """获取相关知识 - - Args: - query: 查询内容 - chat_history: 聊天历史 - - Returns: - Tuple[str, str]: (获取的知识, 知识来源) - """ - # 构建查询上下文 - chat_history_text = await build_readable_messages( - chat_history, - replace_bot_name=True, - merge_messages=False, - timestamp_mode="relative", - read_mark=0.0, - ) - - # 从记忆中获取相关知识 - related_memory = await HippocampusManager.get_instance().get_memory_from_text( - text=f"{query}\n{chat_history_text}", - max_memory_num=3, - max_memory_length=2, - max_depth=3, - fast_retrieval=False, - ) - knowledge_text = "" - sources_text = "无记忆匹配" # 默认值 - if related_memory: - sources = [] - for memory in related_memory: - knowledge_text += memory[1] + "\n" - sources.append(f"记忆片段{memory[0]}") - knowledge_text = knowledge_text.strip() - sources_text = "，".join(sources) - - knowledge_text += "\n现在有以下**知识**可供参考：\n " - knowledge_text += self._lpmm_get_knowledge(query) - knowledge_text += "\n请记住这些**知识**，并根据**知识**回答问题。\n" - - return knowledge_text or "未找到相关知识", sources_text or "无记忆匹配" diff --git a/src/plugins/PFC/pfc_emotion.py b/src/plugins/PFC/pfc_emotion.py index d2496520..c202d04a 100644 --- a/src/plugins/PFC/pfc_emotion.py +++ b/src/plugins/PFC/pfc_emotion.py @@ -71,9 +71,7 @@ class PfcEmotionUpdater: ) current_mood_text_from_manager = self.mood_mng.current_mood.text # 从 MoodManager 获取当前情绪文本 - sender_name_for_prompt = getattr(observation_info, "sender_name", "对方") - if not sender_name_for_prompt: - sender_name_for_prompt = "对方" + sender_name_for_prompt = self.private_name relationship_text_for_prompt = getattr( conversation_info, "relationship_text", "关系一般。" ) # 从 ConversationInfo 获取关系文本 diff --git a/src/plugins/PFC/pfc_utils.py b/src/plugins/PFC/pfc_utils.py index 38ead120..fc5437ab 100644 --- a/src/plugins/PFC/pfc_utils.py +++ b/src/plugins/PFC/pfc_utils.py @@ -8,6 +8,9 @@ from src.plugins.heartFC_chat.heartflow_prompt_builder import prompt_builder # from src.plugins.chat.chat_stream import ChatStream from ..person_info.person_info import person_info_manager import math +from src.plugins.utils.chat_message_builder import build_readable_messages +from .observation_info import ObservationInfo +from src.config.config import global_config logger = get_logger("pfc_utils") @@ -339,3 +342,43 @@ async def adjust_relationship_value_nonlinear(old_value: float, raw_adjustment: value = 0 return value + + +async def build_chat_history_text(observation_info: ObservationInfo, private_name: str) -> str: + """构建聊天历史记录文本 (包含未处理消息)""" + + chat_history_text = "" + try: + if hasattr(observation_info, "chat_history_str") and observation_info.chat_history_str: + chat_history_text = observation_info.chat_history_str + elif hasattr(observation_info, "chat_history") and observation_info.chat_history: + history_slice = observation_info.chat_history[-20:] + chat_history_text = await build_readable_messages( + history_slice, replace_bot_name=True, merge_messages=False, timestamp_mode="relative", read_mark=0.0 + ) + else: + chat_history_text = "还没有聊天记录。\n" + unread_count = getattr(observation_info, "new_messages_count", 0) + unread_messages = getattr(observation_info, "unprocessed_messages", []) + if unread_count > 0 and unread_messages: + bot_qq_str = str(global_config.BOT_QQ) + other_unread_messages = [ + msg for msg in unread_messages if msg.get("user_info", {}).get("user_id") != bot_qq_str + ] + other_unread_count = len(other_unread_messages) + if other_unread_count > 0: + new_messages_str = await build_readable_messages( + other_unread_messages, + replace_bot_name=True, + merge_messages=False, + timestamp_mode="relative", + read_mark=0.0, + ) + chat_history_text += f"\n{new_messages_str}\n------\n" + except AttributeError as e: + logger.warning(f"[私聊][{private_name}] 构建聊天记录文本时属性错误: {e}") + chat_history_text = "[获取聊天记录时出错]\n" + except Exception as e: + logger.error(f"[私聊][{private_name}] 处理聊天记录时发生未知错误: {e}") + chat_history_text = "[处理聊天记录时出错]\n" + return chat_history_text diff --git a/src/plugins/PFC/reply_generator.py b/src/plugins/PFC/reply_generator.py index 01182259..174e3ba0 100644 --- a/src/plugins/PFC/reply_generator.py +++ b/src/plugins/PFC/reply_generator.py @@ -1,4 +1,5 @@ import random + from .pfc_utils import retrieve_contextual_info from src.common.logger_manager import get_logger @@ -9,7 +10,7 @@ from .reply_checker import ReplyChecker from src.individuality.individuality import Individuality from .observation_info import ObservationInfo from .conversation_info import ConversationInfo -from src.plugins.utils.chat_message_builder import build_readable_messages +from .pfc_utils import build_chat_history_text logger = get_logger("reply_generator") @@ -214,25 +215,9 @@ class ReplyGenerator: else: goals_str = "- 目前没有明确对话目标\n" - chat_history_text = observation_info.chat_history_str - if observation_info.new_messages_count > 0 and observation_info.unprocessed_messages: - new_messages_list = observation_info.unprocessed_messages - new_messages_str = await build_readable_messages( - new_messages_list, - replace_bot_name=True, - merge_messages=False, - timestamp_mode="relative", - read_mark=0.0, - ) - chat_history_text += f"\n--- 以下是 {observation_info.new_messages_count} 条新消息 ---\n{new_messages_str}" - elif not chat_history_text: - chat_history_text = "还没有聊天记录。" - else: - chat_history_text += "\n--- 以上均为已读消息，未读消息均已处理完毕 ---\n" + chat_history_text = await build_chat_history_text(observation_info, self.private_name) - sender_name_str = getattr(observation_info, "sender_name", "对方") - if not sender_name_str: - sender_name_str = "对方" + sender_name_str = self.private_name relationship_text_str = getattr(conversation_info, "relationship_text", "你们还不熟悉。") current_emotion_text_str = getattr(conversation_info, "current_emotion_text", "心情平静。") diff --git a/template/bot_config_template.toml b/template/bot_config_template.toml index 41015771..0515edab 100644 --- a/template/bot_config_template.toml +++ b/template/bot_config_template.toml @@ -189,6 +189,7 @@ enable_friend_chat = false # 是否启用好友聊天 talk_allowed_private = [] # 可以回复消息的QQ号 pfc_chatting = false # 是否启用PFC聊天，该功能仅作用于私聊，与回复模式独立 enable_pfc_reply_checker = true # 是否启用 PFC 的回复检查器 +pfc_message_buffer_size = 2 # PFC 聊天消息缓冲数量，有利于使聊天节奏更加紧凑流畅，请根据实际 LLM 响应速度进行调整，默认2条 [idle_chat] enable_idle_chat = false # 是否启用 pfc 主动发言