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

为PFC的输出添加容易区分的聊天流标识

Browse Source
This commit is contained in:
Bakadax
2025-04-28 16:41:39 +08:00
parent 3b88a35a30
commit ef24f013de
12 changed files with 193 additions and 182 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -51,35 +51,35 @@ class Conversation:
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.knowledge_fetcher = KnowledgeFetcher(self.private_name)
self.waiter = Waiter(self.stream_id, self.private_name)
self.direct_sender = DirectMessageSender()
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.bind_to_chat_observer(self.chat_observer)
self.observation_info = ObservationInfo(self.private_name)
self.observation_info.bind_to_chat_observer(self.chat_observer, self.private_name)
# 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(),
@@ -106,17 +106,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
@@ -125,10 +125,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):
@@ -139,7 +139,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
@@ -149,7 +149,7 @@ 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
@@ -162,11 +162,11 @@ 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
@@ -176,12 +176,12 @@ 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)
@@ -198,34 +198,34 @@ 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
@@ -254,7 +254,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
@@ -263,7 +263,7 @@ class Conversation:
):
"""处理规划的行动"""
logger.debug(f"执行行动: {action}, 原因: {reason}")
logger.debug(f"[私聊][{self.private_name}]执行行动: {action}, 原因: {reason}")
# 记录action历史 (逻辑不变)
current_action_record = {
@@ -294,14 +294,14 @@ 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
@@ -315,18 +315,18 @@ 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
@@ -334,7 +334,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}"}
)
@@ -351,7 +351,7 @@ 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}"}
)
@@ -359,7 +359,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 = {
@@ -381,14 +381,14 @@ 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
@@ -402,18 +402,18 @@ 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
@@ -421,7 +421,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}"}
)
@@ -438,7 +438,7 @@ 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}"}
)
@@ -446,7 +446,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 = {
@@ -464,11 +464,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"):
@@ -478,7 +478,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}"}
)
@@ -489,12 +489,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}"}
)
@@ -502,16 +502,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}"}
)
@@ -519,29 +519,29 @@ 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}"}
)
@@ -559,13 +559,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:
@@ -574,10 +574,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)
@@ -587,13 +587,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):
@@ -608,4 +608,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)}")