agentscope Mem0LongTermMemory记忆写入和查询工具2个py

query.py 查询 查询咖啡偏好: 1. 用户表达了咖啡偏好 2. 我喜欢喝美式咖啡不加糖 3. 我喜欢喝美式咖啡不加糖 4. 好的我记住了您喜欢喝不加糖的美式咖啡 5. 我喜欢喝美式咖啡不加糖 6. 好的我记住了您喜欢喝不加糖的美式咖啡 查询周末活动: 1. 周末我喜欢去公园散步 2. 了解了您周末喜欢去公园散步 3. 周末我喜欢去公园散步 4. 了解了您周末喜欢去公园散步 5. 周末我喜欢去公园散步 6. 了解了您周末喜欢去公园散步import asyncio import os import sys sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from agentscope.memory import Mem0LongTermMemory from agentscope.message import Msg from agentscope.model import OpenAIChatModel from agentscope.embedding import DashScopeTextEmbedding from mem0.vector_stores.configs import VectorStoreConfig from dotenv import load_dotenv load_dotenv() async def query_clean(): long_term_memory Mem0LongTermMemory( agent_nameAssistant, user_nameuser_123, modelOpenAIChatModel( model_nameos.environ.get(zhipu_llm), api_keyos.environ.get(zhipu_OPENAI_API_KEY), streamTrue, client_kwargs{base_url: https://open.bigmodel.cn/api/paas/v4} ), embedding_modelDashScopeTextEmbedding( model_nametext-embedding-v2, api_keyos.environ.get(ali_OPENAI_API_KEY), ), vector_store_configVectorStoreConfig( providerqdrant, config{on_disk: True, path: ./memory_data, embedding_model_dims: 1536}, ), ) queries [ (咖啡偏好, 美式咖啡), (周末活动, 公园散步), ] for name, query in queries: print(f\n 查询{name}:) memories_str await long_term_memory.retrieve( msg[Msg(roleuser, contentquery, nameuser)], limit3 ) if memories_str: memory_list memories_str.splitlines() for i, memory in enumerate(memory_list, 1): print(f {i}. {memory}) else: print( 未找到记忆) if __name__ __main__: asyncio.run(query_clean())record.py 写入✅ 成功记录用户偏好 ✅ 成功记录周末活动偏好import asyncio import os import sys sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from agentscope.memory import Mem0LongTermMemory from agentscope.message import Msg from agentscope.model import OpenAIChatModel from agentscope.embedding import DashScopeTextEmbedding from mem0.vector_stores.configs import VectorStoreConfig from dotenv import load_dotenv load_dotenv() async def record_memory(): 记录用户偏好到长期记忆 # 初始化长期记忆 long_term_memory Mem0LongTermMemory( agent_nameAssistant, user_nameuser_123, modelOpenAIChatModel( model_nameos.environ.get(zhipu_llm), api_keyos.environ.get(zhipu_OPENAI_API_KEY), streamTrue, client_kwargs{ base_url: https://open.bigmodel.cn/api/paas/v4 } ), embedding_modelDashScopeTextEmbedding( model_nametext-embedding-v2, api_keyos.environ.get(ali_OPENAI_API_KEY), ), vector_store_configVectorStoreConfig( providerqdrant, config{ on_disk: True, path: ./memory_data, embedding_model_dims: 1536, }, ), ) # 记录用户偏好 user_msg Msg(roleuser, content我喜欢喝美式咖啡不加糖, nameuser) assistant_msg Msg(roleassistant, content好的我记住了您喜欢喝不加糖的美式咖啡, nameassistant) try: await long_term_memory.record( msgs[user_msg, assistant_msg], inferFalse ) print(✅ 成功记录用户偏好) # 记录第二个偏好 user_msg2 Msg(roleuser, content周末我喜欢去公园散步, nameuser) assistant_msg2 Msg(roleassistant, content了解了您周末喜欢去公园散步, nameassistant) await long_term_memory.record( msgs[user_msg2, assistant_msg2], inferFalse ) print(✅ 成功记录周末活动偏好) except Exception as e: print(f❌ 记录失败: {e}) if __name__ __main__: asyncio.run(record_memory())