MemOS-MCP

#!/usr/bin/env python3 """ MemOS 每日回顾模块 生成美观的Markdown格式每日记忆回顾报告 """ import os import json from pathlib import Path from datetime import datetime, timedelta from typing import Dict, Any, List, Optional from mvp_memory import MVPMemoryManager class DailyReviewGenerator: """每日回顾生成器""" def __init__(self, data_dir: str = "./memos_data"): """初始化每日回顾生成器""" self.data_dir = Path(data_dir) self.review_dir = self.data_dir / "daily_reviews" self.review_dir.mkdir(exist_ok=True) # 初始化MVP记忆管理器 self.mvp_manager = MVPMemoryManager(data_dir) def generate_daily_review(self, date: Optional[str] = None) -> str: """生成指定日期的每日回顾""" if date is None: date = datetime.now().strftime("%Y-%m-%d") print(f"📅 正在生成 {date} 的每日回顾...") # 获取当日记忆 daily_memories = self.mvp_manager.get_daily_memories(date) # 生成Markdown内容 markdown_content = self._generate_markdown_report(date, daily_memories) # 保存到文件 review_file = self.review_dir / f"{date}.md" with open(review_file, 'w', encoding='utf-8') as f: f.write(markdown_content) print(f"✅ 每日回顾已保存到: {review_file}") return str(review_file) def _generate_markdown_report(self, date: str, memories: List[Dict[str, Any]]) -> str: """生成Markdown格式的回顾报告""" # 统计信息 total_memories = len(memories) tags_count = {} for memory in memories: if 'metadata' in memory and memory['metadata']: metadata = memory['metadata'] if 'tags' in metadata and metadata['tags']: for tag in metadata['tags']: tags_count[tag] = tags_count.get(tag, 0) + 1 # 生成Markdown内容 markdown = f"""# 📅 {date} 每日记忆回顾 ## 📊 统计概览 - **记忆总数**: {total_memories} 条 - **生成时间**: {datetime.now().strftime("%Y-%m-%d %H:%M:%S")} """ # 标签统计 if tags_count: markdown += "## 🏷️ 标签统计\n\n" sorted_tags = sorted(tags_count.items(), key=lambda x: x[1], reverse=True) for tag, count in sorted_tags: markdown += f"- **{tag}**: {count} 条\n" markdown += "\n" # 记忆详情 if memories: markdown += "## 📝 记忆详情\n\n" for i, memory in enumerate(memories, 1): content = memory.get('content', 'N/A') markdown += f"### {i}. {content[:50]}{'...' if len(content) > 50 else ''}\n\n" markdown += f"**完整内容**: {content}\n\n" if 'metadata' in memory and memory['metadata']: metadata = memory['metadata'] if 'tags' in metadata and metadata['tags']: tags_str = ', '.join(metadata['tags']) markdown += f"**标签**: {tags_str}\n\n" if 'timestamp' in metadata: markdown += f"**时间**: {metadata['timestamp']}\n\n" if 'source' in metadata: markdown += f"**来源**: {metadata['source']}\n\n" markdown += "---\n\n" else: markdown += "## 📝 记忆详情\n\n" markdown += "今日暂无记忆记录。\n\n" # 添加回顾建议 markdown += self._generate_review_suggestions(memories, tags_count) return markdown def _generate_review_suggestions(self, memories: List[Dict[str, Any]], tags_count: Dict[str, int]) -> str: """生成回顾建议""" suggestions = "## 💡 回顾建议\n\n" if not memories: suggestions += "- 今日没有记录任何记忆，建议明天多使用一键记忆功能记录重要信息\n" suggestions += "- 可以尝试记录学习心得、工作进展或生活感悟\n\n" return suggestions # 基于记忆数量的建议 if len(memories) < 3: suggestions += "- 今日记忆较少，建议增加记录频率\n" elif len(memories) > 10: suggestions += "- 今日记忆丰富，建议定期整理和分类\n" else: suggestions += "- 今日记忆数量适中，保持良好的记录习惯\n" # 基于标签的建议 if tags_count: most_common_tag = max(tags_count.items(), key=lambda x: x[1]) suggestions += f"- 今日最关注的主题是「{most_common_tag[0]}」，可以考虑深入学习\n" if len(tags_count) == 1: suggestions += "- 建议扩展记录范围，涵盖更多主题领域\n" else: suggestions += "- 建议为记忆添加标签，便于后续分类和检索\n" suggestions += "\n" return suggestions def generate_weekly_summary(self, end_date: Optional[str] = None) -> str: """生成周总结""" if end_date is None: end_date = datetime.now().strftime("%Y-%m-%d") end_dt = datetime.strptime(end_date, "%Y-%m-%d") start_dt = end_dt - timedelta(days=6) # 7天 print(f"📊 正在生成 {start_dt.strftime('%Y-%m-%d')} 到 {end_date} 的周总结...") # 收集一周的记忆 weekly_memories = [] daily_stats = {} for i in range(7): current_date = (start_dt + timedelta(days=i)).strftime("%Y-%m-%d") daily_memories = self.mvp_manager.get_daily_memories(current_date) weekly_memories.extend(daily_memories) daily_stats[current_date] = len(daily_memories) # 生成周总结Markdown week_range = f"{start_dt.strftime('%Y-%m-%d')}_to_{end_date}" markdown_content = self._generate_weekly_markdown(week_range, weekly_memories, daily_stats) # 保存文件 summary_file = self.review_dir / f"weekly_{week_range}.md" with open(summary_file, 'w', encoding='utf-8') as f: f.write(markdown_content) print(f"✅ 周总结已保存到: {summary_file}") return str(summary_file) def _generate_weekly_markdown(self, week_range: str, memories: List[Dict[str, Any]], daily_stats: Dict[str, int]) -> str: """生成周总结Markdown""" total_memories = len(memories) avg_daily = total_memories / 7 markdown = f"""# 📊 {week_range} 周总结 ## 📈 整体统计 - **总记忆数**: {total_memories} 条 - **日均记忆**: {avg_daily:.1f} 条 - **生成时间**: {datetime.now().strftime("%Y-%m-%d %H:%M:%S")} ## 📅 每日分布 """ for date, count in daily_stats.items(): bar = "█" * min(count, 20) # 最多20个字符的条形图 markdown += f"- **{date}**: {count} 条 {bar}\n" markdown += "\n" # 标签统计 tags_count = {} for memory in memories: if 'metadata' in memory and memory['metadata']: metadata = memory['metadata'] if 'tags' in metadata and metadata['tags']: for tag in metadata['tags']: tags_count[tag] = tags_count.get(tag, 0) + 1 if tags_count: markdown += "## 🏷️ 热门标签\n\n" sorted_tags = sorted(tags_count.items(), key=lambda x: x[1], reverse=True)[:10] for tag, count in sorted_tags: markdown += f"- **{tag}**: {count} 次\n" markdown += "\n" # 周回顾建议 markdown += "## 💭 周回顾\n\n" if total_memories == 0: markdown += "本周没有记录任何记忆，建议下周开始使用MemOS记录重要信息。\n\n" elif total_memories < 7: markdown += "本周记忆较少，建议增加记录频率，每天至少记录一条重要信息。\n\n" else: markdown += "本周记忆记录良好，继续保持这个习惯！\n\n" if tags_count: most_active_tag = max(tags_count.items(), key=lambda x: x[1]) markdown += f"本周最关注的主题是「{most_active_tag[0]}」，可以考虑深入总结相关经验。\n\n" return markdown def backup_memories(self, backup_name: Optional[str] = None) -> str: """备份所有记忆""" if backup_name is None: backup_name = f"backup_{datetime.now().strftime('%Y%m%d_%H%M%S')}" backup_dir = self.data_dir / "backups" / backup_name print(f"💾 正在备份记忆到: {backup_dir}") success = self.mvp_manager.dump_memories(str(backup_dir)) if success: print(f"✅ 记忆备份完成: {backup_dir}") return str(backup_dir) else: print("❌ 记忆备份失败") return "" def main(): """主函数 - 用于测试""" generator = DailyReviewGenerator() # 生成今日回顾 today_review = generator.generate_daily_review() print(f"今日回顾: {today_review}") # 生成周总结 weekly_summary = generator.generate_weekly_summary() print(f"周总结: {weekly_summary}") # 备份记忆 backup_path = generator.backup_memories() print(f"备份路径: {backup_path}") if __name__ == "__main__": main()