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_name="Assistant", user_name="user_123", model=OpenAIChatModel( model_name=os.environ.get("zhipu_llm"), api_key=os.environ.get("zhipu_OPENAI_API_KEY"), stream=True, client_kwargs={"base_url": "https://open.bigmodel.cn/api/paas/v4"} ), embedding_model=DashScopeTextEmbedding( model_name="text-embedding-v2", api_key=os.environ.get("ali_OPENAI_API_KEY"), ), vector_store_config=VectorStoreConfig( provider="qdrant", 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(role="user", content=query, name="user")], limit=3 ) 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_name="Assistant", user_name="user_123", model=OpenAIChatModel( model_name=os.environ.get("zhipu_llm"), api_key=os.environ.get("zhipu_OPENAI_API_KEY"), stream=True, client_kwargs={ "base_url": "https://open.bigmodel.cn/api/paas/v4" } ), embedding_model=DashScopeTextEmbedding( model_name="text-embedding-v2", api_key=os.environ.get("ali_OPENAI_API_KEY"), ), vector_store_config=VectorStoreConfig( provider="qdrant", config={ "on_disk": True, "path": "./memory_data", "embedding_model_dims": 1536, }, ), ) # 记录用户偏好 user_msg = Msg(role="user", content="我喜欢喝美式咖啡，不加糖", name="user") assistant_msg = Msg(role="assistant", content="好的，我记住了您喜欢喝不加糖的美式咖啡", name="assistant") try: await long_term_memory.record( msgs=[user_msg, assistant_msg], infer=False ) print("✅ 成功记录用户偏好") # 记录第二个偏好 user_msg2 = Msg(role="user", content="周末我喜欢去公园散步", name="user") assistant_msg2 = Msg(role="assistant", content="了解了，您周末喜欢去公园散步", name="assistant") await long_term_memory.record( msgs=[user_msg2, assistant_msg2], infer=False ) print("✅ 成功记录周末活动偏好") except Exception as e: print(f"❌ 记录失败: {e}") if __name__ == "__main__": asyncio.run(record_memory())
)