斯特兰兹特工 SDK
使用 Strands
首先,用所有三种长期策略创建一个记忆。您可以使用 AgentCore CLI 或通过以下示例中的 SDK 代码来执行此操作。
例
安装依赖项
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