gardel/Mofox-Core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #879 from Dax233/PFC-test

Browse Source 
PFC 目标相关功能修复 & PFC 的 logger 输出优化 & PFC 启用新(?)的message_sender
This commit is contained in:
墨梓柒
2025-04-28 22:06:12 +08:00
committed by GitHub
parent e93192bbbb 8ff87397b6
commit 200c64646e
13 changed files with 614 additions and 418 deletions
Download Patch File Download Diff File Expand all files Collapse all files

89
src/plugins/PFC/action_planner.py
View File

@@ -36,7 +36,7 @@ PROMPT_INITIAL_REPLY = """{persona_text}。现在你在参与一场QQ私聊，
fetch_knowledge: 需要调取知识，当需要专业知识或特定信息时选择，对方若提到你不太认识的人名或实体也可以尝试选择
listening: 倾听对方发言，当你认为对方话才说到一半，发言明显未结束时选择
direct_reply: 直接回复对方
rethink_goal: 重新思考对话目标，当发现对话目标不再适用或对话卡住时选择，注意私聊的环境是灵活的，有可能需要经常选择
rethink_goal: 思考一个对话目标，当你觉得目前对话需要目标，或当前目标不再适用，或话题卡住时选择。注意私聊的环境是灵活的，有可能需要经常选择
end_conversation: 结束对话，对方长时间没回复或者当你觉得对话告一段落时可以选择
block_and_ignore: 更加极端的结束对话方式，直接结束对话并在一段时间内无视对方所有发言（屏蔽），当对话让你感到十分不适，或你遭到各类骚扰时选择
@@ -69,7 +69,7 @@ fetch_knowledge: 需要调取知识，当需要专业知识或特定信息时选
wait: 暂时不说话，留给对方交互空间，等待对方回复（尤其是在你刚发言后、或上次发言因重复、发言过多被拒时、或不确定做什么时，这是不错的选择）
listening: 倾听对方发言（虽然你刚发过言，但如果对方立刻回复且明显话没说完，可以选择这个）
send_new_message: 发送一条新消息继续对话，允许适当的追问、补充、深入话题，或开启相关新话题。**但是避免在因重复被拒后立即使用，也不要在对方没有回复的情况下过多的“消息轰炸”或重复发言**
rethink_goal: 重新思考对话目标，当发现对话目标不再适用或对话卡住时选择，注意私聊的环境是灵活的，有可能需要经常选择
rethink_goal: 思考一个对话目标，当你觉得目前对话需要目标，或当前目标不再适用，或话题卡住时选择。注意私聊的环境是灵活的，有可能需要经常选择
end_conversation: 结束对话，对方长时间没回复或者当你觉得对话告一段落时可以选择
block_and_ignore: 更加极端的结束对话方式，直接结束对话并在一段时间内无视对方所有发言（屏蔽），当对话让你感到十分不适，或你遭到各类骚扰时选择
@@ -86,7 +86,7 @@ block_and_ignore: 更加极端的结束对话方式，直接结束对话并在
class ActionPlanner:
"""行动规划器"""
def __init__(self, stream_id: str):
def __init__(self, stream_id: str, private_name: str):
self.llm = LLMRequest(
model=global_config.llm_PFC_action_planner,
temperature=global_config.llm_PFC_action_planner["temp"],
@@ -96,7 +96,8 @@ class ActionPlanner:
self.personality_info = Individuality.get_instance().get_prompt(type="personality", x_person=2, level=3)
self.identity_detail_info = Individuality.get_instance().get_prompt(type="identity", x_person=2, level=2)
self.name = global_config.BOT_NICKNAME
self.chat_observer = ChatObserver.get_instance(stream_id)
self.private_name = private_name
self.chat_observer = ChatObserver.get_instance(stream_id, private_name)
# self.action_planner_info = ActionPlannerInfo() # 移除未使用的变量
# 修改 plan 方法签名，增加 last_successful_reply_action 参数
@@ -137,20 +138,24 @@ class ActionPlanner:
)
break
else:
logger.debug("Observation info chat history is empty or not available for bot time check.")
logger.debug(
f"[私聊][{self.private_name}]Observation info chat history is empty or not available for bot time check."
)
except AttributeError:
logger.warning("ObservationInfo object might not have chat_history attribute yet for bot time check.")
logger.warning(
f"[私聊][{self.private_name}]ObservationInfo object might not have chat_history attribute yet for bot time check."
)
except Exception as e:
logger.warning(f"获取 Bot 上次发言时间时出错: {e}")
logger.warning(f"[私聊][{self.private_name}]获取 Bot 上次发言时间时出错: {e}")
# --- 获取超时提示信息 ---
# (这部分逻辑不变)
timeout_context = ""
try:
if hasattr(conversation_info, "goal_list") and conversation_info.goal_list:
last_goal_tuple = conversation_info.goal_list[-1]
if isinstance(last_goal_tuple, tuple) and len(last_goal_tuple) > 0:
last_goal_text = last_goal_tuple[0]
last_goal_dict = conversation_info.goal_list[-1]
if isinstance(last_goal_dict, dict) and "goal" in last_goal_dict:
last_goal_text = last_goal_dict["goal"]
if isinstance(last_goal_text, str) and "分钟，思考接下来要做什么" in last_goal_text:
try:
timeout_minutes_text = last_goal_text.split("，")[0].replace("你等待了", "")
@@ -158,39 +163,48 @@ class ActionPlanner:
except Exception:
timeout_context = "重要提示：对方已经长时间没有回复你的消息了（这可能代表对方繁忙/不想回复/没注意到你的消息等情况，或在对方看来本次聊天已告一段落），请基于此情况规划下一步。\n"
else:
logger.debug("Conversation info goal_list is empty or not available for timeout check.")
logger.debug(
f"[私聊][{self.private_name}]Conversation info goal_list is empty or not available for timeout check."
)
except AttributeError:
logger.warning("ConversationInfo object might not have goal_list attribute yet for timeout check.")
logger.warning(
f"[私聊][{self.private_name}]ConversationInfo object might not have goal_list attribute yet for timeout check."
)
except Exception as e:
logger.warning(f"检查超时目标时出错: {e}")
logger.warning(f"[私聊][{self.private_name}]检查超时目标时出错: {e}")
# --- 构建通用 Prompt 参数 ---
logger.debug(f"开始规划行动：当前目标: {getattr(conversation_info, 'goal_list', '不可用')}")
logger.debug(
f"[私聊][{self.private_name}]开始规划行动：当前目标: {getattr(conversation_info, 'goal_list', '不可用')}"
)
# 构建对话目标 (goals_str)
goals_str = ""
try:
if hasattr(conversation_info, "goal_list") and conversation_info.goal_list:
for goal_reason in conversation_info.goal_list:
if isinstance(goal_reason, tuple) and len(goal_reason) > 0:
goal = goal_reason[0]
reasoning = goal_reason[1] if len(goal_reason) > 1 else "没有明确原因"
elif isinstance(goal_reason, dict):
if isinstance(goal_reason, dict):
goal = goal_reason.get("goal", "目标内容缺失")
reasoning = goal_reason.get("reasoning", "没有明确原因")
else:
goal = str(goal_reason)
reasoning = "没有明确原因"
goal = str(goal) if goal is not None else "目标内容缺失"
reasoning = str(reasoning) if reasoning is not None else "没有明确原因"
goals_str += f"- 目标：{goal}\n 原因：{reasoning}\n"
if not goals_str:
if not goals_str:
goals_str = "- 目前没有明确对话目标，请考虑设定一个。\n"
else:
goals_str = "- 目前没有明确对话目标，请考虑设定一个。\n"
except AttributeError:
logger.warning("ConversationInfo object might not have goal_list attribute yet.")
logger.warning(
f"[私聊][{self.private_name}]ConversationInfo object might not have goal_list attribute yet."
)
goals_str = "- 获取对话目标时出错。\n"
except Exception as e:
logger.error(f"构建对话目标字符串时出错: {e}")
logger.error(f"[私聊][{self.private_name}]构建对话目标字符串时出错: {e}")
goals_str = "- 构建对话目标时出错。\n"
# 获取聊天历史记录 (chat_history_text)
@@ -218,13 +232,15 @@ class ActionPlanner:
)
else:
logger.warning(
"ObservationInfo has new_messages_count > 0 but unprocessed_messages is empty or missing."
f"[私聊][{self.private_name}]ObservationInfo has new_messages_count > 0 but unprocessed_messages is empty or missing."
)
except AttributeError:
logger.warning("ObservationInfo object might be missing expected attributes for chat history.")
logger.warning(
f"[私聊][{self.private_name}]ObservationInfo object might be missing expected attributes for chat history."
)
chat_history_text = "获取聊天记录时出错。\n"
except Exception as e:
logger.error(f"处理聊天记录时发生未知错误: {e}")
logger.error(f"[私聊][{self.private_name}]处理聊天记录时发生未知错误: {e}")
chat_history_text = "处理聊天记录时出错。\n"
# 构建 Persona 文本 (persona_text)
@@ -253,11 +269,13 @@ class ActionPlanner:
if hasattr(conversation_info, "done_action") and conversation_info.done_action:
action_history_list = conversation_info.done_action[-5:]
else:
logger.debug("Conversation info done_action is empty or not available.")
logger.debug(f"[私聊][{self.private_name}]Conversation info done_action is empty or not available.")
except AttributeError:
logger.warning("ConversationInfo object might not have done_action attribute yet.")
logger.warning(
f"[私聊][{self.private_name}]ConversationInfo object might not have done_action attribute yet."
)
except Exception as e:
logger.error(f"访问行动历史时出错: {e}")
logger.error(f"[私聊][{self.private_name}]访问行动历史时出错: {e}")
if not action_history_list:
action_history_summary += "- 还没有执行过行动。\n"
@@ -314,10 +332,10 @@ class ActionPlanner:
# --- 选择 Prompt ---
if last_successful_reply_action in ["direct_reply", "send_new_message"]:
prompt_template = PROMPT_FOLLOW_UP
logger.debug("使用 PROMPT_FOLLOW_UP (追问决策)")
logger.debug(f"[私聊][{self.private_name}]使用 PROMPT_FOLLOW_UP (追问决策)")
else:
prompt_template = PROMPT_INITIAL_REPLY
logger.debug("使用 PROMPT_INITIAL_REPLY (首次/非连续回复决策)")
logger.debug(f"[私聊][{self.private_name}]使用 PROMPT_INITIAL_REPLY (首次/非连续回复决策)")
# --- 格式化最终的 Prompt ---
prompt = prompt_template.format(
@@ -330,13 +348,14 @@ class ActionPlanner:
chat_history_text=chat_history_text if chat_history_text.strip() else "还没有聊天记录。",
)
logger.debug(f"发送到LLM的最终提示词:\n------\n{prompt}\n------")
logger.debug(f"[私聊][{self.private_name}]发送到LLM的最终提示词:\n------\n{prompt}\n------")
try:
content, _ = await self.llm.generate_response_async(prompt)
logger.debug(f"LLM原始返回内容: {content}")
logger.debug(f"[私聊][{self.private_name}]LLM原始返回内容: {content}")
success, result = get_items_from_json(
content,
self.private_name,
"action",
"reason",
default_values={"action": "wait", "reason": "LLM返回格式错误或未提供原因，默认等待"},
@@ -358,14 +377,14 @@ class ActionPlanner:
"block_and_ignore",
]
if action not in valid_actions:
logger.warning(f"LLM返回了未知的行动类型: '{action}'，强制改为 wait")
logger.warning(f"[私聊][{self.private_name}]LLM返回了未知的行动类型: '{action}'，强制改为 wait")
reason = f"(原始行动'{action}'无效，已强制改为wait) {reason}"
action = "wait"
logger.info(f"规划的行动: {action}")
logger.info(f"行动原因: {reason}")
logger.info(f"[私聊][{self.private_name}]规划的行动: {action}")
logger.info(f"[私聊][{self.private_name}]行动原因: {reason}")
return action, reason
except Exception as e:
logger.error(f"规划行动时调用 LLM 或处理结果出错: {str(e)}")
logger.error(f"[私聊][{self.private_name}]规划行动时调用 LLM 或处理结果出错: {str(e)}")
return "wait", f"行动规划处理中发生错误，暂时等待: {str(e)}"

37
src/plugins/PFC/chat_observer.py
View File

@@ -18,7 +18,7 @@ class ChatObserver:
_instances: Dict[str, "ChatObserver"] = {}
@classmethod
def get_instance(cls, stream_id: str) -> "ChatObserver":
def get_instance(cls, stream_id: str, private_name: str) -> "ChatObserver":
"""获取或创建观察器实例
Args:
@@ -28,10 +28,10 @@ class ChatObserver:
ChatObserver: 观察器实例
"""
if stream_id not in cls._instances:
cls._instances[stream_id] = cls(stream_id)
cls._instances[stream_id] = cls(stream_id, private_name)
return cls._instances[stream_id]
def __init__(self, stream_id: str):
def __init__(self, stream_id: str, private_name: str):
"""初始化观察器
Args:
@@ -41,6 +41,7 @@ class ChatObserver:
raise RuntimeError(f"ChatObserver for {stream_id} already exists. Use get_instance() instead.")
self.stream_id = stream_id
self.private_name = private_name
self.message_storage = MongoDBMessageStorage()
# self.last_user_speak_time: Optional[float] = None # 对方上次发言时间
@@ -76,12 +77,12 @@ class ChatObserver:
Returns:
bool: 是否有新消息
"""
logger.debug(f"检查距离上一次观察之后是否有了新消息: {self.last_check_time}")
logger.debug(f"[私聊][{self.private_name}]检查距离上一次观察之后是否有了新消息: {self.last_check_time}")
new_message_exists = await self.message_storage.has_new_messages(self.stream_id, self.last_check_time)
if new_message_exists:
logger.debug("发现新消息")
logger.debug(f"[私聊][{self.private_name}]发现新消息")
self.last_check_time = time.time()
return new_message_exists
@@ -100,7 +101,7 @@ class ChatObserver:
# print(self.notification_manager)
await self.notification_manager.send_notification(notification)
except Exception as e:
logger.error(f"添加消息到历史记录时出错: {e}")
logger.error(f"[私聊][{self.private_name}]添加消息到历史记录时出错: {e}")
print(traceback.format_exc())
# 检查并更新冷场状态
@@ -140,11 +141,13 @@ class ChatObserver:
"""
if self.last_message_time is None:
logger.debug("没有最后消息时间，返回 False")
logger.debug(f"[私聊][{self.private_name}]没有最后消息时间，返回 False")
return False
has_new = self.last_message_time > time_point
logger.debug(f"判断是否在指定时间点后有新消息: {self.last_message_time} > {time_point} = {has_new}")
logger.debug(
f"[私聊][{self.private_name}]判断是否在指定时间点后有新消息: {self.last_message_time} > {time_point} = {has_new}"
)
return has_new
def get_message_history(
@@ -213,7 +216,7 @@ class ChatObserver:
if new_messages:
self.last_message_read = new_messages[-1]["message_id"]
logger.debug(f"获取指定时间点111之前的消息: {new_messages}")
logger.debug(f"[私聊][{self.private_name}]获取指定时间点111之前的消息: {new_messages}")
return new_messages
@@ -226,9 +229,9 @@ class ChatObserver:
# messages = await self._fetch_new_messages_before(start_time)
# for message in messages:
# await self._add_message_to_history(message)
# logger.debug(f"缓冲消息: {messages}")
# logger.debug(f"[私聊][{self.private_name}]缓冲消息: {messages}")
# except Exception as e:
# logger.error(f"缓冲消息出错: {e}")
# logger.error(f"[私聊][{self.private_name}]缓冲消息出错: {e}")
while self._running:
try:
@@ -256,8 +259,8 @@ class ChatObserver:
self._update_complete.set()
except Exception as e:
logger.error(f"更新循环出错: {e}")
logger.error(traceback.format_exc())
logger.error(f"[私聊][{self.private_name}]更新循环出错: {e}")
logger.error(f"[私聊][{self.private_name}]{traceback.format_exc()}")
self._update_complete.set() # 即使出错也要设置完成事件
def trigger_update(self):
@@ -277,7 +280,7 @@ class ChatObserver:
await asyncio.wait_for(self._update_complete.wait(), timeout=timeout)
return True
except asyncio.TimeoutError:
logger.warning(f"等待更新完成超时（{timeout}秒）")
logger.warning(f"[私聊][{self.private_name}]等待更新完成超时（{timeout}秒）")
return False
def start(self):
@@ -287,7 +290,7 @@ class ChatObserver:
self._running = True
self._task = asyncio.create_task(self._update_loop())
logger.debug(f"ChatObserver for {self.stream_id} started")
logger.debug(f"[私聊][{self.private_name}]ChatObserver for {self.stream_id} started")
def stop(self):
"""停止观察器"""
@@ -296,7 +299,7 @@ class ChatObserver:
self._update_complete.set() # 设置完成事件以解除等待
if self._task:
self._task.cancel()
logger.debug(f"ChatObserver for {self.stream_id} stopped")
logger.debug(f"[私聊][{self.private_name}]ChatObserver for {self.stream_id} stopped")
async def process_chat_history(self, messages: list):
"""处理聊天历史
@@ -314,7 +317,7 @@ class ChatObserver:
else:
self.update_user_speak_time(msg["time"])
except Exception as e:
logger.warning(f"处理消息时间时出错: {e}")
logger.warning(f"[私聊][{self.private_name}]处理消息时间时出错: {e}")
continue
def update_check_time(self):

183
src/plugins/PFC/conversation.py
View File

@@ -9,7 +9,8 @@ from src.plugins.utils.chat_message_builder import build_readable_messages, get_
from typing import Dict, Any, Optional
from ..chat.message import Message
from .pfc_types import ConversationState
from .pfc import ChatObserver, GoalAnalyzer, DirectMessageSender
from .pfc import ChatObserver, GoalAnalyzer
from .message_sender import DirectMessageSender
from src.common.logger_manager import get_logger
from .action_planner import ActionPlanner
from .observation_info import ObservationInfo
@@ -29,13 +30,14 @@ logger = get_logger("pfc")
class Conversation:
"""对话类，负责管理单个对话的状态和行为"""
def __init__(self, stream_id: str):
def __init__(self, stream_id: str, private_name: str):
"""初始化对话实例
Args:
stream_id: 聊天流ID
"""
self.stream_id = stream_id
self.private_name = private_name
self.state = ConversationState.INIT
self.should_continue = False
self.ignore_until_timestamp: Optional[float] = None
@@ -47,38 +49,38 @@ class Conversation:
"""初始化实例，注册所有组件"""
try:
self.action_planner = ActionPlanner(self.stream_id)
self.goal_analyzer = GoalAnalyzer(self.stream_id)
self.reply_generator = ReplyGenerator(self.stream_id)
self.knowledge_fetcher = KnowledgeFetcher()
self.waiter = Waiter(self.stream_id)
self.direct_sender = DirectMessageSender()
self.action_planner = ActionPlanner(self.stream_id, self.private_name)
self.goal_analyzer = GoalAnalyzer(self.stream_id, self.private_name)
self.reply_generator = ReplyGenerator(self.stream_id, self.private_name)
self.knowledge_fetcher = KnowledgeFetcher(self.private_name)
self.waiter = Waiter(self.stream_id, self.private_name)
self.direct_sender = DirectMessageSender(self.private_name)
# 获取聊天流信息
self.chat_stream = chat_manager.get_stream(self.stream_id)
self.stop_action_planner = False
except Exception as e:
logger.error(f"初始化对话实例：注册运行组件失败: {e}")
logger.error(traceback.format_exc())
logger.error(f"[私聊][{self.private_name}]初始化对话实例：注册运行组件失败: {e}")
logger.error(f"[私聊][{self.private_name}]{traceback.format_exc()}")
raise
try:
# 决策所需要的信息，包括自身自信和观察信息两部分
# 注册观察器和观测信息
self.chat_observer = ChatObserver.get_instance(self.stream_id)
self.chat_observer = ChatObserver.get_instance(self.stream_id, self.private_name)
self.chat_observer.start()
self.observation_info = ObservationInfo()
self.observation_info = ObservationInfo(self.private_name)
self.observation_info.bind_to_chat_observer(self.chat_observer)
# print(self.chat_observer.get_cached_messages(limit=)
self.conversation_info = ConversationInfo()
except Exception as e:
logger.error(f"初始化对话实例：注册信息组件失败: {e}")
logger.error(traceback.format_exc())
logger.error(f"[私聊][{self.private_name}]初始化对话实例：注册信息组件失败: {e}")
logger.error(f"[私聊][{self.private_name}]{traceback.format_exc()}")
raise
try:
logger.info(f"为 {self.stream_id} 加载初始聊天记录...")
logger.info(f"[私聊][{self.private_name}]为 {self.stream_id} 加载初始聊天记录...")
initial_messages = get_raw_msg_before_timestamp_with_chat( #
chat_id=self.stream_id,
timestamp=time.time(),
@@ -105,17 +107,17 @@ class Conversation:
self.observation_info.last_message_content = last_msg.get("processed_plain_text", "")
logger.info(
f"成功加载 {len(initial_messages)} 条初始聊天记录。最后一条消息时间: {self.observation_info.last_message_time}"
f"[私聊][{self.private_name}]成功加载 {len(initial_messages)} 条初始聊天记录。最后一条消息时间: {self.observation_info.last_message_time}"
)
# 让 ChatObserver 从加载的最后一条消息之后开始同步
self.chat_observer.last_message_time = self.observation_info.last_message_time
self.chat_observer.last_message_read = last_msg # 更新 observer 的最后读取记录
else:
logger.info("没有找到初始聊天记录。")
logger.info(f"[私聊][{self.private_name}]没有找到初始聊天记录。")
except Exception as load_err:
logger.error(f"加载初始聊天记录时出错: {load_err}")
logger.error(f"[私聊][{self.private_name}]加载初始聊天记录时出错: {load_err}")
# 出错也要继续，只是没有历史记录而已
# 组件准备完成，启动该论对话
self.should_continue = True
@@ -124,10 +126,10 @@ class Conversation:
async def start(self):
"""开始对话流程"""
try:
logger.info("对话系统启动中...")
logger.info(f"[私聊][{self.private_name}]对话系统启动中...")
asyncio.create_task(self._plan_and_action_loop())
except Exception as e:
logger.error(f"启动对话系统失败: {e}")
logger.error(f"[私聊][{self.private_name}]启动对话系统失败: {e}")
raise
async def _plan_and_action_loop(self):
@@ -138,7 +140,7 @@ class Conversation:
await asyncio.sleep(30)
continue
elif self.ignore_until_timestamp and time.time() >= self.ignore_until_timestamp:
logger.info(f"忽略时间已到 {self.stream_id}，准备结束对话。")
logger.info(f"[私聊][{self.private_name}]忽略时间已到 {self.stream_id}，准备结束对话。")
self.ignore_until_timestamp = None
self.should_continue = False
continue
@@ -148,7 +150,9 @@ class Conversation:
if hasattr(self.observation_info, "new_messages_count"):
initial_new_message_count = self.observation_info.new_messages_count + 1 # 算上麦麦自己发的那一条
else:
logger.warning("ObservationInfo missing 'new_messages_count' before planning.")
logger.warning(
f"[私聊][{self.private_name}]ObservationInfo missing 'new_messages_count' before planning."
)
# --- 调用 Action Planner ---
# 传递 self.conversation_info.last_successful_reply_action
@@ -161,11 +165,13 @@ class Conversation:
if hasattr(self.observation_info, "new_messages_count"):
current_new_message_count = self.observation_info.new_messages_count
else:
logger.warning("ObservationInfo missing 'new_messages_count' after planning.")
logger.warning(
f"[私聊][{self.private_name}]ObservationInfo missing 'new_messages_count' after planning."
)
if current_new_message_count > initial_new_message_count + 2:
logger.info(
f"规划期间发现新增消息 ({initial_new_message_count} -> {current_new_message_count})，跳过本次行动，重新规划"
f"[私聊][{self.private_name}]规划期间发现新增消息 ({initial_new_message_count} -> {current_new_message_count})，跳过本次行动，重新规划"
)
# 如果规划期间有新消息，也应该重置上次回复状态，因为现在要响应新消息了
self.conversation_info.last_successful_reply_action = None
@@ -175,12 +181,16 @@ class Conversation:
# 包含 send_new_message
if initial_new_message_count > 0 and action in ["direct_reply", "send_new_message"]:
if hasattr(self.observation_info, "clear_unprocessed_messages"):
logger.debug(f"准备执行 {action}，清理 {initial_new_message_count} 条规划时已知的新消息。")
logger.debug(
f"[私聊][{self.private_name}]准备执行 {action}，清理 {initial_new_message_count} 条规划时已知的新消息。"
)
await self.observation_info.clear_unprocessed_messages()
if hasattr(self.observation_info, "new_messages_count"):
self.observation_info.new_messages_count = 0
else:
logger.error("无法清理未处理消息: ObservationInfo 缺少 clear_unprocessed_messages 方法！")
logger.error(
f"[私聊][{self.private_name}]无法清理未处理消息: ObservationInfo 缺少 clear_unprocessed_messages 方法！"
)
await self._handle_action(action, reason, self.observation_info, self.conversation_info)
@@ -188,10 +198,7 @@ class Conversation:
goal_ended = False
if hasattr(self.conversation_info, "goal_list") and self.conversation_info.goal_list:
for goal_item in self.conversation_info.goal_list:
current_goal = None
if isinstance(goal_item, tuple) and len(goal_item) > 0:
current_goal = goal_item[0]
elif isinstance(goal_item, dict):
if isinstance(goal_item, dict):
current_goal = goal_item.get("goal")
if current_goal == "结束对话":
@@ -200,34 +207,38 @@ class Conversation:
if goal_ended:
self.should_continue = False
logger.info("检测到'结束对话'目标，停止循环。")
logger.info(f"[私聊][{self.private_name}]检测到'结束对话'目标，停止循环。")
except Exception as loop_err:
logger.error(f"PFC主循环出错: {loop_err}")
logger.error(traceback.format_exc())
logger.error(f"[私聊][{self.private_name}]PFC主循环出错: {loop_err}")
logger.error(f"[私聊][{self.private_name}]{traceback.format_exc()}")
await asyncio.sleep(1)
if self.should_continue:
await asyncio.sleep(0.1)
logger.info(f"PFC 循环结束 for stream_id: {self.stream_id}")
logger.info(f"[私聊][{self.private_name}]PFC 循环结束 for stream_id: {self.stream_id}")
def _check_new_messages_after_planning(self):
"""检查在规划后是否有新消息"""
# 检查 ObservationInfo 是否已初始化并且有 new_messages_count 属性
if not hasattr(self, "observation_info") or not hasattr(self.observation_info, "new_messages_count"):
logger.warning("ObservationInfo 未初始化或缺少 'new_messages_count' 属性，无法检查新消息。")
logger.warning(
f"[私聊][{self.private_name}]ObservationInfo 未初始化或缺少 'new_messages_count' 属性，无法检查新消息。"
)
return False # 或者根据需要抛出错误
if self.observation_info.new_messages_count > 2:
logger.info(
f"生成/执行动作期间收到 {self.observation_info.new_messages_count} 条新消息，取消当前动作并重新规划"
f"[私聊][{self.private_name}]生成/执行动作期间收到 {self.observation_info.new_messages_count} 条新消息，取消当前动作并重新规划"
)
# 如果有新消息，也应该重置上次回复状态
if hasattr(self, "conversation_info"): # 确保 conversation_info 已初始化
self.conversation_info.last_successful_reply_action = None
else:
logger.warning("ConversationInfo 未初始化，无法重置 last_successful_reply_action。")
logger.warning(
f"[私聊][{self.private_name}]ConversationInfo 未初始化，无法重置 last_successful_reply_action。"
)
return True
return False
@@ -256,7 +267,7 @@ class Conversation:
detailed_plain_text=msg_dict.get("detailed_plain_text", ""),
)
except Exception as e:
logger.warning(f"转换消息时出错: {e}")
logger.warning(f"[私聊][{self.private_name}]转换消息时出错: {e}")
# 可以选择返回 None 或重新抛出异常，这里选择重新抛出以指示问题
raise ValueError(f"无法将字典转换为 Message 对象: {e}") from e
@@ -265,7 +276,7 @@ class Conversation:
):
"""处理规划的行动"""
logger.debug(f"执行行动: {action}, 原因: {reason}")
logger.debug(f"[私聊][{self.private_name}]执行行动: {action}, 原因: {reason}")
# 记录action历史 (逻辑不变)
current_action_record = {
@@ -296,14 +307,18 @@ class Conversation:
while reply_attempt_count < max_reply_attempts and not is_suitable:
reply_attempt_count += 1
logger.info(f"尝试生成追问回复 (第 {reply_attempt_count}/{max_reply_attempts} 次)...")
logger.info(
f"[私聊][{self.private_name}]尝试生成追问回复 (第 {reply_attempt_count}/{max_reply_attempts} 次)..."
)
self.state = ConversationState.GENERATING
# 1. 生成回复 (调用 generate 时传入 action_type)
self.generated_reply = await self.reply_generator.generate(
observation_info, conversation_info, action_type="send_new_message"
)
logger.info(f"第 {reply_attempt_count} 次生成的追问回复: {self.generated_reply}")
logger.info(
f"[私聊][{self.private_name}]第 {reply_attempt_count} 次生成的追问回复: {self.generated_reply}"
)
# 2. 检查回复 (逻辑不变)
self.state = ConversationState.CHECKING
@@ -317,18 +332,20 @@ class Conversation:
retry_count=reply_attempt_count - 1,
)
logger.info(
f"第 {reply_attempt_count} 次追问检查结果: 合适={is_suitable}, 原因='{check_reason}', 需重新规划={need_replan}"
f"[私聊][{self.private_name}]第 {reply_attempt_count} 次追问检查结果: 合适={is_suitable}, 原因='{check_reason}', 需重新规划={need_replan}"
)
if is_suitable:
final_reply_to_send = self.generated_reply
break
elif need_replan:
logger.warning(
f"第 {reply_attempt_count} 次追问检查建议重新规划，停止尝试。原因: {check_reason}"
f"[私聊][{self.private_name}]第 {reply_attempt_count} 次追问检查建议重新规划，停止尝试。原因: {check_reason}"
)
break
except Exception as check_err:
logger.error(f"第 {reply_attempt_count} 次调用 ReplyChecker (追问) 时出错: {check_err}")
logger.error(
f"[私聊][{self.private_name}]第 {reply_attempt_count} 次调用 ReplyChecker (追问) 时出错: {check_err}"
)
check_reason = f"第 {reply_attempt_count} 次检查过程出错: {check_err}"
break
@@ -336,7 +353,7 @@ class Conversation:
if is_suitable:
# 检查是否有新消息
if self._check_new_messages_after_planning():
logger.info("生成追问回复期间收到新消息，取消发送，重新规划行动")
logger.info(f"[私聊][{self.private_name}]生成追问回复期间收到新消息，取消发送，重新规划行动")
conversation_info.done_action[action_index].update(
{"status": "recall", "final_reason": f"有新消息，取消发送追问: {final_reply_to_send}"}
)
@@ -353,7 +370,9 @@ class Conversation:
else:
# 追问失败
logger.warning(f"经过 {reply_attempt_count} 次尝试，未能生成合适的追问回复。最终原因: {check_reason}")
logger.warning(
f"[私聊][{self.private_name}]经过 {reply_attempt_count} 次尝试，未能生成合适的追问回复。最终原因: {check_reason}"
)
conversation_info.done_action[action_index].update(
{"status": "recall", "final_reason": f"追问尝试{reply_attempt_count}次后失败: {check_reason}"}
)
@@ -361,7 +380,7 @@ class Conversation:
self.conversation_info.last_successful_reply_action = None
# 执行 Wait 操作
logger.info("由于无法生成合适追问回复，执行 'wait' 操作...")
logger.info(f"[私聊][{self.private_name}]由于无法生成合适追问回复，执行 'wait' 操作...")
self.state = ConversationState.WAITING
await self.waiter.wait(self.conversation_info)
wait_action_record = {
@@ -383,14 +402,18 @@ class Conversation:
while reply_attempt_count < max_reply_attempts and not is_suitable:
reply_attempt_count += 1
logger.info(f"尝试生成首次回复 (第 {reply_attempt_count}/{max_reply_attempts} 次)...")
logger.info(
f"[私聊][{self.private_name}]尝试生成首次回复 (第 {reply_attempt_count}/{max_reply_attempts} 次)..."
)
self.state = ConversationState.GENERATING
# 1. 生成回复
self.generated_reply = await self.reply_generator.generate(
observation_info, conversation_info, action_type="direct_reply"
)
logger.info(f"第 {reply_attempt_count} 次生成的首次回复: {self.generated_reply}")
logger.info(
f"[私聊][{self.private_name}]第 {reply_attempt_count} 次生成的首次回复: {self.generated_reply}"
)
# 2. 检查回复
self.state = ConversationState.CHECKING
@@ -404,18 +427,20 @@ class Conversation:
retry_count=reply_attempt_count - 1,
)
logger.info(
f"第 {reply_attempt_count} 次首次回复检查结果: 合适={is_suitable}, 原因='{check_reason}', 需重新规划={need_replan}"
f"[私聊][{self.private_name}]第 {reply_attempt_count} 次首次回复检查结果: 合适={is_suitable}, 原因='{check_reason}', 需重新规划={need_replan}"
)
if is_suitable:
final_reply_to_send = self.generated_reply
break
elif need_replan:
logger.warning(
f"第 {reply_attempt_count} 次首次回复检查建议重新规划，停止尝试。原因: {check_reason}"
f"[私聊][{self.private_name}]第 {reply_attempt_count} 次首次回复检查建议重新规划，停止尝试。原因: {check_reason}"
)
break
except Exception as check_err:
logger.error(f"第 {reply_attempt_count} 次调用 ReplyChecker (首次回复) 时出错: {check_err}")
logger.error(
f"[私聊][{self.private_name}]第 {reply_attempt_count} 次调用 ReplyChecker (首次回复) 时出错: {check_err}"
)
check_reason = f"第 {reply_attempt_count} 次检查过程出错: {check_err}"
break
@@ -423,7 +448,7 @@ class Conversation:
if is_suitable:
# 检查是否有新消息
if self._check_new_messages_after_planning():
logger.info("生成首次回复期间收到新消息，取消发送，重新规划行动")
logger.info(f"[私聊][{self.private_name}]生成首次回复期间收到新消息，取消发送，重新规划行动")
conversation_info.done_action[action_index].update(
{"status": "recall", "final_reason": f"有新消息，取消发送首次回复: {final_reply_to_send}"}
)
@@ -440,7 +465,9 @@ class Conversation:
else:
# 首次回复失败
logger.warning(f"经过 {reply_attempt_count} 次尝试，未能生成合适的首次回复。最终原因: {check_reason}")
logger.warning(
f"[私聊][{self.private_name}]经过 {reply_attempt_count} 次尝试，未能生成合适的首次回复。最终原因: {check_reason}"
)
conversation_info.done_action[action_index].update(
{"status": "recall", "final_reason": f"首次回复尝试{reply_attempt_count}次后失败: {check_reason}"}
)
@@ -448,7 +475,7 @@ class Conversation:
self.conversation_info.last_successful_reply_action = None
# 执行 Wait 操作 (保持原有逻辑)
logger.info("由于无法生成合适首次回复，执行 'wait' 操作...")
logger.info(f"[私聊][{self.private_name}]由于无法生成合适首次回复，执行 'wait' 操作...")
self.state = ConversationState.WAITING
await self.waiter.wait(self.conversation_info)
wait_action_record = {
@@ -466,11 +493,11 @@ class Conversation:
try:
# 检查 knowledge_fetcher 是否存在
if not hasattr(self, "knowledge_fetcher"):
logger.error("KnowledgeFetcher 未初始化，无法获取知识。")
logger.error(f"[私聊][{self.private_name}]KnowledgeFetcher 未初始化，无法获取知识。")
raise AttributeError("KnowledgeFetcher not initialized")
knowledge, source = await self.knowledge_fetcher.fetch(knowledge_query, observation_info.chat_history)
logger.info(f"获取到知识: {knowledge[:100]}..., 来源: {source}")
logger.info(f"[私聊][{self.private_name}]获取到知识: {knowledge[:100]}..., 来源: {source}")
if knowledge:
# 确保 knowledge_list 存在
if not hasattr(conversation_info, "knowledge_list"):
@@ -480,7 +507,7 @@ class Conversation:
)
action_successful = True
except Exception as fetch_err:
logger.error(f"获取知识时出错: {fetch_err}")
logger.error(f"[私聊][{self.private_name}]获取知识时出错: {fetch_err}")
conversation_info.done_action[action_index].update(
{"status": "recall", "final_reason": f"获取知识失败: {fetch_err}"}
)
@@ -491,12 +518,12 @@ class Conversation:
try:
# 检查 goal_analyzer 是否存在
if not hasattr(self, "goal_analyzer"):
logger.error("GoalAnalyzer 未初始化，无法重新思考目标。")
logger.error(f"[私聊][{self.private_name}]GoalAnalyzer 未初始化，无法重新思考目标。")
raise AttributeError("GoalAnalyzer not initialized")
await self.goal_analyzer.analyze_goal(conversation_info, observation_info)
action_successful = True
except Exception as rethink_err:
logger.error(f"重新思考目标时出错: {rethink_err}")
logger.error(f"[私聊][{self.private_name}]重新思考目标时出错: {rethink_err}")
conversation_info.done_action[action_index].update(
{"status": "recall", "final_reason": f"重新思考目标失败: {rethink_err}"}
)
@@ -504,16 +531,16 @@ class Conversation:
elif action == "listening":
self.state = ConversationState.LISTENING
logger.info("倾听对方发言...")
logger.info(f"[私聊][{self.private_name}]倾听对方发言...")
try:
# 检查 waiter 是否存在
if not hasattr(self, "waiter"):
logger.error("Waiter 未初始化，无法倾听。")
logger.error(f"[私聊][{self.private_name}]Waiter 未初始化，无法倾听。")
raise AttributeError("Waiter not initialized")
await self.waiter.wait_listening(conversation_info)
action_successful = True # Listening 完成就算成功
except Exception as listen_err:
logger.error(f"倾听时出错: {listen_err}")
logger.error(f"[私聊][{self.private_name}]倾听时出错: {listen_err}")
conversation_info.done_action[action_index].update(
{"status": "recall", "final_reason": f"倾听失败: {listen_err}"}
)
@@ -521,29 +548,31 @@ class Conversation:
elif action == "end_conversation":
self.should_continue = False
logger.info("决定结束对话...")
logger.info(f"[私聊][{self.private_name}]决定结束对话...")
action_successful = True # 标记动作成功
elif action == "block_and_ignore":
logger.info("不想再理你了...")
logger.info(f"[私聊][{self.private_name}]不想再理你了...")
ignore_duration_seconds = 10 * 60
self.ignore_until_timestamp = time.time() + ignore_duration_seconds
logger.info(f"将忽略此对话直到: {datetime.datetime.fromtimestamp(self.ignore_until_timestamp)}")
logger.info(
f"[私聊][{self.private_name}]将忽略此对话直到: {datetime.datetime.fromtimestamp(self.ignore_until_timestamp)}"
)
self.state = ConversationState.IGNORED
action_successful = True # 标记动作成功
else: # 对应 'wait' 动作
self.state = ConversationState.WAITING
logger.info("等待更多信息...")
logger.info(f"[私聊][{self.private_name}]等待更多信息...")
try:
# 检查 waiter 是否存在
if not hasattr(self, "waiter"):
logger.error("Waiter 未初始化，无法等待。")
logger.error(f"[私聊][{self.private_name}]Waiter 未初始化，无法等待。")
raise AttributeError("Waiter not initialized")
_timeout_occurred = await self.waiter.wait(self.conversation_info)
action_successful = True # Wait 完成就算成功
except Exception as wait_err:
logger.error(f"等待时出错: {wait_err}")
logger.error(f"[私聊][{self.private_name}]等待时出错: {wait_err}")
conversation_info.done_action[action_index].update(
{"status": "recall", "final_reason": f"等待失败: {wait_err}"}
)
@@ -561,13 +590,13 @@ class Conversation:
# 重置状态: 对于非回复类动作的成功，清除上次回复状态
if action not in ["direct_reply", "send_new_message"]:
self.conversation_info.last_successful_reply_action = None
logger.debug(f"动作 {action} 成功完成，重置 last_successful_reply_action")
logger.debug(f"[私聊][{self.private_name}]动作 {action} 成功完成，重置 last_successful_reply_action")
# 如果动作是 recall 状态，在各自的处理逻辑中已经更新了 done_action
async def _send_reply(self):
"""发送回复"""
if not self.generated_reply:
logger.warning("没有生成回复内容，无法发送。")
logger.warning(f"[私聊][{self.private_name}]没有生成回复内容，无法发送。")
return
try:
@@ -576,10 +605,10 @@ class Conversation:
# 发送消息 (确保 direct_sender 和 chat_stream 有效)
if not hasattr(self, "direct_sender") or not self.direct_sender:
logger.error("DirectMessageSender 未初始化，无法发送回复。")
logger.error(f"[私聊][{self.private_name}]DirectMessageSender 未初始化，无法发送回复。")
return
if not self.chat_stream:
logger.error("ChatStream 未初始化，无法发送回复。")
logger.error(f"[私聊][{self.private_name}]ChatStream 未初始化，无法发送回复。")
return
await self.direct_sender.send_message(chat_stream=self.chat_stream, content=reply_content)
@@ -589,13 +618,13 @@ class Conversation:
# 暂时注释掉，观察是否影响 ObservationInfo 的更新
# self.chat_observer.trigger_update()
# if not await self.chat_observer.wait_for_update():
# logger.warning("等待 ChatObserver 更新完成超时")
# logger.warning(f"[私聊][{self.private_name}]等待 ChatObserver 更新完成超时")
self.state = ConversationState.ANALYZING # 更新状态
except Exception as e:
logger.error(f"发送消息或更新状态时失败: {str(e)}")
logger.error(traceback.format_exc())
logger.error(f"[私聊][{self.private_name}]发送消息或更新状态时失败: {str(e)}")
logger.error(f"[私聊][{self.private_name}]{traceback.format_exc()}")
self.state = ConversationState.ANALYZING
async def _send_timeout_message(self):
@@ -610,4 +639,4 @@ class Conversation:
chat_stream=self.chat_stream, content="TODO:超时消息", reply_to_message=latest_message
)
except Exception as e:
logger.error(f"发送超时消息失败: {str(e)}")
logger.error(f"[私聊][{self.private_name}]发送超时消息失败: {str(e)}")

58
src/plugins/PFC/message_sender.py
View File

@@ -1,10 +1,14 @@
import time
from typing import Optional
from src.common.logger import get_module_logger
from ..chat.chat_stream import ChatStream
from ..chat.message import Message
from maim_message import Seg
from maim_message import UserInfo, Seg
from src.plugins.chat.message import MessageSending, MessageSet
from src.plugins.chat.message_sender import message_manager
from ..storage.storage import MessageStorage
from ...config.config import global_config
logger = get_module_logger("message_sender")
@@ -12,8 +16,9 @@ logger = get_module_logger("message_sender")
class DirectMessageSender:
"""直接消息发送器"""
def __init__(self):
pass
def __init__(self, private_name: str):
self.private_name = private_name
self.storage = MessageStorage()
async def send_message(
self,
@@ -30,21 +35,44 @@ class DirectMessageSender:
"""
try:
# 创建消息内容
segments = [Seg(type="text", data={"text": content})]
segments = Seg(type="seglist", data=[Seg(type="text", data=content)])
# 检查是否需要引用回复
if reply_to_message:
reply_id = reply_to_message.message_id
message_sending = MessageSending(segments=segments, reply_to_id=reply_id)
else:
message_sending = MessageSending(segments=segments)
# 获取麦麦的信息
bot_user_info = UserInfo(
user_id=global_config.BOT_QQ,
user_nickname=global_config.BOT_NICKNAME,
platform=chat_stream.platform,
)
# 用当前时间作为message_id，和之前那套sender一样
message_id = f"dm{round(time.time(), 2)}"
# 构建消息对象
message = MessageSending(
message_id=message_id,
chat_stream=chat_stream,
bot_user_info=bot_user_info,
sender_info=reply_to_message.message_info.user_info if reply_to_message else None,
message_segment=segments,
reply=reply_to_message,
is_head=True,
is_emoji=False,
thinking_start_time=time.time(),
)
# 处理消息
await message.process()
# 不知道有什么用，先留下来了，和之前那套sender一样
_message_json = message.to_dict()
# 发送消息
message_set = MessageSet(chat_stream, message_sending.message_id)
message_set.add_message(message_sending)
message_manager.add_message(message_set)
logger.info(f"PFC消息已发送: {content}")
message_set = MessageSet(chat_stream, message_id)
message_set.add_message(message)
await message_manager.add_message(message_set)
await self.storage.store_message(message, chat_stream)
logger.info(f"[私聊][{self.private_name}]PFC消息已发送: {content}")
except Exception as e:
logger.error(f"PFC消息发送失败: {str(e)}")
logger.error(f"[私聊][{self.private_name}]PFC消息发送失败: {str(e)}")
raise

365
src/plugins/PFC/observation_info.py
View File

@@ -1,13 +1,12 @@
# Programmable Friendly Conversationalist
# Prefrontal cortex
from typing import List, Optional, Dict, Any, Set
from maim_message import UserInfo
import time
from dataclasses import dataclass, field
from src.common.logger import get_module_logger
from .chat_observer import ChatObserver
from .chat_states import NotificationHandler, NotificationType
from .chat_states import NotificationHandler, NotificationType, Notification
from src.plugins.utils.chat_message_builder import build_readable_messages
import traceback # 导入 traceback 用于调试
logger = get_module_logger("observation_info")
@@ -15,186 +14,287 @@ logger = get_module_logger("observation_info")
class ObservationInfoHandler(NotificationHandler):
"""ObservationInfo的通知处理器"""
def __init__(self, observation_info: "ObservationInfo"):
def __init__(self, observation_info: "ObservationInfo", private_name: str):
"""初始化处理器
Args:
observation_info: 要更新的ObservationInfo实例
private_name: 私聊对象的名称，用于日志记录
"""
self.observation_info = observation_info
# 将 private_name 存储在 handler 实例中
self.private_name = private_name
async def handle_notification(self, notification):
async def handle_notification(self, notification: Notification): # 添加类型提示
# 获取通知类型和数据
notification_type = notification.type
data = notification.data
if notification_type == NotificationType.NEW_MESSAGE:
# 处理新消息通知
logger.debug(f"收到新消息通知data: {data}")
message_id = data.get("message_id")
processed_plain_text = data.get("processed_plain_text")
detailed_plain_text = data.get("detailed_plain_text")
user_info = data.get("user_info")
time_value = data.get("time")
try: # 添加错误处理块
if notification_type == NotificationType.NEW_MESSAGE:
# 处理新消息通知
# logger.debug(f"[私聊][{self.private_name}]收到新消息通知data: {data}") # 可以在需要时取消注释
message_id = data.get("message_id")
processed_plain_text = data.get("processed_plain_text")
detailed_plain_text = data.get("detailed_plain_text")
user_info_dict = data.get("user_info") # 先获取字典
time_value = data.get("time")
message = {
"message_id": message_id,
"processed_plain_text": processed_plain_text,
"detailed_plain_text": detailed_plain_text,
"user_info": user_info,
"time": time_value,
}
# 确保 user_info 是字典类型再创建 UserInfo 对象
user_info = None
if isinstance(user_info_dict, dict):
try:
user_info = UserInfo.from_dict(user_info_dict)
except Exception as e:
logger.error(
f"[私聊][{self.private_name}]从字典创建 UserInfo 时出错: {e}, 字典内容: {user_info_dict}"
)
# 可以选择在这里返回或记录错误，避免后续代码出错
return
elif user_info_dict is not None:
logger.warning(
f"[私聊][{self.private_name}]收到的 user_info 不是预期的字典类型: {type(user_info_dict)}"
)
# 根据需要处理非字典情况，这里暂时返回
return
self.observation_info.update_from_message(message)
message = {
"message_id": message_id,
"processed_plain_text": processed_plain_text,
"detailed_plain_text": detailed_plain_text,
"user_info": user_info_dict, # 存储原始字典或 UserInfo 对象，取决于你的 update_from_message 如何处理
"time": time_value,
}
# 传递 UserInfo 对象（如果成功创建）或原始字典
await self.observation_info.update_from_message(message, user_info) # 修改：传递 user_info 对象
elif notification_type == NotificationType.COLD_CHAT:
# 处理冷场通知
is_cold = data.get("is_cold", False)
self.observation_info.update_cold_chat_status(is_cold, time.time())
elif notification_type == NotificationType.COLD_CHAT:
# 处理冷场通知
is_cold = data.get("is_cold", False)
await self.observation_info.update_cold_chat_status(is_cold, time.time()) # 修改：改为 await 调用
elif notification_type == NotificationType.ACTIVE_CHAT:
# 处理活跃通知
is_active = data.get("is_active", False)
self.observation_info.is_cold = not is_active
elif notification_type == NotificationType.ACTIVE_CHAT:
# 处理活跃通知 (通常由 COLD_CHAT 的反向状态处理)
is_active = data.get("is_active", False)
self.observation_info.is_cold = not is_active
elif notification_type == NotificationType.BOT_SPEAKING:
# 处理机器人说话通知
self.observation_info.is_typing = False
self.observation_info.last_bot_speak_time = time.time()
elif notification_type == NotificationType.BOT_SPEAKING:
# 处理机器人说话通知 (按需实现)
self.observation_info.is_typing = False
self.observation_info.last_bot_speak_time = time.time()
elif notification_type == NotificationType.USER_SPEAKING:
# 处理用户说话通知
self.observation_info.is_typing = False
self.observation_info.last_user_speak_time = time.time()
elif notification_type == NotificationType.USER_SPEAKING:
# 处理用户说话通知
self.observation_info.is_typing = False
self.observation_info.last_user_speak_time = time.time()
elif notification_type == NotificationType.MESSAGE_DELETED:
# 处理消息删除通知
message_id = data.get("message_id")
self.observation_info.unprocessed_messages = [
msg for msg in self.observation_info.unprocessed_messages if msg.get("message_id") != message_id
]
elif notification_type == NotificationType.MESSAGE_DELETED:
# 处理消息删除通知
message_id = data.get("message_id")
# 从 unprocessed_messages 中移除被删除的消息
original_count = len(self.observation_info.unprocessed_messages)
self.observation_info.unprocessed_messages = [
msg for msg in self.observation_info.unprocessed_messages if msg.get("message_id") != message_id
]
if len(self.observation_info.unprocessed_messages) < original_count:
logger.info(f"[私聊][{self.private_name}]移除了未处理的消息 (ID: {message_id})")
elif notification_type == NotificationType.USER_JOINED:
# 处理用户加入通知
user_id = data.get("user_id")
if user_id:
self.observation_info.active_users.add(user_id)
elif notification_type == NotificationType.USER_JOINED:
# 处理用户加入通知 (如果适用私聊场景)
user_id = data.get("user_id")
if user_id:
self.observation_info.active_users.add(str(user_id)) # 确保是字符串
elif notification_type == NotificationType.USER_LEFT:
# 处理用户离开通知
user_id = data.get("user_id")
if user_id:
self.observation_info.active_users.discard(user_id)
elif notification_type == NotificationType.USER_LEFT:
# 处理用户离开通知 (如果适用私聊场景)
user_id = data.get("user_id")
if user_id:
self.observation_info.active_users.discard(str(user_id)) # 确保是字符串
elif notification_type == NotificationType.ERROR:
# 处理错误通知
error_msg = data.get("error", "")
logger.error(f"收到错误通知: {error_msg}")
elif notification_type == NotificationType.ERROR:
# 处理错误通知
error_msg = data.get("error", "未提供错误信息")
logger.error(f"[私聊][{self.private_name}]收到错误通知: {error_msg}")
except Exception as e:
logger.error(f"[私聊][{self.private_name}]处理通知时发生错误: {e}")
logger.error(traceback.format_exc()) # 打印详细堆栈信息
@dataclass
class ObservationInfo:
"""决策信息类，用于收集和管理来自chat_observer的通知信息"""
# --- 修改：添加 private_name 字段 ---
private_name: str = field(init=True) # 让 dataclass 的 __init__ 接收 private_name
# data_list
chat_history: List[str] = field(default_factory=list)
chat_history: List[Dict[str, Any]] = field(default_factory=list) # 修改：明确类型为 Dict
chat_history_str: str = ""
unprocessed_messages: List[Dict[str, Any]] = field(default_factory=list)
unprocessed_messages: List[Dict[str, Any]] = field(default_factory=list) # 修改：明确类型为 Dict
active_users: Set[str] = field(default_factory=set)
# data
last_bot_speak_time: Optional[float] = None
last_user_speak_time: Optional[float] = None
last_message_time: Optional[float] = None
# 添加 last_message_id
last_message_id: Optional[str] = None
last_message_content: str = ""
last_message_sender: Optional[str] = None
bot_id: Optional[str] = None
chat_history_count: int = 0
new_messages_count: int = 0
cold_chat_duration: float = 0.0
cold_chat_start_time: Optional[float] = None # 用于计算冷场持续时间
cold_chat_duration: float = 0.0 # 缓存计算结果
# state
is_typing: bool = False
has_unread_messages: bool = False
is_typing: bool = False # 可能表示对方正在输入
# has_unread_messages: bool = False # 这个状态可以通过 new_messages_count > 0 判断
is_cold_chat: bool = False
changed: bool = False
changed: bool = False # 用于标记状态是否有变化，以便外部模块决定是否重新规划
# #spec
# #spec (暂时注释掉，如果不需要)
# meta_plan_trigger: bool = False
# --- 修改：移除 __post_init__ 的参数 ---
def __post_init__(self):
"""初始化后创建handler"""
self.chat_observer = None
self.handler = ObservationInfoHandler(self)
"""初始化后创建handler并进行必要的设置"""
self.chat_observer: Optional[ChatObserver] = None # 添加类型提示
self.handler = ObservationInfoHandler(self, self.private_name)
def bind_to_chat_observer(self, chat_observer: ChatObserver):
"""绑定到指定的chat_observer
Args:
stream_id: 聊天流ID
chat_observer: 要绑定的 ChatObserver 实例
"""
if self.chat_observer:
logger.warning(f"[私聊][{self.private_name}]尝试重复绑定 ChatObserver")
return
self.chat_observer = chat_observer
self.chat_observer.notification_manager.register_handler(
target="observation_info", notification_type=NotificationType.NEW_MESSAGE, handler=self.handler
)
self.chat_observer.notification_manager.register_handler(
target="observation_info", notification_type=NotificationType.COLD_CHAT, handler=self.handler
)
try:
# 注册关心的通知类型
self.chat_observer.notification_manager.register_handler(
target="observation_info", notification_type=NotificationType.NEW_MESSAGE, handler=self.handler
)
self.chat_observer.notification_manager.register_handler(
target="observation_info", notification_type=NotificationType.COLD_CHAT, handler=self.handler
)
# 可以根据需要注册更多通知类型
# self.chat_observer.notification_manager.register_handler(
# target="observation_info", notification_type=NotificationType.MESSAGE_DELETED, handler=self.handler
# )
logger.info(f"[私聊][{self.private_name}]成功绑定到 ChatObserver")
except Exception as e:
logger.error(f"[私聊][{self.private_name}]绑定到 ChatObserver 时出错: {e}")
self.chat_observer = None # 绑定失败，重置
def unbind_from_chat_observer(self):
"""解除与chat_observer的绑定"""
if self.chat_observer:
self.chat_observer.notification_manager.unregister_handler(
target="observation_info", notification_type=NotificationType.NEW_MESSAGE, handler=self.handler
)
self.chat_observer.notification_manager.unregister_handler(
target="observation_info", notification_type=NotificationType.COLD_CHAT, handler=self.handler
)
self.chat_observer = None
if self.chat_observer and hasattr(self.chat_observer, "notification_manager"): # 增加检查
try:
self.chat_observer.notification_manager.unregister_handler(
target="observation_info", notification_type=NotificationType.NEW_MESSAGE, handler=self.handler
)
self.chat_observer.notification_manager.unregister_handler(
target="observation_info", notification_type=NotificationType.COLD_CHAT, handler=self.handler
)
# 如果注册了其他类型，也要在这里注销
# self.chat_observer.notification_manager.unregister_handler(
# target="observation_info", notification_type=NotificationType.MESSAGE_DELETED, handler=self.handler
# )
logger.info(f"[私聊][{self.private_name}]成功从 ChatObserver 解绑")
except Exception as e:
logger.error(f"[私聊][{self.private_name}]从 ChatObserver 解绑时出错: {e}")
finally: # 确保 chat_observer 被重置
self.chat_observer = None
else:
logger.warning(f"[私聊][{self.private_name}]尝试解绑时 ChatObserver 不存在或无效")
def update_from_message(self, message: Dict[str, Any]):
# 修改：update_from_message 接收 UserInfo 对象
async def update_from_message(self, message: Dict[str, Any], user_info: Optional[UserInfo]):
"""从消息更新信息
Args:
message: 消息数据
message: 消息数据字典
user_info: 解析后的 UserInfo 对象 (可能为 None)
"""
# logger.debug(f"更新信息from_message: {message}")
self.last_message_time = message["time"]
self.last_message_id = message["message_id"]
message_time = message.get("time")
message_id = message.get("message_id")
processed_text = message.get("processed_plain_text", "")
self.last_message_content = message.get("processed_plain_text", "")
# 只有在新消息到达时才更新 last_message 相关信息
if message_time and message_time > (self.last_message_time or 0):
self.last_message_time = message_time
self.last_message_id = message_id
self.last_message_content = processed_text
# 重置冷场计时器
self.is_cold_chat = False
self.cold_chat_start_time = None
self.cold_chat_duration = 0.0
user_info = UserInfo.from_dict(message.get("user_info", {}))
self.last_message_sender = user_info.user_id
if user_info:
sender_id = str(user_info.user_id) # 确保是字符串
self.last_message_sender = sender_id
# 更新发言时间
if sender_id == self.bot_id:
self.last_bot_speak_time = message_time
else:
self.last_user_speak_time = message_time
self.active_users.add(sender_id) # 用户发言则认为其活跃
else:
logger.warning(
f"[私聊][{self.private_name}]处理消息更新时缺少有效的 UserInfo 对象, message_id: {message_id}"
)
self.last_message_sender = None # 发送者未知
if user_info.user_id == self.bot_id:
self.last_bot_speak_time = message["time"]
# 将原始消息字典添加到未处理列表
self.unprocessed_messages.append(message)
self.new_messages_count = len(self.unprocessed_messages) # 直接用列表长度
# logger.debug(f"[私聊][{self.private_name}]消息更新: last_time={self.last_message_time}, new_count={self.new_messages_count}")
self.update_changed() # 标记状态已改变
else:
self.last_user_speak_time = message["time"]
self.active_users.add(user_info.user_id)
self.new_messages_count += 1
self.unprocessed_messages.append(message)
self.update_changed()
# 如果消息时间戳不是最新的，可能不需要处理，或者记录一个警告
pass
# logger.warning(f"[私聊][{self.private_name}]收到过时或无效时间戳的消息: ID={message_id}, time={message_time}")
def update_changed(self):
"""更新changed状态"""
"""标记状态已改变，并重置标记"""
# logger.debug(f"[私聊][{self.private_name}]状态标记为已改变 (changed=True)")
self.changed = True
def update_cold_chat_status(self, is_cold: bool, current_time: float):
async def update_cold_chat_status(self, is_cold: bool, current_time: float):
"""更新冷场状态
Args:
is_cold: 是否冷场
current_time: 当前时间
is_cold: 是否处于冷场状态
current_time: 当前时间戳
"""
self.is_cold_chat = is_cold
if is_cold and self.last_message_time:
self.cold_chat_duration = current_time - self.last_message_time
if is_cold != self.is_cold_chat: # 仅在状态变化时更新
self.is_cold_chat = is_cold
if is_cold:
# 进入冷场状态
self.cold_chat_start_time = (
self.last_message_time or current_time
) # 从最后消息时间开始算，或从当前时间开始
logger.info(f"[私聊][{self.private_name}]进入冷场状态，开始时间: {self.cold_chat_start_time}")
else:
# 结束冷场状态
if self.cold_chat_start_time:
self.cold_chat_duration = current_time - self.cold_chat_start_time
logger.info(f"[私聊][{self.private_name}]结束冷场状态，持续时间: {self.cold_chat_duration:.2f} 秒")
self.cold_chat_start_time = None # 重置开始时间
self.update_changed() # 状态变化，标记改变
# 即使状态没变，如果是冷场状态，也更新持续时间
if self.is_cold_chat and self.cold_chat_start_time:
self.cold_chat_duration = current_time - self.cold_chat_start_time
def get_active_duration(self) -> float:
"""获取当前活跃时长
"""获取当前活跃时长 (距离最后一条消息的时间)
Returns:
float: 最后一条消息到现在的时长（秒）
@@ -204,7 +304,7 @@ class ObservationInfo:
return time.time() - self.last_message_time
def get_user_response_time(self) -> Optional[float]:
"""获取用户响应时间
"""获取用户最后响应时间 (距离用户最后发言的时间)
Returns:
Optional[float]: 用户最后发言到现在的时长（秒），如果没有用户发言则返回None
@@ -214,7 +314,7 @@ class ObservationInfo:
return time.time() - self.last_user_speak_time
def get_bot_response_time(self) -> Optional[float]:
"""获取机器人响应时间
"""获取机器人最后响应时间 (距离机器人最后发言的时间)
Returns:
Optional[float]: 机器人最后发言到现在的时长（秒），如果没有机器人发言则返回None
@@ -224,19 +324,38 @@ class ObservationInfo:
return time.time() - self.last_bot_speak_time
async def clear_unprocessed_messages(self):
"""清空未处理消息列表"""
# 将未处理消息添加到历史记录中
for message in self.unprocessed_messages:
self.chat_history.append(message)
self.chat_history_str = await build_readable_messages(
self.chat_history[-20:] if len(self.chat_history) > 20 else self.chat_history,
replace_bot_name=True,
merge_messages=False,
timestamp_mode="relative",
read_mark=0.0,
)
# 清空未处理消息列表
self.has_unread_messages = False
"""将未处理消息移入历史记录，并更新相关状态"""
if not self.unprocessed_messages:
return # 没有未处理消息，直接返回
# logger.debug(f"[私聊][{self.private_name}]处理 {len(self.unprocessed_messages)} 条未处理消息...")
# 将未处理消息添加到历史记录中 (确保历史记录有长度限制，避免无限增长)
max_history_len = 100 # 示例：最多保留100条历史记录
self.chat_history.extend(self.unprocessed_messages)
if len(self.chat_history) > max_history_len:
self.chat_history = self.chat_history[-max_history_len:]
# 更新历史记录字符串 (只使用最近一部分生成，例如20条)
history_slice_for_str = self.chat_history[-20:]
try:
self.chat_history_str = await build_readable_messages(
history_slice_for_str,
replace_bot_name=True,
merge_messages=False,
timestamp_mode="relative",
read_mark=0.0, # read_mark 可能需要根据逻辑调整
)
except Exception as e:
logger.error(f"[私聊][{self.private_name}]构建聊天记录字符串时出错: {e}")
self.chat_history_str = "[构建聊天记录出错]" # 提供错误提示
# 清空未处理消息列表和计数
# cleared_count = len(self.unprocessed_messages)
self.unprocessed_messages.clear()
self.chat_history_count = len(self.chat_history)
self.new_messages_count = 0
# self.has_unread_messages = False # 这个状态可以通过 new_messages_count 判断
self.chat_history_count = len(self.chat_history) # 更新历史记录总数
# logger.debug(f"[私聊][{self.private_name}]已处理 {cleared_count} 条消息，当前历史记录 {self.chat_history_count} 条。")
self.update_changed() # 状态改变

169
src/plugins/PFC/pfc.py
View File

@@ -1,24 +1,12 @@
# Programmable Friendly Conversationalist
# Prefrontal cortex
# import datetime
# import asyncio
from typing import List, Optional, Tuple, TYPE_CHECKING
from typing import List, Tuple, TYPE_CHECKING
from src.common.logger import get_module_logger
from ..chat.chat_stream import ChatStream
from maim_message import UserInfo, Seg
from ..chat.message import Message
from ..models.utils_model import LLMRequest
from ...config.config import global_config
from src.plugins.chat.message import MessageSending
from ..message.api import global_api
from ..storage.storage import MessageStorage
from .chat_observer import ChatObserver
from .pfc_utils import get_items_from_json
from src.individuality.individuality import Individuality
from .conversation_info import ConversationInfo
from .observation_info import ObservationInfo
import time
from src.plugins.utils.chat_message_builder import build_readable_messages
if TYPE_CHECKING:
@@ -30,7 +18,7 @@ logger = get_module_logger("pfc")
class GoalAnalyzer:
"""对话目标分析器"""
def __init__(self, stream_id: str):
def __init__(self, stream_id: str, private_name: str):
self.llm = LLMRequest(
model=global_config.llm_normal, temperature=0.7, max_tokens=1000, request_type="conversation_goal"
)
@@ -39,7 +27,8 @@ class GoalAnalyzer:
self.identity_detail_info = Individuality.get_instance().get_prompt(type="identity", x_person=2, level=2)
self.name = global_config.BOT_NICKNAME
self.nick_name = global_config.BOT_ALIAS_NAMES
self.chat_observer = ChatObserver.get_instance(stream_id)
self.private_name = private_name
self.chat_observer = ChatObserver.get_instance(stream_id, private_name)
# 多目标存储结构
self.goals = [] # 存储多个目标
@@ -60,16 +49,10 @@ class GoalAnalyzer:
goals_str = ""
if conversation_info.goal_list:
for goal_reason in conversation_info.goal_list:
# 处理字典或元组格式
if isinstance(goal_reason, tuple):
# 假设元组的第一个元素是目标，第二个元素是原因
goal = goal_reason[0]
reasoning = goal_reason[1] if len(goal_reason) > 1 else "没有明确原因"
elif isinstance(goal_reason, dict):
goal = goal_reason.get("goal")
if isinstance(goal_reason, dict):
goal = goal_reason.get("goal", "目标内容缺失")
reasoning = goal_reason.get("reasoning", "没有明确原因")
else:
# 如果是其他类型，尝试转为字符串
goal = str(goal_reason)
reasoning = "没有明确原因"
@@ -81,7 +64,7 @@ class GoalAnalyzer:
goals_str = f"目标：{goal}，产生该对话目标的原因：{reasoning}\n"
# 获取聊天历史记录
chat_history_text = observation_info.chat_history
chat_history_text = observation_info.chat_history_str
if observation_info.new_messages_count > 0:
new_messages_list = observation_info.unprocessed_messages
@@ -140,27 +123,32 @@ class GoalAnalyzer:
输出格式示例：
[
{{
{{
"goal": "回答用户关于Python编程的具体问题",
"reasoning": "用户提出了关于Python的技术问题，需要专业且准确的解答"
}},
{{
}},
{{
"goal": "回答用户关于python安装的具体问题",
"reasoning": "用户提出了关于Python的技术问题，需要专业且准确的解答"
}}
}}
]"""
logger.debug(f"发送到LLM的提示词: {prompt}")
logger.debug(f"[私聊][{self.private_name}]发送到LLM的提示词: {prompt}")
try:
content, _ = await self.llm.generate_response_async(prompt)
logger.debug(f"LLM原始返回内容: {content}")
logger.debug(f"[私聊][{self.private_name}]LLM原始返回内容: {content}")
except Exception as e:
logger.error(f"分析对话目标时出错: {str(e)}")
logger.error(f"[私聊][{self.private_name}]分析对话目标时出错: {str(e)}")
content = ""
# 使用改进后的get_items_from_json函数处理JSON数组
success, result = get_items_from_json(
content, "goal", "reasoning", required_types={"goal": str, "reasoning": str}, allow_array=True
content,
self.private_name,
"goal",
"reasoning",
required_types={"goal": str, "reasoning": str},
allow_array=True,
)
if success:
@@ -169,9 +157,7 @@ class GoalAnalyzer:
# 清空现有目标列表并添加新目标
conversation_info.goal_list = []
for item in result:
goal = item.get("goal", "")
reasoning = item.get("reasoning", "")
conversation_info.goal_list.append((goal, reasoning))
conversation_info.goal_list.append(item)
# 返回第一个目标作为当前主要目标（如果有）
if result:
@@ -179,9 +165,7 @@ class GoalAnalyzer:
return (first_goal.get("goal", ""), "", first_goal.get("reasoning", ""))
else:
# 单个目标的情况
goal = result.get("goal", "")
reasoning = result.get("reasoning", "")
conversation_info.goal_list.append((goal, reasoning))
conversation_info.goal_list.append(result)
return (goal, "", reasoning)
# 如果解析失败，返回默认值
@@ -296,11 +280,12 @@ class GoalAnalyzer:
try:
content, _ = await self.llm.generate_response_async(prompt)
logger.debug(f"LLM原始返回内容: {content}")
logger.debug(f"[私聊][{self.private_name}]LLM原始返回内容: {content}")
# 尝试解析JSON
success, result = get_items_from_json(
content,
self.private_name,
"goal_achieved",
"stop_conversation",
"reason",
@@ -308,7 +293,7 @@ class GoalAnalyzer:
)
if not success:
logger.error("无法解析对话分析结果JSON")
logger.error(f"[私聊][{self.private_name}]无法解析对话分析结果JSON")
return False, False, "解析结果失败"
goal_achieved = result["goal_achieved"]
@@ -318,65 +303,67 @@ class GoalAnalyzer:
return goal_achieved, stop_conversation, reason
except Exception as e:
logger.error(f"分析对话状态时出错: {str(e)}")
logger.error(f"[私聊][{self.private_name}]分析对话状态时出错: {str(e)}")
return False, False, f"分析出错: {str(e)}"
class DirectMessageSender:
"""直接发送消息到平台的发送器"""
# 先注释掉，万一以后出问题了还能开回来（（（
# class DirectMessageSender:
# """直接发送消息到平台的发送器"""
def __init__(self):
self.logger = get_module_logger("direct_sender")
self.storage = MessageStorage()
# def __init__(self, private_name: str):
# self.logger = get_module_logger("direct_sender")
# self.storage = MessageStorage()
# self.private_name = private_name
async def send_via_ws(self, message: MessageSending) -> None:
try:
await global_api.send_message(message)
except Exception as e:
raise ValueError(f"未找到平台：{message.message_info.platform} 的url配置，请检查配置文件") from e
# async def send_via_ws(self, message: MessageSending) -> None:
# try:
# await global_api.send_message(message)
# except Exception as e:
# raise ValueError(f"未找到平台：{message.message_info.platform} 的url配置，请检查配置文件") from e
async def send_message(
self,
chat_stream: ChatStream,
content: str,
reply_to_message: Optional[Message] = None,
) -> None:
"""直接发送消息到平台
# async def send_message(
# self,
# chat_stream: ChatStream,
# content: str,
# reply_to_message: Optional[Message] = None,
# ) -> None:
# """直接发送消息到平台
Args:
chat_stream: 聊天流
content: 消息内容
reply_to_message: 要回复的消息
"""
# 构建消息对象
message_segment = Seg(type="text", data=content)
bot_user_info = UserInfo(
user_id=global_config.BOT_QQ,
user_nickname=global_config.BOT_NICKNAME,
platform=chat_stream.platform,
)
# Args:
# chat_stream: 聊天流
# content: 消息内容
# reply_to_message: 要回复的消息
# """
# # 构建消息对象
# message_segment = Seg(type="text", data=content)
# bot_user_info = UserInfo(
# user_id=global_config.BOT_QQ,
# user_nickname=global_config.BOT_NICKNAME,
# platform=chat_stream.platform,
# )
message = MessageSending(
message_id=f"dm{round(time.time(), 2)}",
chat_stream=chat_stream,
bot_user_info=bot_user_info,
sender_info=reply_to_message.message_info.user_info if reply_to_message else None,
message_segment=message_segment,
reply=reply_to_message,
is_head=True,
is_emoji=False,
thinking_start_time=time.time(),
)
# message = MessageSending(
# message_id=f"dm{round(time.time(), 2)}",
# chat_stream=chat_stream,
# bot_user_info=bot_user_info,
# sender_info=reply_to_message.message_info.user_info if reply_to_message else None,
# message_segment=message_segment,
# reply=reply_to_message,
# is_head=True,
# is_emoji=False,
# thinking_start_time=time.time(),
# )
# 处理消息
await message.process()
# # 处理消息
# await message.process()
_message_json = message.to_dict()
# _message_json = message.to_dict()
# 发送消息
try:
await self.send_via_ws(message)
await self.storage.store_message(message, chat_stream)
logger.success(f"PFC消息已发送: {content}")
except Exception as e:
logger.error(f"PFC消息发送失败: {str(e)}")
# # 发送消息
# try:
# await self.send_via_ws(message)
# await self.storage.store_message(message, chat_stream)
# logger.success(f"[私聊][{self.private_name}]PFC消息已发送: {content}")
# except Exception as e:
# logger.error(f"[私聊][{self.private_name}]PFC消息发送失败: {str(e)}")

9
src/plugins/PFC/pfc_KnowledgeFetcher.py
View File

@@ -13,13 +13,14 @@ logger = get_module_logger("knowledge_fetcher")
class KnowledgeFetcher:
"""知识调取器"""
def __init__(self):
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:
"""获取相关知识
@@ -31,13 +32,13 @@ class KnowledgeFetcher:
str: 构造好的,带相关度的知识
"""
logger.debug("正在从LPMM知识库中获取知识")
logger.debug(f"[私聊][{self.private_name}]正在从LPMM知识库中获取知识")
try:
knowledge_info = qa_manager.get_knowledge(query)
logger.debug(f"LPMM知识库查询结果: {knowledge_info:150}")
logger.debug(f"[私聊][{self.private_name}]LPMM知识库查询结果: {knowledge_info:150}")
return knowledge_info
except Exception as e:
logger.error(f"LPMM知识库搜索工具执行失败: {str(e)}")
logger.error(f"[私聊][{self.private_name}]LPMM知识库搜索工具执行失败: {str(e)}")
return "未找到匹配的知识"
async def fetch(self, query: str, chat_history: List[Message]) -> Tuple[str, str]:

25
src/plugins/PFC/pfc_manager.py
View File

@@ -28,7 +28,7 @@ class PFCManager:
cls._instance = PFCManager()
return cls._instance
async def get_or_create_conversation(self, stream_id: str) -> Optional[Conversation]:
async def get_or_create_conversation(self, stream_id: str, private_name: str) -> Optional[Conversation]:
"""获取或创建对话实例
Args:
@@ -39,11 +39,11 @@ class PFCManager:
"""
# 检查是否已经有实例
if stream_id in self._initializing and self._initializing[stream_id]:
logger.debug(f"会话实例正在初始化中: {stream_id}")
logger.debug(f"[私聊][{private_name}]会话实例正在初始化中: {stream_id}")
return None
if stream_id in self._instances and self._instances[stream_id].should_continue:
logger.debug(f"使用现有会话实例: {stream_id}")
logger.debug(f"[私聊][{private_name}]使用现有会话实例: {stream_id}")
return self._instances[stream_id]
if stream_id in self._instances:
instance = self._instances[stream_id]
@@ -52,28 +52,28 @@ class PFCManager:
and instance.ignore_until_timestamp
and time.time() < instance.ignore_until_timestamp
):
logger.debug(f"会话实例当前处于忽略状态: {stream_id}")
logger.debug(f"[私聊][{private_name}]会话实例当前处于忽略状态: {stream_id}")
# 返回 None 阻止交互。或者可以返回实例但标记它被忽略了喵？
# 还是返回 None 吧喵。
return None
# 检查 should_continue 状态
if instance.should_continue:
logger.debug(f"使用现有会话实例: {stream_id}")
logger.debug(f"[私聊][{private_name}]使用现有会话实例: {stream_id}")
return instance
# else: 实例存在但不应继续
try:
# 创建新实例
logger.info(f"创建新的对话实例: {stream_id}")
logger.info(f"[私聊][{private_name}]创建新的对话实例: {stream_id}")
self._initializing[stream_id] = True
# 创建实例
conversation_instance = Conversation(stream_id)
conversation_instance = Conversation(stream_id, private_name)
self._instances[stream_id] = conversation_instance
# 启动实例初始化
await self._initialize_conversation(conversation_instance)
except Exception as e:
logger.error(f"创建会话实例失败: {stream_id}, 错误: {e}")
logger.error(f"[私聊][{private_name}]创建会话实例失败: {stream_id}, 错误: {e}")
return None
return conversation_instance
@@ -85,20 +85,21 @@ class PFCManager:
conversation: 要初始化的会话实例
"""
stream_id = conversation.stream_id
private_name = conversation.private_name
try:
logger.info(f"开始初始化会话实例: {stream_id}")
logger.info(f"[私聊][{private_name}]开始初始化会话实例: {stream_id}")
# 启动初始化流程
await conversation._initialize()
# 标记初始化完成
self._initializing[stream_id] = False
logger.info(f"会话实例 {stream_id} 初始化完成")
logger.info(f"[私聊][{private_name}]会话实例 {stream_id} 初始化完成")
except Exception as e:
logger.error(f"管理器初始化会话实例失败: {stream_id}, 错误: {e}")
logger.error(traceback.format_exc())
logger.error(f"[私聊][{private_name}]管理器初始化会话实例失败: {stream_id}, 错误: {e}")
logger.error(f"[私聊][{private_name}]{traceback.format_exc()}")
# 清理失败的初始化
async def get_conversation(self, stream_id: str) -> Optional[Conversation]:

15
src/plugins/PFC/pfc_utils.py
View File

@@ -8,6 +8,7 @@ logger = get_module_logger("pfc_utils")
def get_items_from_json(
content: str,
private_name: str,
*items: str,
default_values: Optional[Dict[str, Any]] = None,
required_types: Optional[Dict[str, type]] = None,
@@ -78,9 +79,9 @@ def get_items_from_json(
if valid_items:
return True, valid_items
except json.JSONDecodeError:
logger.debug("JSON数组解析失败，尝试解析单个JSON对象")
logger.debug(f"[私聊][{private_name}]JSON数组解析失败，尝试解析单个JSON对象")
except Exception as e:
logger.debug(f"尝试解析JSON数组时出错: {str(e)}")
logger.debug(f"[私聊][{private_name}]尝试解析JSON数组时出错: {str(e)}")
# 尝试解析JSON对象
try:
@@ -93,10 +94,10 @@ def get_items_from_json(
try:
json_data = json.loads(json_match.group())
except json.JSONDecodeError:
logger.error("提取的JSON内容解析失败")
logger.error(f"[私聊][{private_name}]提取的JSON内容解析失败")
return False, result
else:
logger.error("无法在返回内容中找到有效的JSON")
logger.error(f"[私聊][{private_name}]无法在返回内容中找到有效的JSON")
return False, result
# 提取字段
@@ -106,20 +107,20 @@ def get_items_from_json(
# 验证必需字段
if not all(item in result for item in items):
logger.error(f"JSON缺少必要字段，实际内容: {json_data}")
logger.error(f"[私聊][{private_name}]JSON缺少必要字段，实际内容: {json_data}")
return False, result
# 验证字段类型
if required_types:
for field, expected_type in required_types.items():
if field in result and not isinstance(result[field], expected_type):
logger.error(f"{field} 必须是 {expected_type.__name__} 类型")
logger.error(f"[私聊][{private_name}]{field} 必须是 {expected_type.__name__} 类型")
return False, result
# 验证字符串字段不为空
for field in items:
if isinstance(result[field], str) and not result[field].strip():
logger.error(f"{field} 不能为空")
logger.error(f"[私聊][{private_name}]{field} 不能为空")
return False, result
return True, result

21
src/plugins/PFC/reply_checker.py
View File

@@ -12,12 +12,13 @@ logger = get_module_logger("reply_checker")
class ReplyChecker:
"""回复检查器"""
def __init__(self, stream_id: str):
def __init__(self, stream_id: str, private_name: str):
self.llm = LLMRequest(
model=global_config.llm_PFC_reply_checker, temperature=0.50, max_tokens=1000, request_type="reply_check"
)
self.name = global_config.BOT_NICKNAME
self.chat_observer = ChatObserver.get_instance(stream_id)
self.private_name = private_name
self.chat_observer = ChatObserver.get_instance(stream_id, private_name)
self.max_retries = 3 # 最大重试次数
async def check(
@@ -49,7 +50,9 @@ class ReplyChecker:
# 可以用简单比较，或者更复杂的相似度库 (如 difflib)
# 简单比较：是否完全相同
if reply == bot_messages[0]: # 和最近一条完全一样
logger.warning(f"ReplyChecker 检测到回复与上一条 Bot 消息完全相同: '{reply}'")
logger.warning(
f"[私聊][{self.private_name}]ReplyChecker 检测到回复与上一条 Bot 消息完全相同: '{reply}'"
)
return (
False,
"回复内容与你上一条发言完全相同，请修改，可以选择深入话题或寻找其它话题或等待",
@@ -60,13 +63,13 @@ class ReplyChecker:
# 计算编辑距离相似度，ratio() 返回 0 到 1 之间的浮点数
similarity_ratio = difflib.SequenceMatcher(None, reply, bot_messages[0]).ratio()
logger.debug(f"ReplyChecker - 相似度: {similarity_ratio:.2f}")
logger.debug(f"[私聊][{self.private_name}]ReplyChecker - 相似度: {similarity_ratio:.2f}")
# 设置一个相似度阈值
similarity_threshold = 0.9
if similarity_ratio > similarity_threshold:
logger.warning(
f"ReplyChecker 检测到回复与上一条 Bot 消息高度相似 (相似度 {similarity_ratio:.2f}): '{reply}'"
f"[私聊][{self.private_name}]ReplyChecker 检测到回复与上一条 Bot 消息高度相似 (相似度 {similarity_ratio:.2f}): '{reply}'"
)
return (
False,
@@ -77,8 +80,8 @@ class ReplyChecker:
except Exception as e:
import traceback
logger.error(f"检查回复时出错: 类型={type(e)}, 值={e}")
logger.error(traceback.format_exc()) # 打印详细的回溯信息
logger.error(f"[私聊][{self.private_name}]检查回复时出错: 类型={type(e)}, 值={e}")
logger.error(f"[私聊][{self.private_name}]{traceback.format_exc()}") # 打印详细的回溯信息
prompt = f"""请检查以下回复或消息是否合适：
@@ -118,7 +121,7 @@ class ReplyChecker:
try:
content, _ = await self.llm.generate_response_async(prompt)
logger.debug(f"检查回复的原始返回: {content}")
logger.debug(f"[私聊][{self.private_name}]检查回复的原始返回: {content}")
# 清理内容，尝试提取JSON部分
content = content.strip()
@@ -171,7 +174,7 @@ class ReplyChecker:
return suitable, reason, need_replan
except Exception as e:
logger.error(f"检查回复时出错: {e}")
logger.error(f"[私聊][{self.private_name}]检查回复时出错: {e}")
# 如果出错且已达到最大重试次数，建议重新规划
if retry_count >= self.max_retries:
return False, "多次检查失败，建议重新规划", True

27
src/plugins/PFC/reply_generator.py
View File

@@ -61,7 +61,7 @@ PROMPT_SEND_NEW_MESSAGE = """{persona_text}。现在你在参与一场QQ私聊
class ReplyGenerator:
"""回复生成器"""
def __init__(self, stream_id: str):
def __init__(self, stream_id: str, private_name: str):
self.llm = LLMRequest(
model=global_config.llm_PFC_chat,
temperature=global_config.llm_PFC_chat["temp"],
@@ -71,8 +71,9 @@ class ReplyGenerator:
self.personality_info = Individuality.get_instance().get_prompt(type="personality", x_person=2, level=3)
self.identity_detail_info = Individuality.get_instance().get_prompt(type="identity", x_person=2, level=2)
self.name = global_config.BOT_NICKNAME
self.chat_observer = ChatObserver.get_instance(stream_id)
self.reply_checker = ReplyChecker(stream_id)
self.private_name = private_name
self.chat_observer = ChatObserver.get_instance(stream_id, private_name)
self.reply_checker = ReplyChecker(stream_id, private_name)
# 修改 generate 方法签名，增加 action_type 参数
async def generate(
@@ -89,7 +90,9 @@ class ReplyGenerator:
str: 生成的回复
"""
# 构建提示词
logger.debug(f"开始生成回复 (动作类型: {action_type})：当前目标: {conversation_info.goal_list}")
logger.debug(
f"[私聊][{self.private_name}]开始生成回复 (动作类型: {action_type})：当前目标: {conversation_info.goal_list}"
)
# --- 构建通用 Prompt 参数 ---
# (这部分逻辑基本不变)
@@ -98,15 +101,13 @@ class ReplyGenerator:
goals_str = ""
if conversation_info.goal_list:
for goal_reason in conversation_info.goal_list:
if isinstance(goal_reason, tuple):
goal = goal_reason[0] if len(goal_reason) > 0 else "目标内容缺失"
reasoning = goal_reason[1] if len(goal_reason) > 1 else "没有明确原因"
elif isinstance(goal_reason, dict):
if isinstance(goal_reason, dict):
goal = goal_reason.get("goal", "目标内容缺失")
reasoning = goal_reason.get("reasoning", "没有明确原因")
else:
goal = str(goal_reason)
reasoning = "没有明确原因"
goal = str(goal) if goal is not None else "目标内容缺失"
reasoning = str(reasoning) if reasoning is not None else "没有明确原因"
goals_str += f"- 目标：{goal}\n 原因：{reasoning}\n"
@@ -147,10 +148,10 @@ class ReplyGenerator:
# --- 选择 Prompt ---
if action_type == "send_new_message":
prompt_template = PROMPT_SEND_NEW_MESSAGE
logger.info("使用 PROMPT_SEND_NEW_MESSAGE (追问生成)")
logger.info(f"[私聊][{self.private_name}]使用 PROMPT_SEND_NEW_MESSAGE (追问生成)")
else: # 默认使用 direct_reply 的 prompt
prompt_template = PROMPT_DIRECT_REPLY
logger.info("使用 PROMPT_DIRECT_REPLY (首次/非连续回复生成)")
logger.info(f"[私聊][{self.private_name}]使用 PROMPT_DIRECT_REPLY (首次/非连续回复生成)")
# --- 格式化最终的 Prompt ---
prompt = prompt_template.format(
@@ -158,15 +159,15 @@ class ReplyGenerator:
)
# --- 调用 LLM 生成 ---
logger.debug(f"发送到LLM的生成提示词:\n------\n{prompt}\n------")
logger.debug(f"[私聊][{self.private_name}]发送到LLM的生成提示词:\n------\n{prompt}\n------")
try:
content, _ = await self.llm.generate_response_async(prompt)
logger.debug(f"生成的回复: {content}")
logger.debug(f"[私聊][{self.private_name}]生成的回复: {content}")
# 移除旧的检查新消息逻辑，这应该由 conversation 控制流处理
return content
except Exception as e:
logger.error(f"生成回复时出错: {e}")
logger.error(f"[私聊][{self.private_name}]生成回复时出错: {e}")
return "抱歉，我现在有点混乱，让我重新思考一下..."
# check_reply 方法保持不变

31
src/plugins/PFC/waiter.py
View File

@@ -17,60 +17,63 @@ DESIRED_TIMEOUT_SECONDS = 300
class Waiter:
"""等待处理类"""
def __init__(self, stream_id: str):
self.chat_observer = ChatObserver.get_instance(stream_id)
def __init__(self, stream_id: str, private_name: str):
self.chat_observer = ChatObserver.get_instance(stream_id, private_name)
self.name = global_config.BOT_NICKNAME
self.private_name = private_name
# self.wait_accumulated_time = 0 # 不再需要累加计时
async def wait(self, conversation_info: ConversationInfo) -> bool:
"""等待用户新消息或超时"""
wait_start_time = time.time()
logger.info(f"进入常规等待状态 (超时: {DESIRED_TIMEOUT_SECONDS} 秒)...")
logger.info(f"[私聊][{self.private_name}]进入常规等待状态 (超时: {DESIRED_TIMEOUT_SECONDS} 秒)...")
while True:
# 检查是否有新消息
if self.chat_observer.new_message_after(wait_start_time):
logger.info("等待结束，收到新消息")
logger.info(f"[私聊][{self.private_name}]等待结束，收到新消息")
return False # 返回 False 表示不是超时
# 检查是否超时
elapsed_time = time.time() - wait_start_time
if elapsed_time > DESIRED_TIMEOUT_SECONDS:
logger.info(f"等待超过 {DESIRED_TIMEOUT_SECONDS} 秒...添加思考目标。")
logger.info(f"[私聊][{self.private_name}]等待超过 {DESIRED_TIMEOUT_SECONDS} 秒...添加思考目标。")
wait_goal = {
"goal": f"你等待了{elapsed_time / 60:.1f}分钟，注意可能在对方看来聊天已经结束，思考接下来要做什么",
"reason": "对方很久没有回复你的消息了",
"reasoning": "对方很久没有回复你的消息了",
}
conversation_info.goal_list.append(wait_goal)
logger.info(f"添加目标: {wait_goal}")
logger.info(f"[私聊][{self.private_name}]添加目标: {wait_goal}")
return True # 返回 True 表示超时
await asyncio.sleep(5) # 每 5 秒检查一次
logger.debug("等待中...") # 可以考虑把这个频繁日志注释掉，只在超时或收到消息时输出
logger.debug(
f"[私聊][{self.private_name}]等待中..."
) # 可以考虑把这个频繁日志注释掉，只在超时或收到消息时输出
async def wait_listening(self, conversation_info: ConversationInfo) -> bool:
"""倾听用户发言或超时"""
wait_start_time = time.time()
logger.info(f"进入倾听等待状态 (超时: {DESIRED_TIMEOUT_SECONDS} 秒)...")
logger.info(f"[私聊][{self.private_name}]进入倾听等待状态 (超时: {DESIRED_TIMEOUT_SECONDS} 秒)...")
while True:
# 检查是否有新消息
if self.chat_observer.new_message_after(wait_start_time):
logger.info("倾听等待结束，收到新消息")
logger.info(f"[私聊][{self.private_name}]倾听等待结束，收到新消息")
return False # 返回 False 表示不是超时
# 检查是否超时
elapsed_time = time.time() - wait_start_time
if elapsed_time > DESIRED_TIMEOUT_SECONDS:
logger.info(f"倾听等待超过 {DESIRED_TIMEOUT_SECONDS} 秒...添加思考目标。")
logger.info(f"[私聊][{self.private_name}]倾听等待超过 {DESIRED_TIMEOUT_SECONDS} 秒...添加思考目标。")
wait_goal = {
# 保持 goal 文本一致
"goal": f"你等待了{elapsed_time / 60:.1f}分钟，对方似乎话说一半突然消失了，可能忙去了？也可能忘记了回复？要问问吗？还是结束对话？或继续等待？思考接下来要做什么",
"reason": "对方话说一半消失了，很久没有回复",
"reasoning": "对方话说一半消失了，很久没有回复",
}
conversation_info.goal_list.append(wait_goal)
logger.info(f"添加目标: {wait_goal}")
logger.info(f"[私聊][{self.private_name}]添加目标: {wait_goal}")
return True # 返回 True 表示超时
await asyncio.sleep(5) # 每 5 秒检查一次
logger.debug("倾听等待中...") # 同上，可以考虑注释掉
logger.debug(f"[私聊][{self.private_name}]倾听等待中...") # 同上，可以考虑注释掉

3
src/plugins/chat/bot.py
View File

@@ -38,9 +38,10 @@ class ChatBot:
async def _create_pfc_chat(self, message: MessageRecv):
try:
chat_id = str(message.chat_stream.stream_id)
private_name = str(message.message_info.user_info.user_nickname)
if global_config.enable_pfc_chatting:
await self.pfc_manager.get_or_create_conversation(chat_id)
await self.pfc_manager.get_or_create_conversation(chat_id, private_name)
except Exception as e:
logger.error(f"创建PFC聊天失败: {e}")