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

docs(core): 完善睡眠与主动思考模块的文档注释

Browse Source 
为 `ProactiveThinker`, `SleepManager`, 和 `SleepState` 核心类及其方法添加了详尽的中文文档字符串和内联注释。

本次更新旨在提高代码的可读性和可维护性，详细解释了以下关键组件的职责和工作流程：
- **ProactiveThinker**: 阐明了其作为主动事件处理中心的角色，以及如何与规划器和生成器协作。
- **SleepManager**: 详细描述了其作为睡眠状态机的核心逻辑，包括状态转换的条件和处理流程。
- **SleepState**: 解释了各个睡眠状态的含义以及序列化器的作用，确保状态持久化。

此外，对配置文件 `bot_config_template.toml` 中的相关配置项也补充了更清晰的注释。
This commit is contained in:
minecraft1024a
2025-09-06 13:44:38 +08:00
parent 36b9eae6c8
commit 0b1d365fce
4 changed files with 150 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

59
src/chat/chat_loop/proactive/proactive_thinker.py
View File

@@ -15,28 +15,35 @@ logger = get_logger("hfc")
class ProactiveThinker:
"""
主动思考器，负责处理和执行主动思考事件。
当接收到 ProactiveTriggerEvent 时，它会根据事件内容进行一系列决策和操作，
例如调整情绪、调用规划器生成行动，并最终可能产生一个主动的回复。
"""
def __init__(self, context: HfcContext, cycle_processor: "CycleProcessor"):
"""
初始化主动思考器
初始化主动思考器。
Args:
context: HFC聊天上下文对象
cycle_processor: 循环处理器，用于执行主动思考的结果
context (HfcContext): HFC聊天上下文对象，提供了当前聊天会话的所有背景信息。
cycle_processor (CycleProcessor): 循环处理器，用于执行主动思考后产生的动作。
功能说明:
- 接收主动思考事件并执行思考流程
- 根据事件类型执行不同的前置操作（如修改情绪）
- 调用planner进行决策并由cycle_processor执行
- 接收并处理主动思考事件 (ProactiveTriggerEvent)。
- 在思考前根据事件类型执行预处理操作，如修改当前情绪状态。
- 调用行动规划器 (Action Planner) 来决定下一步应该做什么。
- 如果规划结果是发送消息，则调用生成器API生成回复并发送。
"""
self.context = context
self.cycle_processor = cycle_processor
async def think(self, trigger_event: ProactiveTriggerEvent):
"""
统一的API入口，用于触发主动思考
主动思考的统一入口API。
这是外部触发主动思考时调用的主要方法。
Args:
trigger_event: 描述触发上下文的事件对象
trigger_event (ProactiveTriggerEvent): 描述触发上下文的事件对象，包含了思考的来源和原因。
"""
logger.info(
f"{self.context.log_prefix} 接收到主动思考事件: "
@@ -44,32 +51,38 @@ class ProactiveThinker:
)
try:
# 1. 根据事件类型执行前置操作
# 步骤 1: 根据事件类型执行思考前的准备工作，例如调整情绪。
await self._prepare_for_thinking(trigger_event)
# 2. 执行核心思考逻辑
# 步骤 2: 执行核心的思考和决策逻辑。
await self._execute_proactive_thinking(trigger_event)
except Exception as e:
# 捕获并记录在思考过程中发生的任何异常。
logger.error(f"{self.context.log_prefix} 主动思考 think 方法执行异常: {e}")
logger.error(traceback.format_exc())
async def _prepare_for_thinking(self, trigger_event: ProactiveTriggerEvent):
"""
根据事件类型，执行思考前的准备工作，例如修改情绪
根据事件类型，在正式思考前执行准备工作。
目前主要是处理来自失眠管理器的事件，并据此调整情绪。
Args:
trigger_event: 触发事件
trigger_event (ProactiveTriggerEvent): 触发事件。
"""
# 目前只处理来自失眠管理器(insomnia_manager)的事件
if trigger_event.source != "insomnia_manager":
return
try:
# 动态导入情绪管理器，避免循环依赖
from src.mood.mood_manager import mood_manager
# 获取当前聊天的情绪对象
mood_obj = mood_manager.get_mood_by_chat_id(self.context.stream_id)
new_mood = None
# 根据失眠的不同原因设置对应的情绪
if trigger_event.reason == "low_pressure":
new_mood = "精力过剩，毫无睡意"
elif trigger_event.reason == "random":
@@ -79,6 +92,7 @@ class ProactiveThinker:
elif trigger_event.reason == "post_sleep_insomnia":
new_mood = "可恶，刚刚好像睡着了又醒了，现在睡不着了"
# 如果成功匹配到了新的情绪，则更新情绪状态
if new_mood:
mood_obj.mood_state = new_mood
mood_obj.last_change_time = time.time()
@@ -92,34 +106,39 @@ class ProactiveThinker:
async def _execute_proactive_thinking(self, trigger_event: ProactiveTriggerEvent):
"""
执行主动思考的核心逻辑
执行主动思考的核心逻辑。
它会调用规划器来决定是否要采取行动，以及采取什么行动。
Args:
trigger_event: 触发事件
trigger_event (ProactiveTriggerEvent): 触发事件。
"""
try:
# 直接调用 planner 的 PROACTIVE 模式
# 调用规划器的 PROACTIVE 模式，让其决定下一步的行动
actions, target_message = await self.cycle_processor.action_planner.plan(mode=ChatMode.PROACTIVE)
# 获取第一个规划出的动作作为主要决策
action_result = actions[0] if actions else {}
# 通常只关心规划出的第一个动作
action_result = actions if actions else {}
# 如果决策不是 do_nothing，则执行
# 检查规划出的动作是否是“什么都不做”
if action_result and action_result.get("action_type") != "do_nothing":
# 如果动作是“回复”
if action_result.get("action_type") == "reply":
# 调用生成器API来创建回复内容
success, response_set, _ = await generator_api.generate_reply(
chat_stream=self.context.chat_stream,
reply_message=action_result["action_message"],
available_actions={},
available_actions={}, # 主动回复不考虑工具使用
enable_tool=False,
request_type="chat.replyer.proactive",
request_type="chat.replyer.proactive", # 标记请求类型
from_plugin=False,
)
# 如果成功生成回复，则发送出去
if success and response_set:
await self.cycle_processor.response_handler.send_response(
response_set, time.time(), action_result["action_message"]
)
else:
# 如果规划结果是“什么都不做”，则记录日志
logger.info(f"{self.context.log_prefix} 主动思考决策: 保持沉默")
except Exception as e:

74
src/chat/chat_loop/sleep_manager/sleep_manager.py
View File

@@ -16,29 +16,48 @@ logger = get_logger("sleep_manager")
class SleepManager:
"""
睡眠管理器，核心组件之一，负责管理角色的睡眠周期和状态转换。
它实现了一个状态机，根据预设的时间表、睡眠压力和随机因素，
在不同的睡眠状态（如清醒、准备入睡、睡眠、失眠）之间进行切换。
"""
def __init__(self):
self.time_checker = TimeChecker(self)
self.today_schedule: Optional[List[Dict[str, Any]]] = None
self.last_sleep_log_time = 0
self.sleep_log_interval = 35
"""
初始化睡眠管理器。
"""
self.time_checker = TimeChecker(self) # 时间检查器，用于判断当前是否处于理论睡眠时间
self.today_schedule: Optional[List[Dict[str, Any]]] = None # 当天的日程安排
self.last_sleep_log_time = 0 # 上次记录睡眠日志的时间戳
self.sleep_log_interval = 35 # 睡眠日志记录间隔（秒）
# --- 统一睡眠状态管理 ---
self._current_state: SleepState = SleepState.AWAKE
self._sleep_buffer_end_time: Optional[datetime] = None
self._total_delayed_minutes_today: int = 0
self._last_sleep_check_date: Optional[date] = None
self._last_fully_slept_log_time: float = 0
self._re_sleep_attempt_time: Optional[datetime] = None
self._current_state: SleepState = SleepState.AWAKE # 当前睡眠状态
self._sleep_buffer_end_time: Optional[datetime] = None # 睡眠缓冲结束时间，用于状态转换
self._total_delayed_minutes_today: int = 0 # 今天总共延迟入睡的分钟数
self._last_sleep_check_date: Optional[date] = None # 上次检查睡眠状态的日期
self._last_fully_slept_log_time: float = 0 # 上次完全进入睡眠状态的时间戳
self._re_sleep_attempt_time: Optional[datetime] = None # 被吵醒后，尝试重新入睡的时间点
# 从本地存储加载上一次的睡眠状态
self._load_sleep_state()
def get_current_sleep_state(self) -> SleepState:
"""获取当前的睡眠状态。"""
return self._current_state
def is_sleeping(self) -> bool:
"""判断当前是否处于正在睡觉的状态。"""
return self._current_state == SleepState.SLEEPING
async def update_sleep_state(self, wakeup_manager: Optional["WakeUpManager"] = None):
"""
更新睡眠状态的核心方法，实现状态机的主要逻辑。
该方法会被周期性调用，以检查并更新当前的睡眠状态。
Args:
wakeup_manager (Optional["WakeUpManager"]): 唤醒管理器，用于获取睡眠压力等上下文信息。
"""
# 如果全局禁用了睡眠系统，则强制设置为清醒状态并返回
if not global_config.sleep_system.enable:
if self._current_state != SleepState.AWAKE:
logger.debug("睡眠系统禁用，强制设为 AWAKE")
@@ -48,6 +67,7 @@ class SleepManager:
now = datetime.now()
today = now.date()
# 跨天处理：如果日期变化，重置每日相关的睡眠状态
if self._last_sleep_check_date != today:
logger.info(f"新的一天 ({today})，重置睡眠状态。")
self._total_delayed_minutes_today = 0
@@ -56,9 +76,10 @@ class SleepManager:
self._last_sleep_check_date = today
self._save_sleep_state()
# 检查当前是否处于理论上的睡眠时间段
is_in_theoretical_sleep, activity = self.time_checker.is_in_theoretical_sleep_time(now.time())
# 状态机处理
# --- 状态机核心处理逻辑 ---
if self._current_state == SleepState.AWAKE:
if is_in_theoretical_sleep:
self._handle_awake_to_sleep(now, activity, wakeup_manager)
@@ -76,14 +97,17 @@ class SleepManager:
self._handle_woken_up(now, is_in_theoretical_sleep, wakeup_manager)
def _handle_awake_to_sleep(self, now: datetime, activity: Optional[str], wakeup_manager: Optional["WakeUpManager"]):
"""处理从“清醒”到“准备入睡”的状态转换。"""
if activity:
logger.info(f"进入理论休眠时间 '{activity}'，开始进行睡眠决策...")
else:
logger.info("进入理论休眠时间，开始进行睡眠决策...")
# 如果配置了睡前通知，则发送晚安通知
if wakeup_manager and global_config.sleep_system.enable_pre_sleep_notification:
asyncio.create_task(NotificationSender.send_goodnight_notification(wakeup_manager.context))
# 设置一个随机的缓冲时间，模拟入睡前的准备过程
buffer_seconds = random.randint(1 * 60, 3 * 60)
self._sleep_buffer_end_time = now + timedelta(seconds=buffer_seconds)
self._current_state = SleepState.PREPARING_SLEEP
@@ -91,16 +115,20 @@ class SleepManager:
self._save_sleep_state()
def _handle_preparing_sleep(self, now: datetime, is_in_theoretical_sleep: bool, wakeup_manager: Optional["WakeUpManager"]):
"""处理“准备入睡”状态下的逻辑。"""
# 如果在准备期间离开了理论睡眠时间，则取消入睡
if not is_in_theoretical_sleep:
logger.info("准备入睡期间离开理论休眠时间，取消入睡，恢复清醒。")
self._current_state = SleepState.AWAKE
self._sleep_buffer_end_time = None
self._save_sleep_state()
# 如果缓冲时间结束，则正式进入睡眠状态
elif self._sleep_buffer_end_time and now >= self._sleep_buffer_end_time:
logger.info("睡眠缓冲期结束，正式进入休眠状态。")
self._current_state = SleepState.SLEEPING
self._last_fully_slept_log_time = now.timestamp()
# 设置一个随机的延迟，用于触发“睡后失眠”检查
delay_minutes_range = global_config.sleep_system.insomnia_trigger_delay_minutes
delay_minutes = random.randint(delay_minutes_range[0], delay_minutes_range[1])
self._sleep_buffer_end_time = now + timedelta(minutes=delay_minutes)
@@ -109,40 +137,51 @@ class SleepManager:
self._save_sleep_state()
def _handle_sleeping(self, now: datetime, is_in_theoretical_sleep: bool, activity: Optional[str], wakeup_manager: Optional["WakeUpManager"]):
"""处理“正在睡觉”状态下的逻辑。"""
# 如果理论睡眠时间结束，则自然醒来
if not is_in_theoretical_sleep:
logger.info("理论休眠时间结束，自然醒来。")
self._current_state = SleepState.AWAKE
self._save_sleep_state()
# 检查是否到了触发“睡后失眠”的时间点
elif self._sleep_buffer_end_time and now >= self._sleep_buffer_end_time:
if wakeup_manager:
sleep_pressure = wakeup_manager.context.sleep_pressure
pressure_threshold = global_config.sleep_system.flexible_sleep_pressure_threshold
# 如果睡眠压力低于阈值，则触发失眠
if sleep_pressure < pressure_threshold:
logger.info(f"睡眠压力 ({sleep_pressure:.1f}) 低于阈值 ({pressure_threshold})，触发睡后失眠。")
self._current_state = SleepState.INSOMNIA
# 设置失眠的持续时间
duration_minutes_range = global_config.sleep_system.insomnia_duration_minutes
duration_minutes = random.randint(duration_minutes_range[0], duration_minutes_range[1])
self._sleep_buffer_end_time = now + timedelta(minutes=duration_minutes)
# 发送失眠通知
asyncio.create_task(NotificationSender.send_insomnia_notification(wakeup_manager.context))
logger.info(f"进入失眠状态，将持续 {duration_minutes} 分钟。")
else:
# 睡眠压力正常，不触发失眠，清除检查时间点
logger.info(f"睡眠压力 ({sleep_pressure:.1f}) 正常，未触发睡后失眠。")
self._sleep_buffer_end_time = None
self._save_sleep_state()
else:
# 定期记录睡眠日志
current_timestamp = now.timestamp()
if current_timestamp - self.last_sleep_log_time > self.sleep_log_interval and activity:
logger.info(f"当前处于休眠活动 '{activity}' 中。")
self.last_sleep_log_time = current_timestamp
def _handle_insomnia(self, now: datetime, is_in_theoretical_sleep: bool):
"""处理“失眠”状态下的逻辑。"""
# 如果离开理论睡眠时间，则失眠结束
if not is_in_theoretical_sleep:
logger.info("已离开理论休眠时间，失眠结束，恢复清醒。")
self._current_state = SleepState.AWAKE
self._sleep_buffer_end_time = None
self._save_sleep_state()
# 如果失眠持续时间已过，则恢复睡眠
elif self._sleep_buffer_end_time and now >= self._sleep_buffer_end_time:
logger.info("失眠状态持续时间已过，恢复睡眠。")
self._current_state = SleepState.SLEEPING
@@ -150,17 +189,21 @@ class SleepManager:
self._save_sleep_state()
def _handle_woken_up(self, now: datetime, is_in_theoretical_sleep: bool, wakeup_manager: Optional["WakeUpManager"]):
"""处理“被吵醒”状态下的逻辑。"""
# 如果理论睡眠时间结束，则状态自动结束
if not is_in_theoretical_sleep:
logger.info("理论休眠时间结束，被吵醒的状态自动结束。")
self._current_state = SleepState.AWAKE
self._re_sleep_attempt_time = None
self._save_sleep_state()
# 到了尝试重新入睡的时间点
elif self._re_sleep_attempt_time and now >= self._re_sleep_attempt_time:
logger.info("被吵醒后经过一段时间，尝试重新入睡...")
if wakeup_manager:
sleep_pressure = wakeup_manager.context.sleep_pressure
pressure_threshold = global_config.sleep_system.flexible_sleep_pressure_threshold
# 如果睡眠压力足够，则尝试重新入睡
if sleep_pressure >= pressure_threshold:
logger.info("睡眠压力足够，从被吵醒状态转换到准备入睡。")
buffer_seconds = random.randint(3 * 60, 8 * 60)
@@ -168,6 +211,7 @@ class SleepManager:
self._current_state = SleepState.PREPARING_SLEEP
self._re_sleep_attempt_time = None
else:
# 睡眠压力不足，延迟一段时间后再次尝试
delay_minutes = 15
self._re_sleep_attempt_time = now + timedelta(minutes=delay_minutes)
logger.info(
@@ -176,6 +220,10 @@ class SleepManager:
self._save_sleep_state()
def reset_sleep_state_after_wakeup(self):
"""
当角色被用户消息等外部因素唤醒时调用此方法。
将状态强制转换为 WOKEN_UP，并设置一个延迟，之后会尝试重新入睡。
"""
if self._current_state in [SleepState.PREPARING_SLEEP, SleepState.SLEEPING, SleepState.INSOMNIA]:
logger.info("被唤醒，进入 WOKEN_UP 状态！")
self._current_state = SleepState.WOKEN_UP
@@ -186,12 +234,15 @@ class SleepManager:
self._save_sleep_state()
def get_today_schedule(self) -> Optional[List[Dict[str, Any]]]:
"""获取今天的日程安排。"""
return self.today_schedule
def update_today_schedule(self, schedule: Optional[List[Dict[str, Any]]]):
"""更新今天的日程安排。"""
self.today_schedule = schedule
def _save_sleep_state(self):
"""将当前所有睡眠相关的状态打包并保存到本地存储。"""
state_data = {
"_current_state": self._current_state,
"_sleep_buffer_end_time": self._sleep_buffer_end_time,
@@ -202,6 +253,7 @@ class SleepManager:
SleepStateSerializer.save(state_data)
def _load_sleep_state(self):
"""从本地存储加载并恢复所有睡眠相关的状态。"""
state_data = SleepStateSerializer.load()
self._current_state = state_data["_current_state"]
self._sleep_buffer_end_time = state_data["_sleep_buffer_end_time"]

50
src/chat/chat_loop/sleep_manager/sleep_state.py
View File

@@ -7,26 +7,45 @@ logger = get_logger("sleep_state")
class SleepState(Enum):
"""睡眠状态枚举"""
"""
定义了角色可能处于的几种睡眠状态。
这是一个状态机，用于管理角色的睡眠周期。
"""
AWAKE = auto()
INSOMNIA = auto()
PREPARING_SLEEP = auto()
SLEEPING = auto()
WOKEN_UP = auto()
AWAKE = auto() # 清醒状态
INSOMNIA = auto() # 失眠状态
PREPARING_SLEEP = auto() # 准备入睡状态，一个短暂的过渡期
SLEEPING = auto() # 正在睡觉状态
WOKEN_UP = auto() # 被吵醒状态
class SleepStateSerializer:
"""
睡眠状态序列化器。
负责将内存中的睡眠状态对象持久化到本地存储（如JSON文件），
以及在程序启动时从本地存储中恢复状态。
这样可以确保即使程序重启，角色的睡眠状态也能得以保留。
"""
@staticmethod
def save(state_data: dict):
"""将当前睡眠状态保存到本地存储"""
"""
将当前的睡眠状态数据保存到本地存储。
Args:
state_data (dict): 包含睡眠状态信息的字典。
datetime对象会被转换为时间戳，Enum成员会被转换为其名称字符串。
"""
try:
# 准备要序列化的数据字典
state = {
# 保存当前状态的枚举名称
"current_state": state_data["_current_state"].name,
# 将datetime对象转换为Unix时间戳以便序列化
"sleep_buffer_end_time_ts": state_data["_sleep_buffer_end_time"].timestamp()
if state_data["_sleep_buffer_end_time"]
else None,
"total_delayed_minutes_today": state_data["_total_delayed_minutes_today"],
# 将date对象转换为ISO格式的字符串
"last_sleep_check_date_str": state_data["_last_sleep_check_date"].isoformat()
if state_data["_last_sleep_check_date"]
else None,
@@ -34,6 +53,7 @@ class SleepStateSerializer:
if state_data["_re_sleep_attempt_time"]
else None,
}
# 写入本地存储
local_storage["schedule_sleep_state"] = state
logger.debug(f"已保存睡眠状态: {state}")
except Exception as e:
@@ -41,7 +61,14 @@ class SleepStateSerializer:
@staticmethod
def load() -> dict:
"""从本地存储加载睡眠状态"""
"""
从本地存储加载并解析睡眠状态。
Returns:
dict: 包含恢复后睡眠状态信息的字典。
如果加载失败或没有找到数据，则返回一个默认的清醒状态。
"""
# 定义一个默认的状态，以防加载失败
state_data = {
"_current_state": SleepState.AWAKE,
"_sleep_buffer_end_time": None,
@@ -50,27 +77,34 @@ class SleepStateSerializer:
"_re_sleep_attempt_time": None,
}
try:
# 从本地存储读取数据
state = local_storage["schedule_sleep_state"]
if state and isinstance(state, dict):
# 恢复当前状态枚举
state_name = state.get("current_state")
if state_name and hasattr(SleepState, state_name):
state_data["_current_state"] = SleepState[state_name]
# 从时间戳恢复datetime对象
end_time_ts = state.get("sleep_buffer_end_time_ts")
if end_time_ts:
state_data["_sleep_buffer_end_time"] = datetime.fromtimestamp(end_time_ts)
# 恢复重新入睡尝试时间
re_sleep_ts = state.get("re_sleep_attempt_time_ts")
if re_sleep_ts:
state_data["_re_sleep_attempt_time"] = datetime.fromtimestamp(re_sleep_ts)
# 恢复今日延迟睡眠总分钟数
state_data["_total_delayed_minutes_today"] = state.get("total_delayed_minutes_today", 0)
# 从ISO格式字符串恢复date对象
date_str = state.get("last_sleep_check_date_str")
if date_str:
state_data["_last_sleep_check_date"] = datetime.fromisoformat(date_str).date()
logger.info(f"成功从本地存储加载睡眠状态: {state}")
except Exception as e:
# 如果加载过程中出现任何问题，记录警告并返回默认状态
logger.warning(f"加载睡眠状态失败，将使用默认值: {e}")
return state_data

8
template/bot_config_template.toml
View File

@@ -1,5 +1,5 @@
[inner]
version = "6.7.2"
version = "6.7.3"
#----以下是给开发人员阅读的，如果你只是部署了MoFox-Bot，不需要阅读----
#如果你想要修改配置文件，请递增version的值
@@ -451,7 +451,6 @@ insomnia_chance_normal_pressure = 0.1
sleep_pressure_increment = 1.5
# 睡眠时，每分钟衰减的睡眠压力值
sleep_pressure_decay_rate = 1.5
insomnia_duration_minutes = 30 # 单次失眠状态的持续时间（分钟）
# --- 弹性睡眠与睡前消息 ---
# 是否启用弹性睡眠。启用后，AI不会到点立刻入睡，而是会根据睡眠压力增加5-10分钟的缓冲，并可能因为压力不足而推迟睡眠。
@@ -467,6 +466,11 @@ enable_pre_sleep_notification = false
pre_sleep_notification_groups = []
# 用于生成睡前消息的提示。AI会根据这个提示生成一句晚安问候。
pre_sleep_prompt = "我准备睡觉了，请生成一句简短自然的晚安问候。"
insomnia_duration_minutes = [30, 60] # 单次失眠状态的持续时间范围（分钟）
# --- 睡后失眠 ---
# 入睡后，经过一段延迟后触发失眠判定的延迟时间（分钟），设置为范围以增加随机性
insomnia_trigger_delay_minutes = [15, 45]
[cross_context] # 跨群聊/私聊上下文共享配置
# 这是总开关，用于一键启用或禁用此功能