理赔资料通常散落在申请表、查勘报告、维修报价和往来邮件中。Amazon Bedrock Knowledge Bases 可以把这些存放在 Amazon S3 的文档转化为可检索知识库，再通过 AgenticRetrieveStream API 接收自然语言问题、维护多轮上下文，并返回带来源引用的答案。
真正决定系统能否上线的，不只是“能回答问题”，而是能否限制检索范围、展示证据、隔离不同客户的数据，并在证据不足时拒绝给出武断结论。
从 S3 文档和元数据开始
可以按理赔编号组织对象，但不要只依赖 S3 路径实现业务过滤。为每份文档附加结构化元数据，后续才能按客户、案件状态或文档类型约束检索。
下面以一份查勘报告为例。先准备文件及其元数据侧车文件：
mkdir -p claims/CLM-1001
cp adjuster-report.pdf claims/CLM-1001/adjuster-report.pdf
cat > claims/CLM-1001/adjuster-report.pdf.metadata.json <<'JSON'
{
"metadataAttributes": {
"claim_id": "CLM-1001",
"policyholder_id": "P-42",
"status": "open",
"document_type": "adjuster_report"
}
}
JSON
aws s3 cp claims/CLM-1001/ \
"s3://${CLAIMS_BUCKET}/claims/CLM-1001/" \
--recursive
其中 CLAIMS_BUCKET 需要替换为已经配置为 Knowledge Base 数据源的 S3 存储桶。字段名应尽量稳定，枚举值也要统一，例如不要同时使用 open、OPEN 和 in_progress 表示同一状态。
上传后启动同步任务：
aws bedrock-agent start-ingestion-job \
--knowledge-base-id "$KNOWLEDGE_BASE_ID" \
--data-source-id "$DATA_SOURCE_ID"
可以继续使用 get-ingestion-job 检查任务状态。生产系统还应记录同步批次、失败文档和文档版本，避免回答引用已经失效的理赔材料。
AgenticRetrieveStream 如何串起对话
一次完整请求通常包含四类信息：
- 用户当前问题，例如“这次事故为什么只批准了部分维修费用？”
- Knowledge Base 标识和检索配置。
- 用于延续上下文的会话标识。
- 根据当前登录用户生成的元数据过滤器。
流式接口的价值不只是更快显示文字。服务还可以逐步返回检索结果、引用或安全检查事件，前端能够分别渲染答案和证据，而不是等整个请求结束。
下面给出一个可改造的 Python 调用骨架。它按 AgenticRetrieveStream 的常见 SDK 命名编写；该能力在不同 boto3 版本、区域或发布阶段中的请求字段可能有变化，运行前应升级 SDK，并用脚本打印的输入字段与当前服务模型核对。
import json
import os
import sys
import boto3
REGION = os.getenv("AWS_REGION", "us-east-1")
KB_ID = os.environ["KNOWLEDGE_BASE_ID"]
CLAIM_ID = os.environ["CLAIM_ID"]
SESSION_ID = os.getenv("BEDROCK_SESSION_ID")
client = boto3.client("bedrock-agent-runtime", region_name=REGION)
operation = "AgenticRetrieveStream"
if operation not in client.meta.service_model.operation_names:
raise RuntimeError(
"当前 boto3/区域未公开 AgenticRetrieveStream。"
"请先运行: python -m pip install -U boto3 botocore"
)
model = client.meta.service_model.operation_model(operation)
print("支持的请求字段:", sorted(model.input_shape.members), file=sys.stderr)
request = {
"knowledgeBaseId": KB_ID,
"input": {"text": "这次理赔为什么只批准了部分维修费用？"},
"retrievalConfiguration": {
"vectorSearchConfiguration": {
"filter": {
"equals": {
"key": "claim_id",
"value": CLAIM_ID
}
}
}
}
}
if SESSION_ID:
request["sessionId"] = SESSION_ID
# 使用低层调用可以让代码直接对应 AWS API 操作名。
response = client._make_api_call(operation, request)
# 首次接入时保留原始事件，确认当前 SDK 的文本、引用和会话事件结构。
for event in response["stream"]:
print(json.dumps(event, ensure_ascii=False, default=str))
运行前设置环境变量：
python -m pip install -U boto3 botocore
export AWS_REGION=us-east-1
export KNOWLEDGE_BASE_ID=YOUR_KB_ID
export CLAIM_ID=CLM-1001
python app.py
接入正式界面后，应把事件拆成三条通道：答案文本、引用列表和状态事件。不要通过字符串猜测引用；应读取 API 返回的结构化引用信息，并在界面上显示文档名、相关片段以及可访问的内部文档地址。
多轮追问不能脱离案件边界
用户经常先问“这次赔了多少”，再追问“为什么不是全额”。第二个问题缺少明确主语，因此需要沿用前一次会话的上下文。应用应保存服务返回或使用的会话标识，并在后续请求中继续传入。
但会话上下文不能取代授权检查。一个安全的请求流程应当是：
- 从登录令牌或后端权限系统得到用户可访问的
claim_id。 - 在服务端构造 metadata filter，禁止浏览器自行提交任意案件编号。
- 将同一个过滤条件应用到该会话的每轮检索。
- 用户切换案件时创建新会话，避免旧上下文污染新案件。
如果用户能访问多个案件，可以使用 API 支持的组合过滤器构造允许列表；案件数量较大时，更适合按租户或客户维度隔离数据源，而不是生成一个巨大的 OR 条件。
引用和 Grounding Guardrail 各自解决什么问题
引用回答“这句话依据哪份材料”，Contextual Grounding Guardrail 则用于检查回答是否得到检索上下文支持，以及回答是否真正回应了用户问题。两者应同时存在，但不能互相替代。
可以将处理逻辑设计为：
- 有充分证据：返回答案，并在关键结论后展示引用。
- 检索到了文档，但支撑度不足：明确说明无法从现有材料确认。
- Guardrail 阻止输出：返回稳定的业务提示，不把内部评分、策略文本或原始敏感片段暴露给用户。
- 没有检索结果：建议补充文件或转人工处理，而不是让模型依靠常识补齐理赔事实。
Guardrail 配置应在部署环境中通过 ID 和版本管理，并针对真实问题建立评测集。尤其需要测试金额、日期、责任认定、免赔额和拒赔原因，因为这些字段一旦答错，业务影响远高于普通摘要错误。
上线前检查清单
- S3 文档、元数据文件和 Knowledge Base 同步任务是否一致。
- metadata filter 是否完全由可信后端生成。
- 多轮请求是否复用正确会话，并在切换案件时清空上下文。
- 前端是否展示结构化引用，而不是仅显示模型生成的文档名称。
- Grounding Guardrail 被触发或没有证据时，是否安全降级。
- 日志是否记录请求 ID、案件 ID、引用和 Guardrail 结果，同时避免写入不必要的个人信息。
- 是否用一组已知答案的理赔问题验证召回率、引用准确率和越权访问。
一个可靠的理赔助手不是“把 PDF 接到大模型上”。更稳妥的做法是把 S3 摄取、元数据授权、会话管理、结构化引用和 Grounding Guardrail 看成一条完整链路。只有每个环节都能审计和降级，自然语言查询才适合进入真实理赔流程。