--- name: memos-memory-assistant description: >- 简单的记忆存储与检索助手，提供便捷的记忆功能。 适合需要存储和检索用户偏好的场景。 常见触发场景：记住用户喜好、检索用户信息 trigger_keywords: - 记住 - 记忆 - 想起 - 搜索记忆 - 用户信息 - 偏好记忆 --- ## 详细功能 - **记住信息** - **工具**: remember - **CLI调用**: `python -m skills.memos remember --content "我喜欢苹果"` - **参数**: - content (要记住的内容) - **操作**: 将信息存储到记忆库中 - **检索记忆** - **工具**: recall - **CLI调用**: `python -m skills.memos recall --query "苹果"` - **参数**: - query (搜索关键词) - **操作**: 根据关键词检索相关的记忆信息
"""Simple memory skill for storing and retrieving user memories.""" import os from typing import List, Dict, Optional # Try to import the real MemOS client, fall back to local memory if not available try: from memos.api.client import MemOSClient MEMOS_AVAILABLE = True except ImportError: MEMOS_AVAILABLE = False print("警告: MemOS客户端不可用，将使用本地记忆功能") # Import local memory functions using absolute import try: from agent_chator.memory.local_memory import add_message as local_add_message, search_memory as local_search_memory except ImportError: # Fallback if the module is not available def local_add_message(user_id, conversation_id, messages): print("警告: 本地记忆功能不可用") return {"error": "Local memory not available"} def local_search_memory(query, user_id, conversation_id): print("警告: 本地记忆功能不可用") return {"error": "Local memory not available"} class MemosSkill: """A skill for interacting with MemOS API to store and retrieve memories.""" def __init__(self): self.api_key = os.getenv("MEMOS_API_KEY") if not self.api_key or not MEMOS_AVAILABLE: print("信息: 未设置 MEMOS_API_KEY 或 MemOS不可用，将使用本地记忆功能") self.use_local = True else: self.use_local = False # Use remote if available and key is set if not self.use_local: try: self.client = MemOSClient(api_key=self.api_key) except Exception as e: print(f"警告: 无法初始化MemOS客户端 ({e})，将使用本地记忆功能") self.use_local = True else: self.client = None def add_message(self, user_id: str, conversation_id: str, messages: List[Dict[str, str]]) -> Dict: """ Store a conversation in MemOS to create memories. Args: user_id: The ID of the user conversation_id: The ID of the conversation messages: List of messages in the format {"role": "user|assistant", "content": "message content"} Returns: Response from the MemOS API """ if self.use_local: return local_add_message(user_id=user_id, conversation_id=conversation_id, messages=messages) else: return self.client.add_message(messages=messages, user_id=user_id, conversation_id=conversation_id) def search_memory(self, query: str, user_id: str, conversation_id: str) -> Dict: """ Retrieve relevant memories from MemOS based on a query. Args: query: The search query to find relevant memories user_id: The ID of the user conversation_id: The ID of the current conversation Returns: Response from the MemOS API containing relevant memories """ if self.use_local: return local_search_memory(query=query, user_id=user_id, conversation_id=conversation_id) else: return self.client.search_memory(query=query, user_id=user_id, conversation_id=conversation_id) # Global variable to hold the skill instance _memosskill = None def _get_memosskill(): """Lazy initialization of the MemosSkill instance.""" global _memosskill if _memosskill is None: _memosskill = MemosSkill() return _memosskill # Export the functions for direct use def add_message(user_id: str, conversation_id: str, messages: List[Dict[str, str]]) -> Dict: """ Store a conversation in MemOS to create memories. Args: user_id: The ID of the user conversation_id: The ID of the conversation messages: List of messages in the format {"role": "user|assistant", "content": "message content"} Returns: Response from the MemOS API """ skill = _get_memosskill() return skill.add_message(user_id, conversation_id, messages) def search_memory(query: str, user_id: str, conversation_id: str) -> Dict: """ Retrieve relevant memories from MemOS based on a query. Args: query: The search query to find relevant memories user_id: The ID of the user conversation_id: The ID of the current conversation Returns: Response from the MemOS API containing relevant memories """ skill = _get_memosskill() return skill.search_memory(query, user_id, conversation_id) # Simple convenience functions for common use cases DEFAULT_USER_ID = "default_user" def remember(content: str, user_id: str = DEFAULT_USER_ID) -> Dict: """ Remember something for a user. Args: content: The content to remember user_id: The ID of the user (optional, defaults to 'default_user') Returns: Result of the operation """ messages = [{"role": "user", "content": content}] return add_message(user_id, "default_conversation", messages) def recall(query: str, user_id: str = DEFAULT_USER_ID) -> Dict: """ Recall information for a user based on a query. Args: query: The query to search for user_id: The ID of the user (optional, defaults to 'default_user') Returns: Search results """ return search_memory(query, user_id, "default_conversation")
