更新时间:2026-04-15 GMT+08:00
在智能体中集成记忆库
在智能体中集成记忆库,实现智能体与记忆库的快速对接。
前提条件
- 已创建记忆库,并获取记忆库的api_key。
- 已安装Python,且版本不低于3.10。
- 已安装AgentRunSDK:
pip install AgentRunSDK
在智能体中集成记忆库(操作步骤)
- 在左侧导航栏选择“开发中心 > 组件库 ”,单击“记忆库”页签,进入记忆库界面。
- 在记忆库列表单击操作列的“调用代码”,然后根据开发框架获取环境变量信息。
- 配置环境变量。
# 创建space后,会返回API_KEY,请将其设置为环境变量,用于调用数据面接口时鉴权 export AGENTARTS_MEMORY_API_KEY="你的数据面API密钥"
- 集成记忆库。
import os import time # 确保环境变量已设置 assert os.getenv('AGENTARTS_MEMORY_API_KEY'), "请设置 AGENTARTS_MEMORY_API_KEY" def client_mode_example(): """Client模式完整示例""" print("=== Client模式完整示例 ===") # 2. 创建会话 print("\n2. 创建会话") session_data = client.create_memory_session( space_id=space_id, actor_id="client-test-user", assistant_id="client-test-assistant" ) session_id = session_data.id print(f"✓ Session创建成功: {session_id}") # 3. 发送消息(使用TextMessage对象) print("\n3. 发送对话消息") messages = [ TextMessage( role="user", content="你好,我想了解电商推荐算法,能够根据用户行为进行个性化推荐的算法有哪些?", actor_id="client-test-user" ), TextMessage( role="assistant", content="电商推荐算法主要包括:1) 协同过滤算法,基于用户行为相似性推荐;2) 基于内容的推荐,分析商品特征;3) 深度学习算法,能更准确捕捉用户偏好。推荐组合使用多种算法提升准确率。", actor_id="client-test-assistant" ), TextMessage( role="user", content="我对机器学习的监督学习特别感兴趣,深度学习和传统机器学习有什么区别?", actor_id="client-test-user" ) ] add_result = client.add_messages( space_id=space_id, session_id=session_id, messages=messages ) print(f"✓ 已添加 {len(add_result.items)} 条消息") # 4. 等待记忆系统处理消息 print("\n4. 等待记忆系统处理消息...") time.sleep(30) # 5. 列出记忆 print("\n5. 查询记忆列表") memories = client.list_memories( space_id=space_id, limit=10 ) print(f"✓ 发现 {len(memories.items)} 条记忆") for i, memory in enumerate(memories.items[:3]): print(f" {i+1}. {memory.content[:50]}...") print(f" 策略: {memory.strategy_type}") # 6. 搜索记忆(使用MemorySearchFilter对象) print("\n6. 搜索相关记忆") from hw_agentrun_wrapper.memory.inner.config import MemorySearchFilter search_results = client.search_memories( space_id=space_id, filters=MemorySearchFilter(query="机器学习", top_k=3) ) print(f"✓ 找到 {len(search_results.results)} 条相关记忆") for i, result in enumerate(search_results.results): score = result.get('score', 0) content = result.get('record', {}).get('content', '')[:60] print(f" {i+1}. [{score:.2f}] {content}...") # 7. 获取特定记忆详情 if memories.items: print("\n7. 获取记忆详情") memory_id = memories.items[0].id memory_detail = client.get_memory(space_id, memory_id) print(f"✓ 记忆ID: {memory_detail.id}") print(f" 内容: {memory_detail.content[:80]}...") print(f" 策略: {memory_detail.strategy_type}") return space_id, session_id