更新时间:2026-04-15 GMT+08:00
分享

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内部自动处理。

请求流程

  1. 使用AK/SK创建Space(控制面)。
  2. SDK自动创建API Key并返回api_key(仅创建时可见)和api_key_id。
  3. SDK内部自动使用API Key进行数据面操作。
  4. 配置数据面端点(可选,通过环境变量)。

类型化返回值

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时间。
  • 及时清理测试数据。

相关文档