在大规模运行 RAG（Retrieval-Augmented Generation）系统时，真正影响账单的并不只有生成答案的输出 token。检索结果会被拼接到主模型的输入上下文中，文档片段越多、越长，输入 token 成本就越高，也越容易让模型注意力分散。
一种实用的优化方式是引入“查询感知上下文压缩”：先完成检索，再使用一个成本更低、规模更小的模型，根据用户问题过滤和压缩检索片段，最后才把精简后的上下文交给主模型生成答案。这样可以减少主模型需要处理的输入 token，同时保留与当前问题真正相关的信息。
RAG 成本为什么容易被检索结果放大
典型的 RAG 流程通常是：
- 接收用户问题。
- 从向量数据库或搜索系统中检索若干文档片段。
- 将问题和所有片段拼接成提示词。
- 调用主模型生成答案。
假设一次检索返回 10 个片段，每个片段约 800 个 token，那么主模型可能需要处理约 8,000 个上下文 token。对于一个只需要其中两三个片段的问题，大量输入实际上是冗余的。
冗余上下文会带来三个问题：
- 主模型输入 token 增加，调用成本上升。
- 上下文越长，端到端延迟通常越高。
- 无关内容可能干扰模型判断，降低答案稳定性。
因此，优化重点不应只放在“检索更多内容”，还要考虑“哪些检索内容值得交给主模型”。
查询感知压缩的工作方式
查询感知压缩在检索和生成之间增加一个轻量处理阶段：
用户问题
|
v
检索器 -> 候选文档片段
|
v
小模型进行相关性过滤/压缩
|
v
精简上下文 + 用户问题
|
v
主模型生成答案
这里的小模型可以执行两类任务：
- 过滤：删除与问题无关的文档片段。
- 压缩：从相关片段中提取支持答案所需的句子、事实或字段。
压缩模型不负责最终回答，而是负责控制进入主模型的上下文规模。主模型仍然可以使用精简后的原文证据来完成更复杂的推理和表达。
在 Amazon Bedrock 上，可以将这一阶段实现为一次额外的模型调用。具体模型 ID、区域和请求格式应根据账户中启用的模型进行调整。下面是一个使用 Python 和 boto3 的最小示例，演示如何让轻量模型从检索片段中挑选与问题相关的内容。
一个可改造的 Bedrock 压缩示例
运行前准备：
- 配置 AWS 凭证，例如执行
aws configure，或使用 IAM Role。 - 确认目标区域已启用所选 Amazon Bedrock 模型。
- 将示例中的模型 ID 替换为账户可访问的模型。
- 在生产环境中为输出增加 JSON Schema 校验和失败重试。
import json
import os
import boto3
REGION = os.getenv("AWS_REGION", "us-east-1")
COMPRESSOR_MODEL_ID = os.getenv(
"COMPRESSOR_MODEL_ID",
"anthropic.claude-3-haiku-20240307-v1:0",
)
brt = boto3.client("bedrock-runtime", region_name=REGION)
query = "如何配置订单取消后的退款时限？"
retrieved_chunks = [
{
"id": "policy-12",
"text": "订单取消后，原路退款通常在 3 至 7 个工作日内到账，具体时间取决于支付渠道。",
},
{
"id": "shipping-04",
"text": "仓库每天 18:00 进行当日订单拣选，次日订单进入发货流程。",
},
{
"id": "policy-19",
"text": "如果订单已经发货，用户需要先申请拒收或退货，退款时限按照退货验收规则计算。",
},
]
context = "\n\n".join(
f"[{chunk['id']}] {chunk['text']}" for chunk in retrieved_chunks
)
prompt = f"""你是 RAG 上下文压缩器。
根据用户问题，从候选片段中保留能够直接支持答案的内容。
删除无关片段，不要补充候选片段中不存在的事实。
只输出 JSON，格式为 {{\"relevant_chunks\": [{{\"id\": \"...\", \"text\": \"...\"}}]}}。
用户问题：
{query}
候选片段：
{context}
"""
response = brt.invoke_model(
modelId=COMPRESSOR_MODEL_ID,
contentType="application/json",
accept="application/json",
body=json.dumps({
"anthropic_version": "bedrock-2023-05-31",
"max_tokens": 500,
"temperature": 0,
"messages": [
{"role": "user", "content": prompt}
],
}),
)
payload = json.loads(response["body"].read())
compressed_text = payload["content"][0]["text"]
compressed = json.loads(compressed_text)
print(json.dumps(compressed, ensure_ascii=False, indent=2))
在完整的 RAG 服务中，compressed["relevant_chunks"] 可以被重新拼接到主模型提示词中。建议保留片段 ID，而不是只传递纯文本。这样可以在日志、答案引用和离线评估中追踪压缩模型究竟保留了哪些证据。
主模型阶段可以使用类似下面的上下文：
请仅根据以下资料回答问题。如果资料不足，请明确说明。
问题：如何配置订单取消后的退款时限？
资料：
[policy-12] 订单取消后，原路退款通常在 3 至 7 个工作日内到账，具体时间取决于支付渠道。
[policy-19] 如果订单已经发货，用户需要先申请拒收或退货，退款时限按照退货验收规则计算。
如何判断压缩是否值得
增加压缩模型调用并不一定自动降低总成本。需要比较以下几部分：
总成本 = 压缩模型输入/输出成本
+ 主模型压缩后输入成本
+ 主模型输出成本
+ 额外调用带来的基础设施与延迟成本
当原始检索上下文很短时，压缩调用可能得不偿失。相反，如果检索返回的片段较多，或者主模型输入价格较高，压缩更可能带来明显收益。
可以从以下指标开始评估：
- 压缩前后的输入 token 数。
- 压缩比例：
1 - 压缩后 token / 压缩前 token。 - 主模型调用成本和总调用成本。
- 首 token 延迟与端到端延迟。
- 答案正确率、引用覆盖率和拒答准确率。
- 压缩模型误删关键证据的比例。
评估数据应覆盖简单问题、多跳问题、需要边界条件的问题，以及检索结果包含大量干扰内容的情况。只观察平均 token 节省是不够的；如果压缩让关键答案证据消失，账单下降并不能抵消质量损失。
落地时需要控制的边界
给压缩器清晰的任务边界。 提示词应要求它只筛选和压缩已有内容，不要自行回答问题，也不要补充外部知识。否则，主模型接收到的可能是压缩器的推断，而不是可追溯证据。
保留原始检索结果。 压缩后的上下文用于主模型调用，但原始片段仍应保存在调试日志或可查询的请求记录中，并注意脱敏和访问控制。出现错误答案时，需要能够比较压缩前后的证据。
为不同类型的问题设置不同策略。 简单事实问答可以采用较激进的压缩；合规、医疗、财务或多跳推理场景则应保留更多原文，甚至让压缩器只做排序而不删除内容。
限制压缩器的输出格式和长度。 结构化 JSON、最大输出 token 和温度控制有助于降低解析失败率。生产代码还应处理模型返回非 JSON、空结果、超时和限流等情况。
把压缩放在检索之后，而不是替代检索。 查询感知压缩无法修复召回阶段完全没有找到正确文档的问题。召回数量、分块策略、元数据过滤和排序质量仍然决定了压缩器能否看到正确证据。
一份可执行的采用清单
- 记录当前 RAG 请求的检索片段数量和 token 分布。
- 先用离线数据集测量压缩比例与答案质量变化。
- 选择低成本、低延迟的压缩模型，并验证其区域和模型访问权限。
- 让压缩结果携带文档 ID、片段 ID或引用信息。
- 对压缩为空、格式错误和调用超时设计降级路径。
- 用真实流量比较总成本，而不是只比较主模型输入成本。
- 为高风险问题保留更保守的压缩策略。
查询感知压缩适合被视为 RAG 的一个可观测中间层。它的价值不只是删掉文字，而是让系统根据当前问题决定哪些证据值得进入昂贵的主模型上下文。在检索质量稳定的前提下，先从日志和离线评估开始，再逐步扩大到在线流量，通常是更稳妥的采用路径。