Memory SDK
场景介绍
AgentRunSDK Memory SDK提供两种使用模式:
- Client模式:功能完整的客户端,适合需要控制所有操作的应用场景。
- Session模式:基于绑定的会话进行记忆管理,适合特定用户/对话场景。
原理优势
- Client模式完整案例
#!/usr/bin/env python3 """ Client模式完整使用案例 创建Space -> 创建Session -> 发送Message -> 查询Memory -> 搜索Memory """ import os import time from agentarts.sdk.memory import MemoryClient from agentarts.sdk.memory.inner.config import TextMessage # 确保环境变量已设置 assert os.getenv('HUAWEICLOUD_SDK_AK'), "请设置 HUAWEICLOUD_SDK_AK" assert os.getenv('HUAWEICLOUD_SDK_SK'), "请设置 HUAWEICLOUD_SDK_SK" def client_mode_example(): """Client模式完整示例""" print("=== Client模式完整示例 ===") # 1. 创建Space with MemoryClient() as client: print("1. 创建测试Space") space = client.create_space( name=f"client_test_space_{int(time.time())}", message_ttl_hours=168, description="Client模式测试专用空间", memory_strategies_builtin=["semantic", "user_preference", "episodic"] ) space_id = space.id print(f"✓ Space创建成功: {space_id}") print(f" API Key ID: {space.api_key_id}") # 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 if __name__ == "__main__": client_mode_example() - Session模式完整案例
#!/usr/bin/env python3 """ Session模式完整使用案例 创建Space -> 绑定会话 -> 发送Message -> 查询Memory -> 搜索Memory """ import os import time from agentarts.sdk.memory import MemoryClient from agentarts.sdk.memory.session import MemorySession from agentarts.sdk.memory.inner.config import TextMessage # 确保环境变量已设置 assert os.getenv('HUAWEICLOUD_SDK_AK'), "请设置 HUAWEICLOUD_SDK_AK" assert os.getenv('HUAWEICLOUD_SDK_SK'), "请设置 HUAWEICLOUD_SDK_SK" def session_mode_example(): """Session模式完整示例""" print("=== Session模式完整示例 ===") # 1. 创建Space with MemoryClient() as client: print("1. 创建测试Space") space = client.create_space( name=f"session-test-space-{int(time.time())}", message_ttl_hours=168, description="Session模式测试专用空间", memory_strategies_builtin=["semantic", "user_preference", "episodic"] ) space_id = space.id print(f"✓ Space创建成功: {space_id}") # 2. 创建并绑定会话 print("\n2. 创建并绑定会话") session_obj = MemorySession( space_id=space_id, actor_id="session-test-user", assistant_id="session-test-assistant" ) session_id = session_obj.session_id print(f"✓ Session创建并绑定成功: {session_id}") # 3. 发送对话消息(使用TextMessage对象) print("\n3. 模拟用户与AI助手对话") messages = [ TextMessage( role="user", content="我是一个数据分析师,主要使用Python和SQL进行数据处理" ), TextMessage( role="assistant", content="数据分析师是很有前景的职业!Python方面,pandas、numpy、matplotlib是核心库。" ), TextMessage( role="user", content="我最感兴趣的是可视化和机器学习方向,有什么推荐的学习路径吗?" ), TextMessage( role="assistant", content="可视化推荐学习matplotlib、seaborn、plotly;机器学习可以从scikit-learn入门,然后学习深度学习框架如TensorFlow或PyTorch。" ), TextMessage( role="user", content="我对Python数据分析很熟练,但机器学习经验还比较少,应该从哪里开始?" ) ] add_result = session_obj.add_messages(messages) print(f"✓ 已添加 {len(messages)} 条对话消息") # 4. 等待记忆系统处理 print("\n4. 等待记忆系统生成...") time.sleep(30) # 5. 查询消息(在绑定会话上下文中) print("\n5. 查询当前会话的消息") messages_response = session_obj.list_messages(limit=10) print(f"✓ 会话中有 {messages_response.total} 条消息") # 6. 获取会话中的记忆列表 print("\n6. 查看生成的记忆") memories = session_obj.list_memories(limit=10) memory_list = memories.items print(f"✓ 总共发现 {len(memory_list)} 条记忆") for i, memory in enumerate(memory_list): content = memory.content[:50] strategy = memory.strategy_type or 'unknown' print(f" {i+1}. [{strategy}] {content}...") # 7. 在会话上下文中搜索记忆(使用MemorySearchFilter对象) print("\n7. 搜索Python相关记忆") from hw_agentrun_wrapper.memory.inner.config import MemorySearchFilter search_results = session_obj.search_memories( filters=MemorySearchFilter(query="Python", 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}...") # 8. 获取特定记忆详情 if memory_list: print("\n8. 获取第一条记忆详情") memory_id = memory_list[0].id memory_detail = session_obj.get_memory(memory_id) print(f"✓ 记忆ID: {memory_detail.id}") print(f" 内容: {memory_detail.content[:80]}...") print(f" 策略: {memory_detail.strategy_type}") print(f" 来源策略: {memory_detail.strategy_id or 'N/A'}") return space_id, session_id if __name__ == "__main__": session_mode_example()
前提条件
认证鉴权:
- 华为云访问密钥,用于调用管理面创建space接口认证。获取华为云凭证请参考如何获取华为云AK/SK。
export HUAWEICLOUD_SDK_AK="你的AK" export HUAWEICLOUD_SDK_SK="你的SK"
- 创建space后,会返回HW_API_KEY,可将其设置为环境变量,用于调用数据面接口认证。
export HUAWEICLOUD_SDK_MEMORY_API_KEY="你的数据面API密钥"
也可以在代码中传入。
from agentarts.sdk.memory import MemoryClient client = MemoryClient(api_key="your-memory-api-key")
参数详解
认证信息
- 管理面(控制面):使用AK/SK认证,通过HUAWEICLOUD_SDK_AK和HUAWEICLOUD_SDK_SK环境变量。
- 数据面:使用API_KEY认证,SDK内部自动处理。
请求流程
- 使用AK/SK创建Space(控制面)。
- SDK自动创建API Key并返回api_key(仅创建时可见)和api_key_id。
- SDK内部自动使用API Key进行数据面操作。
- 配置数据面端点(可选,通过环境变量)。
类型化返回值
SDK使用类型化对象作为返回值,便于代码提示和类型检查:
|
返回类型 |
说明 |
重要属性 |
|---|---|---|
|
SpaceInfo |
Space信息 |
id, name, api_key, api_key_id, status, created_at |
|
SpaceListResponse |
Space列表 |
items (SpaceInfo列表), total, limit, offset |
|
SessionInfo |
Session信息 |
id, space_id, actor_id, assistant_id, created_at |
|
MessageInfo |
消息信息 |
id, session_id, role, parts, actor_id, created_at |
|
MessageListResponse |
消息列表 |
items (MessageInfo列表), total, limit, offset |
|
MessageBatchResponse |
批量消息响应 |
items (MessageInfo列表) |
|
MemoryInfo |
记忆信息 |
id, space_id, content, strategy_type, strategy_id, created_at |
|
MemoryListResponse |
记忆列表 |
items (MemoryInfo列表), total, limit, offset |
|
MemorySearchResponse |
搜索结果 |
results (列表,含record和score), total, query |
类型化请求对象
SDK提供类型化的请求对象用于构建参数:
|
请求类型 |
用途 |
|---|---|
|
TextMessage |
文本消息,参数:role, content, actor_id, assistant_id |
|
ToolCallMessage |
工具调用消息,参数:id, name, arguments |
|
ToolResultMessage |
工具结果消息,参数:tool_call_id, content, asset_ref |
|
MemorySearchFilter |
记忆搜索过滤,参数:query, top_k, min_score等 |
|
MemoryListFilter |
记忆列表过滤,参数:strategy_type, actor_id, sort_by等 |
记忆使用建议
- 始终检查环境变量是否正确设置。
- 根据场景选择Client或Session模式。
- 理解记忆生成的延迟时间。
- 合理设置Space的TTL时间。
- 及时清理测试数据。