View a markdown version of this page

Strands 代理软件开发工具包 - Amazon Bedrock AgentCore
Amazon Web Services 文档中描述的 Amazon Web Services 服务或功能可能因区域而异。要查看适用于中国区域的差异,请参阅 中国的 Amazon Web Services 服务入门 (PDF)。

本文属于机器翻译版本。若本译文内容与英语原文存在差异,则一律以英文原文为准。

Strands 代理软件开发工具包

使用 Strands Agents SDK 与代理框架无缝集成,在对话代理中提供自动内存管理和检索。

首先,用所有三种长期策略来创造一段记忆。您可以使用 AgentCore CLI 或通过以下示例中的 SDK 代码来执行此操作。

例
AgentCore CLI
  1. AgentCore CLI 内存命令必须在现有的 agentcore 项目中运行。如果你还没有,请先创建一个项目:

    agentcore create --project-name MemoryProject --no-agent cd MemoryProject

    然后添加内存并部署:

    agentcore add memory --name ComprehensiveAgentMemory \ --strategies SEMANTIC,SUMMARIZATION,USER_PREFERENCE agentcore deploy
Interactive
  1. 运行打开 agentcore TUI,然后选择添加并选择内存:

  2. 输入内存名称:

    内存向导:输入 ComprehensiveAgentMemory 名称
  3. 选择所有三种策略(语义、摘要、用户首选项):

    记忆向导:选择所有三种记忆策略
  4. 查看配置并按 Enter 键确认:

    记忆向导:确认 ComprehensiveAgentMemory 所有策略

    然后运行agentcore deploy以配置内存 Amazon。

安装依赖项

pip install bedrock-agentcore pip install strands-agents

添加短期记忆

from datetime import datetime from strands import Agent from bedrock_agentcore.memory import MemoryClient from bedrock_agentcore.memory.integrations.strands.config import AgentCoreMemoryConfig, RetrievalConfig from bedrock_agentcore.memory.integrations.strands.session_manager import AgentCoreMemorySessionManager client = MemoryClient(region_name="us-east-1") basic_memory = client.create_memory( name="BasicTestMemory", description="Basic memory for testing short-term functionality" ) MEM_ID = basic_memory.get('id') ACTOR_ID = "actor_id_test_%s" % datetime.now().strftime("%Y%m%d%H%M%S") SESSION_ID = "testing_session_id_%s" % datetime.now().strftime("%Y%m%d%H%M%S") # Configure memory agentcore_memory_config = AgentCoreMemoryConfig( memory_id=MEM_ID, session_id=SESSION_ID, actor_id=ACTOR_ID ) # Create session manager session_manager = AgentCoreMemorySessionManager( agentcore_memory_config=agentcore_memory_config, region_name="us-east-1" ) # Create agent agent = Agent( system_prompt="You are a helpful assistant. Use all you know about the user to provide helpful responses.", session_manager=session_manager, ) agent("I like sushi with tuna") # Agent remembers this preference agent("I like pizza") # Agent acknowledges both preferences agent("What should I buy for lunch today?") # Agent suggests options based on remembered preferences

通过策略增加长期记忆

from bedrock_agentcore.memory import MemoryClient from strands import Agent from bedrock_agentcore.memory.integrations.strands.config import AgentCoreMemoryConfig, RetrievalConfig from bedrock_agentcore.memory.integrations.strands.session_manager import AgentCoreMemorySessionManager from datetime import datetime # Create comprehensive memory with all built-in strategies client = MemoryClient(region_name="us-east-1") comprehensive_memory = client.create_memory_and_wait( name="ComprehensiveAgentMemory", description="Full-featured memory with all built-in strategies", strategies=[ { "summaryMemoryStrategy": { "name": "SessionSummarizer", "namespaceTemplates": ["/summaries/{actorId}/{sessionId}/"] } }, { "userPreferenceMemoryStrategy": { "name": "PreferenceLearner", "namespaceTemplates": ["/preferences/{actorId}/"] } }, { "semanticMemoryStrategy": { "name": "FactExtractor", "namespaceTemplates": ["/facts/{actorId}/"] } } ] ) MEM_ID = comprehensive_memory.get('id') ACTOR_ID = "actor_id_test_%s" % datetime.now().strftime("%Y%m%d%H%M%S") SESSION_ID = "testing_session_id_%s" % datetime.now().strftime("%Y%m%d%H%M%S") # Configure memory agentcore_memory_config = AgentCoreMemoryConfig( memory_id=MEM_ID, session_id=SESSION_ID, actor_id=ACTOR_ID ) # Create session manager session_manager = AgentCoreMemorySessionManager( agentcore_memory_config=agentcore_memory_config, region_name="us-east-1" ) # Create agent agent = Agent( system_prompt="You are a helpful assistant. Use all you know about the user to provide helpful responses.", session_manager=session_manager, ) agent("I like sushi with tuna") # Agent remembers this preference agent("I like pizza") # Agent acknowledges both preferences agent("What should I buy for lunch today?") # Agent suggests options based on remembered preferences

消息批处理

当大batch_size于 1 时,消息将在内存中缓冲,并在缓冲区达到配置的大小后通过单个 API 调用发送到 AgentCore 内存。这减少了高吞吐量对话中的 API 请求数量。

重要

使用时batch_size > 1,必须在会话完成close()后使用with屏蔽或呼叫。否则,任何尚未达到批量阈值的缓冲消息都将丢失。

推荐:上下文管理器

from strands import Agent from bedrock_agentcore.memory.integrations.strands.config import AgentCoreMemoryConfig from bedrock_agentcore.memory.integrations.strands.session_manager import AgentCoreMemorySessionManager config = AgentCoreMemoryConfig( memory_id=MEM_ID, session_id=SESSION_ID, actor_id=ACTOR_ID, batch_size=10, # Buffer up to 10 messages before sending ) # The `with` block guarantees all buffered messages are flushed on exit with AgentCoreMemorySessionManager(config, region_name='us-east-1') as session_manager: agent = Agent( system_prompt="You are a helpful assistant.", session_manager=session_manager, ) agent("Hello!") agent("Tell me about AWS") # All remaining buffered messages are automatically flushed here

替代方案:显式 close ()

如果你不能使用with区块,请close()手动调用:

session_manager = AgentCoreMemorySessionManager(config, region_name='us-east-1') try: agent = Agent( system_prompt="You are a helpful assistant.", session_manager=session_manager, ) agent("Hello!") finally: session_manager.close() # Flush any remaining buffered messages

更多示例可在以下网址获得 GitHub:https://github.com/aws/bedrock-agentcore-sdk-python/tree/main/src/bedrock_agentcore/memory/integrations/strands