🤖 自动格式化代码 [skip ci]

src/plugins/PFC/action_planner.py

@@ -1,6 +1,7 @@
 import time
 from typing import Tuple, Optional
 from .pfc_utils import retrieve_contextual_info
+
 # import jieba # 如果需要旧版知识库的回退，可能需要
 # import re    # 如果需要旧版知识库的回退，可能需要
 from src.common.logger_manager import get_logger
@@ -336,10 +337,13 @@ class ActionPlanner:
                         last_action_context += f"- 该行动当前状态: {status}\n"
                         # self.last_successful_action_type = None # 非完成状态，清除记录
 
-        retrieved_memory_str_planner, retrieved_knowledge_str_planner = await retrieve_contextual_info(chat_history_text, self.private_name)
+        retrieved_memory_str_planner, retrieved_knowledge_str_planner = await retrieve_contextual_info(
+            chat_history_text, self.private_name
+        )
         # Optional: 可以加一行日志确认结果，方便调试
-        logger.info(f"[私聊][{self.private_name}] (ActionPlanner) 统一检索完成。记忆: {'有' if '回忆起' in retrieved_memory_str_planner else '无'} / 知识: {'有' if '出错' not in retrieved_knowledge_str_planner and '无相关知识' not in retrieved_knowledge_str_planner else '无'}")
-        
+        logger.info(
+            f"[私聊][{self.private_name}] (ActionPlanner) 统一检索完成。记忆: {'有' if '回忆起' in retrieved_memory_str_planner else '无'} / 知识: {'有' if '出错' not in retrieved_knowledge_str_planner and '无相关知识' not in retrieved_knowledge_str_planner else '无'}"
+        )
 
         # --- 选择 Prompt ---
         if last_successful_reply_action in ["direct_reply", "send_new_message"]:

src/plugins/PFC/pfc_utils.py

@@ -2,9 +2,9 @@ import traceback
 import json
 import re
 from typing import Dict, Any, Optional, Tuple, List, Union
-from src.common.logger_manager import get_logger # 确认 logger 的导入路径
+from src.common.logger_manager import get_logger  # 确认 logger 的导入路径
 from src.plugins.memory_system.Hippocampus import HippocampusManager
-from src.plugins.heartFC_chat.heartflow_prompt_builder import prompt_builder # 确认 prompt_builder 的导入路径
+from src.plugins.heartFC_chat.heartflow_prompt_builder import prompt_builder  # 确认 prompt_builder 的导入路径
 
 logger = get_logger("pfc_utils")
 
@@ -47,33 +47,36 @@ async def retrieve_contextual_info(text: str, private_name: str) -> Tuple[str, s
                 retrieved_memory_str = f"你回忆起：\n{related_memory_info.strip()}\n(以上是你的回忆，供参考)\n"
                 memory_log_msg = f"自动检索到记忆: {related_memory_info.strip()[:100]}..."
             else:
-                 memory_log_msg = "自动检索记忆返回为空。"
+                memory_log_msg = "自动检索记忆返回为空。"
         logger.debug(f"[私聊][{private_name}] (retrieve_contextual_info) 记忆检索: {memory_log_msg}")
 
     except Exception as e:
-        logger.error(f"[私聊][{private_name}] (retrieve_contextual_info) 自动检索记忆时出错: {e}\n{traceback.format_exc()}")
+        logger.error(
+            f"[私聊][{private_name}] (retrieve_contextual_info) 自动检索记忆时出错: {e}\n{traceback.format_exc()}"
+        )
         retrieved_memory_str = "检索记忆时出错。\n"
 
     # 2. 检索知识 (逻辑来自原 action_planner 和 reply_generator)
     try:
         # 使用导入的 prompt_builder 实例及其方法
         knowledge_result = await prompt_builder.get_prompt_info(
-            message=text, threshold=0.38 # threshold 可以根据需要调整
+            message=text,
+            threshold=0.38,  # threshold 可以根据需要调整
         )
         if knowledge_result:
-             retrieved_knowledge_str = knowledge_result # 直接使用返回结果
-             knowledge_log_msg = "自动检索到相关知识。"
+            retrieved_knowledge_str = knowledge_result  # 直接使用返回结果
+            knowledge_log_msg = "自动检索到相关知识。"
         logger.debug(f"[私聊][{private_name}] (retrieve_contextual_info) 知识检索: {knowledge_log_msg}")
 
     except Exception as e:
-        logger.error(f"[私聊][{private_name}] (retrieve_contextual_info) 自动检索知识时出错: {e}\n{traceback.format_exc()}")
+        logger.error(
+            f"[私聊][{private_name}] (retrieve_contextual_info) 自动检索知识时出错: {e}\n{traceback.format_exc()}"
+        )
         retrieved_knowledge_str = "检索知识时出错。\n"
 
     return retrieved_memory_str, retrieved_knowledge_str
 
 
-
-
 def get_items_from_json(
     content: str,
     private_name: str,

src/plugins/PFC/reply_generator.py

@@ -1,8 +1,10 @@
 from .pfc_utils import retrieve_contextual_info
+
 # 可能用于旧知识库提取主题 (如果需要回退到旧方法)
 # import jieba # 如果报错说找不到 jieba，可能需要安装: pip install jieba
 # import re    # 正则表达式库，通常 Python 自带
 from typing import Tuple, List, Dict, Any
+
 # from src.common.logger import get_module_logger
 from src.common.logger_manager import get_logger
 from ..models.utils_model import LLMRequest
@@ -114,7 +116,6 @@ class ReplyGenerator:
         self.chat_observer = ChatObserver.get_instance(stream_id, private_name)
         self.reply_checker = ReplyChecker(stream_id, private_name)
 
-    
     # 修改 generate 方法签名，增加 action_type 参数
     async def generate(
         self, observation_info: ObservationInfo, conversation_info: ConversationInfo, action_type: str
@@ -171,11 +172,15 @@ class ReplyGenerator:
 
         # 构建 Persona 文本 (persona_text)
         persona_text = f"你的名字是{self.name}，{self.personality_info}。"
-        retrieval_context = chat_history_text # 使用前面构建好的 chat_history_text
+        retrieval_context = chat_history_text  # 使用前面构建好的 chat_history_text
         # 调用共享函数进行检索
-        retrieved_memory_str, retrieved_knowledge_str = await retrieve_contextual_info(retrieval_context, self.private_name)
-        logger.info(f"[私聊][{self.private_name}] (ReplyGenerator) 统一检索完成。记忆: {'有' if '回忆起' in retrieved_memory_str else '无'} / 知识: {'有' if '出错' not in retrieved_knowledge_str and '无相关知识' not in retrieved_knowledge_str else '无'}")
-        
+        retrieved_memory_str, retrieved_knowledge_str = await retrieve_contextual_info(
+            retrieval_context, self.private_name
+        )
+        logger.info(
+            f"[私聊][{self.private_name}] (ReplyGenerator) 统一检索完成。记忆: {'有' if '回忆起' in retrieved_memory_str else '无'} / 知识: {'有' if '出错' not in retrieved_knowledge_str and '无相关知识' not in retrieved_knowledge_str else '无'}"
+        )
+
         # --- 修改：构建上次回复失败原因和内容提示 ---
         last_rejection_info_str = ""
         # 检查 conversation_info 是否有上次拒绝的原因和内容，并且它们都不是 None