使用 Moto 模拟 Amazon Bedrock Agents 与 Knowledge Base：完整指南

Mock测试【免费下载链接】motoA library that allows you to easily mock out tests based on AWS infrastructure.项目地址https://gitcode.com/gh_mirrors/mo/moto点击查看免费下载Amazon Bedrock Agentsbedrock-agent是 AWS 上用于构建、编排和部署 AI Agent 的服务而 Knowledge Base知识库为 Agent 提供基于向量检索的 RAG 能力。在本地开发和单元测试中直接依赖真实 AWS 服务既不现实也代价高昂。Moto 是一个允许你轻松模拟基于 AWS 基础设施测试的库本指南将围绕docs/docs/services/bedrock-agent.rst文档结合仓库源码带你完整掌握在 Moto 中如何对 bedrock-agent 服务的 Agent 与 Knowledge Base 进行本地模拟哪些 API 已实现、底层如何工作、如何编写可运行的 boto3 测试代码以及分页、标签、异常等细节行为。一、服务概览已实现与未实现的 APIbedrock-agent.rst文档通过勾选清单的方式标明了 Moto 对 bedrock-agent 服务的支持范围。截至当前仓库版本已实现[X]的 API 共 11 个类别已实现 APIAgent 生命周期create_agent、get_agent、list_agents、delete_agentKnowledge Base 生命周期create_knowledge_base、get_knowledge_base、list_knowledge_bases、delete_knowledge_base标签管理tag_resource、untag_resource、list_tags_for_resource未实现[ ]的 API数量众多包括associate_agent_knowledge_base、create_agent_action_group、create_agent_alias、create_data_source、create_flow、create_prompt、delete_data_source、get_data_source、list_ingestion_jobs、start_ingestion_job、stop_ingestion_job、ingest_knowledge_base_documents、prepare_agent、update_agent、update_knowledge_base、validate_flow_definition等。如果你在测试中调用这些 API会得到未实现相关的错误响应。从实现结构看该模块覆盖了Agent 与知识库资源的创建/查询/列举/删除 标签管理这一核心闭环适合验证资源生命周期与 CRUD 逻辑但暂不支持 Action Group、数据源导入、Flow 编排等更深层编排功能。编写测试前请先确认你的用例落在已实现清单内。二、模块架构与请求处理链路bedrock-agent 后端遵循 Moto 标准的「URL 路由 → 请求响应处理 → 内存后端模型」三层结构相关文件均位于 moto/bedrockagent/ 目录urls.py定义路由。服务端点匹配正则https?://bedrock-agent\.(.)\.amazonaws\.com所有路径{0}/.*$都交给AgentsforBedrockResponse.dispatch处理responses.pyAgentsforBedrockResponse类负责解析 HTTP 请求体JSON、提取参数、调用后端方法并序列化 JSON 响应models.pyAgentsforBedrockBackend后端类与Agent、KnowledgeBase两个资源模型数据全部保存在进程内存中exceptions.py定义ResourceNotFoundException、ConflictException、ValidationException三个客户端异常init.py导出bedrockagent_backends后端注册表。后端实例通过bedrockagent_backends[self.current_account][self.region]见 responses.py按账号与区域隔离这与 Moto 其他服务的多账号、多区域模型保持一致。在 models.py 中后端类内部维护三个核心容器self.agents: dict[str, Agent] {} self.knowledge_bases: dict[str, KnowledgeBase] {} self.tagger TaggingService()即 Agent 与 Knowledge Base 各存于一个以 ID 为键的字典标签统一交给moto.utilities.tagging_service.TaggingService管理。所有数据随 Moto 进程生命周期存续mock_aws装饰器作用域结束即被清空这正是本地模拟的核心语义。三、Agent创建、查询、列举与删除3.1 创建 Agentcreate_agentmodels.py 中的create_agent方法接收以下参数与 AWS SDK 请求字段一一对应参数对应请求字段说明agent_nameagentNameAgent 名称必填agent_resource_role_arnagentResourceRoleArn服务执行角色 ARN必填client_tokenclientToken幂等令牌可选instructioninstruction系统提示词/指令可选foundation_modelfoundationModel基础模型 ID可选descriptiondescription描述可选idle_session_ttl_in_secondsidleSessionTTLInSeconds空闲会话 TTL秒可选customer_encryption_key_arncustomerEncryptionKeyArnKMS 加密密钥 ARN可选tagstags创建时附加的标签可选prompt_override_configurationpromptOverrideConfiguration提示词覆盖配置含promptConfigurations、overrideLambda等可选创建时后端会生成 Agent 对象并注册进self.agents字典。关于生成的资源属性从 models.py 的Agent构造函数可以看到agent_id由agent_name uuid4()[:8]拼接生成8 位随机后缀agent_arn按arn:{partition}:bedrock:{region}:{account_id}:agent/{agent_id}构造其中 partition 由get_partition(region_name)推导如awsagent_status初始化为PREPAREDagent_version初始化为1.0failure_reasons初始为空列表recommended_actions为[action]created_at、updated_at、prepared_at均取当前unix_time()。注意由于 Agent 状态被固定为PREPAREDMoto 中创建出的 Agent 直接处于已就绪状态无需调用prepare_agent该 API 本身也未实现。3.2 查询单个 Agentget_agentget_agent(agent_id)在 models.py 中直接按字典查找若 ID 不存在则抛出ResourceNotFoundException(Agent {agent_id} not found)。响应层 responses.py 从 URL 路径倒数第二段解析agentId返回完整to_dict()序列化结果。3.3 列举 Agentlist_agentslist_agents方法models.py返回所有 Agent 的完整对象列表由paginate装饰器配合PAGINATION_MODEL实现分页详见下文第六节。响应层将其序列化为agentSummaries每个元素为dict_summary()摘要字段agentId、agentName、agentStatus、description、updatedAt、latestAgentVersion并附带nextToken。3.4 删除 Agentdelete_agentdelete_agent方法models.py是异常分支最丰富的实现若 Agent 不存在 →ResourceNotFoundException若 Agent 存在且skip_resource_in_use_checkTrue或Agent 状态为PREPARED→ 将状态置为DELETING从字典中移除返回(agent_id, DELETING)若 Agent 存在但状态既非PREPARED又未跳过检查 → 抛出ConflictException(Agent {agent_id} is in use)。由于 Moto 中新建 Agent 状态恒为PREPARED实际测试里不带skipResourceInUseCheck也可删除成功测试 tests/test_bedrockagent/test_bedrockagent.py 同时演示了两种写法并断言返回agentStatus DELETING。四、Knowledge Base创建、查询、列举与删除4.1 创建 Knowledge Basecreate_knowledge_basecreate_knowledge_basemodels.py的参数包括name、role_arn、knowledge_base_configuration、storage_configuration、client_token、description、tags。其中两个配置字典受到严格的枚举校验见 models.py 的KnowledgeBase构造函数knowledge_base_configuration[type]必须是VECTOR否则抛出ValidationException错误信息明确要求Member must contain type as VECTORstorage_configuration[type]必须是以下四种之一OPENSEARCH_SERVERLESS、PINECONE、REDIS_ENTERPRISE_CLOUD、RDS否则同样抛出ValidationException。测试 tests/test_bedrockagent/test_bedrockagent.py 分别用拼写错误的vECTOR和oPENSEARCH_SERVERLES验证了这两条校验路径都会返回ValidationException——校验是大小写敏感的实际使用时务必使用与 AWS 文档一致的全大写枚举值。创建后生成的知识库属性models.pyknowledge_base_id name uuid4()[:8]knowledge_base_arn arn:{partition}:bedrock:{region}:{account}:knowledge-base/{knowledge_base_id}status初始为Activecreated_at、updated_at为当前时间failure_reasons为空。注意Moto 只做配置校验与内存登记并不会真正连接 OpenSearch Serverless 或 RDS 等存储后端也不会执行真实的向量化与检索这符合模拟库的定位。4.2 查询、列举与删除get_knowledge_basemodels.py按 ID 查询不存在时抛ResourceNotFoundException响应层从 URL 路径最后一段解析 ID返回完整to_dict()list_knowledge_basesmodels.py同样支持分页响应序列化为knowledgeBaseSummaries摘要字段为knowledgeBaseId、name、description、status、updatedAt附带nextTokendelete_knowledge_basemodels.py先置状态DELETING再移除返回(knowledge_base_id, DELETING)ID 不存在时抛ResourceNotFoundException。五、标签管理tag_resource、untag_resource、list_tags_for_resource三个标签 API 都通过统一的_list_arns()辅助方法models.py校验资源 ARN 是否存在——它汇总了当前所有 Agent 与 Knowledge Base 的 ARN。对不存在的 ARN 调用任一标签 API 都会抛出ResourceNotFoundException测试 tests/test_bedrockagent/test_bedrockagent.py 覆盖了三种资源不存在场景。具体行为tag_resource(resource_arn, tags)models.py将dict[str, str]通过TaggingService.convert_dict_to_tags_input转为内部标签结构后合并进 TaggingServiceuntag_resource(resource_arn, tag_keys)models.py按标签键名移除list_tags_for_resource(resource_arn)models.py返回dict[str, str]形式的标签集合。响应层中tag/untag 的resourceArn从路径/tags/之后的部分解析并做 URL 解码unquote见 responses.py。测试还验证了多标签场景给 Agent 打Key1/Key2两个标签后 untagKey1list_tags_for_resource仅剩Key2证明增删改查闭环可用。六、分页行为AgentsforBedrockBackend.PAGINATION_MODELmodels.py为list_agents与list_knowledge_bases各配置了一份分页模型{ input_token: next_token, limit_key: max_results, limit_default: 100, unique_attribute: agent_id, # list_knowledge_bases 为 knowledge_base_id }这意味着请求参数maxResults映射到内部max_results默认每页上限为 100请求参数nextToken映射到内部next_token用于翻页分页游标基于unique_attributeAgent 用agent_id、知识库用knowledge_base_id去重排序保证翻页不重不漏。响应层 responses.py 会把字符串化的maxResults转成int再传给后端并在 JSON 响应中附带nextToken。测试 tests/test_bedrockagent/test_bedrockagent.py 验证了maxResults1时只返回一条且按创建顺序取第一条maxResults4时返回全部两条——分页行为在 Moto 中是确定性的按字典插入顺序这使断言非常稳定。七、异常体系exceptions.py 定义了三个异常全部继承自moto.core.exceptions.JsonRESTErrorHTTP 状态码均为 400异常类触发场景ResourceNotFoundException查询/删除/打标签不存在的 Agent 或 Knowledge BaseConflictException删除处于使用中的 Agent非PREPARED且未跳过检查ValidationExceptionKnowledge Base 配置类型不合法非VECTOR或非四种存储类型之一使用 boto3 客户端调用时这些异常会以botocore.exceptions.ClientError的形式抛给调用方错误码分别是ResourceNotFoundException、ConflictException、ValidationException测试中通过e.value.response[Error][Code]断言。八、完整可运行示例下面是一个覆盖全部已实现 API 的集成示例可直接复制运行需已安装moto与boto3import boto3 from moto import mock_aws from botocore.exceptions import ClientError mock_aws def test_bedrock_agent_crud(): client boto3.client(bedrock-agent, region_nameus-east-1) # 1. 创建 Agent agent client.create_agent( agentNameresearch-agent, agentResourceRoleArnarn:aws:iam::123456789012:role/bedrock-agent-role, instructionYou are a research assistant., foundationModelanthropic.claude-3-sonnet-20240229-v1:0, descriptionResearch assistant agent, idleSessionTTLInSeconds600, tags{Team: AI}, )[agent] agent_id agent[agentId] print(agentId:, agent_id) # 形如 research-agent 8 位随机后缀 print(agentStatus:, agent[agentStatus]) # PREPARED print(agentArn:, agent[agentArn]) # 2. 查询单个 Agent fetched client.get_agent(agentIdagent_id)[agent] assert fetched[agentName] research-agent # 3. 列举 Agent分页 summaries client.list_agents(maxResults1) assert len(summaries[agentSummaries]) 1 # 4. 标签管理 client.tag_resource(resourceArnagent[agentArn], tags{Env: dev}) tags client.list_tags_for_resource(resourceArnagent[agentArn])[tags] assert tags[Env] dev client.untag_resource(resourceArnagent[agentArn], tagKeys[Env]) # 5. 创建 Knowledge BaseVECTOR OpenSearch Serverless kb client.create_knowledge_base( nameproduct-knowledge, descriptionProduct docs KB, roleArnarn:aws:iam::123456789012:role/bedrock-kb-role, knowledgeBaseConfiguration{type: VECTOR}, storageConfiguration{type: OPENSEARCH_SERVERLESS}, tags{Team: AI}, )[knowledgeBase] kb_id kb[knowledgeBaseId] print(kbId:, kb_id) # 形如 product-knowledge 8 位随机后缀 print(kbStatus:, kb[status]) # Active # 6. 查询 / 列举 Knowledge Base assert client.get_knowledge_base(knowledgeBaseIdkb_id)[knowledgeBase][name] product-knowledge assert len(client.list_knowledge_bases()[knowledgeBaseSummaries]) 1 # 7. 删除 assert client.delete_agent(agentIdagent_id, skipResourceInUseCheckTrue)[agentStatus] DELETING assert client.delete_knowledge_base(knowledgeBaseIdkb_id)[status] DELETING # 8. 异常路径 try: client.get_agent(agentIdnonexistent) except ClientError as e: assert e.response[Error][Code] ResourceNotFoundException try: client.create_knowledge_base( namebad-kb, roleArnrole, knowledgeBaseConfiguration{type: TEXT}, storageConfiguration{type: OPENSEARCH_SERVERLESS}, ) except ClientError as e: assert e.response[Error][Code] ValidationException print(all assertions passed)要点说明mock_aws装饰器来自moto顶层导出也可用from moto import mock_aws自动拦截该作用域内所有 AWS 调用无需真实凭证所有资源均保存在 Moto 进程内存中测试结束后自动释放示例中的 ARN、模型 ID 等仅作占位Moto 不做真实性校验因此可以放心使用任意字符串若使用 Standalone Server 模式moto_server -p 5000后通过endpoint_url指向本地服务同一套 boto3 调用逻辑同样适用只需在创建客户端时传入endpoint_urlhttp://127.0.0.1:5000。九、测试依据与进一步阅读仓库为 bedrock-agent 提供了完整的单元测试见 tests/test_bedrockagent/test_bedrockagent.py覆盖了Agent/Knowledge Base 的创建、查询含不存在时的ResourceNotFoundException、列举含maxResults分页边界、删除含返回状态DELETING标签的增删查与资源不存在异常Knowledge Base 配置校验ValidationException。如需深入了解实现建议按以下顺序阅读服务文档docs/docs/services/bedrock-agent.rst本文依据的 API 支持清单后端模型与业务逻辑moto/bedrockagent/models.py请求响应层moto/bedrockagent/responses.py路由与异常moto/bedrockagent/urls.py、moto/bedrockagent/exceptions.py通用基础设施分页装饰器位于 moto/utilities/paginator.py标签服务位于 moto/utilities/tagging_service.py。结合 README.md 与 docs/docs/getting_started.rst 可以了解 Moto 的整体安装与使用方式。当前仓库版本下若你的测试涉及 Action Group、数据源导入ingestion job或 Flow 编排等未实现 API建议在测试设计阶段即避开或等待后续版本支持。赞分享Mock测试【免费下载链接】motoA library that allows you to easily mock out tests based on AWS infrastructure.项目地址https://gitcode.com/gh_mirrors/mo/moto点击查看免费下载相关推荐使用Python SDK操作Amazon Bedrock Agents的完整指南使用Python SDK操作Amazon Bedrock Agents的完整指南 概述 Amazon Bedrock Agents是一项强大的服务允许开发者在示例工程教程后端使用 AWS SDK for JavaScript (v3) 操作 Amazon Bedrock Agents完整实战指南使用 AWS SDK for JavaScript v3 操作 Amazon Bedrock Agents完整实战指南 本指南基于开源仓库 aws doc s示例工程教程后端Semantic Kernel 接入 Amazon Bedrock Agents完整集成指南Semantic Kernel 接入 Amazon Bedrock Agents完整集成指南 导读 Amazon Bedrock Agents 是 AWS 提人工智能大模型AI AgentAgent 框架多智能体RAG上一篇lazy_importer安全最佳实践构建企业级反逆向工程保护系统下一篇如何快速上手FaunaDB JavaScript驱动完整指南与最佳实践创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考