8.0
热点
AI SCORE
技术实践2026-10-05 23:53
LangChain+Bedrock知识库：代理式检索实战对比
AWS ML Blog#LangChain#RAG#AWS Bedrock
Editor brief · 编辑速览
对比单次检索与代理式检索在多跳问题上的效果差异，提供LangChain集成Bedrock知识库的具体实现与成本分析。
当用户向支持助手提问——一个用 LangChain 构建的 RAG 应用，要求在三个维度上对比两款产品——实际上他们同时提出了六个问题。相似性搜索使用单一查询向量来封装所有意图，然后检索器生成这些意图平均值的最佳近似。结果返回的答案很简洁，搜索执行没有报错，相关性得分看起来也很合理。然而，检索到的片段虽然主题相关，却只覆盖了问题实际询问内容的一小部分。
在本文中，我们展示了一个基于 Amazon Bedrock Managed Knowledge Base 和 LangChain 构建的 RAG 应用。我们用标准和智能体式检索两种方式处理同一个多部分问题，并读取 trace 事件来查看模型产生的计划。我们还会讨论两种检索路径的成本差异，以及何时选择更便宜的方案是正确的选择。
Amazon Bedrock Managed Knowledge Base 上提供了智能体式检索功能。与单一搜索不同，Amazon Bedrock Managed Knowledge Base 会规划检索过程。它将问题分解为子查询、执行查询、判断是否有足够的证据支撑，如果不够就再次搜索。langchain-aws 包同时暴露了智能体式和标准检索，因此你可以从 LangChain 应用中使用其中任一种。
Amazon Bedrock Managed Knowledge Base 是 Amazon Bedrock 中的全托管 RAG 能力。它将自管理的向量存储、嵌入模型和重排序模型从 RAG 架构中移除。你只需配置一个数据源，Amazon Bedrock Managed Knowledge Base 就会处理分块、嵌入、存储和检索。本演练使用 Amazon Simple Storage Service（Amazon S3）。
Amazon Bedrock Managed Knowledge Bases 提供了两个 API。我们会在本文中简要讨论它们的差异。Retrieve API 执行一次混合搜索并返回带分数的片段。AgenticRetrieveStream API 则运行一个规划循环，并将步骤以 trace 事件的形式流式返回给你。在 langchain-aws 包中，前者是一个标准的 LangChain 检索器，可以直接插入链中使用。后者是一个直接从知识库执行的函数检索。
下图展示了解决方案架构。应用程序使用 Retrieve API（标准单次）或 AgenticRetrieveStream API（多步规划循环）来查询 Amazon Bedrock Knowledge Bases。两条路径都从知识库返回文档片段，然后应用程序使用这些片段生成有依据的响应。
Figure 1: Solution architecture for querying Amazon Bedrock Knowledge Bases with the Retrieve and AgenticRetrieveStream APIs
以下各节将带你完成创建知识库、用两种检索方法查询知识库，以及读取智能体规划器产生的 trace 事件的过程。
要跟随操作，你需要：
安装以下包。Boto3 版本很重要：agentic_retrieve_stream 在 1.43.32 之前不存在。
langchain-aws>=1.6.3
langchain>=1.0
boto3>=1.43.32
涉及两个身份，明智的做法是有意识地分离它们。知识库假设一个服务角色来读取你的文档并调用嵌入模型。你的应用程序使用 AWS Security Token Service（AWS STS）调用者身份来查询。两者都不需要对方的权限。
如果你允许，Amazon Bedrock 会为你创建服务角色。要提供你自己的角色，需要赋予它一个信任策略，允许 Amazon Bedrock 担任该角色。使用 aws:SourceAccount 和 aws:SourceArn 进行范围限制，这样其他账户就不能将其用作混淆代理：
{
"Version": "2012-10-17",
"Statement": [{
"Effect": "Allow",
"Principal": {"Service": "bedrock.amazonaws.com"},
"Action": "sts:AssumeRole",
"Condition": {
"StringEquals": {"aws:SourceAccount": "111122223333"},
"ArnLike": {
"aws:SourceArn": "arn:aws:bedrock:us-east-1:111122223333:knowledge-base/*"
}
}
}]
}
服务角色还需要对你的存储桶拥有 s3:ListBucket 权限，以及对其内容拥有 s3:GetObject 权限，两者都要以 aws:ResourceAccount 为条件。在创建知识库后，将 knowledge-base/* 通配符范围缩小到特定的知识库 ID。
AWS STS 调用者身份需要另一组不同的权限。bedrock:AgenticRetrieveStream 和 bedrock:InvokeModelWithResponseStream 无法限定到特定知识库 ARN。bedrock:Retrieve 和 bedrock:GetDocumentContent 可以：
{
"Version": "2012-10-17",
"Statement": [
{
"Sid": "AgenticRetrievalAndPlannerModel",
"Effect": "Allow",
"Action": [
"bedrock:AgenticRetrieveStream",
"bedrock:InvokeModelWithResponseStream"
],
"Resource": "*"
},
{
"Sid": "RetrieveAndFullDocumentExpansion",
"Effect": "Allow",
"Action": ["bedrock:Retrieve", "bedrock:GetDocumentContent"],
"Resource": "arn:aws:bedrock:<region>:111122223333:knowledge-base/<knowledge-base-id>"
},
{
"Sid": "GenerateAnswersInTheChains",
"Effect": "Allow",
"Action": ["bedrock:InvokeModel", "bedrock:Converse", "bedrock:ConverseStream"],
"Resource": "*"
}
]
}
bedrock:GetDocumentContent 经常被忽略。当 FullDocumentExpansion 步骤判断某个段落缺少回答问题所需的上下文时，智能体式检索会调用它。只有 bedrock:Retrieve 的策略在规划器需要整篇文档时就会中途失败。
要创建和管理知识库本身，调用角色还需要在 * 上拥有 bedrock:CreateKnowledgeBase 权限，以及在 knowledge-base/* 上拥有 GetKnowledgeBase、UpdateKnowledgeBase、DeleteKnowledgeBase、StartIngestionJob、GetIngestionJob 和 ListIngestionJobs 操作的权限。如果你使用护栏，还需要添加 bedrock:GetGuardrail 和 bedrock:ApplyGuardrail。
运行本演练可能会产生知识库中的文档存储和摄取成本、检索调用费用，以及基础模型（FM）推理费用。
有关定价的更多信息，请参阅 Amazon Bedrock 定价中的 Knowledge Bases 部分。
完成实验后请删除相关资源。
使用 managedKnowledgeBaseConfiguration 创建知识库。将 embeddingModelType 设置为 MANAGED 会使用服务托管的嵌入模型。
import boto3
import os
REGION = os.environ["AWS_REGION"]
bedrock_agent = boto3.client("bedrock-agent", region_name=REGION)
response = bedrock_agent.create_knowledge_base(
name=KB_NAME,
roleArn=KB_ROLE_ARN,
knowledgeBaseConfiguration={
"type": "MANAGED",
"managedKnowledgeBaseConfiguration": {
"embeddingModelType": "MANAGED",
},
},
)
KB_ID = response["knowledgeBase"]["knowledgeBaseId"]
该请求中没有 storageConfiguration。对于自管理知识库，你需要传入一个描述向量存储的配置。Amazon Bedrock Managed Knowledge Base 不需要这个参数，这是 API 中表明 Amazon Bedrock 拥有存储层的最明确信号。
将 S3 存储桶作为数据源附加到知识库，然后启动一个摄取作业。摄取是异步的，因此需要轮询直到作业达到最终状态，而不是固定休眠一段时间然后期待它完成。
import time
SUCCESS_STATES = frozenset({"COMPLETE"})
FAILURE_STATES = frozenset({"FAILED", "STOPPED"})
```python
def wait_for_ingestion(kb_id, ds_id, job_id, timeout_s=1800):
"""轮询摄取任务，直到达到终止状态。"""
deadline = time.time() + timeout_s
while time.time() < deadline:
job = bedrock_agent.get_ingestion_job(
knowledgeBaseId=kb_id,
dataSourceId=ds_id,
ingestionJobId=job_id,
)["ingestionJob"]
status = job["status"]
if status in SUCCESS_STATES:
return job
if status in FAILURE_STATES:
reasons = job.get("failureReasons") or ["no reason reported"]
raise RuntimeError(f"Ingestion job {job_id} finished as {status}: " + "; ".join(reasons))
time.sleep(15)
raise TimeoutError(f"Ingestion job {job_id} did not finish in {timeout_s}s")
完整的数据源配置和错误处理在示例仓库中。
AmazonKnowledgeBasesRetriever 封装了 Retrieve API，行为与任何其他 LangChain 检索器无异。对于 Amazon Bedrock Managed Knowledge Bases，需传入 managedSearchConfiguration。这里是容易踩坑的地方：vectorSearchConfiguration 是早期路径，适用于自行管理向量库的知识库，也是大多数现有示例所展示的方式。
from langchain_aws.retrievers import AmazonKnowledgeBasesRetriever
SIMPLE_QUERY = "What is the restore time objective for the checkout service?"
retriever = AmazonKnowledgeBasesRetriever(
knowledge_base_id=KB_ID,
region_name=REGION,
retrieval_config={
"managedSearchConfiguration": {
"numberOfResults": 5,
}
},
)
docs = retriever.invoke(SIMPLE_QUERY)
每个结果以 LangChain Document 形式返回。相关性分数在 metadata["score"] 中，源文档自身的元数据位于 metadata["source_metadata"] 下（已重命名以避免冲突）。如果需要丢弃低置信度结果，请在检索器上设置 min_score_confidence，而不是事后过滤。
对于意图单一的问题，这是正确的工具。只需一次调用，延迟最低，且你对答案的生成方式保有完全控制权。生产环境中的助手收到的大多数查询都属于这种形态，为它们启动规划循环既浪费资金又浪费时间。
现在用同样的检索器处理一个包含多个部分的问题：
COMPLEX_QUERY = (
"Compare the checkout and inventory services across on-call escalation, backup and "
"restore targets, and deployment rollback procedure. Where do they differ?"
)
docs = retriever.invoke(COMPLEX_QUERY)
返回五个块，按混合分数对该问题的一个向量表示进行排序。
这个问题包含六个意图：三个维度上的两个服务。为每个意图对检索文本进行评分，可以直观衡量单个向量恢复了多少内容。
在五个结果的情况下，代表六个意图的一个向量会遗漏其中两个。十个结果时能覆盖全部六个，但有明显浪费：两个子意图被重复覆盖，一个块则完全没有覆盖。
检索器完成了它的工作。局限性是结构性的：一个向量无法代表六个意图，且流程中没有任何步骤去判断返回的证据是否足以回答问题。
Agentic 检索不是 LangChain 检索器，而是 Amazon Bedrock Managed Knowledge Bases 的一项功能。langchain-aws 包将其暴露为独立函数 agentic_retrieve，因为底层 API 以流式返回结果，不符合同步的 BaseRetriever 接口。AmazonKnowledgeBasesRetriever 上没有任何标志可以切换开启此功能。
from langchain_aws.retrievers.bedrock import agentic_retrieve
result = agentic_retrieve(
knowledge_base_id=KB_ID,
query=COMPLEX_QUERY,
region_name=REGION,
generate_response=True,
number_of_results=10,
)
print(result["generatedResponse"]["answer"])
设置 generate_response=True 时，服务会返回带引用的有依据答案以及检索到的块，因此无需另行调用模型即可获得答案。该函数仅适用于 Amazon Bedrock Managed Knowledge Base。
在内部，服务会规划、检索、评估证据是否充分，必要时进行迭代。辅助函数隐藏了所有这些细节，只返回最终块——这很方便，但意味着你看不到规划过程。
要观察模型如何分解问题，请直接在 bedrock-agent-runtime 客户端上调用 agentic_retrieve_stream。这是本演练中唯一需要绕过 langchain-aws 的地方，因为辅助函数会丢弃跟踪事件，且不暴露 maxAgentIteration 或自定义规划器模型。
runtime = boto3.client("bedrock-agent-runtime", region_name=REGION)
response = runtime.agentic_retrieve_stream(
messages=[{"role": "user", "content": {"text": COMPLEX_QUERY}}],
retrievers=[{
"configuration": {
"knowledgeBase": {
"knowledgeBaseId": KB_ID,
"retrievalOverrides": {"maxNumberOfResults": 10},
}
}
}],
agenticRetrieveConfiguration={
"foundationModelType": "MANAGED",
"rerankingModelType": "MANAGED",
# 5 是 API 默认值。低于 4 时规划器完全停止分解。
"maxAgentIteration": 5,
},
generateResponse=False,
)
for event in response["stream"]:
if "traceEvent" in event:
attrs = event["traceEvent"]["attributes"]
print(f"{attrs.get('step')}: {attrs.get('status')}")
for action in attrs.get("actions", []) or []:
if "retrieve" in action:
query = action["retrieve"].get("inputQuery", {}).get("text", "")
print(f" sub-query: {query}")
elif "result" in event:
for chunk in event["result"].get("results", []):
print(chunk.get("content", {}).get("text", "")[:120])
当你只想要检索行为时，将 generateResponse 设置为 False。API 默认会生成有依据的答案，这会产生额外的模型调用，在检查规划阶段可能并非必要。
跟踪事件的 step 告诉你规划器当前所在位置。SpeculativeRetrieval 在首次规划之前运行以降低延迟，不计入迭代配额。Planning 是模型读取问题和先前结果并发出子查询的阶段。Retrieval 每次子查询触发一次。FullDocumentExpansion 出现在模型判断某个段落缺乏回答问题所需的上下文时，此时会拉取整个文档。每个步骤都携带状态 IN_PROGRESS、SUCCEEDED 或 FAILED，以及一条人类可读的消息。
最终块作为单独的事件类型到达。result 是其自己的事件类型而非第五个步骤，它包含每次迭代中去重后的块，以及启用响应生成时的有依据答案。如前文循环所示，应根据事件 key 进行分支，而非期待某个终止步骤值。
子查询文本是值得记录的部分。它位于 attributes.actions[].retrieve.inputQuery.text，而非顶层跟踪字段，因此只读取 step 和 status 的处理器会显示规划发生了，但不会显示它决定了什么。
下图展示了 Agentic 检索的规划循环，包括推测性检索、规划、子查询检索、评估和可选的重新规划步骤。
Figure 2: Agentic retrieval 规划循环中的步骤
在以此为基础构建之前，有两个细节值得了解。去重仅适用于 result 事件，因此一个块被三个子查询检索到后，在结果中只出现一次，但在各次跟踪中会出现三次。
第二个是关于分数。Retrieve 响应为每个块提供一个类型化的 score 字段，表示其与查询的相关性。Agentic 检索结果携带 content、metadata 和 sourceRetriever，但没有等效的类型化字段。在切换 API 后读取 result["score"] 的代码会得到空值。如果要根据相关性排序或过滤，需要为这种差异做好准备。
在生产环境中，使用 Amazon Bedrock Guardrails 对生成的回答强制执行内容策略和 grounding 检查。两条检索路径都支持护栏。Agentic 检索通过 policyConfiguration.bedrockGuardrailConfiguration 而非 LangChain 检索器接受的 guardrail_config 参数来支持护栏，且仅支持 BLOCK 模式。如果你依赖 MASK 模式，这是继续使用 Retrieve API 的一个理由。
maxAgentIteration 接受二到十，默认为五。保持默认即可。在二或三的情况下，规划器运行一个周期，不发出子查询，只返回推测性检索步骤已经找到的内容。这是单次检索的行为但收取 Agentic 的价格。分解在四时开始。当规划器判断证据充分时，通常会提前停止，因此上限只是一个边界而非目标。
为了解这种方案在大规模场景下的表现，AWS 在 MuSiQue（一个公开的多跳基准测试）上对 AI 智能体式检索进行了评估。评估显示，相较于单次检索，多跳检索的召回率有所提升，且在最困难的问题上收益最大。单跳问题的收益不足五个百分点。这个数字与上述权衡的形态相吻合：当问题有待分解时，分解才有价值。
对于标准检索器，常规的 LangChain Expression Language（LCEL）组合方式直接可用：
from langchain_aws import ChatBedrockConverse
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
def format_docs(docs):
return "\n\n".join(doc.page_content for doc in docs)
llm = ChatBedrockConverse(model=MODEL_ID, region_name=REGION)
prompt = ChatPromptTemplate.from_template(PROMPT_TEMPLATE)
chain = (
{"context": retriever | format_docs, "question": RunnablePassthrough()}
| prompt
| llm
| StrOutputParser()
)
format_docs 的作用比看起来更重要。将 Document 对象直接传入提示词会渲染其 repr，导致元数据噪声混入上下文。
要将 AI 智能体式检索置于同等位置，需要将其包装在 RunnableLambda 中，因为它是一个函数而非检索器：
from langchain_core.runnables import RunnableLambda
def agentic_context(question: str) -> str:
result = agentic_retrieve(
knowledge_base_id=KB_ID,
query=question,
region_name=REGION,
number_of_results=10,
)
return "\n\n".join(
item.get("content", {}).get("text", "")
for item in result.get("results", [])
)
agentic_chain = (
{"context": RunnableLambda(agentic_context), "question": RunnablePassthrough()}
| prompt
| llm
| StrOutputParser()
)
注意，这里 generate_response 是关闭的。该服务可以自行生成答案，但在链内部通常你希望使用自己的提示词和模型，所以只取块并在下游生成。当你需要一次调用且代码更少时使用服务生成，当提示词由你掌控时使用包装版本。
对于简短、范围明确的问题，使用 Retrieve。它更便宜、更快、支持自管知识库，且结果中包含评分。大多数生产流量属于这一类。
当问题涉及多部分、比较性或探索性，或证据跨越多个知识库时，使用 AgenticRetrieveStream。它可以在一个请求中注册多达五个知识库，并使用你为每个知识库附加的自然语言描述来路由子查询。另一个 API 完全无法做到这一点。它的单次调用成本更高，需要多次模型调用，且延迟高于前者。
根据查询形态进行路由而非对所有请求统一使用某一种，是我们推荐的模式。问题上的分类器或启发式方法可以将大部分流量导向低成本路径，将规划器留给真正需要它的查询。
删除知识库、其数据源、S3 对象和 bucket，以及你创建的 IAM 角色。包含文档的知识库会持续产生存储费用。
bedrock_agent.delete_data_source(knowledgeBaseId=KB_ID, dataSourceId=DS_ID)
bedrock_agent.delete_knowledge_base(knowledgeBaseId=KB_ID)
仓库中包含一个清理脚本，还可以清空 bucket 并移除角色。
我们展示了如何使用 LangChain 在 Amazon Bedrock Knowledge Bases 上构建 RAG 应用，以及 AI 智能体式检索如何处理单次检索回答不佳的多部分问题。我们还展示了当前集成中的摩擦点。AI 智能体式检索是一个函数而非 LangChain 检索器，因此需要 RunnableLambda 才能接入链中。显示查询计划的跟踪事件需要直接的 boto3 调用。
AI 智能体式检索以更高的单次调用成本换取多跳问题上的召回率提升，使用内置模型进行查询规划。下一步有用的做法是在将所有请求都经由规划器之前，先测量你自己的查询分布。
入门请参阅 Amazon Bedrock Knowledge Bases 文档和配套示例代码。如需帮助将此应用于你自己的工作负载，请联系你的 AWS 账户团队。