From 6a805b7b22c636781f702b659fa611e7d9e1e7b0 Mon Sep 17 00:00:00 2001 From: SengokuCola <1026294844@qq.com> Date: Wed, 26 Mar 2025 15:04:04 +0800 Subject: [PATCH] =?UTF-8?q?better=20=E6=96=B0=E6=97=A5=E7=A8=8B=E7=B3=BB?= =?UTF-8?q?=E7=BB=9F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../schedule/schedule_generator copy.py | 191 --------------- .../schedule/schedule_generator_pro.py | 222 ++++++++++++++++++ 2 files changed, 222 insertions(+), 191 deletions(-) delete mode 100644 src/plugins/schedule/schedule_generator copy.py create mode 100644 src/plugins/schedule/schedule_generator_pro.py diff --git a/src/plugins/schedule/schedule_generator copy.py b/src/plugins/schedule/schedule_generator copy.py deleted file mode 100644 index eff0a08d6..000000000 --- a/src/plugins/schedule/schedule_generator copy.py +++ /dev/null @@ -1,191 +0,0 @@ -import datetime -import json -import re -import os -import sys -from typing import Dict, Union - - -# 添加项目根目录到 Python 路径 -root_path = os.path.abspath(os.path.join(os.path.dirname(__file__), "../../..")) -sys.path.append(root_path) - -from src.common.database import db # noqa: E402 -from src.common.logger import get_module_logger # noqa: E402 -from src.plugins.schedule.offline_llm import LLMModel # noqa: E402 -from src.plugins.chat.config import global_config # noqa: E402 - -logger = get_module_logger("scheduler") - - -class ScheduleGenerator: - enable_output: bool = True - - def __init__(self): - # 使用离线LLM模型 - self.llm_scheduler = LLMModel(model_name="Pro/deepseek-ai/DeepSeek-V3", temperature=0.9) - self.today_schedule_text = "" - self.today_schedule = {} - self.tomorrow_schedule_text = "" - self.tomorrow_schedule = {} - self.yesterday_schedule_text = "" - self.yesterday_schedule = {} - - async def initialize(self): - today = datetime.datetime.now() - tomorrow = datetime.datetime.now() + datetime.timedelta(days=1) - yesterday = datetime.datetime.now() - datetime.timedelta(days=1) - - self.today_schedule_text, self.today_schedule = await self.generate_daily_schedule(target_date=today) - self.tomorrow_schedule_text, self.tomorrow_schedule = await self.generate_daily_schedule( - target_date=tomorrow, read_only=True - ) - self.yesterday_schedule_text, self.yesterday_schedule = await self.generate_daily_schedule( - target_date=yesterday, read_only=True - ) - - async def generate_daily_schedule( - self, target_date: datetime.datetime = None, read_only: bool = False - ) -> Dict[str, str]: - date_str = target_date.strftime("%Y-%m-%d") - weekday = target_date.strftime("%A") - - schedule_text = str - - existing_schedule = db.schedule.find_one({"date": date_str}) - if existing_schedule: - if self.enable_output: - logger.debug(f"{date_str}的日程已存在:") - schedule_text = existing_schedule["schedule"] - # print(self.schedule_text) - - elif not read_only: - logger.debug(f"{date_str}的日程不存在，准备生成新的日程。") - prompt = ( - f"""我是{global_config.BOT_NICKNAME}，{global_config.PROMPT_SCHEDULE_GEN}，请为我生成{date_str}（{weekday}）的日程安排，包括：""" - + """ - 1. 早上的学习和工作安排 - 2. 下午的活动和任务 - 3. 晚上的计划和休息时间 - 请按照时间顺序列出具体时间点和对应的活动，用一个时间点而不是时间段来表示时间，用JSON格式返回日程表， - 仅返回内容，不要返回注释，不要添加任何markdown或代码块样式，时间采用24小时制， - 格式为{"时间": "活动","时间": "活动",...}。""" - ) - - try: - schedule_text, _ = self.llm_scheduler.generate_response(prompt) - db.schedule.insert_one({"date": date_str, "schedule": schedule_text}) - self.enable_output = True - except Exception as e: - logger.error(f"生成日程失败: {str(e)}") - schedule_text = "生成日程时出错了" - # print(self.schedule_text) - else: - if self.enable_output: - logger.debug(f"{date_str}的日程不存在。") - schedule_text = "忘了" - - return schedule_text, None - - schedule_form = self._parse_schedule(schedule_text) - return schedule_text, schedule_form - - def _parse_schedule(self, schedule_text: str) -> Union[bool, Dict[str, str]]: - """解析日程文本，转换为时间和活动的字典""" - try: - reg = r"\{(.|\r|\n)+\}" - matched = re.search(reg, schedule_text)[0] - schedule_dict = json.loads(matched) - return schedule_dict - except json.JSONDecodeError: - logger.exception("解析日程失败: {}".format(schedule_text)) - return False - - def _parse_time(self, time_str: str) -> str: - """解析时间字符串，转换为时间""" - return datetime.datetime.strptime(time_str, "%H:%M") - - def get_current_task(self) -> str: - """获取当前时间应该进行的任务""" - current_time = datetime.datetime.now().strftime("%H:%M") - - # 找到最接近当前时间的任务 - closest_time = None - min_diff = float("inf") - - # 检查今天的日程 - if not self.today_schedule: - return "摸鱼" - for time_str in self.today_schedule.keys(): - diff = abs(self._time_diff(current_time, time_str)) - if closest_time is None or diff < min_diff: - closest_time = time_str - min_diff = diff - - # 检查昨天的日程中的晚间任务 - if self.yesterday_schedule: - for time_str in self.yesterday_schedule.keys(): - if time_str >= "20:00": # 只考虑晚上8点之后的任务 - # 计算与昨天这个时间点的差异（需要加24小时） - diff = abs(self._time_diff(current_time, time_str)) - if diff < min_diff: - closest_time = time_str - min_diff = diff - return closest_time, self.yesterday_schedule[closest_time] - - if closest_time: - return closest_time, self.today_schedule[closest_time] - return "摸鱼" - - def _time_diff(self, time1: str, time2: str) -> int: - """计算两个时间字符串之间的分钟差""" - if time1 == "24:00": - time1 = "23:59" - if time2 == "24:00": - time2 = "23:59" - t1 = datetime.datetime.strptime(time1, "%H:%M") - t2 = datetime.datetime.strptime(time2, "%H:%M") - diff = int((t2 - t1).total_seconds() / 60) - # 考虑时间的循环性 - if diff < -720: - diff += 1440 # 加一天的分钟 - elif diff > 720: - diff -= 1440 # 减一天的分钟 - # print(f"时间1[{time1}]: 时间2[{time2}]，差值[{diff}]分钟") - return diff - - def print_schedule(self): - """打印完整的日程安排""" - if not self._parse_schedule(self.today_schedule_text): - logger.warning("今日日程有误，将在下次运行时重新生成") - db.schedule.delete_one({"date": datetime.datetime.now().strftime("%Y-%m-%d")}) - else: - logger.info("=== 今日日程安排 ===") - for time_str, activity in self.today_schedule.items(): - logger.info(f"时间[{time_str}]: 活动[{activity}]") - logger.info("==================") - self.enable_output = False - - -async def main(): - # 使用示例 - scheduler = ScheduleGenerator() - await scheduler.initialize() - scheduler.print_schedule() - print("\n当前任务：") - print(await scheduler.get_current_task()) - - print("昨天日程：") - print(scheduler.yesterday_schedule) - print("今天日程：") - print(scheduler.today_schedule) - print("明天日程：") - print(scheduler.tomorrow_schedule) - -# 当作为组件导入时使用的实例 -bot_schedule = ScheduleGenerator() - -if __name__ == "__main__": - import asyncio - # 当直接运行此文件时执行 - asyncio.run(main()) diff --git a/src/plugins/schedule/schedule_generator_pro.py b/src/plugins/schedule/schedule_generator_pro.py new file mode 100644 index 000000000..5a2c2a687 --- /dev/null +++ b/src/plugins/schedule/schedule_generator_pro.py @@ -0,0 +1,222 @@ +import datetime +import json +import re +import os +import sys +from typing import Dict, Union +# 添加项目根目录到 Python 路径 +root_path = os.path.abspath(os.path.join(os.path.dirname(__file__), "../../..")) +sys.path.append(root_path) + +from src.common.database import db # noqa: E402 +from src.common.logger import get_module_logger # noqa: E402 +from src.plugins.schedule.offline_llm import LLMModel # noqa: E402 + +logger = get_module_logger("scheduler") + + +class ScheduleGenerator: + enable_output: bool = True + + def __init__(self, name: str = "bot_name", personality: str = "你是一个爱国爱党的新时代青年", behavior: str = "你非常外向，喜欢尝试新事物和人交流"): + # 使用离线LLM模型 + self.llm_scheduler = LLMModel(model_name="Pro/deepseek-ai/DeepSeek-V3", temperature=0.9) + + self.today_schedule_text = "" + self.today_done_list = [] + + self.yesterday_schedule_text = "" + self.yesterday_done_list = [] + + self.name = name + self.personality = personality + self.behavior = behavior + + self.start_time = datetime.datetime.now() + + async def mai_schedule_start(self): + """启动日程系统，每5分钟执行一次move_doing，并在日期变化时重新检查日程""" + try: + logger.info(f"日程系统启动/刷新时间: {self.start_time.strftime('%Y-%m-%d %H:%M:%S')}") + # 初始化日程 + await self.check_and_create_today_schedule() + self.print_schedule() + + while True: + current_time = datetime.datetime.now() + + # 检查是否需要重新生成日程（日期变化） + if current_time.date() != self.start_time.date(): + logger.info("检测到日期变化，重新生成日程") + self.start_time = current_time + await self.check_and_create_today_schedule() + self.print_schedule() + + # 执行当前活动 + current_activity = await self.move_doing() + logger.info(f"当前活动: {current_activity}") + + # 等待5分钟 + await asyncio.sleep(300) # 300秒 = 5分钟 + + except Exception as e: + logger.error(f"日程系统运行时出错: {str(e)}") + logger.exception("详细错误信息：") + + async def check_and_create_today_schedule(self): + """检查昨天的日程，并确保今天有日程安排 + + Returns: + tuple: (today_schedule_text, today_schedule) 今天的日程文本和解析后的日程字典 + """ + today = datetime.datetime.now() + yesterday = today - datetime.timedelta(days=1) + + # 先检查昨天的日程 + self.yesterday_schedule_text, self.yesterday_done_list = self.load_schedule_from_db(yesterday) + if self.yesterday_schedule_text: + logger.debug(f"已加载{yesterday.strftime('%Y-%m-%d')}的日程") + + # 检查今天的日程 + self.today_schedule_text, self.today_done_list = self.load_schedule_from_db(today) + if not self.today_schedule_text: + logger.info(f"{today.strftime('%Y-%m-%d')}的日程不存在，准备生成新的日程") + self.today_schedule_text = await self.generate_daily_schedule(target_date=today) + + self.save_today_schedule_to_db() + + def construct_daytime_prompt(self, target_date: datetime.datetime): + date_str = target_date.strftime("%Y-%m-%d") + weekday = target_date.strftime("%A") + + prompt = f"我是{self.name}，{self.personality}，{self.behavior}" + prompt += f"我昨天的日程是：{self.yesterday_schedule_text}\n" + prompt += f"请为我生成{date_str}（{weekday}）的日程安排，结合我的个人特点和行为习惯\n" + prompt += "推测我的日程安排，包括我一天都在做什么，有什么发现和思考，具体一些，详细一些，记得写明时间\n" + prompt += "直接返回我的日程，不要输出其他内容：" + return prompt + + def construct_doing_prompt(self,time: datetime.datetime): + now_time = time.strftime("%H:%M") + previous_doing = self.today_done_list[-20:] if len(self.today_done_list) > 20 else self.today_done_list + prompt = f"我是{self.name}，{self.personality}，{self.behavior}" + prompt += f"我今天的日程是：{self.today_schedule_text}\n" + prompt += f"我之前做了的事情是：{previous_doing}\n" + prompt += f"现在是{now_time}，结合我的个人特点和行为习惯," + prompt += "推测我现在做什么，具体一些，详细一些\n" + prompt += "直接返回我在做的事情，不要输出其他内容：" + return prompt + + async def generate_daily_schedule( + self, target_date: datetime.datetime = None,) -> Dict[str, str]: + daytime_prompt = self.construct_daytime_prompt(target_date) + daytime_response, _ = await self.llm_scheduler.generate_response(daytime_prompt) + return daytime_response + + def _time_diff(self, time1: str, time2: str) -> int: + """计算两个时间字符串之间的分钟差""" + if time1 == "24:00": + time1 = "23:59" + if time2 == "24:00": + time2 = "23:59" + t1 = datetime.datetime.strptime(time1, "%H:%M") + t2 = datetime.datetime.strptime(time2, "%H:%M") + diff = int((t2 - t1).total_seconds() / 60) + # 考虑时间的循环性 + if diff < -720: + diff += 1440 # 加一天的分钟 + elif diff > 720: + diff -= 1440 # 减一天的分钟 + # print(f"时间1[{time1}]: 时间2[{time2}]，差值[{diff}]分钟") + return diff + + def print_schedule(self): + """打印完整的日程安排""" + if not self.today_schedule_text: + logger.warning("今日日程有误，将在下次运行时重新生成") + db.schedule.delete_one({"date": datetime.datetime.now().strftime("%Y-%m-%d")}) + else: + logger.info("=== 今日日程安排 ===") + logger.info(self.today_schedule_text) + logger.info("==================") + self.enable_output = False + + async def update_today_done_list(self): + # 更新数据库中的 today_done_list + today_str = datetime.datetime.now().strftime("%Y-%m-%d") + existing_schedule = db.schedule.find_one({"date": today_str}) + + if existing_schedule: + # 更新数据库中的 today_done_list + db.schedule.update_one( + {"date": today_str}, + {"$set": {"today_done_list": self.today_done_list}} + ) + logger.debug(f"已更新{today_str}的已完成活动列表") + else: + logger.warning(f"未找到{today_str}的日程记录") + + async def move_doing(self): + current_time = datetime.datetime.now() + time_str = current_time.strftime("%H:%M") + doing_prompt = self.construct_doing_prompt(current_time) + doing_response, _ = await self.llm_scheduler.generate_response(doing_prompt) + self.today_done_list.append(current_time,time_str + "在" + doing_response) + + await self.update_today_done_list() + + return doing_response + + + + + def save_today_schedule_to_db(self): + """保存日程到数据库，同时初始化 today_done_list""" + date_str = datetime.datetime.now().strftime("%Y-%m-%d") + schedule_data = { + "date": date_str, + "schedule": self.today_schedule_text, + "today_done_list": self.today_done_list if hasattr(self, 'today_done_list') else [] + } + # 使用 upsert 操作，如果存在则更新，不存在则插入 + db.schedule.update_one( + {"date": date_str}, + {"$set": schedule_data}, + upsert=True + ) + logger.debug(f"已保存{date_str}的日程到数据库") + + def load_schedule_from_db(self, date: datetime.datetime): + """从数据库加载日程，同时加载 today_done_list""" + date_str = date.strftime("%Y-%m-%d") + existing_schedule = db.schedule.find_one({"date": date_str}) + + if existing_schedule: + schedule_text = existing_schedule["schedule"] + return schedule_text, existing_schedule.get("today_done_list", []) + else: + logger.debug(f"{date_str}的日程不存在") + return None, None + +async def main(): + # 使用示例 + scheduler = ScheduleGenerator(name="麦麦", personality="你叫麦麦，你19岁，是一个大二的女大学生，你有一头黑色短发，你会刷贴吧，你现在在学习心理学", behavior="你比较内向") + await scheduler.check_and_create_today_schedule() + scheduler.print_schedule() + print("\n当前任务：") + print(await scheduler.get_current_task()) + + print("昨天日程：") + print(scheduler.yesterday_schedule) + print("今天日程：") + print(scheduler.today_schedule) + print("明天日程：") + print(scheduler.tomorrow_schedule) + +# 当作为组件导入时使用的实例 +bot_schedule = ScheduleGenerator() + +if __name__ == "__main__": + import asyncio + # 当直接运行此文件时执行 + asyncio.run(main())