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

fix: 优化scheduler_generator日志&格式

Browse Source
This commit is contained in:
AL76
2025-03-09 22:40:32 +08:00
parent 31a5514a0f
commit 980cde7230
1 changed files with 52 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files

102
src/plugins/schedule/schedule_generator.py
View File

@@ -13,21 +13,21 @@ from ..models.utils_model import LLM_request
driver = get_driver() driver = get_driver()
config = driver.config config = driver.config
Database.initialize( Database.initialize(
host= config.MONGODB_HOST, host=config.MONGODB_HOST,
port= int(config.MONGODB_PORT), port=int(config.MONGODB_PORT),
db_name= config.DATABASE_NAME, db_name=config.DATABASE_NAME,
username= config.MONGODB_USERNAME, username=config.MONGODB_USERNAME,
password= config.MONGODB_PASSWORD, password=config.MONGODB_PASSWORD,
auth_source=config.MONGODB_AUTH_SOURCE auth_source=config.MONGODB_AUTH_SOURCE
) )
class ScheduleGenerator: class ScheduleGenerator:
def __init__(self): def __init__(self):
#根据global_config.llm_normal这一字典配置指定模型 # 根据global_config.llm_normal这一字典配置指定模型
# self.llm_scheduler = LLMModel(model = global_config.llm_normal,temperature=0.9) # self.llm_scheduler = LLMModel(model = global_config.llm_normal,temperature=0.9)
self.llm_scheduler = LLM_request(model = global_config.llm_normal,temperature=0.9) self.llm_scheduler = LLM_request(model=global_config.llm_normal, temperature=0.9)
self.db = Database.get_instance() self.db = Database.get_instance()
self.today_schedule_text = "" self.today_schedule_text = ""
self.today_schedule = {} self.today_schedule = {}
@@ -35,39 +35,41 @@ class ScheduleGenerator:
self.tomorrow_schedule = {} self.tomorrow_schedule = {}
self.yesterday_schedule_text = "" self.yesterday_schedule_text = ""
self.yesterday_schedule = {} self.yesterday_schedule = {}
async def initialize(self): async def initialize(self):
today = datetime.datetime.now() today = datetime.datetime.now()
tomorrow = datetime.datetime.now() + datetime.timedelta(days=1) tomorrow = datetime.datetime.now() + datetime.timedelta(days=1)
yesterday = 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.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.tomorrow_schedule_text, self.tomorrow_schedule = await self.generate_daily_schedule(target_date=tomorrow,
self.yesterday_schedule_text, self.yesterday_schedule = await self.generate_daily_schedule(target_date=yesterday,read_only=True) read_only=True)
self.yesterday_schedule_text, self.yesterday_schedule = await self.generate_daily_schedule(
async def generate_daily_schedule(self, target_date: datetime.datetime = None,read_only:bool = False) -> Dict[str, str]: 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") date_str = target_date.strftime("%Y-%m-%d")
weekday = target_date.strftime("%A") weekday = target_date.strftime("%A")
schedule_text = str schedule_text = str
existing_schedule = self.db.db.schedule.find_one({"date": date_str}) existing_schedule = self.db.db.schedule.find_one({"date": date_str})
if existing_schedule: if existing_schedule:
print(f"{date_str}的日程已存在:") logger.info(f"{date_str}的日程已存在:")
schedule_text = existing_schedule["schedule"] schedule_text = existing_schedule["schedule"]
# print(self.schedule_text) # print(self.schedule_text)
elif read_only == False: elif read_only == False:
print(f"{date_str}的日程不存在，准备生成新的日程。") logger.info(f"{date_str}的日程不存在，准备生成新的日程。")
prompt = f"""我是{global_config.BOT_NICKNAME}，{global_config.PROMPT_SCHEDULE_GEN}，请为我生成{date_str}（{weekday}）的日程安排，包括："""+\ prompt = f"""我是{global_config.BOT_NICKNAME}，{global_config.PROMPT_SCHEDULE_GEN}，请为我生成{date_str}（{weekday}）的日程安排，包括：""" + \
""" """
1. 早上的学习和工作安排 1. 早上的学习和工作安排
2. 下午的活动和任务 2. 下午的活动和任务
3. 晚上的计划和休息时间 3. 晚上的计划和休息时间
请按照时间顺序列出具体时间点和对应的活动，用一个时间点而不是时间段来表示时间，用JSON格式返回日程表，仅返回内容，不要返回注释，时间采用24小时制，格式为{"时间": "活动","时间": "活动",...}。""" 请按照时间顺序列出具体时间点和对应的活动，用一个时间点而不是时间段来表示时间，用JSON格式返回日程表，仅返回内容，不要返回注释，时间采用24小时制，格式为{"时间": "活动","时间": "活动",...}。"""
try: try:
schedule_text, _ = await self.llm_scheduler.generate_response(prompt) schedule_text, _ = await self.llm_scheduler.generate_response(prompt)
self.db.db.schedule.insert_one({"date": date_str, "schedule": schedule_text}) self.db.db.schedule.insert_one({"date": date_str, "schedule": schedule_text})
@@ -76,36 +78,35 @@ class ScheduleGenerator:
schedule_text = "生成日程时出错了" schedule_text = "生成日程时出错了"
# print(self.schedule_text) # print(self.schedule_text)
else: else:
print(f"{date_str}的日程不存在。") logger.info(f"{date_str}的日程不存在。")
schedule_text = "忘了" schedule_text = "忘了"
return schedule_text,None return schedule_text, None
schedule_form = self._parse_schedule(schedule_text) schedule_form = self._parse_schedule(schedule_text)
return schedule_text,schedule_form return schedule_text, schedule_form
def _parse_schedule(self, schedule_text: str) -> Union[bool, Dict[str, str]]: def _parse_schedule(self, schedule_text: str) -> Union[bool, Dict[str, str]]:
"""解析日程文本，转换为时间和活动的字典""" """解析日程文本，转换为时间和活动的字典"""
try: try:
schedule_dict = json.loads(schedule_text) schedule_dict = json.loads(schedule_text)
return schedule_dict return schedule_dict
except json.JSONDecodeError as e: except json.JSONDecodeError as e:
print(schedule_text) logger.exception("解析日程失败: {}".format(schedule_text))
print(f"解析日程失败: {str(e)}")
return False return False
def _parse_time(self, time_str: str) -> str: def _parse_time(self, time_str: str) -> str:
"""解析时间字符串，转换为时间""" """解析时间字符串，转换为时间"""
return datetime.datetime.strptime(time_str, "%H:%M") return datetime.datetime.strptime(time_str, "%H:%M")
def get_current_task(self) -> str: def get_current_task(self) -> str:
"""获取当前时间应该进行的任务""" """获取当前时间应该进行的任务"""
current_time = datetime.datetime.now().strftime("%H:%M") current_time = datetime.datetime.now().strftime("%H:%M")
# 找到最接近当前时间的任务 # 找到最接近当前时间的任务
closest_time = None closest_time = None
min_diff = float('inf') min_diff = float('inf')
# 检查今天的日程 # 检查今天的日程
if not self.today_schedule: if not self.today_schedule:
return "摸鱼" return "摸鱼"
@@ -114,7 +115,7 @@ class ScheduleGenerator:
if closest_time is None or diff < min_diff: if closest_time is None or diff < min_diff:
closest_time = time_str closest_time = time_str
min_diff = diff min_diff = diff
# 检查昨天的日程中的晚间任务 # 检查昨天的日程中的晚间任务
if self.yesterday_schedule: if self.yesterday_schedule:
for time_str in self.yesterday_schedule.keys(): for time_str in self.yesterday_schedule.keys():
@@ -125,17 +126,17 @@ class ScheduleGenerator:
closest_time = time_str closest_time = time_str
min_diff = diff min_diff = diff
return closest_time, self.yesterday_schedule[closest_time] return closest_time, self.yesterday_schedule[closest_time]
if closest_time: if closest_time:
return closest_time, self.today_schedule[closest_time] return closest_time, self.today_schedule[closest_time]
return "摸鱼" return "摸鱼"
def _time_diff(self, time1: str, time2: str) -> int: def _time_diff(self, time1: str, time2: str) -> int:
"""计算两个时间字符串之间的分钟差""" """计算两个时间字符串之间的分钟差"""
if time1=="24:00": if time1 == "24:00":
time1="23:59" time1 = "23:59"
if time2=="24:00": if time2 == "24:00":
time2="23:59" time2 = "23:59"
t1 = datetime.datetime.strptime(time1, "%H:%M") t1 = datetime.datetime.strptime(time1, "%H:%M")
t2 = datetime.datetime.strptime(time2, "%H:%M") t2 = datetime.datetime.strptime(time2, "%H:%M")
diff = int((t2 - t1).total_seconds() / 60) diff = int((t2 - t1).total_seconds() / 60)
@@ -146,17 +147,18 @@ class ScheduleGenerator:
diff -= 1440 # 减一天的分钟 diff -= 1440 # 减一天的分钟
# print(f"时间1[{time1}]: 时间2[{time2}]，差值[{diff}]分钟") # print(f"时间1[{time1}]: 时间2[{time2}]，差值[{diff}]分钟")
return diff return diff
def print_schedule(self): def print_schedule(self):
"""打印完整的日程安排""" """打印完整的日程安排"""
if not self._parse_schedule(self.today_schedule_text): if not self._parse_schedule(self.today_schedule_text):
print("今日日程有误，将在下次运行时重新生成") logger.warning("今日日程有误，将在下次运行时重新生成")
self.db.db.schedule.delete_one({"date": datetime.datetime.now().strftime("%Y-%m-%d")}) self.db.db.schedule.delete_one({"date": datetime.datetime.now().strftime("%Y-%m-%d")})
else: else:
print("\n=== 今日日程安排 ===") logger.info("\n=== 今日日程安排 ===")
for time_str, activity in self.today_schedule.items(): for time_str, activity in self.today_schedule.items():
print(f"时间[{time_str}]: 活动[{activity}]") logger.info(f"时间[{time_str}]: 活动[{activity}]")
print("==================\n") logger.info("==================\n")
# def main(): # def main():
# # 使用示例 # # 使用示例
@@ -165,7 +167,7 @@ class ScheduleGenerator:
# scheduler.print_schedule() # scheduler.print_schedule()
# print("\n当前任务：") # print("\n当前任务：")
# print(scheduler.get_current_task()) # print(scheduler.get_current_task())
# print("昨天日程：") # print("昨天日程：")
# print(scheduler.yesterday_schedule) # print(scheduler.yesterday_schedule)
# print("今天日程：") # print("今天日程：")
@@ -175,5 +177,5 @@ class ScheduleGenerator:
# if __name__ == "__main__": # if __name__ == "__main__":
# main() # main()
bot_schedule = ScheduleGenerator() bot_schedule = ScheduleGenerator()