mirror of https://github.com/Mai-with-u/MaiBot.git
究极的可维护性和可读性
114514
parent
7737a95e40
commit
acbabdc24b
|
|
@ -1,12 +1,6 @@
|
|||
import time
|
||||
from typing import Tuple, Optional
|
||||
from src.plugins.memory_system.Hippocampus import HippocampusManager
|
||||
|
||||
# --- NEW IMPORT ---
|
||||
# 从 heartflow 导入知识检索和数据库查询函数/实例
|
||||
from src.plugins.heartFC_chat.heartflow_prompt_builder import prompt_builder
|
||||
|
||||
# --- END NEW IMPORT ---
|
||||
from .pfc_utils import retrieve_contextual_info
|
||||
# import jieba # 如果需要旧版知识库的回退，可能需要
|
||||
# import re # 如果需要旧版知识库的回退，可能需要
|
||||
from src.common.logger_manager import get_logger
|
||||
|
|
@ -128,41 +122,6 @@ class ActionPlanner:
|
|||
self.private_name = private_name
|
||||
self.chat_observer = ChatObserver.get_instance(stream_id, private_name)
|
||||
|
||||
# _get_memory_info 保持不变
|
||||
async def _get_memory_info(self, text: str) -> str:
|
||||
"""根据文本自动检索相关记忆"""
|
||||
memory_prompt = ""
|
||||
related_memory_info = ""
|
||||
try:
|
||||
related_memory = await HippocampusManager.get_instance().get_memory_from_text(
|
||||
text=text,
|
||||
max_memory_num=2, # 最多获取 2 条记忆
|
||||
max_memory_length=2, # 每条记忆长度限制（这个参数含义可能需确认）
|
||||
max_depth=3, # 搜索深度
|
||||
fast_retrieval=False, # 是否快速检索
|
||||
)
|
||||
if related_memory:
|
||||
for memory in related_memory:
|
||||
# memory[0] 是记忆ID, memory[1] 是记忆内容
|
||||
related_memory_info += memory[1] + "\n" # 将记忆内容拼接起来
|
||||
if related_memory_info:
|
||||
memory_prompt = f"你回忆起：\n{related_memory_info.strip()}\n(以上是你的回忆，供参考)\n"
|
||||
logger.debug(
|
||||
f"[私聊]决策层[{self.private_name}]自动检索到记忆: {related_memory_info.strip()[:100]}..."
|
||||
)
|
||||
else:
|
||||
logger.debug(f"[私聊]决策层[{self.private_name}]自动检索记忆返回为空。")
|
||||
else:
|
||||
logger.debug(f"[私聊]决策层[{self.private_name}]未自动检索到相关记忆。")
|
||||
except Exception as e:
|
||||
logger.error(f"[私聊]决策层[{self.private_name}]自动检索记忆时出错: {e}")
|
||||
# memory_prompt = "检索记忆时出错。\n" # 可以选择是否提示错误
|
||||
return memory_prompt
|
||||
|
||||
# --- REMOVED _get_prompt_info_old ---
|
||||
|
||||
# --- REMOVED _get_prompt_info ---
|
||||
|
||||
# 修改 plan 方法签名，增加 last_successful_reply_action 参数
|
||||
async def plan(
|
||||
self,
|
||||
|
|
@ -377,38 +336,10 @@ class ActionPlanner:
|
|||
last_action_context += f"- 该行动当前状态: {status}\n"
|
||||
# self.last_successful_action_type = None # 非完成状态，清除记录
|
||||
|
||||
retrieved_memory_str_planner = ""
|
||||
retrieved_knowledge_str_planner = ""
|
||||
retrieval_context = chat_history_text # 使用聊天记录作为检索上下文
|
||||
if retrieval_context and retrieval_context != "还没有聊天记录。" and retrieval_context != "[构建聊天记录出错]":
|
||||
try:
|
||||
# 调用本地的 _get_memory_info
|
||||
logger.debug(f"[私聊][{self.private_name}] (ActionPlanner) 开始自动检索记忆...")
|
||||
retrieved_memory_str_planner = await self._get_memory_info(text=retrieval_context)
|
||||
logger.info(
|
||||
f"[私聊][{self.private_name}] (ActionPlanner) 自动检索记忆 {'完成' if retrieved_memory_str_planner else '无结果'}。"
|
||||
)
|
||||
|
||||
# --- MODIFIED KNOWLEDGE RETRIEVAL ---
|
||||
# 调用导入的 prompt_builder.get_prompt_info
|
||||
logger.debug(f"[私聊][{self.private_name}] (ActionPlanner) 开始自动检索知识 (使用导入函数)...")
|
||||
# 使用导入的 prompt_builder 实例及其方法
|
||||
retrieved_knowledge_str_planner = await prompt_builder.get_prompt_info(
|
||||
message=retrieval_context, threshold=0.38
|
||||
)
|
||||
# --- END MODIFIED KNOWLEDGE RETRIEVAL ---
|
||||
logger.info(
|
||||
f"[私聊][{self.private_name}] (ActionPlanner) 自动检索知识 {'完成' if retrieved_knowledge_str_planner else '无结果'}。"
|
||||
)
|
||||
|
||||
except Exception as retrieval_err:
|
||||
logger.error(f"[私聊][{self.private_name}] (ActionPlanner) 自动检索时出错: {retrieval_err}")
|
||||
retrieved_memory_str_planner = "检索记忆时出错。\n"
|
||||
retrieved_knowledge_str_planner = "检索知识时出错。\n"
|
||||
else:
|
||||
logger.debug(f"[私聊][{self.private_name}] (ActionPlanner) 无有效聊天记录，跳过自动检索。")
|
||||
retrieved_memory_str_planner = "无聊天记录无法检索记忆。\n"
|
||||
retrieved_knowledge_str_planner = "无聊天记录无法检索知识。\n"
|
||||
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 '无'}")
|
||||
|
||||
|
||||
# --- 选择 Prompt ---
|
||||
if last_successful_reply_action in ["direct_reply", "send_new_message"]:
|
||||
|
|
|
|||
|
|
@ -1,9 +1,77 @@
|
|||
import traceback
|
||||
import json
|
||||
import re
|
||||
from typing import Dict, Any, Optional, Tuple, List, Union
|
||||
from src.common.logger import get_module_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 的导入路径
|
||||
|
||||
logger = get_logger("pfc_utils")
|
||||
|
||||
|
||||
async def retrieve_contextual_info(text: str, private_name: str) -> Tuple[str, str]:
|
||||
"""
|
||||
根据输入文本检索相关的记忆和知识。
|
||||
|
||||
Args:
|
||||
text: 用于检索的上下文文本 (例如聊天记录)。
|
||||
private_name: 私聊对象的名称，用于日志记录。
|
||||
|
||||
Returns:
|
||||
Tuple[str, str]: (检索到的记忆字符串, 检索到的知识字符串)
|
||||
"""
|
||||
retrieved_memory_str = "无相关记忆。"
|
||||
retrieved_knowledge_str = "无相关知识。"
|
||||
memory_log_msg = "未自动检索到相关记忆。"
|
||||
knowledge_log_msg = "未自动检索到相关知识。"
|
||||
|
||||
if not text or text == "还没有聊天记录。" or text == "[构建聊天记录出错]":
|
||||
logger.debug(f"[私聊][{private_name}] (retrieve_contextual_info) 无有效上下文，跳过检索。")
|
||||
return retrieved_memory_str, retrieved_knowledge_str
|
||||
|
||||
# 1. 检索记忆 (逻辑来自原 _get_memory_info)
|
||||
try:
|
||||
related_memory = await HippocampusManager.get_instance().get_memory_from_text(
|
||||
text=text,
|
||||
max_memory_num=2,
|
||||
max_memory_length=2,
|
||||
max_depth=3,
|
||||
fast_retrieval=False,
|
||||
)
|
||||
if related_memory:
|
||||
related_memory_info = ""
|
||||
for memory in related_memory:
|
||||
related_memory_info += memory[1] + "\n"
|
||||
if related_memory_info:
|
||||
# 注意：原版提示信息可以根据需要调整
|
||||
retrieved_memory_str = f"你回忆起：\n{related_memory_info.strip()}\n(以上是你的回忆，供参考)\n"
|
||||
memory_log_msg = f"自动检索到记忆: {related_memory_info.strip()[:100]}..."
|
||||
else:
|
||||
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()}")
|
||||
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 可以根据需要调整
|
||||
)
|
||||
if knowledge_result:
|
||||
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()}")
|
||||
retrieved_knowledge_str = "检索知识时出错。\n"
|
||||
|
||||
return retrieved_memory_str, retrieved_knowledge_str
|
||||
|
||||
|
||||
logger = get_module_logger("pfc_utils")
|
||||
|
||||
|
||||
def get_items_from_json(
|
||||
|
|
|
|||
|
|
@ -1,16 +1,10 @@
|
|||
# 用于访问记忆系统
|
||||
from src.plugins.memory_system.Hippocampus import HippocampusManager
|
||||
|
||||
# --- NEW IMPORT ---
|
||||
# 从 heartflow 导入知识检索和数据库查询函数/实例
|
||||
from src.plugins.heartFC_chat.heartflow_prompt_builder import prompt_builder
|
||||
|
||||
# --- END NEW IMPORT ---
|
||||
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 import get_module_logger
|
||||
from src.common.logger_manager import get_logger
|
||||
from ..models.utils_model import LLMRequest
|
||||
from ...config.config import global_config
|
||||
from .chat_observer import ChatObserver
|
||||
|
|
@ -20,7 +14,7 @@ from .observation_info import ObservationInfo
|
|||
from .conversation_info import ConversationInfo
|
||||
from src.plugins.utils.chat_message_builder import build_readable_messages
|
||||
|
||||
logger = get_module_logger("reply_generator")
|
||||
logger = get_logger("reply_generator")
|
||||
|
||||
# --- 定义 Prompt 模板 ---
|
||||
|
||||
|
|
@ -120,39 +114,7 @@ class ReplyGenerator:
|
|||
self.chat_observer = ChatObserver.get_instance(stream_id, private_name)
|
||||
self.reply_checker = ReplyChecker(stream_id, private_name)
|
||||
|
||||
# _get_memory_info 保持不变，因为它不是与 heartflow 重复的部分
|
||||
async def _get_memory_info(self, text: str) -> str:
|
||||
"""根据文本自动检索相关记忆"""
|
||||
memory_prompt = ""
|
||||
related_memory_info = ""
|
||||
try:
|
||||
related_memory = await HippocampusManager.get_instance().get_memory_from_text(
|
||||
text=text,
|
||||
max_memory_num=2, # 最多获取 2 条记忆
|
||||
max_memory_length=2, # 每条记忆长度限制（这个参数含义可能需确认）
|
||||
max_depth=3, # 搜索深度
|
||||
fast_retrieval=False, # 是否快速检索
|
||||
)
|
||||
if related_memory:
|
||||
for memory in related_memory:
|
||||
# memory[0] 是记忆ID, memory[1] 是记忆内容
|
||||
related_memory_info += memory[1] + "\n" # 将记忆内容拼接起来
|
||||
if related_memory_info:
|
||||
memory_prompt = f"你回忆起：\n{related_memory_info.strip()}\n(以上是你的回忆，不一定是目前聊天里的人说的，回忆中别人说的事情也不一定是准确的，请记住)\n"
|
||||
logger.debug(f"[私聊][{self.private_name}]自动检索到记忆: {related_memory_info.strip()[:100]}...")
|
||||
else:
|
||||
logger.debug(f"[私聊][{self.private_name}]自动检索记忆返回为空。")
|
||||
else:
|
||||
logger.debug(f"[私聊][{self.private_name}]未自动检索到相关记忆。")
|
||||
except Exception as e:
|
||||
logger.error(f"[私聊][{self.private_name}]自动检索记忆时出错: {e}")
|
||||
# memory_prompt = "检索记忆时出错。\n" # 可以选择是否提示错误
|
||||
return memory_prompt
|
||||
|
||||
# --- REMOVED _get_prompt_info_old ---
|
||||
|
||||
# --- REMOVED _get_prompt_info ---
|
||||
|
||||
|
||||
# 修改 generate 方法签名，增加 action_type 参数
|
||||
async def generate(
|
||||
self, observation_info: ObservationInfo, conversation_info: ConversationInfo, action_type: str
|
||||
|
|
@ -209,43 +171,11 @@ class ReplyGenerator:
|
|||
|
||||
# 构建 Persona 文本 (persona_text)
|
||||
persona_text = f"你的名字是{self.name}，{self.personality_info}。"
|
||||
retrieved_memory_str = ""
|
||||
retrieved_knowledge_str = ""
|
||||
# 使用 chat_history_text 作为检索的上下文，因为它包含了最近的对话和新消息
|
||||
retrieval_context = chat_history_text
|
||||
if retrieval_context and retrieval_context != "还没有聊天记录。" and retrieval_context != "[构建聊天记录出错]":
|
||||
try:
|
||||
# 提取记忆 (调用本地的 _get_memory_info)
|
||||
logger.debug(f"[私聊][{self.private_name}]开始自动检索记忆...")
|
||||
retrieved_memory_str = await self._get_memory_info(text=retrieval_context)
|
||||
if retrieved_memory_str:
|
||||
logger.info(f"[私聊][{self.private_name}]自动检索到记忆片段。")
|
||||
else:
|
||||
logger.info(f"[私聊][{self.private_name}]未自动检索到相关记忆。")
|
||||
|
||||
# --- MODIFIED KNOWLEDGE RETRIEVAL ---
|
||||
# 提取知识 (调用导入的 prompt_builder.get_prompt_info)
|
||||
logger.debug(f"[私聊][{self.private_name}]开始自动检索知识 (使用导入函数)...")
|
||||
# 使用导入的 prompt_builder 实例及其方法
|
||||
retrieved_knowledge_str = await prompt_builder.get_prompt_info(
|
||||
message=retrieval_context, threshold=0.38
|
||||
)
|
||||
# --- END MODIFIED KNOWLEDGE RETRIEVAL ---
|
||||
|
||||
if retrieved_knowledge_str:
|
||||
logger.info(f"[私聊][{self.private_name}]自动检索到相关知识。")
|
||||
else:
|
||||
logger.info(f"[私聊][{self.private_name}]未自动检索到相关知识。")
|
||||
|
||||
except Exception as retrieval_err:
|
||||
logger.error(f"[私聊][{self.private_name}]在自动检索记忆/知识时发生错误: {retrieval_err}")
|
||||
retrieved_memory_str = "检索记忆时出错。\n"
|
||||
retrieved_knowledge_str = "检索知识时出错。\n"
|
||||
else:
|
||||
logger.debug(f"[私聊][{self.private_name}]聊天记录为空或无效，跳过自动记忆/知识检索。")
|
||||
retrieved_memory_str = "无聊天记录，无法自动检索记忆。\n"
|
||||
retrieved_knowledge_str = "无聊天记录，无法自动检索知识。\n"
|
||||
|
||||
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 '无'}")
|
||||
|
||||
# --- 修改：构建上次回复失败原因和内容提示 ---
|
||||
last_rejection_info_str = ""
|
||||
# 检查 conversation_info 是否有上次拒绝的原因和内容，并且它们都不是 None
|
||||
|
|
|
|||
Loading…
Reference in New Issue