Building a Home Life Assistant with Memory

pip install MemoryOS -U

import os
import uuid
from openai import OpenAI
from memos.api.client import MemOSClient

os.environ["MEMOS_API_KEY"] = "mpg-xx" # Get MemOS_API_KEY from cloud service console
os.environ["OPENAI_API_KEY"] = "sk-xx" # Replace with your own API_KEY

conversation_counter = 0

def generate_conversation_id():
    global conversation_counter
    conversation_counter += 1
    return f"conversation_{conversation_counter:03d}"

class HomeAssistant:    
    def __init__(self):
        self.memos_client = MemOSClient(api_key=os.getenv("MEMOS_API_KEY"))
        self.openai_client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
    
    def search_memory(self, query, user_id, conversation_id):
        """Search relevant memories based on query"""
        response = self.memos_client.search_memory(query, user_id, conversation_id)

        return [memory_detail.memory_value for memory_detail in response.data.memory_detail_list]

    def add_message(self, messages, user_id, conversation_id):
        """Add messages to MemOS so they can be processed into memories"""
        self.memos_client.add_message(messages, user_id, conversation_id)

    def get_message(self, user_id, conversation_id):
        """Retrieve the raw messages stored in MemOS (for debugging/inspection)"""
        response = self.memos_client.get_message(user_id, conversation_id)
        
        return response.data.message_detail_list

    def build_system_prompt(self, memories):
        """Builds a system prompt containing formatted memories"""
        base_prompt = """
          You are a knowledgeable and considerate home life assistant.
          You can leverage conversation memories to provide more personalized responses.
          Use these memories to understand the user’s context, preferences, and past interactions.
          If memory content is provided, naturally reference it when relevant, but do not explicitly state you have memory functions.
        """

        if memories:
            # Format memories as a numbered list
            formatted_memories = "## Memories:\n"
            for i, memory in enumerate(memories, 1):
                formatted_memories += f"{i}. {memory}\n"
            
            return f"{base_prompt}\n\n{formatted_memories}"
        else:
            return base_prompt
        

    def chat(self, query, user_id, conversation_id):
        """Main chat function handling memory-integrated conversation"""
        # 1. Search relevant memories
        memories = self.search_memory(query, user_id, conversation_id)
        
        # Build system prompt including memories
        system_prompt = self.build_system_prompt(memories)
        
        # 2. Use OpenAI to generate response
        response = self.openai_client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": query}
            ]
        )
        answer = response.choices[0].message.content

        # 3. Save dialogue into memory
        messages = [
            {"role": "user", "content": query},
            {"role": "assistant", "content": answer}
        ]
        self.memos_client.add_message(messages, user_id, conversation_id)
        
        return answer

ai_assistant = HomeAssistant()
user_id = "memos_home_management_user_123"

def demo_questions():
    return [
      "What plans do I have for the weekend?",
      "Help me plan next week’s housework distribution"
    ]

def pre_configured_conversations():
    """Return pre-configured dialogue pairs"""
    return [
        {
            "user": "Take the kids to the zoo on Saturday, please remember it.",
        },
        {
            "user": "For future reminders or plans, please first list three key points, then add one short suggestion.",
        }
    ]

def execute_pre_conversations():
    """Execute pre-configured dialogues"""
    conversations = pre_configured_conversations()
    conversation_id = generate_conversation_id()

    print(f"\n🔄 Executing pre-configured dialogues（conversation_id={conversation_id}）...")
    print("=" * 60)
    
    for i, conv in enumerate(conversations, 1):
        print(f"\n💬 Dialogue {i}")
        print(f"👤 User: {conv['user']}")
        
        # Execute dialogue
        answer = ai_assistant.chat(conv['user'], user_id, conversation_id)
        print(f"🤖 Assistant: {answer}")
        print("-" * 40)
    
    print("\n✅ Pre-configured dialogues completed!")
    print("=" * 60)

def main():    
    print("🏠 Welcome to the example of MemOS applied in a home assistant!")
    print("💡 With the power of MemOS, your product can deliver a real butler-like experience! 😊 \n")
    
    # Ask whether to execute pre-configured dialogues first
    while True:
        pre_chat = input("🤔 Would you like to execute the pre-configured dialogues first? This will consume 2 add calls and 2 search calls. Proceed? (y/n): ").strip().lower()
        
        if pre_chat in ['y', 'yes']:
            execute_pre_conversations()
            break
        elif pre_chat in ['n', 'no']:
            print("📝 Starting a new dialogue...")
            break
        else:
            print("⚠️  Please enter 'y' for yes or 'n' for no")

    print("\n⚡️ Each question you enter next will take place in a brand-new conversation (with a new conversation ID). MemOS will automatically recall your historical behavioral memories across conversations to provide you with continuous and personalized service.")
    print("\n🎯 Here are some sample questions you can continue to ask the assistant:")
    for i, question in enumerate(demo_questions(), 1):
      print(f"  {i}. {question}")

    while True:
        user_query = input("\n🤔 Please enter your question (or type 'exit' to quit): ").strip()
        
        if user_query.lower() in ['quit', 'exit', 'q']:
            print("👋 Thank you for using the home assistant!")
            break
        
        if not user_query:
            continue
        
        print("🤖 Processing...")
        conversation_id = generate_conversation_id()
        answer = ai_assistant.chat(user_query, user_id, conversation_id)
        print(f"\n💬 conversation_id: {conversation_id}\n💡 [Assistant]: {answer}\n")
        print("-" * 60)


if __name__ == "__main__":
    main()