在 AWS 上构建企业级 RAG：结合托管知识库、智能解析与智能体检索
- 作者
- 姓名
- Nino
- 职业
- Senior Tech Editor
企业级人工智能助手需要的远不止一个大语言模型。它们需要访问正确的文档、可靠的检索机制以及用户能够验证的答案。当开发复杂的生产级应用程序时，开发者往往会遇到基础设施瓶颈，从而拖慢上市时间。为了应对高并发和多模型切换的生产工作负载，通过 n1n.ai 接入稳定高效的 API 聚合服务可以确保系统具备极低的延迟与高可用性。
试想向内部助手提出这样一个问题：哪个团队拥有我们的机器学习平台，其年度云预算是多少，我们的采购政策是否允许年度预付？这个答案可能分散在团队所有权文档、预算文件和费用政策中。基础的检索管道可能只会找到其中一个，从而导致答案不完整。
Amazon Bedrock 托管知识库通过托管的数据摄取与检索、原生数据连接器、智能解析以及智能体检索，帮助简化企业级检索增强生成（RAG）的开发。在本教程中，我们将探讨这些核心功能，并构建一个从多个企业文档中检索证据的 Python 演示程序。在构建大规模 AI 架构时，许多团队也会将 n1n.ai 作为统一的 API 网关来优化推理成本。
企业级 RAG 架构解析
RAG 将文档检索与生成式 AI 模型结合起来。应用程序不再仅仅依赖训练期间学到的信息，而是从配置的来源检索相关信息并利用其生成答案。一个典型的 RAG 管道包括：
- 数据源连接
- 文档解析与分块
- 嵌入与索引
- 检索与重排
- 答案生成
- 来源引用与评估
独立管理每个组件会增加运营复杂度。Amazon Bedrock 托管知识库提供了一种托管的方法来进行知识摄取和检索，帮助开发者专注于应用程序本身，而不必从头构建每个 RAG 组件。
原生数据连接器与智能解析
企业知识通常分布在不同的平台中。托管知识库支持多种原生数据连接器，包括 Amazon S3、Microsoft SharePoint、Atlassian Confluence、Google Drive、Microsoft OneDrive 以及网页爬虫。这些连接器可以减少将企业内容引入知识库所需的自定义摄取代码量。
并非所有企业数据都是纯文本。PDF、报告、演示文稿和技术文档可能包含表格、图表、截图或复杂的排版。智能解析有助于处理这些支持的内容，以便在检索过程中更有效地使用它们。
实现智能体检索
传统检索通常只搜索与查询直接相关的内容。这对于简单直接的问题非常有效，但更复杂的问题可能需要多次搜索。智能体检索使用基础模型在配置的知识库检索器之间规划和精炼检索过程。它可以将复杂问题分解为子查询，迭代检索信息，并评估是否有足够的证据可用。
项目环境搭建
我们将使用存储在 S3 存储桶中的三个虚拟文档来创建一个企业政策助手。
entrise-rag-demo/
├── documents/
│ ├── team-ownership.md
│ ├── cloud-budget.md
│ └── expense-policy.md
└── query_enterprise_kb.py
创建 documents/team-ownership.md 文件：
# ML Platform Team Ownership
The ML Platform team owns the internal machine learning platform and its shared deployment infrastructure.
创建 documents/cloud-budget.md 文件：
# ML Platform Cloud Budget
The ML Platform team has an illustrative annual cloud infrastructure budget of $120,000.
创建 documents/expense-policy.md 文件：
# Cloud Expense and Prepayment Policy
Annual cloud prepayment may be considered when the purchase meets procurement requirements.
使用 Python 和 Boto3 实现检索
首先安装适用于 Python 的 AWS SDK：
python -m pip install --upgrade boto3
创建 query_enterprise_kb.py 文件来处理智能体检索流：
import os
import boto3
REGION = os.getenv("AWS_REGION", "us-east-1")
KNOWLEDGE_BASE_ID = os.environ["KNOWLEDGE_BASE_ID"]
client = boto3.client(
"bedrock-agent-runtime",
region_name=REGION
)
question = (
"Which team owns the ML platform, what is its annual "
"cloud budget, and does the expense policy allow annual "
"prepayment? Explain the conditions and identify what "
"must happen before a commitment is made."
)
response = client.agentic_retrieve_stream(
messages=[
{
"role": "user",
"content": {
"text": question
}
}
],
retrievers=[
{
"description": "Enterprise policies and platform documents",
"configuration": {
"knowledgeBase": {
"knowledgeBaseId": KNOWLEDGE_BASE_ID
}
}
}
],
agenticRetrieveConfiguration={
"foundationModelType": "MANAGED",
"rerankingModelType": "MANAGED"
},
generateResponse=True
)
for event in response["stream"]:
if "responseEvent" in event:
print(
event["responseEvent"].get("text", ""),
end="",
flush=True
)
elif "result" in event:
result = event["result"]
print("\n\n--- Retrieved Sources ---")
for index, item in enumerate(result.get("results", []), start=1):
content = item.get("content", {})
print(f"\nSource {index}: {content.get('text', '')}")
生产环境最佳实践
在将企业级 RAG 应用程序部署到生产环境之前，请考虑以下几个关键支柱：
- 访问控制：确保用户只能检索其有权访问的文档。
- 可观测性与路由管理：将 AWS 日志与诸如 n1n.ai 之类的智能聚合网关相结合，以安全地监控执行指标、延迟和令牌消耗。
- 提示词注入防御：将检索到的内容视为不受信任的输入，防止文档覆盖应用程序的安全规则。
获取免费 API 密钥请访问 n1n.ai