# Memory SDK #### 场景介绍 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}") os.environ["HUAWEICLOUD_SDK_MEMORY_API_KEY"] = space.api_key # 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 agentarts.sdk.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}") os.environ["HUAWEICLOUD_SDK_MEMORY_API_KEY"] = space.api_key # 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 agentarts.sdk.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](https://support.huaweicloud.com/highcode-agentarts/agentarts_10_047.html#agentarts_10_047__section13365141032517)。 ``` export HUAWEICLOUD_SDK_AK="您的AK" export HUAWEICLOUD_SDK_SK="您的SK" ``` - 创建space后，会返回apikey，可将其设置为环境变量，用于调用数据面接口认证。 ``` 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内部自动处理。 **请求流程** 1. 使用AK/SK创建Space（控制面）。 2. SDK自动创建API Key并返回api_key（仅创建时可见）和api_key_id。 3. SDK内部自动使用API Key进行数据面操作。 4. 配置数据面访问端点（私网访问场景下，通过环境变量AGENTARTS_MEMORY_DATA_ENDPOINT配置，例如https://memory-{space_id}.cn-southwest-2.huaweicloud-agentarts.com）。 **类型化返回值** 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时间。 - 及时清理测试数据。