Let the Financial Assistant Understand Customer Preferences Behind Behaviors

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 FinancialManagementAssistant:
    """AI financial management assistant with memory capability"""
    
    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 build_system_prompt(self, memories):
        """Construct a system prompt including formatted memories"""
        base_prompt = """
          You are a knowledgeable and professional financial management assistant.
          You can access conversational memories to help you provide more personalized answers.
          Use memories to understand the user’s background, preferences, and past interactions.
          If memories are provided, naturally reference them when relevant, but do not explicitly mention having memories.
        """

        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 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 chat(self, query, user_id, conversation_id):
        """Main chat function for handling conversations with memory integration"""
        # 1) Search relevant memories
        memories = self.search_memory(query, user_id, conversation_id)
        
        # Build system prompt with memories
        system_prompt = self.build_system_prompt(memories)
        
        # 2) Use OpenAI to generate an answer
        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 the interaction back to MemOS
        messages = [
            {"role": "user", "content": query},
            {"role": "assistant", "content": answer}
        ]
        self.memos_client.add_message(messages, user_id, conversation_id)
        
        return answer

ai_assistant = FinancialManagementAssistant()
user_id = "memos_financial_management_user_123"

def demo_questions():
    return [
      "What is my risk preference?",
      "Recommend some investments suitable for me"
    ]

def preset_user_behaviors():
    """Show preset user behavior memories"""
    conversation_id = generate_conversation_id()
    print(f"\n📊 Preset user behavior memories（conversation_id={conversation_id}）:")
    print("=" * 60)

    behaviors = [{
      "role": "user",
      "content": "Clicked 'Retirement Finance' ad to enter app"
    }, {
      "role": "user",
      "content": "Browsed and bookmarked low-risk funds"
    }]
    
    for i, behavior in enumerate(behaviors, 1):
        print(f"{i}. {behavior['content']}")
    ai_assistant.add_message(behaviors, user_id, conversation_id)
    
    print("=" * 60)
    print("💡 The above behavioral memories have been recorded by MemOS. The assistant will provide personalized recommendations based on them.")

def main():
    print("💰 Welcome to see how MemOS is used in a financial management assistant!")
    print("💡 With MemOS, your financial assistant becomes smarter and more caring! 😊 \n")
    
    # Ask whether to preload user behavior memories (consumes 1 add quota)
    while True:
        pre_chat = input("🤔 Do you want to preload user behavior memories? This will consume 1 add quota. Proceed? (y/n): ").strip().lower()
        
        if pre_chat in ['y', 'yes']:
            preset_user_behaviors()
            break
        elif pre_chat in ['n', 'no']:
            print("📝 Starting a new conversation...")
            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 example 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("👋 Thanks for using the financial management 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()