掘金 人工智能 18小时前
理解AWS AgentCore - Long term Memory
index_new5.html
../../../zaker_core/zaker_tpl_static/wap/tpl_guoji1.html

 

AWS AgentCore是一套强大的服务组合,旨在提升AI代理的能力。它提供了AgentCore Runtime以支持低延迟、无服务器环境,并处理多模式和长时间运行的代理。AgentCore Memory负责管理短期和长期记忆,为模型提供上下文并支持代理从过往交互中学习。AgentCore Observability则提供代理执行的可视化和调试功能。此外,AgentCore Identity确保AI代理能安全访问AWS及第三方服务,AgentCore Gateway能将现有API和Lambda函数转化为代理就绪工具,AgentCore Browser支持Web自动化,代码解释器则提供安全的代码运行环境。其中,Memory功能尤为关键,它支持摘要、语义记忆和用户偏好等策略,实现个性化和上下文感知的代理交互。

🎯 **AgentCore的全面能力**:AWS AgentCore提供了一系列服务,包括支持多种代理框架和多模式工作负载的Runtime,管理会话和长期记忆的Memory,以及提供执行可视化的Observability。此外,它还支持代理安全访问AWS和第三方服务的Identity,将API和Lambda函数转化为工具的Gateway,用于Web自动化的Browser,以及安全代码运行环境的代码解释器,共同构建强大的AI代理。

🧠 **AgentCore Memory的双重记忆机制**:AgentCore Memory支持短期和长期记忆。短期记忆以事件形式存储交互,维持即时上下文,支持实时对话。长期记忆则提取关键洞察,如对话摘要、事实知识或用户偏好,并跨会话保留,实现个性化推荐。例如,代理会记住用户喜欢的鞋子品牌,并在后续对话中提供相关推荐。

📊 **记忆策略与命名空间**:Memory功能通过策略来智能地从对话中提取信息。支持的策略包括摘要、语义记忆(事实和知识)以及用户偏好。这些策略通过命名空间(如`user/{actorId}/preferences`)来组织存储,确保信息被结构化并易于检索。这些策略可以由AgentCore管理,也可自定义以适应特定领域或用例。

🔄 **Memory交互流程的自动化**:当代理与用户交互时,对话历史被保存为短期事件。后台的异步提取过程会根据预设的内存策略(如用户偏好)分析这些事件,生成长期记忆。代理随后可以通过调用API(如`ListEvents`和`RetrieveMemoryRecords`)来检索短期和长期记忆,从而提供上下文感知和个性化的帮助,无需用户重复提供信息。

💡 **代码示例展示了长期记忆的实现**:提供的Python代码示例详细展示了如何使用`MemoryClient`创建带有用户偏好策略的长期记忆,模拟用户对话并将其保存到短期记忆,然后通过检索机制获取用户偏好信息。最后,展示了如何将此Memory工具集成到AI代理中,使其能够提供个性化的餐饮推荐。

参考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提供两方面支持:

Memory交互流程

Memory API的交互流程如下:

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")

代码说明:

理解long term memory

理解长期记忆策略此记忆创建的关键区别在于添加了记忆策略。让我们分解一下这些组件:

"userPreferenceMemoryStrategy": {"name": "UserPreferences","description": "捕获用户偏好","namespaces": ["user/{actorId}/preferences"]}
"namespaces": ["user/{actorId}/preferences"]

此记忆策略创建了一个更复杂的记忆系统,它不仅能记住对话,还能理解和组织对话中的重要信息,以供将来使用。当我们将对话保存到配置了提取策略的Memory资源时,系统会自动处理这些信息以实现长期保留,而无需额外的代码。此自动化过程确保即使短期对话记录过期,重要信息仍可保存在长期记忆中。

Fish AI Reader

Fish AI Reader

AI辅助创作,多种专业模板,深度分析,高质量内容生成。从观点提取到深度思考,FishAI为您提供全方位的创作支持。新版本引入自定义参数,让您的创作更加个性化和精准。

FishAI

FishAI

鱼阅,AI 时代的下一个智能信息助手,助你摆脱信息焦虑

联系邮箱 441953276@qq.com

相关标签

AWS AgentCore AI代理 记忆管理 长期记忆 个性化推荐
相关文章