MemOS-MCP

#!/usr/bin/env python3 """ MemOS 高级使用示例 演示更复杂的记忆管理和AI应用场景 """ import os import json import datetime from pathlib import Path from usage_examples import SimpleMemOS, load_env_file class AdvancedMemOS(SimpleMemOS): """扩展的MemOS，支持更多高级功能""" def add_learning_note(self, topic, content, difficulty="medium"): """添加学习笔记""" metadata = { "type": "learning", "difficulty": difficulty, "date": datetime.datetime.now().isoformat() } return self.add_memory( content=f"学习笔记 - {topic}: {content}", tags=["学习", topic, difficulty], metadata=metadata ) def add_work_task(self, task, priority="normal", deadline=None): """添加工作任务""" metadata = { "type": "work_task", "priority": priority, "deadline": deadline, "status": "pending" } return self.add_memory( content=f"工作任务: {task}", tags=["工作", "任务", priority], metadata=metadata ) def add_personal_insight(self, insight, category="general"): """添加个人见解""" metadata = { "type": "insight", "category": category, "date": datetime.datetime.now().isoformat() } return self.add_memory( content=f"个人见解: {insight}", tags=["见解", category], metadata=metadata ) def get_memories_by_type(self, memory_type): """根据类型获取记忆""" all_memories = self.get_all_memories() return [m for m in all_memories if m['metadata'].get('type') == memory_type] def get_learning_summary(self): """获取学习总结""" learning_memories = self.get_memories_by_type("learning") if not learning_memories: return "暂无学习记录" # 构建学习总结的提示 learning_content = "\n".join([ f"- {m['content']}" for m in learning_memories ]) prompt = f"""基于以下学习记录，生成一个简洁的学习总结： {learning_content} 请总结： 1. 主要学习的主题 2. 学习的重点内容 3. 建议的下一步学习方向""" return self.chat_with_memory(prompt, use_memory=False) def get_work_dashboard(self): """获取工作仪表板""" work_memories = self.get_memories_by_type("work_task") if not work_memories: return "暂无工作任务" # 按优先级分组 high_priority = [m for m in work_memories if "high" in m['tags']] normal_priority = [m for m in work_memories if "normal" in m['tags']] low_priority = [m for m in work_memories if "low" in m['tags']] dashboard = "📊 工作仪表板\n" dashboard += f"🔴 高优先级任务: {len(high_priority)}个\n" dashboard += f"🟡 普通优先级任务: {len(normal_priority)}个\n" dashboard += f"🟢 低优先级任务: {len(low_priority)}个\n\n" if high_priority: dashboard += "🔴 高优先级任务:\n" for task in high_priority[:3]: # 只显示前3个 dashboard += f" - {task['content']}\n" return dashboard def smart_reminder(self, query): """智能提醒功能""" # 搜索相关记忆 memories = self.search_memories(query, limit=5) if not memories: return f"没有找到与'{query}'相关的记忆。" # 构建提醒内容 reminder_prompt = f"""用户询问: {query} 相关记忆: {chr(10).join([f"- {m['content']}" for m in memories])} 请基于这些记忆，给出一个有用的提醒或建议。""" return self.chat_with_memory(reminder_prompt, use_memory=False) def intelligent_query(self, query, max_memories=5): """智能查询功能 - 为MCP调用设计""" # 搜索相关记忆 memories = self.search_memories(query, limit=max_memories) if not memories: return { "relevant_memories": [], "llm_context": f"没有找到与'{query}'相关的记忆。这可能是一个新的话题。", "query_summary": f"查询: {query} (无相关记忆)" } # 构建记忆摘要 memory_summary = "\n".join([ f"记忆{i+1}: {m['content']} (相关度: {m['score']:.2f})" for i, m in enumerate(memories) ]) # 让LLM组织上下文（移除硬编码提示词） context_prompt = f"""用户查询: {query} 相关记忆: {memory_summary}""" llm_context = self.chat_with_memory(context_prompt, use_memory=False) return { "relevant_memories": [ { "content": m['content'], "score": m['score'], "tags": m.get('tags', []) } for m in memories ], "llm_context": llm_context, "query_summary": f"查询: {query} (找到{len(memories)}条相关记忆)" } def demo_knowledge_management(): """演示知识管理场景""" print("📚 知识管理演示") print("=" * 50) memos = AdvancedMemOS("./knowledge_base") # 添加学习笔记 print("\n📝 添加学习笔记...") memos.add_learning_note("Python", "学习了装饰器的使用，可以用来增强函数功能", "medium") memos.add_learning_note("机器学习", "了解了梯度下降算法的原理和应用", "hard") memos.add_learning_note("Docker", "学会了容器化部署，提高了开发效率", "easy") # 添加个人见解 print("\n💡 添加个人见解...") memos.add_personal_insight("代码复用是提高开发效率的关键", "编程") memos.add_personal_insight("持续学习比一次性掌握更重要", "学习方法") # 获取学习总结 print("\n📊 学习总结:") summary = memos.get_learning_summary() print(summary) def demo_work_management(): """演示工作管理场景""" print("\n💼 工作管理演示") print("=" * 50) memos = AdvancedMemOS("./work_space") # 添加工作任务 print("\n📋 添加工作任务...") memos.add_work_task("完成MemOS项目文档", "high", "2024-01-20") memos.add_work_task("准备下周的技术分享", "normal", "2024-01-18") memos.add_work_task("整理代码仓库", "low") # 获取工作仪表板 print("\n📊 工作仪表板:") dashboard = memos.get_work_dashboard() print(dashboard) def demo_personal_assistant(): """演示个人助手场景""" print("\n🤖 个人助手演示") print("=" * 50) memos = AdvancedMemOS("./personal_assistant") # 添加各种类型的记忆 memos.add_memory("我每天早上7点起床，喜欢先喝一杯水", tags=["生活习惯", "健康"]) memos.add_memory("我的生日是3月15日", tags=["个人信息"]) memos.add_memory("我正在学习AI和机器学习", tags=["学习", "AI"]) memos.add_memory("我喜欢在周末看电影和读书", tags=["兴趣爱好"]) # 智能提醒测试 print("\n🔔 智能提醒测试:") queries = [ "我的生活习惯", "学习计划", "周末安排" ] for query in queries: print(f"\n查询: {query}") reminder = memos.smart_reminder(query) print(f"提醒: {reminder}") def demo_interactive_assistant(): """演示交互式个人助手""" print("\n🎯 交互式个人助手") print("=" * 50) print("命令:") print(" add:内容 - 添加记忆") print(" learn:主题:内容 - 添加学习笔记") print(" work:任务:优先级 - 添加工作任务") print(" remind:查询 - 智能提醒") print(" summary - 学习总结") print(" dashboard - 工作仪表板") print(" quit - 退出") memos = AdvancedMemOS("./interactive_assistant") while True: user_input = input("\n> ").strip() if user_input.lower() == 'quit': break if user_input.startswith('add:'): content = user_input[4:].strip() memory_id = memos.add_memory(content) print(f"✅ 已添加记忆 #{memory_id}") elif user_input.startswith('learn:'): parts = user_input[6:].split(':', 1) if len(parts) == 2: topic, content = parts memory_id = memos.add_learning_note(topic.strip(), content.strip()) print(f"✅ 已添加学习笔记 #{memory_id}") else: print("❌ 格式错误，请使用: learn:主题:内容") elif user_input.startswith('work:'): parts = user_input[5:].split(':', 1) if len(parts) == 2: task, priority = parts memory_id = memos.add_work_task(task.strip(), priority.strip()) print(f"✅ 已添加工作任务 #{memory_id}") else: print("❌ 格式错误，请使用: work:任务:优先级") elif user_input.startswith('remind:'): query = user_input[7:].strip() reminder = memos.smart_reminder(query) print(f"🔔 {reminder}") elif user_input == 'summary': summary = memos.get_learning_summary() print(f"📊 {summary}") elif user_input == 'dashboard': dashboard = memos.get_work_dashboard() print(dashboard) else: # 普通对话 response = memos.chat_with_memory(user_input, use_memory=True) print(f"🤖 {response}") if __name__ == "__main__": # 加载环境变量 load_env_file() # 运行演示 demo_knowledge_management() demo_work_management() demo_personal_assistant() # 可选：运行交互式助手 # demo_interactive_assistant()