BruceSun
/
MaiBot
mirror of https://github.com/Mai-with-u/MaiBot.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

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

pull/937/head
github-actions[bot] 2025-05-09 03:50:04 +00:00
parent 84edc85413
commit 200fa96702
3 changed files with 148 additions and 104 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -6,7 +6,7 @@ from typing import Dict, Any, Optional
from src.common.logger_manager import get_logger
from maim_message import UserInfo
from src.plugins.chat.chat_stream import chat_manager, ChatStream
from ..chat.message import Message # 假设 Message 类型被 _convert_to_message 使用
from ..chat.message import Message # 假设 Message 类型被 _convert_to_message 使用
from src.config.config import global_config
from ..person_info.person_info import person_info_manager
from ..person_info.relationship_manager import relationship_manager
@ -28,7 +28,7 @@ from .PFC_idle.idle_chat import IdleChat
from .waiter import Waiter
from .reply_checker import ReplyChecker
from .conversation_loop import run_conversation_loop
from .conversation_loop import run_conversation_loop
from rich.traceback import install
@ -50,7 +50,7 @@ class Conversation:
self.stream_id: str = stream_id
self.private_name: str = private_name
self.state: ConversationState = ConversationState.INIT
self.should_continue: bool = False
self.should_continue: bool = False
self.ignore_until_timestamp: Optional[float] = None
self.generated_reply: str = ""
self.chat_stream: Optional[ChatStream] = None
@ -73,15 +73,14 @@ class Conversation:
self.conversation_info: Optional[ConversationInfo] = None
self.reply_checker: Optional[ReplyChecker] = None
self._initialized: bool = False
self._initialized: bool = False
self.bot_qq_str: Optional[str] = str(global_config.BOT_QQ) if global_config.BOT_QQ else None
if not self.bot_qq_str:
logger.error(f"[私聊][{self.private_name}] 严重错误：未能从配置中获取 BOT_QQ ID！")
#确保这个属性被正确初始化
self.consecutive_llm_action_failures: int = 0 # LLM相关动作连续失败的计数器
# 确保这个属性被正确初始化
self.consecutive_llm_action_failures: int = 0 # LLM相关动作连续失败的计数器
async def start(self):
"""
@ -102,22 +101,22 @@ class Conversation:
logger.info(f"[私聊][{self.private_name}] 规划循环任务已创建。")
except Exception as task_err:
logger.error(f"[私聊][{self.private_name}] 创建规划循环任务时出错: {task_err}")
await self.stop() # 发生错误时尝试停止
await self.stop() # 发生错误时尝试停止
async def stop(self):
"""
停止对话实例并清理相关资源。
"""
logger.info(f"[私聊][{self.private_name}] 正在停止对话实例: {self.stream_id}")
self.should_continue = False # 设置标志以退出循环
self.should_continue = False # 设置标志以退出循环
# 最终关系评估
if (
self._initialized # 确保已初始化
self._initialized # 确保已初始化
and self.relationship_updater
and self.conversation_info
and self.observation_info
and self.chat_observer # 确保所有需要的组件都存在
and self.chat_observer # 确保所有需要的组件都存在
):
try:
logger.info(f"[私聊][{self.private_name}] 准备执行最终关系评估...")
@ -135,18 +134,17 @@ class Conversation:
# 停止其他组件
if self.idle_chat:
await self.idle_chat.decrement_active_instances() # 减少活跃实例计数
await self.idle_chat.decrement_active_instances() # 减少活跃实例计数
logger.debug(f"[私聊][{self.private_name}] 已减少IdleChat活跃实例计数")
if self.observation_info and self.chat_observer: # 确保二者都存在
self.observation_info.unbind_from_chat_observer() # 解绑
if self.mood_mng and hasattr(self.mood_mng, "stop_mood_update") and self.mood_mng._running: # type: ignore
self.mood_mng.stop_mood_update() # type: ignore
if self.observation_info and self.chat_observer: # 确保二者都存在
self.observation_info.unbind_from_chat_observer() # 解绑
if self.mood_mng and hasattr(self.mood_mng, "stop_mood_update") and self.mood_mng._running: # type: ignore
self.mood_mng.stop_mood_update() # type: ignore
logger.debug(f"[私聊][{self.private_name}] MoodManager 后台更新已停止。")
self._initialized = False # 标记为未初始化
self._initialized = False # 标记为未初始化
logger.info(f"[私聊][{self.private_name}] 对话实例 {self.stream_id} 已停止。")
def _convert_to_message(self, msg_dict: Dict[str, Any]) -> Optional[Message]:
"""将从数据库或其他来源获取的消息字典转换为内部使用的 Message 对象"""
# 这个方法似乎没有被其他内部方法调用，但为了完整性暂时保留
@ -169,7 +167,7 @@ class Conversation:
logger.warning(
f"[私聊][{self.private_name}] 从字典创建 UserInfo 时出错: {e}, dict: {user_info_dict}"
)
if not user_info: # 如果没有有效的 UserInfo，则无法创建 Message 对象
if not user_info: # 如果没有有效的 UserInfo，则无法创建 Message 对象
logger.warning(
f"[私聊][{self.private_name}] 消息缺少有效的 UserInfo，无法转换。 msg_id: {msg_dict.get('message_id')}"
)
@ -177,14 +175,14 @@ class Conversation:
# 创建并返回 Message 对象
return Message(
message_id=msg_dict.get("message_id", f"gen_{time.time()}"), # 提供默认 message_id
message_id=msg_dict.get("message_id", f"gen_{time.time()}"), # 提供默认 message_id
chat_stream=chat_stream_to_use,
time=msg_dict.get("time", time.time()), # 提供默认时间
time=msg_dict.get("time", time.time()), # 提供默认时间
user_info=user_info,
processed_plain_text=msg_dict.get("processed_plain_text", ""), # 提供默认文本
detailed_plain_text=msg_dict.get("detailed_plain_text", ""), # 提供默认详细文本
processed_plain_text=msg_dict.get("processed_plain_text", ""), # 提供默认文本
detailed_plain_text=msg_dict.get("detailed_plain_text", ""), # 提供默认详细文本
)
except Exception as e:
logger.error(f"[私聊][{self.private_name}] 转换消息时出错: {e}")
logger.error(f"[私聊][{self.private_name}] {traceback.format_exc()}")
return None # 出错时返回 None
return None # 出错时返回 None

203
src/plugins/PFC/conversation_loop.py
View File

@ -7,8 +7,8 @@ from dateutil import tz
from src.common.logger_manager import get_logger
from src.config.config import global_config
from .pfc_types import ConversationState # 需要导入 ConversationState
from . import actions # 需要导入 actions 模块
from .pfc_types import ConversationState # 需要导入 ConversationState
from . import actions # 需要导入 actions 模块
if TYPE_CHECKING:
from .conversation import Conversation
@ -22,7 +22,8 @@ if TIME_ZONE is None:
logger.error(f"配置的时区 '{configured_tz}' 无效，将使用默认时区 'Asia/Shanghai'")
TIME_ZONE = tz.gettz("Asia/Shanghai")
MAX_CONSECUTIVE_LLM_ACTION_FAILURES = 3 # 可配置的最大LLM连续失败次数
MAX_CONSECUTIVE_LLM_ACTION_FAILURES = 3 # 可配置的最大LLM连续失败次数
async def run_conversation_loop(conversation_instance: "Conversation"):
"""
@ -39,9 +40,11 @@ async def run_conversation_loop(conversation_instance: "Conversation"):
while conversation_instance.should_continue:
loop_iter_start_time = time.time()
current_force_reflect_and_act = _force_reflect_and_act_next_iter
_force_reflect_and_act_next_iter = False
_force_reflect_and_act_next_iter = False
logger.debug(f"[私聊][{conversation_instance.private_name}] 开始新一轮循环迭代 ({loop_iter_start_time:.2f}), force_reflect_next_iter: {current_force_reflect_and_act}, consecutive_llm_failures: {conversation_instance.consecutive_llm_action_failures}")
logger.debug(
f"[私聊][{conversation_instance.private_name}] 开始新一轮循环迭代 ({loop_iter_start_time:.2f}), force_reflect_next_iter: {current_force_reflect_and_act}, consecutive_llm_failures: {conversation_instance.consecutive_llm_action_failures}"
)
try:
global TIME_ZONE
@ -100,6 +103,7 @@ async def run_conversation_loop(conversation_instance: "Conversation"):
)
if not isinstance(numeric_relationship_value, (int, float)):
from bson.decimal128 import Decimal128
if isinstance(numeric_relationship_value, Decimal128):
numeric_relationship_value = float(numeric_relationship_value.to_decimal())
else:
@ -150,7 +154,9 @@ async def run_conversation_loop(conversation_instance: "Conversation"):
f"[私聊][{conversation_instance.private_name}] (Loop) ActionPlanner.plan 完成，初步规划动作: {action}"
)
current_unprocessed_messages_after_plan = getattr(conversation_instance.observation_info, "unprocessed_messages", [])
current_unprocessed_messages_after_plan = getattr(
conversation_instance.observation_info, "unprocessed_messages", []
)
new_messages_during_action_planning: List[Dict[str, Any]] = []
other_new_messages_during_action_planning: List[Dict[str, Any]] = []
@ -162,21 +168,23 @@ async def run_conversation_loop(conversation_instance: "Conversation"):
new_messages_during_action_planning.append(msg_ap)
if sender_id_ap != conversation_instance.bot_qq_str:
other_new_messages_during_action_planning.append(msg_ap)
new_msg_count_action_planning = len(new_messages_during_action_planning)
other_new_msg_count_action_planning = len(other_new_messages_during_action_planning)
if conversation_instance.conversation_info and other_new_msg_count_action_planning > 0:
pass
pass
should_interrupt_action_planning: bool = False
interrupt_reason_action_planning: str = ""
if action in ["wait", "listening"] and new_msg_count_action_planning > 0:
should_interrupt_action_planning = True
interrupt_reason_action_planning = f"规划 {action} 期间收到 {new_msg_count_action_planning} 条新消息"
elif other_new_msg_count_action_planning > 2:
elif other_new_msg_count_action_planning > 2:
should_interrupt_action_planning = True
interrupt_reason_action_planning = f"规划 {action} 期间收到 {other_new_msg_count_action_planning} 条来自他人的新消息"
interrupt_reason_action_planning = (
f"规划 {action} 期间收到 {other_new_msg_count_action_planning} 条来自他人的新消息"
)
if should_interrupt_action_planning:
logger.info(
@ -190,7 +198,10 @@ async def run_conversation_loop(conversation_instance: "Conversation"):
"final_reason": interrupt_reason_action_planning,
}
if conversation_instance.conversation_info:
if not hasattr(conversation_instance.conversation_info, "done_action") or conversation_instance.conversation_info.done_action is None:
if (
not hasattr(conversation_instance.conversation_info, "done_action")
or conversation_instance.conversation_info.done_action is None
):
conversation_instance.conversation_info.done_action = []
conversation_instance.conversation_info.done_action.append(cancel_record_ap)
conversation_instance.conversation_info.last_successful_reply_action = None
@ -204,9 +215,11 @@ async def run_conversation_loop(conversation_instance: "Conversation"):
f"[私聊][{conversation_instance.private_name}] (Loop) 动作 '{action}' 需要LLM生成，进入监控执行模式..."
)
llm_call_start_time = time.time()
if conversation_instance.conversation_info:
conversation_instance.conversation_info.other_new_messages_during_planning_count = other_new_msg_count_action_planning
conversation_instance.conversation_info.other_new_messages_during_planning_count = (
other_new_msg_count_action_planning
)
llm_action_task = asyncio.create_task(
actions.handle_action(
@ -219,126 +232,156 @@ async def run_conversation_loop(conversation_instance: "Conversation"):
)
interrupted_by_new_messages = False
llm_action_completed_successfully = False
action_outcome_processed = False # Flag to ensure we process outcome only once
llm_action_completed_successfully = False
action_outcome_processed = False # Flag to ensure we process outcome only once
while not llm_action_task.done() and not action_outcome_processed:
try:
# Shield the task so wait_for timeout doesn't cancel it directly
await asyncio.wait_for(asyncio.shield(llm_action_task), timeout=1.5)
# If wait_for completes without timeout, the shielded task is done (or errored/cancelled by itself)
action_outcome_processed = True # Outcome will be processed outside this loop
action_outcome_processed = True # Outcome will be processed outside this loop
except asyncio.TimeoutError:
# Shielded task didn't finish in 1.5s. This is our chance to check messages.
current_time_for_check = time.time()
logger.debug(f"[私聊][{conversation_instance.private_name}] (Loop) LLM Monitor polling. llm_call_start_time: {llm_call_start_time:.2f}, current_check_time: {current_time_for_check:.2f}. Task still running, checking for new messages.")
current_unprocessed_messages_during_llm = getattr(conversation_instance.observation_info, "unprocessed_messages", [])
logger.debug(
f"[私聊][{conversation_instance.private_name}] (Loop) LLM Monitor polling. llm_call_start_time: {llm_call_start_time:.2f}, current_check_time: {current_time_for_check:.2f}. Task still running, checking for new messages."
)
current_unprocessed_messages_during_llm = getattr(
conversation_instance.observation_info, "unprocessed_messages", []
)
other_new_messages_this_check: List[Dict[str, Any]] = []
logger.debug(f"[私聊][{conversation_instance.private_name}] (Loop) Checking unprocessed_messages (count: {len(current_unprocessed_messages_during_llm)}):")
logger.debug(
f"[私聊][{conversation_instance.private_name}] (Loop) Checking unprocessed_messages (count: {len(current_unprocessed_messages_during_llm)}):"
)
for msg_llm in current_unprocessed_messages_during_llm:
msg_time_llm = msg_llm.get("time")
sender_id_info_llm = msg_llm.get("user_info", {})
sender_id_llm = str(sender_id_info_llm.get("user_id")) if sender_id_info_llm else None
is_new_enough = msg_time_llm and msg_time_llm >= llm_call_start_time
is_other_sender = sender_id_llm != conversation_instance.bot_qq_str
time_str_for_log = f"{msg_time_llm:.2f}" if msg_time_llm is not None else "N/A"
logger.debug(f" - Msg ID: {msg_llm.get('message_id')}, Time: {time_str_for_log}, Sender: {sender_id_llm}. New enough? {is_new_enough}. Other sender? {is_other_sender}.")
logger.debug(
f" - Msg ID: {msg_llm.get('message_id')}, Time: {time_str_for_log}, Sender: {sender_id_llm}. New enough? {is_new_enough}. Other sender? {is_other_sender}."
)
if is_new_enough and is_other_sender:
other_new_messages_this_check.append(msg_llm)
logger.debug(f"[私聊][{conversation_instance.private_name}] (Loop) Found {len(other_new_messages_this_check)} 'other_new_messages_this_check'.")
logger.debug(
f"[私聊][{conversation_instance.private_name}] (Loop) Found {len(other_new_messages_this_check)} 'other_new_messages_this_check'."
)
if len(other_new_messages_this_check) > 2:
logger.info(
f"[私聊][{conversation_instance.private_name}] (Loop) LLM动作 '{action}' 执行期间收到 {len(other_new_messages_this_check)} 条来自他人的新消息，将取消LLM任务。"
)
if not llm_action_task.done(): # Check again before cancelling
llm_action_task.cancel() # Now we explicitly cancel the original task
interrupted_by_new_messages = True
action_outcome_processed = True # We've made a decision, exit monitoring
if not llm_action_task.done(): # Check again before cancelling
llm_action_task.cancel() # Now we explicitly cancel the original task
interrupted_by_new_messages = True
action_outcome_processed = True # We've made a decision, exit monitoring
# else: continue polling if not enough new messages
# Shield ensures CancelledError from llm_action_task itself is caught by the outer try/except
# After the monitoring loop (either task finished, or we decided to cancel it)
# Await the task properly to get its result or handle its exception/cancellation
action_final_status_in_history = "unknown"
try:
await llm_action_task # This will re-raise CancelledError if we cancelled it, or other exceptions
await llm_action_task # This will re-raise CancelledError if we cancelled it, or other exceptions
# If no exception, it means the task completed.
# actions.handle_action updates done_action, so we check its status.
if conversation_instance.conversation_info and conversation_instance.conversation_info.done_action:
# Check if done_action is not empty
if conversation_instance.conversation_info.done_action:
action_final_status_in_history = conversation_instance.conversation_info.done_action[-1].get("status", "unknown")
action_final_status_in_history = conversation_instance.conversation_info.done_action[
-1
].get("status", "unknown")
if action_final_status_in_history in ["done", "done_no_reply"]:
logger.debug(f"[私聊][{conversation_instance.private_name}] (Loop) LLM动作 '{action}' 任务最终成功完成 (status: {action_final_status_in_history})。")
conversation_instance.consecutive_llm_action_failures = 0
logger.debug(
f"[私聊][{conversation_instance.private_name}] (Loop) LLM动作 '{action}' 任务最终成功完成 (status: {action_final_status_in_history})。"
)
conversation_instance.consecutive_llm_action_failures = 0
llm_action_completed_successfully = True
else:
logger.warning(f"[私聊][{conversation_instance.private_name}] (Loop) LLM动作 '{action}' 任务完成但未成功 (status: {action_final_status_in_history})。")
if not interrupted_by_new_messages:
conversation_instance.consecutive_llm_action_failures += 1
logger.warning(
f"[私聊][{conversation_instance.private_name}] (Loop) LLM动作 '{action}' 任务完成但未成功 (status: {action_final_status_in_history})。"
)
if not interrupted_by_new_messages:
conversation_instance.consecutive_llm_action_failures += 1
except asyncio.CancelledError:
logger.info(f"[私聊][{conversation_instance.private_name}] (Loop) LLM动作 '{action}' 任务最终确认被取消。")
if not interrupted_by_new_messages:
conversation_instance.consecutive_llm_action_failures += 1
logger.warning(f"[私聊][{conversation_instance.private_name}] (Loop) LLM任务因外部原因取消，连续失败次数: {conversation_instance.consecutive_llm_action_failures}")
else: # interrupted_by_new_messages is True
logger.info(f"[私聊][{conversation_instance.private_name}] (Loop) LLM任务因新消息被内部逻辑取消，不计为LLM失败。")
except Exception as e_llm_final:
logger.error(f"[私聊][{conversation_instance.private_name}] (Loop) LLM动作 '{action}' 任务执行时发生最终错误: {e_llm_final}")
logger.error(traceback.format_exc())
conversation_instance.state = ConversationState.ERROR
logger.info(
f"[私聊][{conversation_instance.private_name}] (Loop) LLM动作 '{action}' 任务最终确认被取消。"
)
if not interrupted_by_new_messages:
conversation_instance.consecutive_llm_action_failures += 1
logger.warning(
f"[私聊][{conversation_instance.private_name}] (Loop) LLM任务因外部原因取消，连续失败次数: {conversation_instance.consecutive_llm_action_failures}"
)
else: # interrupted_by_new_messages is True
logger.info(
f"[私聊][{conversation_instance.private_name}] (Loop) LLM任务因新消息被内部逻辑取消，不计为LLM失败。"
)
except Exception as e_llm_final:
logger.error(
f"[私聊][{conversation_instance.private_name}] (Loop) LLM动作 '{action}' 任务执行时发生最终错误: {e_llm_final}"
)
logger.error(traceback.format_exc())
conversation_instance.state = ConversationState.ERROR
if not interrupted_by_new_messages:
conversation_instance.consecutive_llm_action_failures += 1
# --- Post LLM Action Task Handling ---
if not llm_action_completed_successfully:
if not llm_action_completed_successfully:
if conversation_instance.consecutive_llm_action_failures >= MAX_CONSECUTIVE_LLM_ACTION_FAILURES:
logger.error(f"[私聊][{conversation_instance.private_name}] (Loop) LLM相关动作连续失败或被取消 {conversation_instance.consecutive_llm_action_failures} 次。将强制等待并重置计数器。")
action = "wait" # Force action to wait
logger.error(
f"[私聊][{conversation_instance.private_name}] (Loop) LLM相关动作连续失败或被取消 {conversation_instance.consecutive_llm_action_failures} 次。将强制等待并重置计数器。"
)
action = "wait" # Force action to wait
reason = f"LLM连续失败{conversation_instance.consecutive_llm_action_failures}次，强制等待"
conversation_instance.consecutive_llm_action_failures = 0
conversation_instance.consecutive_llm_action_failures = 0
if conversation_instance.conversation_info:
conversation_instance.conversation_info.last_successful_reply_action = None
conversation_instance.conversation_info.last_successful_reply_action = None
logger.info(f"[私聊][{conversation_instance.private_name}] (Loop) 执行强制等待动作...")
await actions.handle_action(
conversation_instance,
action,
reason,
action,
reason,
conversation_instance.observation_info,
conversation_instance.conversation_info,
)
_force_reflect_and_act_next_iter = False
conversation_instance.state = ConversationState.ANALYZING
await asyncio.sleep(1)
continue
else:
_force_reflect_and_act_next_iter = False
conversation_instance.state = ConversationState.ANALYZING
logger.info(f"[私聊][{conversation_instance.private_name}] (Loop) LLM动作中断/失败，准备重新规划。Interrupted by new msgs: {interrupted_by_new_messages}, Consecutive LLM Failures: {conversation_instance.consecutive_llm_action_failures}")
await asyncio.sleep(1)
continue
else:
conversation_instance.state = ConversationState.ANALYZING
logger.info(
f"[私聊][{conversation_instance.private_name}] (Loop) LLM动作中断/失败，准备重新规划。Interrupted by new msgs: {interrupted_by_new_messages}, Consecutive LLM Failures: {conversation_instance.consecutive_llm_action_failures}"
)
await asyncio.sleep(0.1)
continue
else:
else:
logger.debug(f"[私聊][{conversation_instance.private_name}] (Loop) 执行非LLM类动作 '{action}'...")
conversation_instance.consecutive_llm_action_failures = 0
logger.debug(
f"[私聊][{conversation_instance.private_name}] (Loop) 执行非LLM类动作 '{action}'..."
f"[私聊][{conversation_instance.private_name}] (Loop) 重置 consecutive_llm_action_failures due to non-LLM action."
)
conversation_instance.consecutive_llm_action_failures = 0
logger.debug(f"[私聊][{conversation_instance.private_name}] (Loop) 重置 consecutive_llm_action_failures due to non-LLM action.")
if conversation_instance.conversation_info:
conversation_instance.conversation_info.other_new_messages_during_planning_count = other_new_msg_count_action_planning
conversation_instance.conversation_info.other_new_messages_during_planning_count = (
other_new_msg_count_action_planning
)
await actions.handle_action(
conversation_instance,
action,
@ -352,12 +395,14 @@ async def run_conversation_loop(conversation_instance: "Conversation"):
if conversation_instance.conversation_info and conversation_instance.conversation_info.done_action:
if conversation_instance.conversation_info.done_action:
last_action_record = conversation_instance.conversation_info.done_action[-1]
if (
last_action_record.get("action") == "send_new_message"
and last_action_record.get("status") == "done_no_reply"
and last_action_record.get("status") == "done_no_reply"
):
logger.info(f"[私聊][{conversation_instance.private_name}] (Loop) 检测到 ReplyGenerator 决定不发送消息，下一轮将强制反思。")
logger.info(
f"[私聊][{conversation_instance.private_name}] (Loop) 检测到 ReplyGenerator 决定不发送消息，下一轮将强制反思。"
)
_force_reflect_and_act_next_iter = True
goal_ended: bool = False
@ -375,9 +420,9 @@ async def run_conversation_loop(conversation_instance: "Conversation"):
if current_goal == "结束对话":
goal_ended = True
last_action_record_for_end_check = {}
last_action_record_for_end_check = {}
if conversation_instance.conversation_info and conversation_instance.conversation_info.done_action:
if conversation_instance.conversation_info.done_action:
if conversation_instance.conversation_info.done_action:
last_action_record_for_end_check = conversation_instance.conversation_info.done_action[-1]
action_ended: bool = (

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

@ -343,6 +343,7 @@ async def adjust_relationship_value_nonlinear(old_value: float, raw_adjustment:
return value
async def build_chat_history_text(observation_info: ObservationInfo, private_name: str) -> str:
"""构建聊天历史记录文本 (包含未处理消息)"""
@ -380,4 +381,4 @@ async def build_chat_history_text(observation_info: ObservationInfo, private_nam
except Exception as e:
logger.error(f"[私聊][{private_name}] 处理聊天记录时发生未知错误: {e}")
chat_history_text = "[处理聊天记录时出错]\n"
return chat_history_text
return chat_history_text