MemOS-MCP

#!/usr/bin/env python3 """ MemOS SDK 集成示例 演示如何在实际项目中集成Memory Ops SDK """ import sys import json import time from pathlib import Path from datetime import datetime # 添加父目录到Python路径 sys.path.insert(0, str(Path(__file__).parent.parent)) from memory_ops_sdk import MemoryOps class SmartNoteTaker: """智能笔记应用示例""" def __init__(self, memory_dir: str = "./smart_notes"): """初始化智能笔记应用""" self.mem = MemoryOps(data_dir=memory_dir, verbose=True) self.session_id = datetime.now().strftime("%Y%m%d_%H%M%S") print(f"📝 智能笔记应用启动 (会话ID: {self.session_id})") def add_note(self, content: str, category: str = "general", priority: str = "normal") -> bool: """添加笔记""" metadata = { "session_id": self.session_id, "category": category, "priority": priority, "app": "SmartNoteTaker" } tags = [category, priority, "note"] return self.mem.add(content, tags=tags, metadata=metadata) def search_notes(self, query: str, limit: int = 5) -> List[Dict]: """搜索笔记""" return self.mem.query(query, limit=limit) def rate_note(self, note_id: str, helpful: bool) -> bool: """评价笔记有用性""" feedback = "👍" if helpful else "👎" return self.mem.feedback(note_id, feedback) def get_daily_summary(self) -> Dict: """获取今日笔记摘要""" stats = self.mem.stats() return { "total_notes": len(self.mem.dump()), "session_id": self.session_id, "system_health": self.mem.health(), "stats": stats } def close(self): """关闭应用""" self.mem.close() print("📝 智能笔记应用已关闭") class LearningAssistant: """学习助手示例""" def __init__(self, memory_dir: str = "./learning_memories"): """初始化学习助手""" self.mem = MemoryOps(data_dir=memory_dir, verbose=True) print("🎓 学习助手启动") def record_learning(self, topic: str, content: str, difficulty: str = "medium") -> bool: """记录学习内容""" metadata = { "topic": topic, "difficulty": difficulty, "learning_date": datetime.now().isoformat(), "app": "LearningAssistant" } tags = ["learning", topic, difficulty] return self.mem.add(content, tags=tags, metadata=metadata) def review_topic(self, topic: str) -> List[Dict]: """复习特定主题""" print(f"📚 复习主题: {topic}") results = self.mem.query(topic, limit=10) # 按时间排序，最新的在前面 results.sort(key=lambda x: x.get('metadata', {}).get('learning_date', ''), reverse=True) return results def mark_mastered(self, memory_id: str) -> bool: """标记已掌握""" return self.mem.feedback(memory_id, "👍") def mark_need_review(self, memory_id: str) -> bool: """标记需要复习""" return self.mem.feedback(memory_id, "👎") def get_learning_progress(self) -> Dict: """获取学习进度""" stats = self.mem.stats() memories = self.mem.dump() # 分析学习主题 topics = {} for memory in memories: for tag in memory.get('tags', []): if tag not in ['learning', 'easy', 'medium', 'hard']: topics[tag] = topics.get(tag, 0) + 1 return { "total_learning_records": len(memories), "topics_covered": len(topics), "topic_distribution": topics, "system_stats": stats } def close(self): """关闭学习助手""" self.mem.close() print("🎓 学习助手已关闭") def demo_smart_note_taker(): """演示智能笔记应用""" print("\n📝 智能笔记应用演示") print("-" * 30) app = SmartNoteTaker() try: # 添加一些笔记 app.add_note("今天学习了Python装饰器的用法", "技术", "high") app.add_note("需要复习数据结构中的树结构", "学习", "medium") app.add_note("明天要参加项目会议", "工作", "high") app.add_note("周末计划去图书馆", "生活", "low") # 搜索笔记 results = app.search_notes("Python") print(f"搜索'Python'找到 {len(results)} 条笔记") # 评价笔记 if results: app.rate_note(results[0].get('id'), True) print("已对第一条笔记给出好评") # 获取摘要 summary = app.get_daily_summary() print(f"今日笔记摘要: {summary['total_notes']} 条笔记") finally: app.close() def demo_learning_assistant(): """演示学习助手""" print("\n🎓 学习助手演示") print("-" * 30) assistant = LearningAssistant() try: # 记录学习内容 assistant.record_learning("Python", "学习了列表推导式的高级用法", "medium") assistant.record_learning("算法", "掌握了快速排序的实现原理", "hard") assistant.record_learning("Python", "了解了生成器和迭代器的区别", "medium") assistant.record_learning("数据库", "学习了SQL的JOIN操作", "easy") # 复习主题 python_memories = assistant.review_topic("Python") print(f"Python相关学习记录: {len(python_memories)} 条") # 标记掌握情况 if python_memories: assistant.mark_mastered(python_memories[0].get('id')) print("已标记第一条Python记录为已掌握") # 获取学习进度 progress = assistant.get_learning_progress() print(f"学习进度: 共 {progress['total_learning_records']} 条记录") print(f"涉及主题: {list(progress['topic_distribution'].keys())}") finally: assistant.close() def demo_simple_integration(): """演示简单集成""" print("\n⚡ 简单集成演示") print("-" * 30) # 最简单的使用方式 with MemoryOps(data_dir="./simple_demo", verbose=True) as mem: # 模拟一个简单的问答系统 questions_answers = [ ("什么是Python?", "Python是一种高级编程语言"), ("如何安装Python?", "可以从官网下载安装包或使用包管理器"), ("Python有什么特点?", "语法简洁、易学易用、功能强大") ] # 添加问答对 for question, answer in questions_answers: content = f"问题: {question}\n答案: {answer}" mem.add(content, tags=["问答", "Python"]) # 模拟用户提问 user_question = "Python语言特点" results = mem.query(user_question, limit=3) print(f"用户问题: {user_question}") print(f"找到 {len(results)} 个相关答案:") for i, result in enumerate(results, 1): content = result.get('content', '') score = result.get('score', 0) print(f" {i}. {content[:100]}... (相关度: {score:.3f})") def demo_batch_operations(): """演示批量操作""" print("\n📦 批量操作演示") print("-" * 30) with MemoryOps(data_dir="./batch_demo", verbose=False) as mem: # 批量添加数据 print("批量添加编程概念...") concepts = [ ("变量", "用于存储数据的容器"), ("函数", "可重用的代码块"), ("类", "对象的蓝图或模板"), ("继承", "子类获得父类特性的机制"), ("多态", "同一接口的不同实现"), ("封装", "隐藏内部实现细节"), ("抽象", "简化复杂系统的过程") ] for concept, description in concepts: content = f"{concept}: {description}" mem.add(content, tags=["编程概念", concept]) # 批量查询 queries = ["变量", "面向对象", "函数"] for query in queries: results = mem.query(query, limit=2) print(f"查询'{query}': {len(results)} 个结果") # 获取统计信息 stats = mem.stats() print(f"系统状态: {stats.get('status', {}).get('status', '未知')}") def main(): """主演示函数""" print("🚀 MemOS SDK 集成示例演示") print("=" * 50) try: demo_smart_note_taker() demo_learning_assistant() demo_simple_integration() demo_batch_operations() print("\n🎉 所有集成示例演示完成!") print("\n💡 提示:") print("- 可以根据这些示例创建自己的应用") print("- SDK支持上下文管理器，自动管理资源") print("- 所有操作都有错误处理和日志记录") print("- 支持元数据和标签系统，便于组织记忆") except Exception as e: print(f"\n❌ 演示失败: {e}") import traceback traceback.print_exc() if __name__ == "__main__": main()