Amazon Kendra 从 2026 年 7 月 30 日起，将不再向新客户开放。如果您想使用该服务，请在 7 月 30 日之前注册。有关类似的功能 Amazon Kendra，请浏览 Amazon Bedrock 知识库。了解更多。

本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。

Amazon Kendra   可用性变更

概述

经过仔细考虑，我们决定 Amazon Kendra 进入维护模式，自 2026 年 6 月 30 日起生效。截至该日期，该服务将不会有新的功能或功能开发，并且从2026年7月30日起，该服务将停止接受新客户。

在维护模式下，该服务仍 AWS 将得到全面支持，并将继续为现有客户提供错误修复和安全更新，但是将不再考虑新功能请求。

我们建议客户迁移他们的 Kendra 应用程序，并在亚马逊 Bedrock 托管知识库 (BMKB) 上实现任何新的搜索应用程序，以获得与 Kendra 相似的功能，并为生成人工智能和代理人工智能用例提供更高级的功能。Bedrock Managed Knowledge Base 是一个完全托管的 RAG 解决方案，具有内置连接器、智能解析、带混合搜索的托管矢量存储，以及生成响应（使用检索和生成 API）和跨多个知识库进行多步推理的能力（使用代理检索 API）。BMKB 还使您能够调整分块策略和嵌入模型以针对您的特定应用程序进行优化，还可以选择用于生成响应的基础模型。这种全新的功能和灵活性具有可预测的成本，具体取决于所摄取的知识库的大小、执行的查询数量和LLM的使用情况。

迁移指南

对于大多数企业搜索和 RAG 工作负载，只要仔细规划，就可以实现从 Amazon Kendra Amazon Bedrock 托管知识库 (BMKB) 迁移。并非所有 Kendra 功能都可直接在 Bedrock 托管知识库中获得，但许多功能可以通过变通方法实现。本指南为现有的 Kendra 客户提供了全面的分步迁移路径，帮助他们将其应用程序过渡到 BMKB，包括架构映射、API 转换、代码示例、功能差距分析和推荐的变通方法。

Amazon Bedrock 托管知识库功能

BMKB 对整个 RAG 管道进行端到端管理。它支持嵌入模型，包括亚马逊 Titan 文本嵌入 V2、Cohere Embed Embed English v3、Cohere Embed 多语言 v4 和 Amazon Nova 多模式嵌入模型，所有模型都固定在 1024 个维度，使用 float32 向量。托管矢量存储完全由 Bedrock 运营，无需配置或管理 OpenSearch Aurora 或其他矢量数据库。分块策略包括默认（固定大小，大约 300 个标记）、 Fixed-size （可配置的 maxTokens 和 OverlPercence）、分层（具有级别配置的父子策略）和 “不分块”；托管知识库不支持语义分块。

BMKB 目前支持七个数据源连接器：亚马逊 S3、Confluence、微 SharePoint软、Web Crawler、谷歌云端硬盘、微 OneDrive软和一个自定义连接器。该服务始终执行混合搜索（关键字加语义），并且不提供仅限语义的搜索模式。

迁移步骤

Amazon Bedrock 托管知识库设置

步骤 1：配置 IAM 角色

创建一个 IAM 角色以授予 Bedrock 访问您的数据源和调用嵌入模型的权限。信任策略必须允许 bedrock.amazonaws.com 担任该角色，并且权限策略必须包括对您的 S3 存储桶和所选嵌入模型的访问权限。

IAM 配置（Python 代码）：

import boto3
import json

iam = boto3.client('iam')

trust_policy = {
    "Version": "2012-10-17",
    "Statement": [{
        "Effect": "Allow",
        "Principal": {"Service": "bedrock.amazonaws.com"},
        "Action": "sts:AssumeRole"
    }]
}

iam.create_role(
    RoleName="BedrockKBRole",
    AssumeRolePolicyDocument=json.dumps(trust_policy),
    Description="Role for Bedrock Managed Knowledge Base"
)

permissions_policy = {
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": ["s3:GetObject", "s3:ListBucket"],
            "Resource": [
                "arn:aws:s3:::your-bucket-name",
                "arn:aws:s3:::your-bucket-name/*"
            ]
        },
        {
            "Effect": "Allow",
            "Action": ["bedrock:InvokeModel"],
            "Resource": ["arn:aws:bedrock:*::foundation-model/amazon.titan-embed-text-v2:0"]
        }
    ]
}

iam.put_role_policy(
    RoleName="BedrockKBRole",
    PolicyName="BedrockKBPermissions",
    PolicyDocument=json.dumps(permissions_policy)
)
                

步骤 2：创建托管知识库

使用类型为 MANAGED 的 CreateKnowledgeBase API 来创建知识库：

import boto3

bedrock_agent = boto3.client("bedrock-agent", region_name="us-east-1")

response = bedrock_agent.create_knowledge_base(
    name="my-managed-kb",
    description="Migrated from Kendra index",
    roleArn="arn:aws:iam::123456789012:role/BedrockKBRole",
    knowledgeBaseConfiguration={
        "type": "MANAGED",
        "managedKnowledgeBaseConfiguration": {
            "embeddingModelArn": "arn:aws:bedrock:us-east-1::foundation-model/amazon.titan-embed-text-v2:0",
            "embeddingModelConfiguration": {
                "bedrockEmbeddingModelConfiguration": {
                    "embeddingDataType": "FLOAT32"
                }
            }
        }
    }
)

kb_id = response["knowledgeBase"]["knowledgeBaseId"]
print(f"Created knowledge base: {kb_id}")
                

步骤 3：配置数据源

使用托管连接器配置创建 S3 数据源：

response = bedrock_agent.create_data_source(
    knowledgeBaseId=kb_id,
    name="my-s3-data-source",
    description="Product documentation from S3",
    dataSourceConfiguration={
        "type": "MANAGED_KNOWLEDGE_BASE_CONNECTOR",
        "managedKnowledgeBaseConnectorConfiguration": {
            "connectorParameters": {
                "type": "S3",
                "version": "1",
                "connectionConfiguration": {
                    "bucketName": "your-bucket-name",
                    "bucketOwnerAccountId": "123456789012"
                },
                "filterConfiguration": {
                    "inclusionPrefixes": ["documents/"]
                }
            }
        }
    },
    vectorIngestionConfiguration={
        "parsingConfiguration": {
            "parsingStrategy": "SMART_PARSING"
        }
    }
)

data_source_id = response["dataSource"]["dataSourceId"]
print(f"Created data source: {data_source_id}")
                

步骤 4：启动并监控摄取

触发文档摄取并轮询完成情况：

import time

ingestion_response = bedrock_agent.start_ingestion_job(
    knowledgeBaseId=kb_id,
    dataSourceId=data_source_id,
    description="Initial ingestion"
)

ingestion_job_id = ingestion_response["ingestionJob"]["ingestionJobId"]
print(f"Started ingestion job: {ingestion_job_id}")

while True:
    job_response = bedrock_agent.get_ingestion_job(
        knowledgeBaseId=kb_id,
        dataSourceId=data_source_id,
        ingestionJobId=ingestion_job_id
    )
    job = job_response["ingestionJob"]
    status = job["status"]
    stats = job.get("statistics", {})

    print(f"Status: {status} | Scanned: {stats.get('numberOfDocumentsScanned', 0)} | "
          f"Indexed: {stats.get('numberOfNewDocumentsIndexed', 0)} | "
          f"Failed: {stats.get('numberOfDocumentsFailed', 0)}")

    if status == "COMPLETE":
        print("Ingestion complete!")
        break
    elif status in ("FAILED", "STOPPED"):
        print(f"Ingestion {status}: {job.get('failureReasons', [])}")
        break

    time.sleep(30)
                

API 迁移映射和代码示例

API 操作映射

迁移检索 API

之前（Kendra）：

kendra_client = boto3.client("kendra")

response = kendra_client.retrieve(
    IndexId="your-kendra-index-id",
    QueryText="How do I configure VPC endpoints?",
    AttributeFilter={
        "EqualsTo": {
            "Key": "_category",
            "Value": {"StringValue": "networking"}
        }
    },
    PageSize=10
)

for item in response["ResultItems"]:
    print(item["DocumentTitle"])
    print(item["Content"])
            

之后（BMKB）：

bedrock_runtime = boto3.client("bedrock-agent-runtime", region_name="us-east-1")

response = bedrock_runtime.retrieve(
    knowledgeBaseId="your-kb-id",
    retrievalQuery={
        "text": "How do I configure VPC endpoints?"
    },
    retrievalConfiguration={
        "managedSearchConfiguration": {
            "numberOfResults": 10,
            "filter": {
                "equals": {
                    "key": "category",
                    "value": "networking"
                }
            }
        }
    }
)

for result in response["retrievalResults"]:
    print(f"Score: {result['score']}")
    print(f"Content: {result['content']['text']}")
    print(f"Source: {result['location']['s3Location']['uri']}")
            

主要区别在于：查询文本从顶级参数移动到嵌套检索Query.text 字段； AttributeFilter 变为托管中的筛选器SearchConfiguration；结果包括相关性分数字段。

用 RetrieveAndGenerate 于 RAG

BMKB 提供了原生 RAG 功能，无需在检索后单独调用 LLM：

response = bedrock_runtime.retrieve_and_generate(
    input={
        "text": "Explain how to configure VPC endpoints for S3 access"
    },
    retrieveAndGenerateConfiguration={
        "type": "KNOWLEDGE_BASE",
        "knowledgeBaseConfiguration": {
            "knowledgeBaseId": "your-kb-id",
            "modelArn": "arn:aws:bedrock:us-east-1::foundation-model/anthropic.claude-3-sonnet-20240229-v1:0",
            "retrievalConfiguration": {
                "managedSearchConfiguration": {
                    "numberOfResults": 5
                }
            }
        }
    }
)

# Generated answer with citations
print(response["output"]["text"])

# Source citations
for citation in response.get("citations", []):
    for ref in citation.get("retrievedReferences", []):
        print(f"Source: {ref['location']['s3Location']['uri']}")
            

此 API 返回生成的自然语言响应以及指向源文档的引文，提供了 Kendra 原生不提供的内置 RAG。

元数据筛选器语法翻译

功能差距和解决方法

并非所有 Kendra 功能都在 BMKB 中可用。查询建议、分面搜索、自定义同义词、拼写检查、增量学习和文档丰富是需要在 BMKB 中变通解决方法的功能。本节讨论这些解决方法。

数据源迁移策略

连接器覆盖差距

Kendra 支持 32 个本机连接器，而 BMKB 支持 7 个。对于 BMKB 不直接支持的数据源，推荐的方法是将内容导出到 Amazon S3，然后在 BMKB 中配置 S3 数据源。

不支持的连接器的迁移模式：创建一个自动管道（使用 Lamb AWS da、Step Fun ctions 或 A EventBridge mazon Scheduler），该管道通过其 API 定期从源系统提取内容，将包含相应元数据 JSON sidecar 文件的文档写入 S3 存储桶，并触发 BMKB 摄取任务。这复制了 Kendra 连接器的定期同步行为。

元数据迁移

Kendra 文档属性必须转换为 BMKB 元数据格式。在 Kendra 中，属性是在索引级别定义的，并在摄取期间附加到文档中。在 BMKB 中，元数据是通过与源文档一起存储在 S3 中的.metadata.json sidecar 文件定义的，每个文件的最大大小为 10 KB。每个属性都必须键入为字符串、数字或布尔值。

分块策略选择

从 Kendra（内部处理分块）迁移时，必须明确选择 BMKB 的分块策略。对于大多数迁移方案，包含 200 个代币和 30% 重叠的 Fixed-size策略提供了一个很好的起点。如果您的文档具有清晰的分层结构（章节、小节、小节），请考虑分层分块，以改善对大背景和特定细节的检索。

测试和验证

要评估性能，请并行运行 Kendra 和 BMKB。向两项服务发送相同的查询，并使用以下维度比较结果：相关性质量（由NDCG或MRR与黄金测试集进行衡量）、延迟（p50、p95、p99响应时间）、吞吐量（负载下每秒查询次数）和完整性（检索到的预期文档的百分比）。

创建评估检索质量的测试工具：

def compare_retrieval(query, kendra_index_id, bmkb_kb_id):
    # Query Kendra
    kendra_results = kendra_client.retrieve(
        IndexId=kendra_index_id,
        QueryText=query,
        PageSize=10
    )

    # Query BMKB
    bmkb_results = bedrock_runtime.retrieve(
        knowledgeBaseId=bmkb_kb_id,
        retrievalQuery={"text": query},
        retrievalConfiguration={
            "managedSearchConfiguration": {"numberOfResults": 10}
        }
    )

    # Compare overlap in top-10 results
    kendra_docs = {r["DocumentId"] for r in kendra_results["ResultItems"]}
    bmkb_docs = {r["location"]["s3Location"]["uri"] for r in bmkb_results["retrievalResults"]}

    overlap = len(kendra_docs.intersection(bmkb_docs))
    print(f"Query: {query}")
    print(f"Result overlap: {overlap}/10 documents in common")
    return overlap
        

在将生产流量切换到 BMKB 之前，请验证以下几点：所有数据源均已采集并处于最新状态，没有失败的文档；元数据筛选器为所有应用程序筛选模式生成预期结果；访问控制变通办法正确限制未经授权的访问；相关性基准达到或超过 Kendra 基准质量；应用程序错误处理正确处理 BMKB 响应格式；为 BMKB API 错误和延迟配置了监控和警报。

Summary

从 Amazon Kendra Bedrock Managed Knowledge Base 迁移需要做两项主要工作：将数据源重新采集到 BMKB 中，以及重写应用程序代码以使用 BMKB API。虽然 BMKB 引入了包括 RetrieveAndGenerate 代理检索在内的强大 RAG-native 功能，但使用企业搜索功能（例如分面、查询建议、自定义同义词和增量学习）的客户将需要按照本指南中的说明实施变通方案。

如有任何其他问题，请联系AWS Support。