使用 CDK 构建托管式亚马逊Bedrock知识库的智能体检索系统

企业级的检索增强生成技术早已超越了单纯的向量搜索和传统的相似度匹配。现代生产环境中的应用程序要求具备复杂的多源路由、严苛的合规保障、全面的可观测性以及零触碰的底层基础设施交付。当在亚马逊云科技上架构复杂的生成式人工智能工作负载时，如果缺乏合适的自动化手段，同时管理多个数据源与多轮推理智能体会迅速演变成一场基础设施的噩梦。本次技术深度剖析将探讨如何利用 AWS Cloud Development Kit (CDK) 编排托管式 Amazon Bedrock 知识库，从而构建企业级的智能体检索流水线。通过像 n1n.ai 这样统一的平台来集中管理访问模式，开发者可以进一步优化下游模型推理的成本与延迟，同时专注于业务逻辑和路由语义。

架构概览：多源智能体路由设计

传统的单索引检索增强生成系统在面对包含离散代码库、内部工程维基以及财务资产负债表等不同企业领域时往往会失效。当被强制处理需要不同领域专业知识的查询时，单体向量数据库经常会出现幻觉或不恰当地混合上下文。

为了解决这一痛点，我们的目标架构实现了一个配备语义路由器的自主智能体。该智能体会评估传入的用户查询，检查元数据配置文件，并将检索请求分发到两个专门的 Amazon Bedrock 知识库之一。路由层位于 AgentCore Gateway 的下游，该网关充当一个安全且具备速率限制的边界，在调用通过 n1n.ai 提供的高吞吐量端点（如 Claude 3.5 Sonnet 或 DeepSeek-V3）之前，严格执行 IAM 策略和有效负载净化。

[用户查询] --> [AgentCore Gateway] --> [自主路由器智能体]
                                                |
                      +-------------------------+-------------------------+
                      |                                                   |
                      v                                                   v
        [知识库 A：工程技术]                        [知识库 B：法律与财务]

基础设施即代码：使用 AWS CDK 构建核心资源

部署有状态的无服务器人工智能流水线需要绝对的可复现性。AWS CDK 允许我们在单个经过版本控制的 Python 应用程序中定义 Amazon Bedrock 知识库、关联的 OpenSearch Serverless 集合、S3 数据桶、IAM 执行角色以及 Lambda 编排函数。

以下是一个生产就绪的 CDK 栈代码片段，它配置了一个加密的 S3 数据桶、一个 OpenSearch Serverless 向量集合，并将其通过优化分块策略直接链接到 Amazon Bedrock 知识库资源。

from aws_cdk import (
    Duration,
    RemovalPolicy,
    Stack,
    aws_s3 as s3,
    aws_bedrock as bedrock,
    aws_iam as iam,
    aws_opensearchserverless as oss
)
from constructs import Construct

class BedrockAgenticRagStack(Stack):
    def __init__(self, scope: Construct, id: str, **kwargs) -> None:
        super().__init__(scope, id, **kwargs)

        # 1. S3 数据源存储桶
        source_bucket = s3.Bucket(
            self, "KnowledgeBaseBucket",
            encryption=s3.BucketEncryption.S3_MANAGED,
            removal_policy=RemovalPolicy.DESTROY,
            auto_delete_objects=True
        )

        # 2. Bedrock 知识库所需的 IAM 角色
        kb_role = iam.Role(
            self, "BedrockKBRole",
            assumed_by=iam.ServicePrincipal("bedrock.amazonaws.com")
        )
        source_bucket.grant_read(kb_role)

        # 3. OpenSearch Serverless 集合向量存储
        vector_collection = oss.CfnCollection(
            self, "VectorCollection",
            name="enterprise-kb-store",
            type="VECTORSEARCH"
        )

        # 4. Amazon Bedrock 知识库资源
        knowledge_base = bedrock.CfnKnowledgeBase(
            self, "EnterpriseKnowledgeBase",
            name="EngineeringAndComplianceKB",
            role_arn=kb_role.role_arn,
            knowledge_base_configuration=bedrock.CfnKnowledgeBase.KnowledgeBaseConfigurationProperty(
                type="VECTOR",
                vector_knowledge_base_configuration=bedrock.CfnKnowledgeBase.VectorKnowledgeBaseConfigurationProperty(
                    embedding_model_arn=f"arn:aws:bedrock:us-east-1::foundation-model/amazon.titan-embed-text-v2:0"
                )
            ),
            storage_configuration=bedrock.CfnKnowledgeBase.StorageConfigurationProperty(
                type="OPENSEARCH_SERVERLESS",
                opensearch_serverless_configuration=bedrock.CfnKnowledgeBase.OpenSearchServerlessConfigurationProperty(
                    collection_arn=vector_collection.attr_arn,
                    vector_index_name="bedrock-knowledge-base-index",
                    field_mapping=bedrock.CfnKnowledgeBase.OpenSearchServerlessFieldMappingProperty(
                        vector_field="vector",
                        text_field="text",
                        metadata_field="metadata"
                    )
                )
            )
        )
        knowledge_base.add_dependency(vector_collection)

实现智能体路由与网关集成

一旦基础架构实例化完成，接下来的关键阶段就是编写自主路由逻辑。当请求到达 AgentCore Gateway 时，它们会被运行 LangChain 或自定义编排循环的 AWS Lambda 函数拦截。该循环评估提示词是需要来自知识库 A 的技术文档，还是来自知识库 B 的合规数据。

为了保持最佳性能并在扩展至数千个并发企业用户时防止速率限制瓶颈，集成一个高效的 API 网关（例如 n1n.ai）可确保基础模型端点的高可用性和智能负载均衡。

以下是一个 Python 实现代码片段，展示了路由智能体如何与 Bedrock 的检索并生成 API 进行交互，并增强了降级处理能力：

import boto3
import json

bedrock_agent_runtime = boto3.client('bedrock-agent-runtime', region_name='us-east-1')

def evaluate_and_route_query(query_text: str, kb_id_engineering: str, kb_id_legal: str) -> dict:
    # 简单的基于关键词的分类器或大模型路由器
    target_kb = kb_id_legal if any(keyword in query_text.lower() for keyword in ["compliance", "contract", "policy"]) else kb_id_engineering
    
    try:
        response = bedrock_agent_runtime.retrieve_and_generate(
            input={
                'text': query_text
            },
            retrieveAndGenerateConfiguration={
                'type': 'KNOWLEDGE_BASE',
                'knowledgeBaseConfiguration': {
                    'knowledgeBaseId': target_kb,
                    'modelArn': 'arn:aws:bedrock:us-east-1::foundation-model/anthropic.claude-3-5-sonnet-20241022-v2:0'
                }
            }
        }
        
        return {
            "status": "success",
            "output": response.get('output', {}).get('text'),
            "citations": response.get('citations', [])
        }
    except Exception as e:
        print(f"执行智能体检索时发生错误: {str(e)}")
        return {
            "status": "error",
            "message": str(e)
        }

七层可观测性与评估体系

任何企业级人工智能应用如果缺少严苛的可观测性都是不完整的。部署我们的 CDK 架构纳入了七个不同的监控层，以确保系统健康、安全和输出准确性：

基础设施指标：跟踪 CPU、内存、Lambda 执行持续时间以及 OpenSearch Serverless 计算容量单位的 CloudWatch 指标。
网关延迟追踪：跨 AgentCore Gateway 和内部微服务的 AWS X-Ray 分布式追踪。
Token 消耗与成本审计：对提示词和补全 Token 进行精细化跟踪，并辅以成本优化洞察。
检索相关性评分：评估托管知识库返回文本片段的余弦相似度分数和上下文精确度。
模型幻觉检测：衡量输出内容相对于检索到的源引用是否忠实可靠的自动化检查。
护栏策略违规监控：监控 Amazon Bedrock 护栏是否存在有毒内容、隐私泄露和提示词注入尝试。
持续评估流水线：使用黄金测试数据集进行定期的评估运行，以随时间推移衡量检索准确率的退化情况。

总结与后续展望

利用托管式 Amazon Bedrock 知识库和 AWS CDK 构建企业级智能体检索流水线，省去了维护自定义向量索引和手动分块流水线的沉重负担。通过将多源路由、健壮的基础架构自动化以及像 n1n.ai 这样用于优化推理路由的企业级工具结合在一起，工程团队可以迅速部署安全、带引用且具有极高可扩展性的生成式人工智能应用程序。

Get a free API key at n1n.ai

参考来源：https://aws.amazon.com/blogs/machine-learning/build-agentic-retrieval-on-a-managed-amazon-bedrock-knowledge-base-with-cdk/