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

refactor(schedule): 重构睡眠系统为状态机模型

Browse Source 
将原有的睡眠、失眠、唤醒等分散的布尔标记逻辑重构为一个统一的睡眠状态机（SleepState），以提高代码的可读性、可维护性和可扩展性。

主要变更：
- 引入 `SleepState` 枚举，包含 `AWAKE`, `INSOMNIA`, `PREPARING_SLEEP`, `SLEEPING`, `WOKEN_UP` 状态。
- 在 `ScheduleManager` 中实现 `_update_sleep_state` 作为核心状态机，统一管理所有状态转换。
- 将原有的失眠判断逻辑从 `WakeUpManager` 移至 `ScheduleManager` 的状态机内部，与弹性睡眠决策合并，简化了模块职责。
- `heartFC_chat.py` 中的聊天循环现在直接查询 `ScheduleManager` 的当前状态，而不是处理多个独立的布尔值，使逻辑更清晰。
- 删除了 `WakeUpManager` 中与失眠相关的配置和方法，因为它现在由 `ScheduleManager` 统一管理。
- 删除了配置中已废弃的 `enable_is_sleep` 选项。
This commit is contained in:
minecraft1024a
2025-08-29 17:40:22 +08:00
parent 01455156a7
commit 41565eb144
4 changed files with 150 additions and 183 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
src/chat/chat_loop/heartFC_chat.py
View File

@@ -8,7 +8,7 @@ from src.config.config import global_config
from src.person_info.relationship_builder_manager import relationship_builder_manager
from src.chat.express.expression_learner import expression_learner_manager
from src.plugin_system.base.component_types import ChatMode
from src.schedule.schedule_manager import schedule_manager
from src.schedule.schedule_manager import schedule_manager, SleepState
from src.plugin_system.apis import message_api
from .hfc_context import HfcContext
@@ -196,30 +196,14 @@ class HeartFChatting:
- FOCUS模式：直接处理所有消息并检查退出条件
- NORMAL模式：检查进入FOCUS模式的条件，并通过normal_mode_handler处理消息
"""
is_sleeping = schedule_manager.is_sleeping(self.wakeup_manager)
# --- 失眠状态管理 ---
if self.context.is_in_insomnia and time.time() > self.context.insomnia_end_time:
# 失眠状态结束
self.context.is_in_insomnia = False
await self.proactive_thinker.trigger_goodnight_thinking()
if is_sleeping and not self.context.was_sleeping:
# 刚刚进入睡眠状态，进行一次入睡检查
if self.wakeup_manager and self.wakeup_manager.check_for_insomnia():
# 触发失眠
self.context.is_in_insomnia = True
duration = global_config.sleep_system.insomnia_duration_minutes * 60
self.context.insomnia_end_time = time.time() + duration
# 判断失眠原因并触发思考
reason = "random"
if self.context.sleep_pressure < global_config.sleep_system.sleep_pressure_threshold:
reason = "low_pressure"
await self.proactive_thinker.trigger_insomnia_thinking(reason)
# --- 核心状态更新 ---
await schedule_manager._update_sleep_state(self.wakeup_manager)
current_sleep_state = schedule_manager.get_current_sleep_state()
is_sleeping = current_sleep_state == SleepState.SLEEPING
is_in_insomnia = current_sleep_state == SleepState.INSOMNIA
# 核心修复：在睡眠模式(包括失眠)下获取消息时，不过滤命令消息，以确保@消息能被接收
filter_command_flag = not is_sleeping
filter_command_flag = not (is_sleeping or is_in_insomnia)
recent_messages = message_api.get_messages_by_time_in_chat(
chat_id=self.context.stream_id,
@@ -239,16 +223,17 @@ class HeartFChatting:
self.context.last_read_time = time.time()
# 处理唤醒度逻辑
if is_sleeping:
if current_sleep_state in [SleepState.SLEEPING, SleepState.PREPARING_SLEEP, SleepState.INSOMNIA]:
self._handle_wakeup_messages(recent_messages)
# 再次检查睡眠状态，因为_handle_wakeup_messages可能会触发唤醒
current_is_sleeping = schedule_manager.is_sleeping(self.wakeup_manager)
if not self.context.is_in_insomnia and current_is_sleeping:
# 仍然在睡眠，跳过本轮的消息处理
# 再次获取最新状态，因为 handle_wakeup 可能导致状态变为 WOKEN_UP
current_sleep_state = schedule_manager.get_current_sleep_state()
if current_sleep_state == SleepState.SLEEPING:
# 只有在纯粹的 SLEEPING 状态下才跳过消息处理
return has_new_messages
else:
# 从睡眠中被唤醒，需要继续处理本轮消息
if current_sleep_state == SleepState.WOKEN_UP:
logger.info(f"{self.context.log_prefix} 从睡眠中被唤醒，将处理积压的消息。")
self.context.last_wakeup_time = time.time()

44
src/chat/chat_loop/wakeup_manager.py
View File

@@ -41,13 +41,6 @@ class WakeUpManager:
self.enabled = sleep_config.enable
self.angry_prompt = sleep_config.angry_prompt
# 失眠系统参数
self.insomnia_enabled = sleep_config.enable_insomnia_system
self.sleep_pressure_threshold = sleep_config.sleep_pressure_threshold
self.deep_sleep_threshold = sleep_config.deep_sleep_threshold
self.insomnia_chance_low_pressure = sleep_config.insomnia_chance_low_pressure
self.insomnia_chance_normal_pressure = sleep_config.insomnia_chance_normal_pressure
self._load_wakeup_state()
def _get_storage_key(self) -> str:
@@ -220,39 +213,4 @@ class WakeUpManager:
"wakeup_threshold": self.wakeup_threshold,
"is_angry": self.is_angry,
"angry_remaining_time": max(0, self.angry_duration - (time.time() - self.angry_start_time)) if self.is_angry else 0
}
def check_for_insomnia(self) -> bool:
"""
在尝试入睡时检查是否会失眠
Returns:
bool: 如果失眠则返回 True，否则返回 False
"""
if not self.insomnia_enabled:
return False
import random
pressure = self.context.sleep_pressure
# 压力过高，深度睡眠，极难失眠
if pressure > self.deep_sleep_threshold:
return False
# 根据睡眠压力决定失眠概率
from src.schedule.schedule_manager import schedule_manager
if pressure < self.sleep_pressure_threshold:
# 压力不足型失眠
if schedule_manager._is_in_voluntary_delay:
logger.debug(f"{self.context.log_prefix} 处于主动延迟睡眠期间，跳过压力不足型失眠判断。")
elif random.random() < self.insomnia_chance_low_pressure:
logger.info(f"{self.context.log_prefix} 睡眠压力不足 ({pressure:.1f})，触发失眠！")
return True
else:
# 压力正常，随机失眠
if random.random() < self.insomnia_chance_normal_pressure:
logger.info(f"{self.context.log_prefix} 睡眠压力正常 ({pressure:.1f})，触发随机失眠！")
return True
return False
}

1
src/config/official_configs.py
View File

@@ -529,7 +529,6 @@ class ScheduleConfig(ValidatedConfigBase):
enable: bool = Field(default=True, description="启用")
guidelines: Optional[str] = Field(default=None, description="指导方针")
enable_is_sleep: bool = Field(default=True, description="让AI会根据日程表睡觉和苏醒")
class DependencyManagementConfig(ValidatedConfigBase):

243
src/schedule/schedule_manager.py
View File

@@ -2,7 +2,8 @@ import orjson
import asyncio
import random
from datetime import datetime, time, timedelta
from typing import Optional, List, Dict, Any
from enum import Enum, auto
from typing import Optional, List, Dict, Any, TYPE_CHECKING
from lunar_python import Lunar
from pydantic import BaseModel, ValidationError, validator
@@ -19,6 +20,9 @@ from src.manager.async_task_manager import AsyncTask, async_task_manager
from src.manager.local_store_manager import local_storage
from src.plugin_system.apis import send_api, generator_api
if TYPE_CHECKING:
from src.chat.chat_loop.wakeup_manager import WakeUpManager
logger = get_logger("schedule_manager")
@@ -121,6 +125,14 @@ class ScheduleData(BaseModel):
# 检查是否所有分钟都被覆盖
return all(covered)
class SleepState(Enum):
"""睡眠状态枚举"""
AWAKE = auto() # 完全清醒
INSOMNIA = auto() # 失眠（在理论睡眠时间内保持清醒）
PREPARING_SLEEP = auto() # 准备入睡（缓冲期）
SLEEPING = auto() # 正在休眠
WOKEN_UP = auto() # 被吵醒
class ScheduleManager:
def __init__(self):
self.today_schedule: Optional[List[Dict[str, Any]]] = None
@@ -131,14 +143,12 @@ class ScheduleManager:
self.sleep_log_interval = 35 # 日志记录间隔，单位秒
self.schedule_generation_running = False # 防止重复生成任务
# 弹性睡眠相关状态
self._is_preparing_sleep: bool = False
# --- 统一睡眠状态管理 ---
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[datetime.date] = None
self._last_fully_slept_log_time: float = 0
self._is_in_voluntary_delay: bool = False # 新增：标记是否处于主动延迟睡眠状态
self._is_woken_up: bool = False # 新增：标记是否被吵醒
self._load_sleep_state()
@@ -406,147 +416,162 @@ class ScheduleManager:
continue
return None
def is_sleeping(self, wakeup_manager: Optional["WakeUpManager"] = None) -> bool:
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):
"""
通过关键词匹配、唤醒度、睡眠压力等综合判断是否处于休眠时间。
新增弹性睡眠机制，允许在压力低时延迟入睡，并在入睡前发送通知。
核心状态机：根据当前情况更新睡眠状态
"""
# --- 基础检查 ---
if not global_config.schedule.enable_is_sleep:
return False
if not self.today_schedule:
return False
if not global_config.sleep_system.enable or not self.today_schedule:
if self._current_state != SleepState.AWAKE:
logger.debug("睡眠系统禁用或无日程，强制设为 AWAKE")
self._current_state = SleepState.AWAKE
return
now = datetime.now()
today = now.date()
# --- 每日状态重置 ---
if self._last_sleep_check_date != today:
logger.info(f"新的一天 ({today})，重置弹性睡眠状态。")
logger.info(f"新的一天 ({today})，重置睡眠状态为 AWAKE。")
self._total_delayed_minutes_today = 0
self._is_preparing_sleep = False
self._current_state = SleepState.AWAKE
self._sleep_buffer_end_time = None
self._last_sleep_check_date = today
self._is_in_voluntary_delay = False
self._save_sleep_state()
# --- 检查是否在“准备入睡”的缓冲期 ---
if self._is_preparing_sleep and self._sleep_buffer_end_time:
if now >= self._sleep_buffer_end_time:
current_timestamp = now.timestamp()
if current_timestamp - self._last_fully_slept_log_time > 45:
logger.info("睡眠缓冲期结束，正式进入休眠状态。")
self._last_fully_slept_log_time = current_timestamp
return True
else:
remaining_seconds = (self._sleep_buffer_end_time - now).total_seconds()
logger.debug(f"处于入睡缓冲期，剩余 {remaining_seconds:.1f} 秒。")
return False
# --- 判断当前是否为理论上的睡眠时间 ---
is_in_theoretical_sleep, activity = self._is_in_theoretical_sleep_time(now.time())
if not is_in_theoretical_sleep:
# 如果不在理论睡眠时间，确保重置准备状态
if self._is_preparing_sleep:
logger.info("已离开理论休眠时间，取消“准备入睡”状态。")
self._is_preparing_sleep = False
self._sleep_buffer_end_time = None
self._is_in_voluntary_delay = False
self._is_woken_up = False # 离开睡眠时间，重置唤醒状态
# ===================================
# 状态机核心逻辑
# ===================================
# 状态：清醒 (AWAKE)
if self._current_state == SleepState.AWAKE:
if is_in_theoretical_sleep:
logger.info(f"进入理论休眠时间 '{activity}'，开始进行睡眠决策...")
# --- 合并后的失眠与弹性睡眠决策逻辑 ---
sleep_pressure = wakeup_manager.context.sleep_pressure if wakeup_manager else 999
pressure_threshold = global_config.sleep_system.flexible_sleep_pressure_threshold
# 决策1：因睡眠压力低而延迟入睡（原弹性睡眠）
if sleep_pressure < pressure_threshold and self._total_delayed_minutes_today < global_config.sleep_system.max_sleep_delay_minutes:
delay_minutes = 15
self._total_delayed_minutes_today += delay_minutes
self._sleep_buffer_end_time = now + timedelta(minutes=delay_minutes)
self._current_state = SleepState.INSOMNIA
logger.info(f"睡眠压力 ({sleep_pressure:.1f}) 低于阈值 ({pressure_threshold})，进入失眠状态，延迟入睡 {delay_minutes} 分钟。")
# 发送睡前通知
if global_config.sleep_system.enable_pre_sleep_notification:
asyncio.create_task(self._send_pre_sleep_notification())
# 决策2：进入正常的入睡准备流程
else:
buffer_seconds = random.randint(5 * 60, 10 * 60)
self._sleep_buffer_end_time = now + timedelta(seconds=buffer_seconds)
self._current_state = SleepState.PREPARING_SLEEP
logger.info(f"睡眠压力正常或已达今日最大延迟，进入准备入睡状态，将在 {buffer_seconds / 60:.1f} 分钟内入睡。")
# 发送睡前通知
if global_config.sleep_system.enable_pre_sleep_notification:
asyncio.create_task(self._send_pre_sleep_notification())
self._save_sleep_state()
return False
# --- 处理唤醒状态 ---
if self._is_woken_up:
current_timestamp = now.timestamp()
if current_timestamp - self.last_sleep_log_time > self.sleep_log_interval:
logger.info(f"在休眠活动 '{activity}' 期间，但已被唤醒，保持清醒状态。")
self.last_sleep_log_time = current_timestamp
return False
# 状态：失眠 (INSOMNIA)
elif self._current_state == SleepState.INSOMNIA:
if not is_in_theoretical_sleep:
logger.info("已离开理论休眠时间，失眠结束，恢复清醒。")
self._current_state = SleepState.AWAKE
self._save_sleep_state()
# TODO: 添加从失眠到准备入睡的转换逻辑
# --- 核心：弹性睡眠逻辑 ---
if global_config.schedule.enable_flexible_sleep and not self._is_preparing_sleep:
# 首次进入理论睡眠时间，触发弹性判断
logger.info(f"进入理论休眠时间 '{activity}'，开始弹性睡眠判断...")
# 1. 获取睡眠压力
sleep_pressure = wakeup_manager.context.sleep_pressure if wakeup_manager else 999
pressure_threshold = global_config.schedule.flexible_sleep_pressure_threshold
# 2. 判断是否延迟
if sleep_pressure < pressure_threshold and self._total_delayed_minutes_today < global_config.schedule.max_sleep_delay_minutes:
delay_minutes = 15 # 每次延迟15分钟
self._total_delayed_minutes_today += delay_minutes
self._sleep_buffer_end_time = now + timedelta(minutes=delay_minutes)
self._is_in_voluntary_delay = True # 标记进入主动延迟
logger.info(f"睡眠压力 ({sleep_pressure:.1f}) 低于阈值 ({pressure_threshold})，延迟入睡 {delay_minutes} 分钟。今日已累计延迟 {self._total_delayed_minutes_today} 分钟。")
# 状态：准备入睡 (PREPARING_SLEEP)
elif self._current_state == SleepState.PREPARING_SLEEP:
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()
self._save_sleep_state()
# 状态：休眠中 (SLEEPING)
elif self._current_state == SleepState.SLEEPING:
if not is_in_theoretical_sleep:
logger.info("理论休眠时间结束，自然醒来。")
self._current_state = SleepState.AWAKE
self._save_sleep_state()
else:
# 3. 计算5-10分钟的入睡缓冲
self._is_in_voluntary_delay = False # 非主动延迟
buffer_seconds = random.randint(5 * 60, 10 * 60)
self._sleep_buffer_end_time = now + timedelta(seconds=buffer_seconds)
logger.info(f"睡眠压力正常或已达今日最大延迟，将在 {buffer_seconds / 60:.1f} 分钟内入睡。")
# 记录日志
current_timestamp = now.timestamp()
if current_timestamp - self.last_sleep_log_time > self.sleep_log_interval:
logger.info(f"当前处于休眠活动 '{activity}' 中。")
self.last_sleep_log_time = current_timestamp
# 4. 发送睡前通知
if global_config.schedule.enable_pre_sleep_notification:
asyncio.create_task(self._send_pre_sleep_notification())
self._is_preparing_sleep = True
self._save_sleep_state()
return False # 进入准备阶段，但尚未正式入睡
# --- 经典模式或已在弹性睡眠流程中 ---
current_timestamp = now.timestamp()
if current_timestamp - self.last_sleep_log_time > self.sleep_log_interval:
logger.info(f"当前处于休眠活动 '{activity}' 中 (经典模式)。")
self.last_sleep_log_time = current_timestamp
return True
# 状态：被吵醒 (WOKEN_UP)
elif self._current_state == SleepState.WOKEN_UP:
if not is_in_theoretical_sleep:
logger.info("理论休眠时间结束，被吵醒的状态自动结束。")
self._current_state = SleepState.AWAKE
self._save_sleep_state()
# TODO: 添加重新入睡的逻辑
def reset_sleep_state_after_wakeup(self):
"""被唤醒后重置睡眠状态"""
if self._is_preparing_sleep or self.is_sleeping():
logger.info("被唤醒，重置所有睡眠准备状态，恢复清醒！")
self._is_preparing_sleep = False
"""被唤醒后，将状态切换到 WOKEN_UP"""
if self._current_state in [SleepState.PREPARING_SLEEP, SleepState.SLEEPING, SleepState.INSOMNIA]:
logger.info("被唤醒，进入 WOKEN_UP 状态！")
self._current_state = SleepState.WOKEN_UP
self._sleep_buffer_end_time = None
self._is_in_voluntary_delay = False
self._is_woken_up = True # 标记为已被唤醒
self._save_sleep_state()
def _is_in_theoretical_sleep_time(self, now_time: time) -> (bool, Optional[str]):
"""检查当前时间是否落在日程表的任何一个睡眠活动中"""
sleep_keywords = ["休眠", "睡觉", "梦乡"]
for event in self.today_schedule:
try:
activity = event.get("activity", "").strip()
time_range = event.get("time_range")
if self.today_schedule:
for event in self.today_schedule:
try:
activity = event.get("activity", "").strip()
time_range = event.get("time_range")
if not activity or not time_range:
if not activity or not time_range:
continue
if any(keyword in activity for keyword in sleep_keywords):
start_str, end_str = time_range.split('-')
start_time = datetime.strptime(start_str.strip(), "%H:%M").time()
end_time = datetime.strptime(end_str.strip(), "%H:%M").time()
if start_time <= end_time: # 同一天
if start_time <= now_time < end_time:
return True, activity
else: # 跨天
if now_time >= start_time or now_time < end_time:
return True, activity
except (ValueError, KeyError, AttributeError) as e:
logger.warning(f"解析日程事件时出错: {event}, 错误: {e}")
continue
if any(keyword in activity for keyword in sleep_keywords):
start_str, end_str = time_range.split('-')
start_time = datetime.strptime(start_str.strip(), "%H:%M").time()
end_time = datetime.strptime(end_str.strip(), "%H:%M").time()
if start_time <= end_time: # 同一天
if start_time <= now_time < end_time:
return True, activity
else: # 跨天
if now_time >= start_time or now_time < end_time:
return True, activity
except (ValueError, KeyError, AttributeError) as e:
logger.warning(f"解析日程事件时出错: {event}, 错误: {e}")
continue
return False, None
async def _send_pre_sleep_notification(self):
"""异步生成并发送睡前通知"""
try:
groups = global_config.schedule.pre_sleep_notification_groups
prompt = global_config.schedule.pre_sleep_prompt
groups = global_config.sleep_system.pre_sleep_notification_groups
prompt = global_config.sleep_system.pre_sleep_prompt
if not groups:
logger.info("未配置睡前通知的群组，跳过发送。")