随着大语言模型技术的快速迭代2026年的AI应用开发已经进入了Agentic AI时代。传统基于简单问答的LLM应用难以满足复杂业务需求而具备自主规划、工具调用、多步推理能力的智能体系统正成为技术主流。在实际项目中开发者面临的最大挑战不再是模型能力本身而是如何有效管理智能体执行过程中产生的海量上下文信息——从简单的对话历史扩展到工具定义、执行轨迹、多智能体协作等复杂结构。本文将从工程实践角度系统梳理Agent框架、Prompt工程、多智能体协作和RAG技术的最新发展路线。不同于单纯的概念介绍我们将通过完整的AWS Bedrock实战案例展示如何构建具备上下文感知、记忆管理和工具集成能力的生产级智能体应用。无论是正在规划大模型转型的团队还是希望深入理解Agentic AI技术栈的开发者都能通过本文掌握从环境搭建到生产部署的全链路实践方案。1. 理解上下文工程Agentic AI的核心挑战在传统基于大语言模型的聊天应用中上下文管理相对简单——应用层维护完整的对话历史每次请求携带全部上下文即可。但当应用场景演进到Agentic AI时上下文复杂度发生了质的变化。智能体具备任务分解、多步骤执行、工具调用等高级能力这些能力的实现都依赖于更加丰富和复杂的上下文结构。1.1 单Agent场景的上下文膨胀在单Agent架构中除了基础的对话历史还新增了多个关键上下文模块。工具定义需要集成到Context中让Agent知道可以调用哪些外部功能每一次工具调用的完整历史请求参数和返回结果都需要记录系统还需要支持多步骤的推理链存储和复杂任务的分解执行过程。以一个具体例子说明这种上下文增长假设一个数据分析任务被Agent分解成10个子任务每个子任务需要调用两次工具数据清洗和统计分析。这个任务总共会产生41条新增的上下文记录1条初始的任务分解思考加上10个子任务各自产生的4条记录。这种指数级增长对上下文管理提出了全新挑战。1.2 多Agent系统的上下文协同在多Agent系统中问题变得更加复杂。每个Agent都需要维护单Agent所需的完整上下文模块而多个Agent之间还需要进行协作和信息共享。这导致整个系统的上下文总量呈现倍增关系。医疗诊断场景中分诊Agent、检查Agent和诊断Agent需要共享患者信息、检查结果和诊断建议形成复杂的上下文依赖网络。1.3 上下文管理面临的实际挑战模型上下文窗口的物理限制是最直接的挑战。即使是支持超长上下文的模型在处理500页PDF文档这样的大规模内容时也无法一次性加载全部信息。更长的上下文还带来成本压力——大模型的定价通常与处理的token数量成正比过长的上下文会大幅增加调用成本。从工程角度看还存在性能和准确率的隐性下降。多步工具调用导致的过长上下文不仅降低模型响应速度还可能引发Lost in the Middle问题——模型在过长的上下文中迷失方向无法准确捕捉关键信息。上下文工程正是为了解决这些挑战而生的系统性方法论。2. 环境准备与AWS Bedrock基础配置在开始构建智能体应用前需要先完成开发环境的基础配置。AWS Bedrock提供了完整的托管服务让开发者能够通过API访问主流大语言模型同时享受企业级的安全性和可靠性保障。2.1 AWS账户和权限配置首先需要确保AWS账户已开通Bedrock服务访问权限。由于Bedrock目前主要在海外区域可用建议使用us-east-1或us-west-2区域进行开发测试。# 安装AWS CLI并配置凭证 aws configure # 设置默认区域 aws configure set region us-east-1 # 验证Bedrock服务访问权限 aws bedrock list-foundation-models --region us-east-12.2 安装必要的开发工具链对于Python开发环境需要安装boto3和相关的AI开发库# 创建虚拟环境 python -m venv agentic-ai-env source agentic-ai-env/bin/activate # Linux/Mac # agentic-ai-env\Scripts\activate # Windows # 安装核心依赖 pip install boto3 strands-agents langchain2.3 Bedrock模型访问配置在代码中配置Bedrock客户端确保能够正常调用Claude等主流模型import boto3 from botocore.config import Config # 配置Bedrock客户端 bedrock_config Config( region_nameus-east-1, retries{ max_attempts: 10, mode: adaptive } ) bedrock_runtime boto3.client(bedrock-runtime, configbedrock_config) # 测试模型连接 def test_bedrock_connection(): try: response bedrock_runtime.converse( modelIdanthropic.claude-3-sonnet-20240229-v1:0, messages[{ role: user, content: [{text: Hello, test connection}] }] ) print(Bedrock连接测试成功) return True except Exception as e: print(f连接测试失败: {e}) return False3. 构建基础智能体从对话管理到工具集成有了基础环境后我们开始构建第一个具备完整上下文管理能力的智能体。使用Strands Agents框架可以快速实现生产就绪的解决方案。3.1 创建基础对话智能体Strands Agents提供了简洁的API来创建功能完整的智能体。以下代码展示如何创建一个具备滑动窗口上下文管理的客服助手from strands import Agent from strands.agent.conversation_manager import SlidingWindowConversationManager # 配置对话管理器 - 保持最近20轮对话 conversation_manager SlidingWindowConversationManager( window_size20, should_truncate_resultsTrue # 自动截断过长的工具结果 ) # 创建智能体实例 customer_service_agent Agent( conversation_managerconversation_manager, system_prompt你是一个专业的客服助手擅长解决用户问题并提供准确信息。请保持友好、专业的语气。, modelanthropic.claude-3-sonnet-20240229-v1:0 ) # 测试对话功能 def test_agent_conversation(): # 第一轮对话 response1 customer_service_agent.invoke(你好我的订单状态如何查询) print(f助手: {response1}) # 第二轮对话 - 依赖上下文 response2 customer_service_agent.invoke(订单号是12345能具体告诉我物流信息吗) print(f助手: {response2}) # 查看当前对话历史长度 print(f当前对话轮次: {len(customer_service_agent.messages)})3.2 实现工具集成能力智能体的核心价值在于能够调用外部工具完成任务。下面实现一个数据分析智能体集成pandas数据处理工具from strands import Agent, tool import pandas as pd import json tool def load_csv_data(file_path: str) - str: 加载CSV文件并返回基本信息 try: df pd.read_csv(file_path) info { row_count: len(df), column_count: len(df.columns), columns: list(df.columns), sample_data: df.head(3).to_dict(records) } return json.dumps(info, ensure_asciiFalse) except Exception as e: return f错误: {str(e)} tool def analyze_data_statistics(file_path: str, column_name: str) - str: 分析指定列的统计信息 try: df pd.read_csv(file_path) if column_name not in df.columns: return f列 {column_name} 不存在 stats { mean: df[column_name].mean(), median: df[column_name].median(), std: df[column_name].std(), min: df[column_name].min(), max: df[column_name].max() } return json.dumps(stats, ensure_asciiFalse) except Exception as e: return f分析错误: {str(e)} # 创建数据分析智能体 data_analysis_agent Agent( tools[load_csv_data, analyze_data_statistics], system_prompt你是一个数据分析专家可以帮助用户加载和分析CSV文件数据。, modelanthropic.claude-3-sonnet-20240229-v1:0 ) # 测试工具调用 def test_data_analysis(): response data_analysis_agent.invoke( 请加载data/sales.csv文件并分析销售额列的统计信息 ) print(f分析结果: {response})3.3 配置Prompt Cache优化成本对于需要频繁调用相同工具定义的场景使用Prompt Cache可以显著降低推理成本。Bedrock的Converse API原生支持缓存机制import boto3 def create_cached_agent_request(): 创建支持缓存的智能体请求 bedrock_runtime boto3.client(bedrock-runtime, region_nameus-east-1) response bedrock_runtime.converse( modelIdanthropic.claude-3-7-sonnet-20250219-v1:0, # 系统提示设置缓存点 system[ { text: 你是一个专业的代码助手擅长分析代码质量和性能优化。 }, { cachePoint: {type: default} } ], # 工具配置设置缓存点 toolConfig{ tools: [ { toolSpec: { name: code_review, description: 代码审查和质量分析, inputSchema: { json: { type: object, properties: { code: {type: string}, language: {type: string} } } } } } ], cachePoint: {type: default} }, messages[ { role: user, content: [ {text: 请分析这段Python代码的质量问题} ] } ] ) # 检查缓存使用情况 usage response[usage] print(f缓存读取tokens: {usage.get(cacheReadInputTokens, 0)}) print(f标准输入tokens: {usage.get(inputTokens, 0)}) return response4. 实现高级记忆管理跨会话上下文持久化基础对话管理只能处理单次会话的上下文而真正的智能体需要具备跨会话的记忆能力。AWS Bedrock AgentCore Memory提供了企业级的记忆管理解决方案。4.1 配置AgentCore Memory服务首先需要初始化记忆管理客户端并创建记忆实例import boto3 from datetime import datetime class MemoryManager: def __init__(self, region_nameus-west-2): self.client boto3.client(bedrock-agentcore-memory, region_nameregion_name) def create_memory_instance(self, name, description): 创建记忆实例 response self.client.create_memory( Namename, Descriptiondescription, Strategies[ { UserPreferenceMemoryStrategy: { Name: UserPreferences, Namespaces: [/users/{actorId}] } }, { SemanticMemoryStrategy: { Name: SemanticFacts, Namespaces: [/knowledge/general] } } ] ) return response[MemoryId] def store_user_preference(self, memory_id, actor_id, preference_data): 存储用户偏好记忆 response self.client.put_memory_item( MemoryIdmemory_id, ActorIdactor_id, Namespacef/users/{actor_id}, Content{ Text: preference_data, Timestamp: datetime.utcnow().isoformat() } ) return response def retrieve_relevant_memories(self, memory_id, actor_id, query, max_results5): 检索相关记忆 response self.client.get_memory_items( MemoryIdmemory_id, ActorIdactor_id, Queryquery, MaxResultsmax_results ) return response.get(Items, []) # 使用示例 memory_manager MemoryManager() memory_id memory_manager.create_memory_instance( CustomerServiceMemory, 客服智能体的长期记忆存储 )4.2 集成记忆管理的智能体将记忆管理功能集成到智能体中实现真正的个性化服务from strands import Agent, HookProvider, HookRegistry, MessageAddedEvent, AfterInvocationEvent class MemoryEnhancedAgent: def __init__(self, memory_manager, memory_id): self.memory_manager memory_manager self.memory_id memory_id self.agent self._create_enhanced_agent() def _create_enhanced_agent(self): 创建具备记忆能力的智能体 memory_hooks MemoryHookProvider(self.memory_manager, self.memory_id) return Agent( hooks[memory_hooks], system_prompt你是一个个性化助手能够记住用户的偏好和历史交互。, modelanthropic.claude-3-sonnet-20240229-v1:0 ) def invoke(self, message, actor_iddefault_user): 调用智能体并自动管理记忆 # 设置当前用户上下文 self.agent.state[actor_id] actor_id response self.agent.invoke(message) return response class MemoryHookProvider(HookProvider): 记忆管理钩子提供者 def __init__(self, memory_manager, memory_id): self.memory_manager memory_manager self.memory_id memory_id def register_hooks(self, registry: HookRegistry): registry.add_callback(MessageAddedEvent, self._on_message_added) registry.add_callback(AfterInvocationEvent, self._after_invocation) def _on_message_added(self, event: MessageAddedEvent): 消息添加时自动检索相关记忆 messages event.agent.messages if messages and messages[-1][role] user: user_message messages[-1][content][0].get(text, ) actor_id event.agent.state.get(actor_id, default_user) # 检索相关记忆 memories self.memory_manager.retrieve_relevant_memories( self.memory_id, actor_id, user_message ) if memories: # 将记忆注入上下文 context \n.join([item[Content][Text] for item in memories]) enhanced_message f{user_message}\n\n相关历史信息:{context} messages[-1][content][0][text] enhanced_message def _after_invocation(self, event: AfterInvocationEvent): 调用完成后保存重要信息到记忆 messages event.agent.messages if len(messages) 2: user_msg messages[-2][content][0][text] assistant_msg messages[-1][content][0][text] actor_id event.agent.state.get(actor_id, default_user) # 判断是否需要保存为长期记忆 if self._should_save_as_memory(user_msg, assistant_msg): self.memory_manager.store_user_preference( self.memory_id, actor_id, f用户询问: {user_msg} - 助手回复: {assistant_msg} ) def _should_save_as_memory(self, user_msg, assistant_msg): 判断对话内容是否需要保存为长期记忆 memory_keywords [记住, 偏好, 喜欢, 经常, 总是] return any(keyword in user_msg for keyword in memory_keywords)5. 多智能体系统架构与协作实践单个智能体的能力有限复杂任务需要多个智能体协作完成。多智能体系统通过分工协作能够处理更加复杂的业务场景。5.1 设计多智能体协作架构以电商客服系统为例设计包含分诊、查询、处理三个智能体的协作架构from strands import Agent from typing import Dict, List import asyncio class MultiAgentOrchestrator: 多智能体协调器 def __init__(self): self.agents self._initialize_agents() self.communication_protocol self._setup_communication() def _initialize_agents(self) - Dict[str, Agent]: 初始化各个智能体 return { triage_agent: Agent( system_prompt你是分诊助手负责分析用户问题类型并路由到合适的处理单元。, modelanthropic.claude-3-haiku-20240307-v1:0 ), query_agent: Agent( system_prompt你是查询专家擅长从数据库和知识库中检索信息。, modelanthropic.claude-3-sonnet-20240229-v1:0 ), processing_agent: Agent( system_prompt你是处理专家负责执行具体的业务操作和问题解决。, modelanthropic.claude-3-sonnet-20240229-v1:0 ) } def _setup_communication(self): 设置智能体间通信协议 return { message_format: { from_agent: str, to_agent: str, message_type: str, # query, result, error content: dict, timestamp: str }, routing_rules: { order_status: query_agent, refund_request: processing_agent, product_info: query_agent, complaint: processing_agent } } async def process_user_request(self, user_message: str) - str: 处理用户请求的多智能体协作流程 # 第一步分诊智能体分析问题类型 triage_result await self._call_agent( triage_agent, f分析用户问题类型: {user_message}. 返回JSON格式: {{problem_type: 类型, priority: 优先级}} ) # 解析分诊结果 problem_info self._parse_triage_result(triage_result) target_agent self.communication_protocol[routing_rules].get( problem_info[problem_type], processing_agent ) # 第二步目标智能体处理具体问题 final_result await self._call_agent( target_agent, f用户问题: {user_message}. 问题类型: {problem_info[problem_type]} ) return final_result async def _call_agent(self, agent_name: str, message: str) - str: 调用指定智能体 agent self.agents[agent_name] response agent.invoke(message) return response def _parse_triage_result(self, result: str) - dict: 解析分诊结果 # 简化实现实际项目中需要更健壮的JSON解析 import json try: return json.loads(result) except: return {problem_type: general, priority: normal} # 使用多智能体系统 async def test_multi_agent_system(): orchestrator MultiAgentOrchestrator() test_cases [ 我的订单12345到哪里了, 我想申请商品退货, 这个产品的规格参数是什么 ] for case in test_cases: result await orchestrator.process_user_request(case) print(f问题: {case}) print(f处理结果: {result}\n)5.2 实现智能体间上下文共享多智能体协作的关键在于上下文的有效共享。通过AgentCore Gateway实现工具和上下文的统一管理import boto3 from typing import Dict, Any class AgentGatewayManager: 智能体网关管理器 def __init__(self, region_nameus-west-2): self.client boto3.client(bedrock-agentcore-gateway, region_nameregion_name) self.gateway_id self._create_gateway() def _create_gateway(self) - str: 创建工具网关 response self.client.create_gateway( NameMultiAgentGateway, Description多智能体系统工具网关 ) return response[GatewayId] def register_shared_tools(self): 注册共享工具到网关 tools [ { name: order_lookup, description: 订单信息查询工具, inputSchema: { type: object, properties: { order_id: {type: string}, customer_id: {type: string} } }, tags: [order, query, shared] }, { name: customer_info, description: 客户信息查询工具, inputSchema: { type: object, properties: { customer_id: {type: string}, info_type: {type: string} } }, tags: [customer, query, shared] } ] for tool_def in tools: self.client.register_tool( GatewayIdself.gateway_id, ToolDefinitiontool_def ) def search_relevant_tools(self, query: str, max_results: int 3) - List[Dict]: 基于语义搜索相关工具 response self.client.search_tools( GatewayIdself.gateway_id, Queryquery, MaxResultsmax_results ) return response.get(Tools, []) def get_tool_context(self, tool_names: List[str]) - Dict[str, Any]: 获取工具定义上下文 context {} for tool_name in tool_names: tools self.search_relevant_tools(tool_name, 1) if tools: context[tool_name] tools[0] return context # 集成网关的多智能体协调器 class GatewayEnhancedOrchestrator(MultiAgentOrchestrator): 增强的多智能体协调器 def __init__(self): super().__init__() self.gateway_manager AgentGatewayManager() self.gateway_manager.register_shared_tools() async def process_user_request(self, user_message: str) - str: 增强的处理流程集成工具网关 # 分析需要的工具 required_tools self._analyze_required_tools(user_message) tool_context self.gateway_manager.get_tool_context(required_tools) # 将工具上下文注入到智能体请求中 enhanced_message f{user_message}\n\n可用工具: {tool_context} return await super().process_user_request(enhanced_message) def _analyze_required_tools(self, message: str) - List[str]: 分析消息需要的工具 tool_mapping { 订单: [order_lookup], 客户: [customer_info], 支付: [payment_tools], 物流: [shipping_tools] } required_tools [] for keyword, tools in tool_mapping.items(): if keyword in message: required_tools.extend(tools) return list(set(required_tools)) # 去重6. RAG系统进阶从基础检索到智能体增强检索增强生成RAG是解决模型知识滞后性的关键技术。随着Agentic AI的发展RAG系统也演进为更加智能的架构。6.1 实现模块化RAG系统模块化RAG将检索流程分解为可重用的组件提高系统的灵活性和可维护性from typing import List, Dict, Any import hashlib from langchain.vectorstores import FAISS from langchain.embeddings import BedrockEmbeddings from langchain.text_splitter import RecursiveCharacterTextSplitter class ModularRAGSystem: 模块化RAG系统 def __init__(self, bedrock_regionus-east-1): self.embedding_model BedrockEmbeddings( clientboto3.client(bedrock-runtime, region_namebedrock_region), model_idamazon.titan-embed-text-v1 ) self.vector_store None self.text_splitter RecursiveCharacterTextSplitter( chunk_size1000, chunk_overlap200 ) def ingest_documents(self, documents: List[str], metadata: List[Dict] None): 文档摄取和处理 if metadata is None: metadata [{}] * len(documents) # 文本分割 chunks [] metadatas [] for doc, meta in zip(documents, metadata): doc_chunks self.text_splitter.split_text(doc) chunks.extend(doc_chunks) metadatas.extend([meta] * len(doc_chunks)) # 创建向量存储 self.vector_store FAISS.from_texts( chunks, self.embedding_model, metadatasmetadatas ) def retrieve_relevant_context(self, query: str, k: int 5) - List[Dict]: 检索相关上下文 if self.vector_store is None: return [] # 相似性检索 docs self.vector_store.similarity_search(query, kk) results [] for doc in docs: results.append({ content: doc.page_content, metadata: doc.metadata, score: self._calculate_relevance_score(query, doc.page_content) }) # 按相关性排序 results.sort(keylambda x: x[score], reverseTrue) return results def _calculate_relevance_score(self, query: str, context: str) - float: 计算查询与上下文的相关性分数 # 简化实现实际项目可以使用更复杂的算法 query_words set(query.lower().split()) context_words set(context.lower().split()) intersection query_words.intersection(context_words) if not query_words: return 0.0 return len(intersection) / len(query_words) def generate_response(self, query: str, context: List[Dict]) - str: 基于检索结果生成响应 # 组装上下文 context_text \n\n.join([f参考{i1}: {item[content]} for i, item in enumerate(context)]) prompt f基于以下参考信息回答问题。如果参考信息不足以回答问题请如实说明。 参考信息: {context_text} 问题: {query} 请给出详细、准确的回答: # 调用LLM生成回答 bedrock_runtime boto3.client(bedrock-runtime, region_nameus-east-1) response bedrock_runtime.converse( modelIdanthropic.claude-3-sonnet-20240229-v1:0, messages[{role: user, content: [{text: prompt}]}] ) return response[output][message][content] # 使用示例 def test_modular_rag(): rag_system ModularRAGSystem() # 模拟文档数据 documents [ AWS Bedrock是亚马逊提供的完全托管服务让开发者能够通过API访问基础模型。, Claude模型由Anthropic开发在推理和代码生成方面表现优秀。, 多智能体系统通过分工协作能够处理复杂的业务场景。 ] rag_system.ingest_documents(documents) # 测试检索和生成 query AWS Bedrock是什么服务 context rag_system.retrieve_relevant_context(query) response rag_system.generate_response(query, context) print(f问题: {query}) print(f检索到的上下文数量: {len(context)}) print(f生成的回答: {response})6.2 实现智能体增强的Self-RAGSelf-RAG通过让模型自主决定检索时机和策略实现更加智能的检索增强class SelfRAGAgent: 自优化RAG智能体 def __init__(self, rag_system: ModularRAGSystem): self.rag_system rag_system self.retrieval_threshold 0.3 # 检索阈值 def should_retrieve(self, query: str, conversation_history: List[str]) - bool: 判断是否需要检索 # 基于查询复杂度和历史上下文判断 complexity_score self._calculate_query_complexity(query) history_relevance self._calculate_history_relevance(query, conversation_history) # 如果查询复杂或历史上下文不足需要检索 return complexity_score 0.5 or history_relevance self.retrieval_threshold def _calculate_query_complexity(self, query: str) - float: 计算查询复杂度 # 基于查询长度、专业术语数量等 words query.split() complexity_indicators [如何, 为什么, 步骤, 方法, 比较] score min(len(words) / 20, 1.0) # 长度因素 if any(indicator in query for indicator in complexity_indicators): score 0.3 return min(score, 1.0) def _calculate_history_relevance(self, query: str, history: List[str]) - float: 计算历史上下文相关性 if not history: return 0.0 # 简化实现检查历史中是否包含相关关键词 query_keywords set(query.lower().split()) history_text .join(history).lower() matches sum(1 for keyword in query_keywords if keyword in history_text) return matches / len(query_keywords) if query_keywords else 0.0 def process_query(self, query: str, conversation_history: List[str] None) - str: 处理用户查询 if conversation_history is None: conversation_history [] if self.should_retrieve(query, conversation_history): # 需要检索增强 context self.rag_system.retrieve_relevant_context(query) response self.rag_system.generate_response(query, context) response_source retrieval_augmented else: # 直接基于模型知识回答 response self._generate_direct_response(query) response_source model_knowledge # 记录决策日志 self._log_decision(query, response_source, len(context) if context in locals() else 0) return response def _generate_direct_response(self, query: str) - str: 基于模型知识直接生成回答 bedrock_runtime boto3.client(bedrock-runtime, region_nameus-east-1) response bedrock_runtime.converse( modelIdanthropic.claude-3-sonnet-20240229-v1:0, messages[{role: user, content: [{text: query}]}] ) return response[output][message][content] def _log_decision(self, query: str, source: str, context_count: int): 记录检索决策日志 print(f查询: {query} | 来源: {source} | 上下文数量: {context_count}) # 测试Self-RAG def test_self_rag(): rag_system ModularRAGSystem() self_rag_agent SelfRAGAgent(rag_system) # 简单问题 - 可能不需要检索 simple_query 你好今天天气怎么样 response1 self_rag_agent.process_query(simple_query) print(f简单问题回答: {response1}) # 复杂问题 - 需要检索 complex_query 请详细解释AWS Bedrock的架构设计和最佳实践 response2 self_rag_agent.process_query(complex_query) print(f复杂问题回答: {response2})7. 生产环境部署与运维最佳实践将智能体系统部署到生产环境需要考虑性能、安全、监控等多个方面。AWS Bedrock AgentCore提供了企业级的部署解决方案。7.1 配置生产环境安全策略生产环境的安全配置是首要任务需要确保数据隔离和访问控制# security-policy.yaml AWSTemplateFormatVersion: 2010-09-09 Resources: AgentExecutionRole: Type: AWS::IAM::Role Properties: RoleName: BedrockAgentExecutionRole AssumeRolePolicyDocument: Version: 2012-10-17 Statement: - Effect: Allow Principal: Service: bedrock.amazonaws.com Action: sts:AssumeRole Policies: - PolicyName: BedrockAgentBasicPolicy PolicyDocument: Version: 2012-10-17 Statement: - Effect: Allow Action: - bedrock:InvokeModel - bedrock:InvokeModelWithResponseStream Resource: * - Effect: Allow Action: - s3:GetObject - s3:ListBucket Resource: - !Sub arn:aws:s3:::${DataBucket} - !Sub arn:aws:s3:::${DataBucket}/* DataEncryptionKey: Type: AWS::KMS::Key Properties: Description: Encryption key for agent data KeyPolicy: Version: 2012-10-17 Statement: - Effect: Allow Principal: AWS: !Sub arn:aws:iam::${AWS::AccountId}:root Action: kms:* Resource: * Parameters: DataBucket: Type: String Description: S3 bucket for agent data storage7.2 实现监控和可观测性智能体系统的可观测性对于生产运维至关重要需要监控性能指标和业务指标import boto3 import time from datetime import datetime from typing import Dict, Any class AgentMonitoring: 智能体监控系统 def __init__(self): self.cloudwatch boto3.client(cloudwatch) self.xray boto3.client(xray) self.metrics_namespace BedrockAgents def record_invocation_metrics(self, agent_name: str, duration: float, success: bool, token_usage: Dict[str, int]): 记录调用指标 metrics_data [ { MetricName: InvocationDuration, Dimensions: [ {Name: AgentName, Value: agent_name}, {Name: Success, Value: str(success)} ], Value: duration, Unit: Milliseconds }, { MetricName: InvocationCount, Dimensions: [{Name: AgentName, Value: agent_name}], Value: 1, Unit: Count }, { MetricName: InputTokens, Dimensions: [{Name: AgentName, Value: agent_name}], Value: token_usage.get(input_tokens, 0), Unit: Count }, { MetricName: OutputTokens,
AWS Bedrock实战：构建具备上下文感知与记忆管理的智能体系统
发布时间：2026/7/14 2:37:34
随着大语言模型技术的快速迭代2026年的AI应用开发已经进入了Agentic AI时代。传统基于简单问答的LLM应用难以满足复杂业务需求而具备自主规划、工具调用、多步推理能力的智能体系统正成为技术主流。在实际项目中开发者面临的最大挑战不再是模型能力本身而是如何有效管理智能体执行过程中产生的海量上下文信息——从简单的对话历史扩展到工具定义、执行轨迹、多智能体协作等复杂结构。本文将从工程实践角度系统梳理Agent框架、Prompt工程、多智能体协作和RAG技术的最新发展路线。不同于单纯的概念介绍我们将通过完整的AWS Bedrock实战案例展示如何构建具备上下文感知、记忆管理和工具集成能力的生产级智能体应用。无论是正在规划大模型转型的团队还是希望深入理解Agentic AI技术栈的开发者都能通过本文掌握从环境搭建到生产部署的全链路实践方案。1. 理解上下文工程Agentic AI的核心挑战在传统基于大语言模型的聊天应用中上下文管理相对简单——应用层维护完整的对话历史每次请求携带全部上下文即可。但当应用场景演进到Agentic AI时上下文复杂度发生了质的变化。智能体具备任务分解、多步骤执行、工具调用等高级能力这些能力的实现都依赖于更加丰富和复杂的上下文结构。1.1 单Agent场景的上下文膨胀在单Agent架构中除了基础的对话历史还新增了多个关键上下文模块。工具定义需要集成到Context中让Agent知道可以调用哪些外部功能每一次工具调用的完整历史请求参数和返回结果都需要记录系统还需要支持多步骤的推理链存储和复杂任务的分解执行过程。以一个具体例子说明这种上下文增长假设一个数据分析任务被Agent分解成10个子任务每个子任务需要调用两次工具数据清洗和统计分析。这个任务总共会产生41条新增的上下文记录1条初始的任务分解思考加上10个子任务各自产生的4条记录。这种指数级增长对上下文管理提出了全新挑战。1.2 多Agent系统的上下文协同在多Agent系统中问题变得更加复杂。每个Agent都需要维护单Agent所需的完整上下文模块而多个Agent之间还需要进行协作和信息共享。这导致整个系统的上下文总量呈现倍增关系。医疗诊断场景中分诊Agent、检查Agent和诊断Agent需要共享患者信息、检查结果和诊断建议形成复杂的上下文依赖网络。1.3 上下文管理面临的实际挑战模型上下文窗口的物理限制是最直接的挑战。即使是支持超长上下文的模型在处理500页PDF文档这样的大规模内容时也无法一次性加载全部信息。更长的上下文还带来成本压力——大模型的定价通常与处理的token数量成正比过长的上下文会大幅增加调用成本。从工程角度看还存在性能和准确率的隐性下降。多步工具调用导致的过长上下文不仅降低模型响应速度还可能引发Lost in the Middle问题——模型在过长的上下文中迷失方向无法准确捕捉关键信息。上下文工程正是为了解决这些挑战而生的系统性方法论。2. 环境准备与AWS Bedrock基础配置在开始构建智能体应用前需要先完成开发环境的基础配置。AWS Bedrock提供了完整的托管服务让开发者能够通过API访问主流大语言模型同时享受企业级的安全性和可靠性保障。2.1 AWS账户和权限配置首先需要确保AWS账户已开通Bedrock服务访问权限。由于Bedrock目前主要在海外区域可用建议使用us-east-1或us-west-2区域进行开发测试。# 安装AWS CLI并配置凭证 aws configure # 设置默认区域 aws configure set region us-east-1 # 验证Bedrock服务访问权限 aws bedrock list-foundation-models --region us-east-12.2 安装必要的开发工具链对于Python开发环境需要安装boto3和相关的AI开发库# 创建虚拟环境 python -m venv agentic-ai-env source agentic-ai-env/bin/activate # Linux/Mac # agentic-ai-env\Scripts\activate # Windows # 安装核心依赖 pip install boto3 strands-agents langchain2.3 Bedrock模型访问配置在代码中配置Bedrock客户端确保能够正常调用Claude等主流模型import boto3 from botocore.config import Config # 配置Bedrock客户端 bedrock_config Config( region_nameus-east-1, retries{ max_attempts: 10, mode: adaptive } ) bedrock_runtime boto3.client(bedrock-runtime, configbedrock_config) # 测试模型连接 def test_bedrock_connection(): try: response bedrock_runtime.converse( modelIdanthropic.claude-3-sonnet-20240229-v1:0, messages[{ role: user, content: [{text: Hello, test connection}] }] ) print(Bedrock连接测试成功) return True except Exception as e: print(f连接测试失败: {e}) return False3. 构建基础智能体从对话管理到工具集成有了基础环境后我们开始构建第一个具备完整上下文管理能力的智能体。使用Strands Agents框架可以快速实现生产就绪的解决方案。3.1 创建基础对话智能体Strands Agents提供了简洁的API来创建功能完整的智能体。以下代码展示如何创建一个具备滑动窗口上下文管理的客服助手from strands import Agent from strands.agent.conversation_manager import SlidingWindowConversationManager # 配置对话管理器 - 保持最近20轮对话 conversation_manager SlidingWindowConversationManager( window_size20, should_truncate_resultsTrue # 自动截断过长的工具结果 ) # 创建智能体实例 customer_service_agent Agent( conversation_managerconversation_manager, system_prompt你是一个专业的客服助手擅长解决用户问题并提供准确信息。请保持友好、专业的语气。, modelanthropic.claude-3-sonnet-20240229-v1:0 ) # 测试对话功能 def test_agent_conversation(): # 第一轮对话 response1 customer_service_agent.invoke(你好我的订单状态如何查询) print(f助手: {response1}) # 第二轮对话 - 依赖上下文 response2 customer_service_agent.invoke(订单号是12345能具体告诉我物流信息吗) print(f助手: {response2}) # 查看当前对话历史长度 print(f当前对话轮次: {len(customer_service_agent.messages)})3.2 实现工具集成能力智能体的核心价值在于能够调用外部工具完成任务。下面实现一个数据分析智能体集成pandas数据处理工具from strands import Agent, tool import pandas as pd import json tool def load_csv_data(file_path: str) - str: 加载CSV文件并返回基本信息 try: df pd.read_csv(file_path) info { row_count: len(df), column_count: len(df.columns), columns: list(df.columns), sample_data: df.head(3).to_dict(records) } return json.dumps(info, ensure_asciiFalse) except Exception as e: return f错误: {str(e)} tool def analyze_data_statistics(file_path: str, column_name: str) - str: 分析指定列的统计信息 try: df pd.read_csv(file_path) if column_name not in df.columns: return f列 {column_name} 不存在 stats { mean: df[column_name].mean(), median: df[column_name].median(), std: df[column_name].std(), min: df[column_name].min(), max: df[column_name].max() } return json.dumps(stats, ensure_asciiFalse) except Exception as e: return f分析错误: {str(e)} # 创建数据分析智能体 data_analysis_agent Agent( tools[load_csv_data, analyze_data_statistics], system_prompt你是一个数据分析专家可以帮助用户加载和分析CSV文件数据。, modelanthropic.claude-3-sonnet-20240229-v1:0 ) # 测试工具调用 def test_data_analysis(): response data_analysis_agent.invoke( 请加载data/sales.csv文件并分析销售额列的统计信息 ) print(f分析结果: {response})3.3 配置Prompt Cache优化成本对于需要频繁调用相同工具定义的场景使用Prompt Cache可以显著降低推理成本。Bedrock的Converse API原生支持缓存机制import boto3 def create_cached_agent_request(): 创建支持缓存的智能体请求 bedrock_runtime boto3.client(bedrock-runtime, region_nameus-east-1) response bedrock_runtime.converse( modelIdanthropic.claude-3-7-sonnet-20250219-v1:0, # 系统提示设置缓存点 system[ { text: 你是一个专业的代码助手擅长分析代码质量和性能优化。 }, { cachePoint: {type: default} } ], # 工具配置设置缓存点 toolConfig{ tools: [ { toolSpec: { name: code_review, description: 代码审查和质量分析, inputSchema: { json: { type: object, properties: { code: {type: string}, language: {type: string} } } } } } ], cachePoint: {type: default} }, messages[ { role: user, content: [ {text: 请分析这段Python代码的质量问题} ] } ] ) # 检查缓存使用情况 usage response[usage] print(f缓存读取tokens: {usage.get(cacheReadInputTokens, 0)}) print(f标准输入tokens: {usage.get(inputTokens, 0)}) return response4. 实现高级记忆管理跨会话上下文持久化基础对话管理只能处理单次会话的上下文而真正的智能体需要具备跨会话的记忆能力。AWS Bedrock AgentCore Memory提供了企业级的记忆管理解决方案。4.1 配置AgentCore Memory服务首先需要初始化记忆管理客户端并创建记忆实例import boto3 from datetime import datetime class MemoryManager: def __init__(self, region_nameus-west-2): self.client boto3.client(bedrock-agentcore-memory, region_nameregion_name) def create_memory_instance(self, name, description): 创建记忆实例 response self.client.create_memory( Namename, Descriptiondescription, Strategies[ { UserPreferenceMemoryStrategy: { Name: UserPreferences, Namespaces: [/users/{actorId}] } }, { SemanticMemoryStrategy: { Name: SemanticFacts, Namespaces: [/knowledge/general] } } ] ) return response[MemoryId] def store_user_preference(self, memory_id, actor_id, preference_data): 存储用户偏好记忆 response self.client.put_memory_item( MemoryIdmemory_id, ActorIdactor_id, Namespacef/users/{actor_id}, Content{ Text: preference_data, Timestamp: datetime.utcnow().isoformat() } ) return response def retrieve_relevant_memories(self, memory_id, actor_id, query, max_results5): 检索相关记忆 response self.client.get_memory_items( MemoryIdmemory_id, ActorIdactor_id, Queryquery, MaxResultsmax_results ) return response.get(Items, []) # 使用示例 memory_manager MemoryManager() memory_id memory_manager.create_memory_instance( CustomerServiceMemory, 客服智能体的长期记忆存储 )4.2 集成记忆管理的智能体将记忆管理功能集成到智能体中实现真正的个性化服务from strands import Agent, HookProvider, HookRegistry, MessageAddedEvent, AfterInvocationEvent class MemoryEnhancedAgent: def __init__(self, memory_manager, memory_id): self.memory_manager memory_manager self.memory_id memory_id self.agent self._create_enhanced_agent() def _create_enhanced_agent(self): 创建具备记忆能力的智能体 memory_hooks MemoryHookProvider(self.memory_manager, self.memory_id) return Agent( hooks[memory_hooks], system_prompt你是一个个性化助手能够记住用户的偏好和历史交互。, modelanthropic.claude-3-sonnet-20240229-v1:0 ) def invoke(self, message, actor_iddefault_user): 调用智能体并自动管理记忆 # 设置当前用户上下文 self.agent.state[actor_id] actor_id response self.agent.invoke(message) return response class MemoryHookProvider(HookProvider): 记忆管理钩子提供者 def __init__(self, memory_manager, memory_id): self.memory_manager memory_manager self.memory_id memory_id def register_hooks(self, registry: HookRegistry): registry.add_callback(MessageAddedEvent, self._on_message_added) registry.add_callback(AfterInvocationEvent, self._after_invocation) def _on_message_added(self, event: MessageAddedEvent): 消息添加时自动检索相关记忆 messages event.agent.messages if messages and messages[-1][role] user: user_message messages[-1][content][0].get(text, ) actor_id event.agent.state.get(actor_id, default_user) # 检索相关记忆 memories self.memory_manager.retrieve_relevant_memories( self.memory_id, actor_id, user_message ) if memories: # 将记忆注入上下文 context \n.join([item[Content][Text] for item in memories]) enhanced_message f{user_message}\n\n相关历史信息:{context} messages[-1][content][0][text] enhanced_message def _after_invocation(self, event: AfterInvocationEvent): 调用完成后保存重要信息到记忆 messages event.agent.messages if len(messages) 2: user_msg messages[-2][content][0][text] assistant_msg messages[-1][content][0][text] actor_id event.agent.state.get(actor_id, default_user) # 判断是否需要保存为长期记忆 if self._should_save_as_memory(user_msg, assistant_msg): self.memory_manager.store_user_preference( self.memory_id, actor_id, f用户询问: {user_msg} - 助手回复: {assistant_msg} ) def _should_save_as_memory(self, user_msg, assistant_msg): 判断对话内容是否需要保存为长期记忆 memory_keywords [记住, 偏好, 喜欢, 经常, 总是] return any(keyword in user_msg for keyword in memory_keywords)5. 多智能体系统架构与协作实践单个智能体的能力有限复杂任务需要多个智能体协作完成。多智能体系统通过分工协作能够处理更加复杂的业务场景。5.1 设计多智能体协作架构以电商客服系统为例设计包含分诊、查询、处理三个智能体的协作架构from strands import Agent from typing import Dict, List import asyncio class MultiAgentOrchestrator: 多智能体协调器 def __init__(self): self.agents self._initialize_agents() self.communication_protocol self._setup_communication() def _initialize_agents(self) - Dict[str, Agent]: 初始化各个智能体 return { triage_agent: Agent( system_prompt你是分诊助手负责分析用户问题类型并路由到合适的处理单元。, modelanthropic.claude-3-haiku-20240307-v1:0 ), query_agent: Agent( system_prompt你是查询专家擅长从数据库和知识库中检索信息。, modelanthropic.claude-3-sonnet-20240229-v1:0 ), processing_agent: Agent( system_prompt你是处理专家负责执行具体的业务操作和问题解决。, modelanthropic.claude-3-sonnet-20240229-v1:0 ) } def _setup_communication(self): 设置智能体间通信协议 return { message_format: { from_agent: str, to_agent: str, message_type: str, # query, result, error content: dict, timestamp: str }, routing_rules: { order_status: query_agent, refund_request: processing_agent, product_info: query_agent, complaint: processing_agent } } async def process_user_request(self, user_message: str) - str: 处理用户请求的多智能体协作流程 # 第一步分诊智能体分析问题类型 triage_result await self._call_agent( triage_agent, f分析用户问题类型: {user_message}. 返回JSON格式: {{problem_type: 类型, priority: 优先级}} ) # 解析分诊结果 problem_info self._parse_triage_result(triage_result) target_agent self.communication_protocol[routing_rules].get( problem_info[problem_type], processing_agent ) # 第二步目标智能体处理具体问题 final_result await self._call_agent( target_agent, f用户问题: {user_message}. 问题类型: {problem_info[problem_type]} ) return final_result async def _call_agent(self, agent_name: str, message: str) - str: 调用指定智能体 agent self.agents[agent_name] response agent.invoke(message) return response def _parse_triage_result(self, result: str) - dict: 解析分诊结果 # 简化实现实际项目中需要更健壮的JSON解析 import json try: return json.loads(result) except: return {problem_type: general, priority: normal} # 使用多智能体系统 async def test_multi_agent_system(): orchestrator MultiAgentOrchestrator() test_cases [ 我的订单12345到哪里了, 我想申请商品退货, 这个产品的规格参数是什么 ] for case in test_cases: result await orchestrator.process_user_request(case) print(f问题: {case}) print(f处理结果: {result}\n)5.2 实现智能体间上下文共享多智能体协作的关键在于上下文的有效共享。通过AgentCore Gateway实现工具和上下文的统一管理import boto3 from typing import Dict, Any class AgentGatewayManager: 智能体网关管理器 def __init__(self, region_nameus-west-2): self.client boto3.client(bedrock-agentcore-gateway, region_nameregion_name) self.gateway_id self._create_gateway() def _create_gateway(self) - str: 创建工具网关 response self.client.create_gateway( NameMultiAgentGateway, Description多智能体系统工具网关 ) return response[GatewayId] def register_shared_tools(self): 注册共享工具到网关 tools [ { name: order_lookup, description: 订单信息查询工具, inputSchema: { type: object, properties: { order_id: {type: string}, customer_id: {type: string} } }, tags: [order, query, shared] }, { name: customer_info, description: 客户信息查询工具, inputSchema: { type: object, properties: { customer_id: {type: string}, info_type: {type: string} } }, tags: [customer, query, shared] } ] for tool_def in tools: self.client.register_tool( GatewayIdself.gateway_id, ToolDefinitiontool_def ) def search_relevant_tools(self, query: str, max_results: int 3) - List[Dict]: 基于语义搜索相关工具 response self.client.search_tools( GatewayIdself.gateway_id, Queryquery, MaxResultsmax_results ) return response.get(Tools, []) def get_tool_context(self, tool_names: List[str]) - Dict[str, Any]: 获取工具定义上下文 context {} for tool_name in tool_names: tools self.search_relevant_tools(tool_name, 1) if tools: context[tool_name] tools[0] return context # 集成网关的多智能体协调器 class GatewayEnhancedOrchestrator(MultiAgentOrchestrator): 增强的多智能体协调器 def __init__(self): super().__init__() self.gateway_manager AgentGatewayManager() self.gateway_manager.register_shared_tools() async def process_user_request(self, user_message: str) - str: 增强的处理流程集成工具网关 # 分析需要的工具 required_tools self._analyze_required_tools(user_message) tool_context self.gateway_manager.get_tool_context(required_tools) # 将工具上下文注入到智能体请求中 enhanced_message f{user_message}\n\n可用工具: {tool_context} return await super().process_user_request(enhanced_message) def _analyze_required_tools(self, message: str) - List[str]: 分析消息需要的工具 tool_mapping { 订单: [order_lookup], 客户: [customer_info], 支付: [payment_tools], 物流: [shipping_tools] } required_tools [] for keyword, tools in tool_mapping.items(): if keyword in message: required_tools.extend(tools) return list(set(required_tools)) # 去重6. RAG系统进阶从基础检索到智能体增强检索增强生成RAG是解决模型知识滞后性的关键技术。随着Agentic AI的发展RAG系统也演进为更加智能的架构。6.1 实现模块化RAG系统模块化RAG将检索流程分解为可重用的组件提高系统的灵活性和可维护性from typing import List, Dict, Any import hashlib from langchain.vectorstores import FAISS from langchain.embeddings import BedrockEmbeddings from langchain.text_splitter import RecursiveCharacterTextSplitter class ModularRAGSystem: 模块化RAG系统 def __init__(self, bedrock_regionus-east-1): self.embedding_model BedrockEmbeddings( clientboto3.client(bedrock-runtime, region_namebedrock_region), model_idamazon.titan-embed-text-v1 ) self.vector_store None self.text_splitter RecursiveCharacterTextSplitter( chunk_size1000, chunk_overlap200 ) def ingest_documents(self, documents: List[str], metadata: List[Dict] None): 文档摄取和处理 if metadata is None: metadata [{}] * len(documents) # 文本分割 chunks [] metadatas [] for doc, meta in zip(documents, metadata): doc_chunks self.text_splitter.split_text(doc) chunks.extend(doc_chunks) metadatas.extend([meta] * len(doc_chunks)) # 创建向量存储 self.vector_store FAISS.from_texts( chunks, self.embedding_model, metadatasmetadatas ) def retrieve_relevant_context(self, query: str, k: int 5) - List[Dict]: 检索相关上下文 if self.vector_store is None: return [] # 相似性检索 docs self.vector_store.similarity_search(query, kk) results [] for doc in docs: results.append({ content: doc.page_content, metadata: doc.metadata, score: self._calculate_relevance_score(query, doc.page_content) }) # 按相关性排序 results.sort(keylambda x: x[score], reverseTrue) return results def _calculate_relevance_score(self, query: str, context: str) - float: 计算查询与上下文的相关性分数 # 简化实现实际项目可以使用更复杂的算法 query_words set(query.lower().split()) context_words set(context.lower().split()) intersection query_words.intersection(context_words) if not query_words: return 0.0 return len(intersection) / len(query_words) def generate_response(self, query: str, context: List[Dict]) - str: 基于检索结果生成响应 # 组装上下文 context_text \n\n.join([f参考{i1}: {item[content]} for i, item in enumerate(context)]) prompt f基于以下参考信息回答问题。如果参考信息不足以回答问题请如实说明。 参考信息: {context_text} 问题: {query} 请给出详细、准确的回答: # 调用LLM生成回答 bedrock_runtime boto3.client(bedrock-runtime, region_nameus-east-1) response bedrock_runtime.converse( modelIdanthropic.claude-3-sonnet-20240229-v1:0, messages[{role: user, content: [{text: prompt}]}] ) return response[output][message][content] # 使用示例 def test_modular_rag(): rag_system ModularRAGSystem() # 模拟文档数据 documents [ AWS Bedrock是亚马逊提供的完全托管服务让开发者能够通过API访问基础模型。, Claude模型由Anthropic开发在推理和代码生成方面表现优秀。, 多智能体系统通过分工协作能够处理复杂的业务场景。 ] rag_system.ingest_documents(documents) # 测试检索和生成 query AWS Bedrock是什么服务 context rag_system.retrieve_relevant_context(query) response rag_system.generate_response(query, context) print(f问题: {query}) print(f检索到的上下文数量: {len(context)}) print(f生成的回答: {response})6.2 实现智能体增强的Self-RAGSelf-RAG通过让模型自主决定检索时机和策略实现更加智能的检索增强class SelfRAGAgent: 自优化RAG智能体 def __init__(self, rag_system: ModularRAGSystem): self.rag_system rag_system self.retrieval_threshold 0.3 # 检索阈值 def should_retrieve(self, query: str, conversation_history: List[str]) - bool: 判断是否需要检索 # 基于查询复杂度和历史上下文判断 complexity_score self._calculate_query_complexity(query) history_relevance self._calculate_history_relevance(query, conversation_history) # 如果查询复杂或历史上下文不足需要检索 return complexity_score 0.5 or history_relevance self.retrieval_threshold def _calculate_query_complexity(self, query: str) - float: 计算查询复杂度 # 基于查询长度、专业术语数量等 words query.split() complexity_indicators [如何, 为什么, 步骤, 方法, 比较] score min(len(words) / 20, 1.0) # 长度因素 if any(indicator in query for indicator in complexity_indicators): score 0.3 return min(score, 1.0) def _calculate_history_relevance(self, query: str, history: List[str]) - float: 计算历史上下文相关性 if not history: return 0.0 # 简化实现检查历史中是否包含相关关键词 query_keywords set(query.lower().split()) history_text .join(history).lower() matches sum(1 for keyword in query_keywords if keyword in history_text) return matches / len(query_keywords) if query_keywords else 0.0 def process_query(self, query: str, conversation_history: List[str] None) - str: 处理用户查询 if conversation_history is None: conversation_history [] if self.should_retrieve(query, conversation_history): # 需要检索增强 context self.rag_system.retrieve_relevant_context(query) response self.rag_system.generate_response(query, context) response_source retrieval_augmented else: # 直接基于模型知识回答 response self._generate_direct_response(query) response_source model_knowledge # 记录决策日志 self._log_decision(query, response_source, len(context) if context in locals() else 0) return response def _generate_direct_response(self, query: str) - str: 基于模型知识直接生成回答 bedrock_runtime boto3.client(bedrock-runtime, region_nameus-east-1) response bedrock_runtime.converse( modelIdanthropic.claude-3-sonnet-20240229-v1:0, messages[{role: user, content: [{text: query}]}] ) return response[output][message][content] def _log_decision(self, query: str, source: str, context_count: int): 记录检索决策日志 print(f查询: {query} | 来源: {source} | 上下文数量: {context_count}) # 测试Self-RAG def test_self_rag(): rag_system ModularRAGSystem() self_rag_agent SelfRAGAgent(rag_system) # 简单问题 - 可能不需要检索 simple_query 你好今天天气怎么样 response1 self_rag_agent.process_query(simple_query) print(f简单问题回答: {response1}) # 复杂问题 - 需要检索 complex_query 请详细解释AWS Bedrock的架构设计和最佳实践 response2 self_rag_agent.process_query(complex_query) print(f复杂问题回答: {response2})7. 生产环境部署与运维最佳实践将智能体系统部署到生产环境需要考虑性能、安全、监控等多个方面。AWS Bedrock AgentCore提供了企业级的部署解决方案。7.1 配置生产环境安全策略生产环境的安全配置是首要任务需要确保数据隔离和访问控制# security-policy.yaml AWSTemplateFormatVersion: 2010-09-09 Resources: AgentExecutionRole: Type: AWS::IAM::Role Properties: RoleName: BedrockAgentExecutionRole AssumeRolePolicyDocument: Version: 2012-10-17 Statement: - Effect: Allow Principal: Service: bedrock.amazonaws.com Action: sts:AssumeRole Policies: - PolicyName: BedrockAgentBasicPolicy PolicyDocument: Version: 2012-10-17 Statement: - Effect: Allow Action: - bedrock:InvokeModel - bedrock:InvokeModelWithResponseStream Resource: * - Effect: Allow Action: - s3:GetObject - s3:ListBucket Resource: - !Sub arn:aws:s3:::${DataBucket} - !Sub arn:aws:s3:::${DataBucket}/* DataEncryptionKey: Type: AWS::KMS::Key Properties: Description: Encryption key for agent data KeyPolicy: Version: 2012-10-17 Statement: - Effect: Allow Principal: AWS: !Sub arn:aws:iam::${AWS::AccountId}:root Action: kms:* Resource: * Parameters: DataBucket: Type: String Description: S3 bucket for agent data storage7.2 实现监控和可观测性智能体系统的可观测性对于生产运维至关重要需要监控性能指标和业务指标import boto3 import time from datetime import datetime from typing import Dict, Any class AgentMonitoring: 智能体监控系统 def __init__(self): self.cloudwatch boto3.client(cloudwatch) self.xray boto3.client(xray) self.metrics_namespace BedrockAgents def record_invocation_metrics(self, agent_name: str, duration: float, success: bool, token_usage: Dict[str, int]): 记录调用指标 metrics_data [ { MetricName: InvocationDuration, Dimensions: [ {Name: AgentName, Value: agent_name}, {Name: Success, Value: str(success)} ], Value: duration, Unit: Milliseconds }, { MetricName: InvocationCount, Dimensions: [{Name: AgentName, Value: agent_name}], Value: 1, Unit: Count }, { MetricName: InputTokens, Dimensions: [{Name: AgentName, Value: agent_name}], Value: token_usage.get(input_tokens, 0), Unit: Count }, { MetricName: OutputTokens,
)
