---
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")