传统 RAG 检索通常把用户问题编码成一个向量，再用它搜索知识库。这种方式处理单一事实问题很直接，但遇到包含多个约束、比较项或时间范围的复合问题时，一个查询向量很难同时准确表达所有意图。Amazon Bedrock Managed Knowledge Base 的 Agentic Retrieval 针对这一缺口，引入查询规划和流式执行，让检索过程能够拆题、分步搜索，并通过 trace 暴露中间决策。
为什么一次 Retrieve 不够
假设用户提出下面的问题：
比较 2023 年和 2024 年退款政策的变化，并说明企业客户是否受到影响。
这个问题至少包含三个检索目标：
- 找到 2023 年退款政策；
- 找到 2024 年退款政策；
- 识别企业客户条款，并比较前后变化。
标准 Retrieve API 通常把整段文本作为一个检索查询。它速度快、行为容易预测，但返回结果可能集中在语义最强的部分，例如只命中 2024 年政策，遗漏旧版本或企业客户条款。
Agentic Retrieval 的差异不只是“多查几次”。它会围绕原始问题制定检索步骤，再逐步返回结果和执行轨迹。应用因此可以观察系统拆出了哪些子问题、调用了哪些检索步骤，以及最终证据来自哪里。
这并不意味着 Agentic Retrieval 应该替代所有向量检索。简单的产品编号查询、文档定位和单一事实问答，通常仍适合标准 Retrieve；复合比较、跨文档归纳和多约束问题，才更能体现 agentic 模式的价值。
AgenticRetrieveStream 的调用形态
AgenticRetrieveStream 是流式 API。客户端提交知识库标识、用户输入以及相关检索配置，然后持续消费服务端事件。与等待一个完整 JSON 响应相比，流式接口有两个工程影响：
- 结果和 trace 可能分散在多个事件中，不能假设一次读取就得到完整答案；
- 客户端必须区分内容事件、轨迹事件、引用或检索结果事件，以及异常事件。
具体字段会随 AWS SDK 和服务模型版本变化。接入前应升级 boto3，并直接检查本地 Botocore 模型，避免依赖手写且可能过期的请求结构：
python -m pip install --upgrade boto3 botocore
python - <<'PY'
import boto3
client = boto3.client("bedrock-agent-runtime", region_name="us-east-1")
model = client.meta.service_model
operation_name = "AgenticRetrieveStream"
if operation_name not in model.operation_names:
raise SystemExit(
"当前 SDK 尚未包含 AgenticRetrieveStream，请升级 boto3/botocore，"
"并确认该 API 已在目标区域开放。"
)
operation = model.operation_model(operation_name)
print("Input members:", sorted(operation.input_shape.members))
print("Output members:", sorted(operation.output_shape.members))
PY
这段命令不猜测请求字段，而是读取当前安装版本中的正式服务模型。部署环境也应执行同样的版本检查，因为开发机支持新操作，并不代表 Lambda 层或容器镜像里的 SDK 同样支持。
可以这样实践：构造请求并解析流式 trace
下面是一个可改造的 Python 客户端骨架。由于来源摘要没有给出所有字段的精确名称，示例会从 Botocore 模型校验请求；你需要根据脚本打印出的 input members，把 request 调整为当前 SDK 接受的结构。
运行前设置 BEDROCK_KB_ID 和 AWS 区域，并确保调用身份拥有对应的 Bedrock Knowledge Base 权限。
import json
import os
import boto3
from botocore.exceptions import ClientError, ParamValidationError
REGION = os.getenv("AWS_REGION", "us-east-1")
KNOWLEDGE_BASE_ID = os.environ["BEDROCK_KB_ID"]
QUESTION = os.getenv(
"QUESTION",
"比较 2023 年和 2024 年退款政策，并说明企业客户是否受到影响。",
)
client = boto3.client("bedrock-agent-runtime", region_name=REGION)
operation_name = "AgenticRetrieveStream"
if operation_name not in client.meta.service_model.operation_names:
raise RuntimeError("当前 boto3/botocore 版本不支持 AgenticRetrieveStream")
# 这是接入模板。请按照当前 SDK 的 input shape 调整字段名称和嵌套结构。
request = {
"knowledgeBaseId": KNOWLEDGE_BASE_ID,
"input": {"text": QUESTION},
}
def walk(value, path="event"):
"""递归打印流式事件，便于识别 trace、结果和异常的实际结构。"""
if isinstance(value, dict):
for key, child in value.items():
walk(child, f"{path}.{key}")
elif isinstance(value, list):
for index, child in enumerate(value):
walk(child, f"{path}[{index}]")
else:
print(f"{path}: {value}")
try:
response = client.agentic_retrieve_stream(**request)
event_stream = response.get("stream") or response.get("body")
if event_stream is None:
print(json.dumps(response, ensure_ascii=False, default=str, indent=2))
raise RuntimeError("响应中未找到事件流，请检查当前 SDK 的 output shape")
for event in event_stream:
print("\n--- stream event ---")
walk(event)
except ParamValidationError as exc:
shape = client.meta.service_model.operation_model(operation_name).input_shape
print("可用顶层字段:", sorted(shape.members))
raise SystemExit(f"请求结构与当前 SDK 不匹配: {exc}")
except ClientError as exc:
error = exc.response.get("Error", {})
raise SystemExit(
f"Bedrock 调用失败: {error.get('Code')} - {error.get('Message')}"
)
首次联调时，建议保留完整事件并写入受控日志，再根据实际事件类型建立显式分发器。例如将 trace 发送到调试日志，将检索结果转换为统一的证据对象，将可展示文本推送到 WebSocket 或 Server-Sent Events 客户端。
不要在生产环境无条件记录 trace。轨迹里可能包含用户问题、拆分后的查询和文档片段，应执行脱敏、访问控制和日志保留期限管理。
Trace 应该回答哪些问题
Trace 的价值不只是调试请求失败。它可以帮助团队判断 agentic 检索是否真的改善了答案：
- 查询规划是否覆盖了原问题中的每个约束；
- 不同子查询是否只返回了同一批重复文档；
- 最终结果是否引用了每个比较项的证据；
- 某个子查询无结果时，系统是否仍生成了过度确定的结论；
- 拆分查询带来的额外延迟和调用成本是否可以接受。
解析 trace 时，不要把事件顺序硬编码成“规划、检索、结束”三个固定位置。流式协议可能插入多个检索步骤或错误事件，更稳妥的做法是按事件键或 SDK 定义的事件类型分派，并忽略客户端暂时不认识的可选事件。这样服务端增加新事件后，旧客户端不会立即崩溃。
Retrieve 与 AgenticRetrieveStream 怎么选
可以从问题复杂度、延迟预算和可观测性三个维度做决定。
选择标准 Retrieve 的典型场景包括：
- 用户输入通常是关键词、编号或单一事实问题；
- 应用已经自行完成查询拆分；
- 需要稳定的低延迟，并能接受一次召回的边界；
- 只需要检索结果，不需要观察规划过程。
选择 AgenticRetrieveStream 的典型场景包括：
- 问题经常包含比较、条件、时间范围或多个实体；
- 证据分布在不同文档或文档版本中；
- 应用希望边执行边展示状态或结果；
- 团队需要通过 trace 审计查询规划和检索步骤。
上线前不要只比较最终答案的主观流畅度。准备一组带证据标注的复合问题，同时记录召回覆盖率、引用正确率、首事件延迟、总延迟、错误率和单次请求成本。Agentic Retrieval 增加了规划能力，也增加了调用链长度和运行时不确定性。最稳妥的落地方式通常是按问题复杂度路由：简单问题继续使用 Retrieve，只有检测到多意图或多约束时才进入 AgenticRetrieveStream。