让理财助手读懂客户行为背后的偏好

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" # 从云服务控制台获取MemOS_API_KEY
os.environ["OPENAI_API_KEY"] = "sk-xx" # 替换为你自己的API_KEY

conversation_counter = 0

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

class FinancialManagementAssistant:
    """AI财务管理助手，具备记忆能力"""
    
    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):
        """查询相关记忆"""
        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):
        """构建包含格式化记忆的系统提示"""
        base_prompt = """
          你是一位知识丰富且专业贴心的理财管理助手。
          你可以访问对话记忆，帮助你提供更个性化的回答。
          使用记忆来理解用户的背景、偏好和过往互动。
          如果提供了记忆，请在相关时自然地引用它们，但不要明确提及拥有记忆
        """

        if memories:
            # 将记忆格式化为编号列表
            formatted_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):
        """添加消息"""
        self.memos_client.add_message(messages, user_id, conversation_id)

    def get_message(self, user_id, conversation_id):
        """获取消息"""
        response = self.memos_client.get_message(user_id, conversation_id)
        
        return response.data.message_detail_list

    def chat(self, query, user_id, conversation_id):
        """处理包含记忆集成的对话的主要聊天函数"""
        # 1. 搜索相关记忆
        memories = self.search_memory(query, user_id, conversation_id)
        
        # 构建包含记忆的系统提示
        system_prompt = self.build_system_prompt(memories)
        
        # 2. 使用OpenAI生成回答
        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. 将对话保存到记忆中
        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 [
      '我的风险偏好是什么',
      "推荐一些适合我的投资"
    ]

def preset_user_behaviors():
    """显示预设的用户行为记忆"""
    conversation_id = generate_conversation_id()
    
    print(f"\n📊 已预设的用户行为记忆 (conversation_id={conversation_id}):")
    print("=" * 60)

    behaviors = [{
      "role": "user",
      "content": "点击「养老理财」广告进入 APP"
    }, {
      "role": "user",
      "content": "浏览并收藏低风险基金"
    }]
    
    for i, behavior in enumerate(behaviors, 1):
        print(f"{i}. {behavior['content']}")
    ai_assistant.add_message(behaviors, user_id, conversation_id)
    
    print("=" * 60)
    print("💡 以上行为记忆已被MemOS记录，助手会基于这些信息提供个性化建议")

def main():
    print("💰 欢迎查看MemOS在理财管理助手中的使用示例！")
    print("💡 在MemOS的加持下，让您的理财助手更智能、更贴心！ 😊 \n")
    
    # 询问用户是否要先执行预配置对话
    while True:
        pre_chat = input("🤔 您想先预加载用户行为记忆吗？预计消耗1次add额度，是否执行？(y/n): ").strip().lower()
        
        if pre_chat in ['y', 'yes', '是', 'Y']:
            preset_user_behaviors()
            break
        elif pre_chat in ['n', 'no', '否', 'N']:
            print("📝 开始全新对话...")
            break
        else:
            print("⚠️  请输入 'y' 表示是或 'n' 表示否")

    print("\n⚡️ 接下来您输入的每一个问题，都将会在一个全新的会话中展开（新的 conversation id）。MemOS将会跨会话自动召回您的历史行为记忆，为您提供连续、个性化的服务。")    
    print("\n🎯 以下是一些示例问题，您可以继续跟助手对话:")
    for i, question in enumerate(demo_questions(), 1):
      print(f"  {i}. {question}")

    while True:
        user_query = input("\n🤔 请输入您的问题 (或输入 'exit' 退出): ").strip()
        
        if user_query.lower() in ['quit', 'exit', 'q', '退出']:
            print("👋 感谢使用理财管理助手！")
            break
        
        if not user_query:
            continue
        
        print("🤖 正在处理...")
        conversation_id = generate_conversation_id()
        answer = ai_assistant.chat(user_query, user_id, conversation_id)
        print(f"\n💬 conversation_id: {conversation_id}\n💡 [助手]: {answer}\n")
        print("-" * 60)


if __name__ == "__main__":
    main()