View a markdown version of this page

斯特兰兹特工 SDK - Amazon Bedrock AgentCore

斯特兰兹特工 SDK

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

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

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

    agentcore create --name my-agent --no-agent cd my-agent

    然后添加内存并部署:

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

  2. 输入内存名称:

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

    内存向导:选择所有三种内存策略
  4. 查看配置并按 Enter 进行确认:

    记忆向导: ComprehensiveAgentMemory 使用所有策略进行确认

    然后运行agentcore deploy在中配置内存 AWS。

安装依赖项

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

替代方案:显式关闭 ()

如果您无法使用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