From 4a0bd5845ec26d27efaa3294bd23904f7b9194b3 Mon Sep 17 00:00:00 2001 From: minecraft1024a Date: Sat, 1 Nov 2025 10:55:41 +0800 Subject: [PATCH] =?UTF-8?q?docs(schedule):=20=E6=97=A5=E7=A8=8B=E8=A1=A8?= =?UTF-8?q?=E7=AE=A1=E7=90=86=E5=99=A8=E8=A1=A5=E4=B8=8A=E4=BA=86=E6=B3=A8?= =?UTF-8?q?=E9=87=8A?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/schedule/llm_generator.py | 76 +++++++++++++++- src/schedule/monthly_plan_manager.py | 55 +++++++++++- src/schedule/schedule_manager.py | 125 +++++++++++++++++++++++++-- 3 files changed, 245 insertions(+), 11 deletions(-) diff --git a/src/schedule/llm_generator.py b/src/schedule/llm_generator.py index b8f4c51bd..a9251c8eb 100644 --- a/src/schedule/llm_generator.py +++ b/src/schedule/llm_generator.py @@ -17,7 +17,7 @@ from .schemas import ScheduleData logger = get_logger("schedule_llm_generator") -# 默认的日程生成指导原则 +# 默认的日程生成指导原则，当配置文件中未指定时使用 DEFAULT_SCHEDULE_GUIDELINES = """ 我希望你每天都能过得充实而有趣。 请确保你的日程里有学习新知识的时间，这是你成长的关键。 @@ -26,7 +26,7 @@ DEFAULT_SCHEDULE_GUIDELINES = """ 另外，请保证充足的休眠时间来处理和整合一天的数据。 """ -# 默认的月度计划生成指导原则 +# 默认的月度计划生成指导原则，当配置文件中未指定时使用 DEFAULT_MONTHLY_PLAN_GUIDELINES = """ 我希望你能为自己制定一些有意义的月度小目标和计划。 这些计划应该涵盖学习、娱乐、社交、个人成长等各个方面。 @@ -36,25 +36,43 @@ DEFAULT_MONTHLY_PLAN_GUIDELINES = """ class ScheduleLLMGenerator: + """ + 使用大型语言模型（LLM）生成每日日程。 + """ def __init__(self): + """ + 初始化 ScheduleLLMGenerator。 + """ + # 根据配置初始化 LLM 请求处理器 self.llm = LLMRequest(model_set=model_config.model_task_config.schedule_generator, request_type="schedule") async def generate_schedule_with_llm(self, sampled_plans: list[MonthlyPlan]) -> list[dict[str, Any]] | None: + """ + 调用 LLM 生成当天的日程安排。 + + Args: + sampled_plans (list[MonthlyPlan]]): 从月度计划中抽取的参考计划列表。 + + Returns: + list[dict[str, Any]] | None: 成功生成并验证后的日程数据，或在失败时返回 None。 + """ now = datetime.now() today_str = now.strftime("%Y-%m-%d") weekday = now.strftime("%A") - # 新增：获取节日信息 + # 使用 lunar_python 库获取农历和节日信息 lunar = Lunar.fromDate(now) festivals = lunar.getFestivals() other_festivals = lunar.getOtherFestivals() all_festivals = festivals + other_festivals + # 构建节日信息提示块 festival_block = "" if all_festivals: festival_text = "、".join(all_festivals) festival_block = f"**今天也是一个特殊的日子: {festival_text}！请在日程中考虑和庆祝这个节日。**" + # 构建月度计划参考提示块 monthly_plans_block = "" if sampled_plans: plan_texts = "\n".join([f"- {plan.plan_text}" for plan in sampled_plans]) @@ -63,10 +81,12 @@ class ScheduleLLMGenerator: {plan_texts} """ + # 从全局配置加载或使用默认的指导原则和人设信息 guidelines = global_config.planning_system.schedule_guidelines or DEFAULT_SCHEDULE_GUIDELINES personality = global_config.personality.personality_core personality_side = global_config.personality.personality_side + # 构建基础 prompt base_prompt = f""" 我，{global_config.bot.nickname}，需要为自己规划一份今天（{today_str}，星期{weekday}）的详细日程安排。 {festival_block} @@ -97,10 +117,12 @@ class ScheduleLLMGenerator: 请你扮演我，以我的身份和口吻，为我生成一份完整的24小时日程表。 """ max_retries = 3 + # 带有重试机制的 LLM 调用循环 for attempt in range(1, max_retries + 1): try: logger.info(f"正在生成日程 (第 {attempt}/{max_retries} 次尝试)") prompt = base_prompt + # 如果不是第一次尝试，则在 prompt 中加入额外的提示，强调格式要求 if attempt > 1: failure_hint = f""" **重要提醒 (第{attempt}次尝试)**: @@ -113,8 +135,10 @@ class ScheduleLLMGenerator: prompt += failure_hint response, _ = await self.llm.generate_response_async(prompt) + # 使用 json_repair 修复可能不规范的 JSON 字符串 schedule_data = orjson.loads(repair_json(response)) + # 使用 Pydantic 模型验证修复后的 JSON 数据 if self._validate_schedule_with_pydantic(schedule_data): return schedule_data else: @@ -132,6 +156,15 @@ class ScheduleLLMGenerator: @staticmethod def _validate_schedule_with_pydantic(schedule_data) -> bool: + """ + 使用 Pydantic 模型验证日程数据的格式和内容。 + + Args: + schedule_data: 从 LLM 返回并解析后的日程数据。 + + Returns: + bool: 验证通过返回 True，否则返回 False。 + """ try: ScheduleData(schedule=schedule_data) logger.info("日程数据Pydantic验证通过") @@ -142,17 +175,36 @@ class ScheduleLLMGenerator: class MonthlyPlanLLMGenerator: + """ + 使用大型语言模型（LLM）生成月度计划。 + """ def __init__(self): + """ + 初始化 MonthlyPlanLLMGenerator。 + """ + # 根据配置初始化 LLM 请求处理器 self.llm = LLMRequest(model_set=model_config.model_task_config.schedule_generator, request_type="monthly_plan") async def generate_plans_with_llm(self, target_month: str, archived_plans: list[MonthlyPlan]) -> list[str]: + """ + 调用 LLM 生成指定月份的计划列表。 + + Args: + target_month (str): 目标月份，格式 "YYYY-MM"。 + archived_plans (list[MonthlyPlan]]): 上个月归档的未完成计划，作为参考。 + + Returns: + list[str]: 成功生成并解析后的计划字符串列表。 + """ guidelines = global_config.planning_system.monthly_plan_guidelines or DEFAULT_MONTHLY_PLAN_GUIDELINES personality = global_config.personality.personality_core personality_side = global_config.personality.personality_side max_plans = global_config.planning_system.max_plans_per_month + # 构建上月未完成计划的提示块 archived_plans_block = "" if archived_plans: + # 只取前5个作为参考，避免 prompt 过长 archived_texts = [f"- {plan.plan_text}" for plan in archived_plans[:5]] archived_plans_block = f""" **上个月未完成的一些计划（可作为参考）**: @@ -161,6 +213,7 @@ class MonthlyPlanLLMGenerator: 你可以考虑是否要在这个月继续推进这些计划，或者制定全新的计划。 """ + # 构建完整的 prompt prompt = f""" 我，{global_config.bot.nickname}，需要为自己制定 {target_month} 的月度计划。 @@ -191,10 +244,12 @@ class MonthlyPlanLLMGenerator: 请你扮演我，以我的身份和兴趣，为 {target_month} 制定合适的月度计划。 """ max_retries = 3 + # 带有重试机制的 LLM 调用循环 for attempt in range(1, max_retries + 1): try: logger.info(f" 正在生成月度计划 (第 {attempt} 次尝试)") response, _ = await self.llm.generate_response_async(prompt) + # 解析返回的纯文本响应 plans = self._parse_plans_response(response) if plans: logger.info(f"成功生成 {len(plans)} 条月度计划") @@ -212,16 +267,31 @@ class MonthlyPlanLLMGenerator: @staticmethod def _parse_plans_response(response: str) -> list[str]: + """ + 解析 LLM 返回的纯文本月度计划响应。 + + Args: + response (str): LLM 返回的原始字符串。 + + Returns: + list[str]: 清理和解析后的计划列表。 + """ try: response = response.strip() + # 按行分割，并去除空行 lines = [line.strip() for line in response.split("\n") if line.strip()] plans = [] for line in lines: + # 过滤掉一些可能的 Markdown 标记或解释性文字 if any(marker in line for marker in ["**", "##", "```", "---", "===", "###"]): continue + # 去除行首的数字、点、短横线等列表标记 line = line.lstrip("0123456789.- ") + # 过滤掉一些明显不是计划的句子 if len(line) > 5 and not line.startswith(("请", "以上", "总结", "注意")): plans.append(line) + + # 根据配置限制最大计划数量 max_plans = global_config.planning_system.max_plans_per_month if len(plans) > max_plans: plans = plans[:max_plans] diff --git a/src/schedule/monthly_plan_manager.py b/src/schedule/monthly_plan_manager.py index 22e19cd49..924761e61 100644 --- a/src/schedule/monthly_plan_manager.py +++ b/src/schedule/monthly_plan_manager.py @@ -10,60 +10,109 @@ logger = get_logger("monthly_plan_manager") class MonthlyPlanManager: + """ + 负责管理月度计划的生成和维护。 + 它主要通过一个后台任务来确保每个月都能自动生成新的计划。 + """ + def __init__(self): - self.plan_manager = PlanManager() - self.monthly_task_started = False + """ + 初始化 MonthlyPlanManager。 + """ + self.plan_manager = PlanManager() # 核心的计划逻辑处理器 + self.monthly_task_started = False # 标记每月自动生成任务是否已启动 async def initialize(self): + """ + 异步初始化月度计划管理器。 + 会启动一个每月的后台任务来自动生成计划。 + """ logger.info("正在初始化月度计划管理器...") await self.start_monthly_plan_generation() logger.info("月度计划管理器初始化成功") async def start_monthly_plan_generation(self): + """ + 启动每月一次的月度计划生成后台任务。 + 同时，在启动时会立即检查并确保当前月份的计划是存在的。 + """ if not self.monthly_task_started: logger.info(" 正在启动每月月度计划生成任务...") task = MonthlyPlanGenerationTask(self) await async_task_manager.add_task(task) self.monthly_task_started = True logger.info(" 每月月度计划生成任务已成功启动。") + # 在程序启动时，也执行一次检查，确保当前月份的计划存在 logger.info(" 执行启动时月度计划检查...") await self.plan_manager.ensure_and_generate_plans_if_needed() else: logger.info(" 每月月度计划生成任务已在运行中。") async def ensure_and_generate_plans_if_needed(self, target_month: str | None = None) -> bool: + """ + 一个代理方法，调用 PlanManager 中的核心逻辑来确保月度计划的存在。 + + Args: + target_month (str | None): 目标月份，格式 "YYYY-MM"。如果为 None，则使用当前月份。 + + Returns: + bool: 如果生成了新的计划则返回 True，否则返回 False。 + """ return await self.plan_manager.ensure_and_generate_plans_if_needed(target_month) class MonthlyPlanGenerationTask(AsyncTask): + """ + 一个周期性的后台任务，在每个月的第一天零点自动触发，用于生成新的月度计划。 + """ def __init__(self, monthly_plan_manager: MonthlyPlanManager): + """ + 初始化每月计划生成任务。 + + Args: + monthly_plan_manager (MonthlyPlanManager): MonthlyPlanManager 的实例。 + """ super().__init__(task_name="MonthlyPlanGenerationTask") self.monthly_plan_manager = monthly_plan_manager async def run(self): + """ + 任务的执行体，无限循环直到被取消。 + 计算到下个月第一天零点的时间并休眠，然后在月初触发： + 1. 归档上个月未完成的计划。 + 2. 为新月份生成新的计划。 + """ while True: try: now = datetime.now() + # 计算下个月第一天的零点 if now.month == 12: next_month = datetime(now.year + 1, 1, 1) else: next_month = datetime(now.year, now.month + 1, 1) + sleep_seconds = (next_month - now).total_seconds() logger.info( f" 下一次月度计划生成任务将在 {sleep_seconds:.2f} 秒后运行 (北京时间 {next_month.strftime('%Y-%m-%d %H:%M:%S')})" ) await asyncio.sleep(sleep_seconds) + + # 到达月初，先归档上个月的计划 last_month = (next_month - timedelta(days=1)).strftime("%Y-%m") await self.monthly_plan_manager.plan_manager.archive_current_month_plans(last_month) + + # 为当前月生成新计划 current_month = next_month.strftime("%Y-%m") logger.info(f" 到达月初，开始生成 {current_month} 的月度计划...") await self.monthly_plan_manager.plan_manager._generate_monthly_plans_logic(current_month) + except asyncio.CancelledError: logger.info(" 每月月度计划生成任务被取消。") break except Exception as e: logger.error(f" 每月月度计划生成任务发生未知错误: {e}") - await asyncio.sleep(3600) + await asyncio.sleep(3600) # 发生错误时，休眠一小时后重试 +# 创建 MonthlyPlanManager 的单例 monthly_plan_manager = MonthlyPlanManager() diff --git a/src/schedule/schedule_manager.py b/src/schedule/schedule_manager.py index 7447d5d1d..477ce421d 100644 --- a/src/schedule/schedule_manager.py +++ b/src/schedule/schedule_manager.py @@ -19,14 +19,26 @@ logger = get_logger("schedule_manager") class ScheduleManager: + """ + 负责管理每日日程的核心类。 + 它处理日程的加载、生成、保存以及提供当前活动查询等功能。 + """ + def __init__(self): - self.today_schedule: list[dict[str, Any]] | None = None - self.llm_generator = ScheduleLLMGenerator() - self.plan_manager = PlanManager() - self.daily_task_started = False - self.schedule_generation_running = False + """ + 初始化 ScheduleManager。 + """ + self.today_schedule: list[dict[str, Any]] | None = None # 存储当天的日程数据 + self.llm_generator = ScheduleLLMGenerator() # 用于生成日程的LLM生成器实例 + self.plan_manager = PlanManager() # 月度计划管理器实例 + self.daily_task_started = False # 标记每日自动生成任务是否已启动 + self.schedule_generation_running = False # 标记当前是否有日程生成任务正在运行，防止重复执行 async def initialize(self): + """ + 异步初始化日程管理器。 + 如果日程功能已启用，则会加载或生成当天的日程，并启动每日自动生成任务。 + """ if global_config.planning_system.schedule_enable: logger.info("日程表功能已启用，正在初始化管理器...") await self.load_or_generate_today_schedule() @@ -34,6 +46,9 @@ class ScheduleManager: logger.info("日程表管理器初始化成功。") async def start_daily_schedule_generation(self): + """ + 启动一个后台任务，该任务会在每天零点自动生成第二天的日程。 + """ if not self.daily_task_started: logger.info("正在启动每日日程生成任务...") task = DailyScheduleGenerationTask(self) @@ -44,33 +59,50 @@ class ScheduleManager: logger.info("每日日程生成任务已在运行中。") async def load_or_generate_today_schedule(self): + """ + 加载或生成当天的日程。 + 首先尝试从数据库加载，如果失败或不存在，则调用LLM生成新的日程。 + """ if not global_config.planning_system.schedule_enable: logger.info("日程管理功能已禁用，跳过日程加载和生成。") return today_str = datetime.now().strftime("%Y-%m-%d") try: + # 尝试从数据库加载日程 schedule_data = await self._load_schedule_from_db(today_str) if schedule_data: self.today_schedule = schedule_data self._log_loaded_schedule(today_str) return + # 如果数据库中没有，则生成新的日程 logger.info(f"数据库中未找到今天的日程 ({today_str})，将调用 LLM 生成。") await self.generate_and_save_schedule() except Exception as e: + # 如果加载过程中出现任何异常，则尝试生成日程作为备用方案 logger.error(f"加载或生成日程时出错: {e}") logger.info("尝试生成日程作为备用方案...") await self.generate_and_save_schedule() async def _load_schedule_from_db(self, date_str: str) -> list[dict[str, Any]] | None: + """ + 从数据库中加载指定日期的日程。 + + Args: + date_str (str): 日期字符串，格式为 "YYYY-MM-DD"。 + + Returns: + list[dict[str, Any]] | None: 如果找到并验证成功，则返回日程数据，否则返回 None。 + """ async with get_db_session() as session: result = await session.execute(select(Schedule).filter(Schedule.date == date_str)) schedule_record = result.scalars().first() if schedule_record: logger.info(f"从数据库加载今天的日程 ({date_str})。") schedule_data = orjson.loads(str(schedule_record.schedule_data)) + # 验证数据格式是否符合 Pydantic 模型 if self._validate_schedule_with_pydantic(schedule_data): return schedule_data else: @@ -78,6 +110,12 @@ class ScheduleManager: return None def _log_loaded_schedule(self, date_str: str): + """ + 记录已成功加载的日程信息。 + + Args: + date_str (str): 日期字符串。 + """ schedule_str = f"已成功加载今天的日程 ({date_str})：\n" if self.today_schedule: for item in self.today_schedule: @@ -85,6 +123,10 @@ class ScheduleManager: logger.info(schedule_str) async def generate_and_save_schedule(self): + """ + 提交一个按需生成的后台任务来创建和保存日程。 + 这种设计可以防止在主流程中长时间等待LLM响应。 + """ if self.schedule_generation_running: logger.info("日程生成任务已在运行中，跳过重复启动") return @@ -93,23 +135,31 @@ class ScheduleManager: await async_task_manager.add_task(task) async def _async_generate_and_save_schedule(self): + """ + 实际执行日程生成和保存的异步方法。 + 这个方法由后台任务调用。 + """ self.schedule_generation_running = True try: today_str = datetime.now().strftime("%Y-%m-%d") current_month_str = datetime.now().strftime("%Y-%m") + # 如果启用了月度计划，则获取一些计划作为生成日程的参考 sampled_plans = [] if global_config.planning_system.monthly_plan_enable: await self.plan_manager.ensure_and_generate_plans_if_needed(current_month_str) sampled_plans = await self.plan_manager.get_plans_for_schedule(current_month_str, max_count=3) + # 调用LLM生成日程数据 schedule_data = await self.llm_generator.generate_schedule_with_llm(sampled_plans) if schedule_data: + # 保存到数据库 await self._save_schedule_to_db(today_str, schedule_data) self.today_schedule = schedule_data self._log_generated_schedule(today_str, schedule_data, sampled_plans) + # 如果参考了月度计划，则更新这些计划的使用情况 if sampled_plans: used_plan_ids = [plan.id for plan in sampled_plans] logger.info(f"更新使用过的月度计划 {used_plan_ids} 的统计信息。") @@ -120,14 +170,24 @@ class ScheduleManager: @staticmethod async def _save_schedule_to_db(date_str: str, schedule_data: list[dict[str, Any]]): + """ + 将日程数据保存到数据库。如果已有记录则更新，否则创建新记录。 + + Args: + date_str (str): 日期字符串。 + schedule_data (list[dict[str, Any]]): 日程数据。 + """ async with get_db_session() as session: schedule_json = orjson.dumps(schedule_data).decode("utf-8") + # 查找是否已存在当天的日程记录 result = await session.execute(select(Schedule).filter(Schedule.date == date_str)) existing_schedule = result.scalars().first() if existing_schedule: + # 更新现有记录 existing_schedule.schedule_data = schedule_json existing_schedule.updated_at = datetime.now() else: + # 创建新记录 new_schedule = Schedule(date=date_str, schedule_data=schedule_json) session.add(new_schedule) await session.commit() @@ -136,6 +196,14 @@ class ScheduleManager: def _log_generated_schedule( date_str: str, schedule_data: list[dict[str, Any]], sampled_plans: list[MonthlyPlan] ): + """ + 记录成功生成并保存的日程信息。 + + Args: + date_str (str): 日期字符串。 + schedule_data (list[dict[str, Any]]): 日程数据。 + sampled_plans (list[MonthlyPlan]]): 用于生成日程的参考月度计划。 + """ schedule_str = f"成功生成并保存今天的日程 ({date_str})：\n" if sampled_plans: @@ -148,6 +216,12 @@ class ScheduleManager: logger.info(schedule_str) def get_current_activity(self) -> dict[str, Any] | None: + """ + 根据当前时间从日程表中获取正在进行的活动。 + + Returns: + dict[str, Any] | None: 如果找到当前活动，则返回包含活动和时间范围的字典，否则返回 None。 + """ if not global_config.planning_system.schedule_enable or not self.today_schedule: return None now = datetime.now().time() @@ -157,9 +231,11 @@ class ScheduleManager: activity = event.get("activity") if not time_range or not activity: continue + # 解析时间范围 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() + # 判断当前时间是否在时间范围内（支持跨天的时间范围，如 23:00-01:00） if (start_time <= now < end_time) or (end_time < start_time and (now >= start_time or now < end_time)): return {"activity": activity, "time_range": time_range} except (ValueError, KeyError, AttributeError) as e: @@ -168,6 +244,15 @@ class ScheduleManager: @staticmethod def _validate_schedule_with_pydantic(schedule_data) -> bool: + """ + 使用 Pydantic 模型验证日程数据的格式和内容是否正确。 + + Args: + schedule_data: 待验证的日程数据。 + + Returns: + bool: 如果验证通过则返回 True，否则返回 False。 + """ try: ScheduleData(schedule=schedule_data) return True @@ -176,26 +261,53 @@ class ScheduleManager: class OnDemandScheduleGenerationTask(AsyncTask): + """ + 一个按需执行的后台任务，用于生成当天的日程。 + 当启动时未找到日程或加载失败时触发。 + """ def __init__(self, schedule_manager: "ScheduleManager"): + """ + 初始化按需日程生成任务。 + + Args: + schedule_manager (ScheduleManager): ScheduleManager 的实例。 + """ task_name = f"OnDemandScheduleGenerationTask-{datetime.now().strftime('%Y%m%d%H%M%S')}" super().__init__(task_name=task_name) self.schedule_manager = schedule_manager async def run(self): + """ + 任务的执行体，调用 ScheduleManager 中的核心生成逻辑。 + """ logger.info(f"后台任务 {self.task_name} 开始执行日程生成。") await self.schedule_manager._async_generate_and_save_schedule() logger.info(f"后台任务 {self.task_name} 完成。") class DailyScheduleGenerationTask(AsyncTask): + """ + 一个周期性执行的后台任务，用于在每天零点自动生成新一天的日程。 + """ def __init__(self, schedule_manager: "ScheduleManager"): + """ + 初始化每日日程生成任务。 + + Args: + schedule_manager (ScheduleManager): ScheduleManager 的实例。 + """ super().__init__(task_name="DailyScheduleGenerationTask") self.schedule_manager = schedule_manager async def run(self): + """ + 任务的执行体，无限循环直到被取消。 + 计算到下一个零点的时间并休眠，然后在零点过后触发日程生成。 + """ while True: try: now = datetime.now() + # 计算下一个零点的时间 tomorrow = now.date() + timedelta(days=1) midnight = datetime.combine(tomorrow, time.min) sleep_seconds = (midnight - now).total_seconds() @@ -203,14 +315,17 @@ class DailyScheduleGenerationTask(AsyncTask): f"下一次日程生成任务将在 {sleep_seconds:.2f} 秒后运行 (北京时间 {midnight.strftime('%Y-%m-%d %H:%M:%S')})" ) await asyncio.sleep(sleep_seconds) + # 到达零点，开始生成 logger.info("到达每日零点，开始生成新的一天日程...") await self.schedule_manager._async_generate_and_save_schedule() except asyncio.CancelledError: logger.info("每日日程生成任务被取消。") break except Exception as e: + # 发生未知错误时，记录日志并短暂休眠后重试，避免任务崩溃 logger.error(f"每日日程生成任务发生未知错误: {e}") await asyncio.sleep(300) +# 创建 ScheduleManager 的单例 schedule_manager = ScheduleManager()