memory sdk-j9九游会登录

场景介绍

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}")
          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。

  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使用类型化对象作为返回值，便于代码提示和类型检查：

类型化请求对象

sdk提供类型化的请求对象用于构建参数：

记忆使用建议

• 始终检查环境变量是否正确设置。
• 根据场景选择client或session模式。
• 理解记忆生成的延迟时间。
• 合理设置space的ttl时间。
• 及时清理测试数据。