参考Blog:aws.amazon.com/blogs/aws/i…
AWS AgentCore是一系列服务的组合,包括:
AgentCore Runtime – 提供具有会话隔离功能的低延迟无服务器环境,支持任何代理框架(包括流行的开源框架、工具和模型),并处理多模式工作负载和长时间运行的代理。 AgentCore Memory - 管理会话和长期记忆,为模型提供相关上下文,同时帮助代理从过去的交互中学习。 AgentCore Observability — 通过元数据标记、自定义评分、轨迹检查和故障排除/调试过滤器提供代理执行的逐步可视化。 AgentCore Identity – 使 AI 代理能够代表用户或在预先授权的用户同意下自行安全地访问 AWS 服务以及第三方工具和服务,例如 GitHub、Salesforce 和 Slack。 AgentCore Gateway – 将现有 API 和 AWS Lambda 函数转换为代理就绪工具,提供跨协议(包括 MCP)和运行时发现的统一访问。 AgentCore Browser – 提供托管的 Web 浏览器实例来扩展代理的 Web 自动化工作流程。 AgentCore 代码解释器 - 提供一个隔离的环境来运行代理生成的代码。
Memory说明
Agentcore对于Memory提供两方面支持:
短期记忆
- 代理会将交互作为短期事件存储在 AgentCore Memory 中。之后,代理可以检索这些事件,以便无需你重复之前的信息即可进行对话。短期记忆可以捕获原始交互事件,维护即时上下文,支持实时对话,丰富长期记忆系统,并支持构建高级上下文解决方案,例如多步骤任务完成、会话中知识积累和上下文感知决策。
长期记忆
长期记忆存储从原始代理交互中提取的结构化信息,这些信息会在多个会话中保留。长期记忆不会保存所有原始对话数据,而是仅保留关键洞察,例如对话摘要、事实和知识(语义记忆)或用户偏好。例如,如果客户在聊天中告诉代理他们喜欢的鞋子品牌,AI 代理会将其存储为长期记忆。之后,即使在不同的对话中,代理也能记住并推荐该鞋子品牌,从而使互动更加个性化和更具相关性。
在 AgentCore Memory 中,你可以添加内存策略作为 CreateMemory 操作决定从原始对话中提取哪些信息。这些策略是一种配置,可以智能地从交互中捕获主要概念(在 CreateEvent 操作中以事件形式发送),并将其持久化。目前支持的策略包括:
- 摘要语义记忆(事实和知识)用户偏好
AgentCore Memory 支持两种方式将这些策略添加到内存中
- 内置策略(由 AgentCore 管理并在服务管理账户中运行)自定义策略,允许您附加到系统提示并选择模型。这允许您根据特定领域或用例定制内存提取和整合。
Memory交互流程
Memory API的交互流程如下:
- 创建短期和长期记忆,并添加策略来创建长期记忆开始会话:当Sarah 发起对话时,代理会创建一个新的、唯一的会话 ID 来单独跟踪此交互捕获对话历史记录:当Sarah解释她的问题时,每条消息(她的问题和代理的回复)都使用 CreateEvent 操作保存为一个事件,确保按顺序记录完整的对话。生成长期记忆:在后台,异步提取过程每隔几轮运行一次。此过程使用内置或自定义内存策略(您在通过 CreateMemory 操作设置 AgentCore Memory 时已配置)分析近期原始事件,以提取长期记忆,例如摘要、语义事实或用户偏好,然后将其存储以备将来使用。从短期记忆中检索过去的互动:为了提供情境感知的帮助,agent调用ListEvents来加载对话历史记录。这有助于agent了解Sarah之前提出过哪些问题。利用长期记忆提供个性化帮助:代理调用 RetrieveMemoryRecords,它会对提取的长期记忆进行语义搜索,以查找有关 Sarah 的偏好、订单历史记录或过往顾虑的相关洞察。这使得代理能够提供高度个性化的帮助,而无需要求 Sarah 重复她在之前的聊天中已经分享过的信息。
Long term memory示例代码
import timeimport loggingimport osimport timefrom datetime import datetimefrom bedrock_agentcore.memory import MemoryClientfrom bedrock_agentcore.memory.constants import StrategyTypefrom botocore.exceptions import ClientErrorfrom strands import tool, Agentfrom strands_tools.agent_core_memory import AgentCoreMemoryToolProvider## Environment set uplogging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s", datefmt="%Y-%m-%d %H:%M:%S")logger = logging.getLogger("culinary-memory")region = os.getenv('AWS_REGION', 'us-east-1')## Creating Memory with Long-Term Strategiesclient = MemoryClient(region_name=region)memory_name = "CulinaryAssistant"memory_id = Nonetry: print("Creating Long-Term Memory...") # We use a more descriptive name for our long-term memory resource memory_name = memory_name # Create memory with user preference strategy memory = client.create_memory_and_wait( name=memory_name, description="Culinary Assistant Agent with long term memory", strategies=[{ StrategyType.USER_PREFERENCE.value: { "name": "UserPreferences", "description": "Captures user preferences", "namespaces": ["user/{actorId}/preferences"] } }], event_expiry_days=7, max_wait=300, poll_interval=10 ) memory_id = memory['id'] print(f"Memory created successfully with ID: {memory_id}")except ClientError as e: if e.response['Error']['Code'] == 'ValidationException' and "already exists" in str(e): # If memory already exists, retrieve its ID memories = client.list_memories() memory_id = next((m['id'] for m in memories if m['id'].startswith(memory_name)), None) logger.info(f"Memory already exists. Using existing memory ID: {memory_id}")except Exception as e: # Handle any errors during memory creation logger.info(f"❌ ERROR: {e}") import traceback traceback.print_exc() # Cleanup on error - delete the memory if it was partially created if memory_id: try: client.delete_memory_and_wait(memory_id=memory_id) logger.info(f"Cleaned up memory: {memory_id}") except Exception as cleanup_error: logger.info(f"Failed to clean up memory: {cleanup_error}")## Saving Previous Conversations to Memoryactor_id = f"user-{datetime.now().strftime('%Y%m%d%H%M%S')}"session_id = f"foodie-{datetime.now().strftime('%Y%m%d%H%M%S')}"namespace = f"user/{actor_id}/preferences"previous_messages = [ ("Hi, I'm John", "USER"), ("Hi John, how can I help you with food recommendations today?", "ASSISTANT"), ("I'm looking for some vegetarian dishes to try this weekend.", "USER"), ("That sounds great! I'd be happy to help with vegetarian recommendations. Do you have any specific ingredients or cuisine types you prefer?", "ASSISTANT"), ("Yes, I really like tofu and fresh vegetables in my dishes", "USER"), ("Perfect! Tofu and fresh vegetables make for excellent vegetarian meals. I can suggest some stir-fries, Buddha bowls, or tofu curries. Do you have any other preferences?", "ASSISTANT"), ("I also really enjoy Italian cuisine. I love pasta dishes and would like them to be vegetarian-friendly.", "USER"), ("Excellent! Italian cuisine has wonderful vegetarian options. I can recommend pasta primavera, mushroom risotto, eggplant parmesan, or penne arrabbiata. The combination of Italian flavors with vegetarian ingredients creates delicious meals!", "ASSISTANT"), ("I spent 2 hours looking through cookbooks but couldn't find inspiring vegetarian Italian recipes", "USER"), ("I'm sorry you had trouble finding inspiring recipes! Let me help you with some creative vegetarian Italian dishes. How about stuffed bell peppers with Italian herbs and rice, spinach and ricotta cannelloni, or a Mediterranean vegetable lasagna?", "ASSISTANT"), ("Hey, I appreciate food assistants with good taste", "USER"), ("Ha! I definitely try to bring good taste to the table! Speaking of which, shall we explore some more vegetarian Italian recipes that might inspire you?", "ASSISTANT")]print("\nHydrating short term memory with previous conversations...")# Save the conversation history to short-term memoryinitial = client.create_event( memory_id=memory_id, actor_id=actor_id, session_id=session_id, messages=previous_messages,)print("✓ Conversation saved in short term memory")events = client.list_events( memory_id=memory_id, actor_id=actor_id, session_id=session_id, max_results=5)events## Retrieving Long-Term Memories# Adding a 30s wait to ensure the memory extraction has time to process the eventtime.sleep(30)try: # Query the memory system for food preferences food_preferences = client.retrieve_memories( memory_id=memory_id, namespace=namespace, query="food preferences", top_k=3 # Return up to 3 most relevant results ) if food_preferences: print(f"Retrieved {len(food_preferences)} relevant preference records:") for i, record in enumerate(food_preferences): print(f"\nMemory {i+1}:") print(f"- Content: {record.get('content', 'Not specified')}") else: print("No matching preference records found.")except Exception as e: print(f"Error retrieving preference records: {e}")## Creating the agentsystem_prompt = f"""You are the Culinary Assistant, a sophisticated restaurant recommendation assistant.PURPOSE:- Help users discover restaurants based on their preferences- Remember user preferences throughout the conversation- Provide personalized dining recommendationsYou have access to a Memory tool that enables you to:- Store user preferences (dietary restrictions, favorite cuisines, budget preferences, etc.)- Retrieve previously stored information to personalize recommendations"""provider = AgentCoreMemoryToolProvider( memory_id=memory_id, actor_id=actor_id, session_id=session_id, namespace=namespace)agent = Agent(tools=provider.tools, model="us.anthropic.claude-3-7-sonnet-20250219-v1:0",system_prompt=system_prompt)agent("Give me restaurant recommendations in Irvine based on my food preferences")
代码说明:
创建了具有用户偏好策略的记忆,跟用户偏好相关的内容会被存入长期记忆中
模拟了一段历史对话,包括:
- 用户自我介绍对食物的偏好
将对话保存到短期记忆
创建agent并赋予agent memory tool工具,与agent交互
理解long term memory
理解长期记忆策略此记忆创建的关键区别在于添加了记忆策略。让我们分解一下这些组件:
用户偏好记忆策略
- 此策略自动识别并从对话中提取用户偏好:
"userPreferenceMemoryStrategy": {"name": "UserPreferences","description": "捕获用户偏好","namespaces": ["user/{actorId}/preferences"]}
记忆命名空间
- namespaces 参数定义了提取信息的存储位置
"namespaces": ["user/{actorId}/preferences"]
此记忆策略创建了一个更复杂的记忆系统,它不仅能记住对话,还能理解和组织对话中的重要信息,以供将来使用。当我们将对话保存到配置了提取策略的Memory资源时,系统会自动处理这些信息以实现长期保留,而无需额外的代码。此自动化过程确保即使短期对话记录过期,重要信息仍可保存在长期记忆中。