突破传统 RAG 上限：在 AWS 上构建任务感知知识压缩

2026-07-28 21 预计阅读时间: 1 分钟
来源: aws.amazon.com AI 摘要 Original link

Disclaimer: This article is an AI-assisted summary. Read it together with the original source when precision matters. The summary may omit context, version differences, or edge cases and is not official documentation.

预计阅读时间：11 分钟

传统检索增强生成（RAG）擅长回答“哪份文档提到了这个术语”，但面对需要横跨数百份报告进行比较、归因和趋势分析的任务时，很快会碰到上下文窗口、检索召回率和推理成本的共同上限。任务感知知识压缩（Task-Aware Knowledge Compression，TAKC）换了一个切入点：不要等到查询到来后再临时拼接文档，而是围绕明确任务，提前把整个知识库压缩成可复用的任务表示，并按不同保真度缓存。

RAG 为什么会在分析型任务上失速

典型 RAG 请求通常包含四步：切分文档、向量检索、重排片段、把少量片段交给模型。这个流程隐含了一个假设：答案主要存在于少数局部片段中。

分析型任务往往不满足这个假设。例如：

  • 比较过去八个季度中所有业务线的风险变化；
  • 从合同库中归纳反复出现的责任限制条款；
  • 综合数百份事故报告，判断哪些控制措施最有效；
  • 根据政策、工单和审计记录解释某项指标异常。

这些问题的证据分散在大量文档中。即使检索器成功找回十几个“最相似”片段，也可能漏掉低相似度但决定结论的反例。单纯扩大 top_k 又会增加 token、延迟和噪声，使模型把计算预算浪费在重复内容上。

TAKC 的关键变化是把部分计算从在线查询阶段前移到离线构建阶段。系统预先知道目标任务，例如“季度风险比较”或“合同义务抽取”，因此可以保留与该任务有关的实体、指标、时间、结论、例外和来源，而压缩掉版式、重复措辞及无关叙述。

一份知识库，多档保真度

任务感知压缩不是生成一篇不可追溯的长摘要。更稳妥的实现会为同一任务准备多个层级：

层级 典型内容 适合的问题
L0 概览 主题、关键指标、主要结论 快速问答、查询分类
L1 结构化摘要 按实体、时间或业务单元组织的事实 跨文档比较、趋势分析
L2 证据包 关键原文、来源标识、页码或段落 归因、审计、结论验证
原始层 完整文档 高风险复核、补充证据

查询路由器根据任务类型、风险等级和用户要求选择层级。一个普通的趋势问题可以命中 L1；要求“逐条列出依据”的问题应读取 L2；法律、合规或财务结论则可能需要回到原始层。

在 AWS 上可以这样实践：将原始文件和压缩产物放入 Amazon S3，以对象版本或内容哈希管理更新；使用批处理作业完成解析和压缩；把结构化表示写入适合过滤查询的存储；在线服务根据路由结果加载相应层级，再调用模型生成答案。具体计算服务、模型和数据库应根据现有平台约束选择，不能仅凭摘要断言某个固定组件是唯一实现。

多层缓存带来的收益不只是减少 token。它还让系统能够分别设置更新频率：概览可在文档变更后立即重建，昂贵的高保真证据包则按批次刷新。代价是缓存失效、版本一致性和访问控制会变成正式的工程问题。

可以这样实践：最小可运行的压缩与路由器

下面是一个仅依赖 Python 标准库的示例。它不调用真实大模型，而用关键词和句子评分模拟“面向风险分析任务的压缩”，便于先验证分层格式、缓存键和路由规则。生产环境可以把 compress_for_task 替换为受约束的模型调用，并将缓存目录替换为 S3。

将代码保存为 takc_demo.py，直接运行 python takc_demo.py：

from __future__ import annotations

import hashlib
import json
from pathlib import Path

DOCUMENTS = {
    "q1.txt": "Revenue grew 12%. Supplier delays affected two regions. No security incidents were reported.",
    "q2.txt": "Revenue grew 8%. Supplier delays expanded to four regions. One critical security incident occurred.",
    "q3.txt": "Revenue grew 10%. A second supplier reduced delivery delays. Security controls were upgraded.",
}

TASKS = {
    "risk_trend": ["risk", "delay", "incident", "security", "supplier", "control"],
    "revenue_trend": ["revenue", "grew", "declined", "quarter"],
}

CACHE_DIR = Path(".takc-cache")
CACHE_DIR.mkdir(exist_ok=True)


def split_sentences(text: str) -> list[str]:
    return [part.strip() for part in text.split(".") if part.strip()]


def compress_for_task(documents: dict[str, str], task: str) -> dict:
    keywords = TASKS[task]
    evidence = []

    for source, text in documents.items():
        for sentence in split_sentences(text):
            score = sum(word in sentence.lower() for word in keywords)
            if score:
                evidence.append({
                    "source": source,
                    "text": sentence,
                    "score": score,
                })

    evidence.sort(key=lambda item: (-item["score"], item["source"]))
    return {
        "task": task,
        "l0": [item["text"] for item in evidence[:2]],
        "l1": evidence[:6],
        "l2": evidence,
    }


def cache_key(documents: dict[str, str], task: str) -> str:
    payload = json.dumps(
        {"documents": documents, "task": task},
        sort_keys=True,
    ).encode()
    return hashlib.sha256(payload).hexdigest()


def load_or_build(documents: dict[str, str], task: str) -> dict:
    path = CACHE_DIR / f"{cache_key(documents, task)}.json"
    if path.exists():
        return json.loads(path.read_text(encoding="utf-8"))

    compressed = compress_for_task(documents, task)
    path.write_text(json.dumps(compressed, indent=2), encoding="utf-8")
    return compressed


def route(query: str) -> tuple[str, str]:
    text = query.lower()
    task = "revenue_trend" if "revenue" in text else "risk_trend"
    if any(word in text for word in ("evidence", "source", "audit")):
        tier = "l2"
    elif any(word in text for word in ("compare", "trend", "change")):
        tier = "l1"
    else:
        tier = "l0"
    return task, tier


if __name__ == "__main__":
    query = "Compare the supplier risk trend and include evidence"
    task, tier = route(query)
    knowledge = load_or_build(DOCUMENTS, task)
    print(json.dumps({
        "query": query,
        "task": task,
        "tier": tier,
        "context": knowledge[tier],
    }, indent=2))

这个例子有三个值得保留到生产系统的设计点：缓存键同时包含知识库版本和任务；压缩结果保留 source；路由器输出任务与保真度，而不是直接生成答案。真正接入模型时，还应要求模型输出结构化 JSON，并校验每条事实是否关联有效来源。

在 AWS 上落地时要补齐的控制面

从本地原型迁移到 AWS，可以采用如下对象布局。这里是建议性的实践结构，并非对原实现细节的断言：

s3://enterprise-knowledge/raw/{tenant}/{document-version}/...
s3://enterprise-knowledge/compressed/{tenant}/{task}/{kb-version}/l0.json
s3://enterprise-knowledge/compressed/{tenant}/{task}/{kb-version}/l1.json
s3://enterprise-knowledge/compressed/{tenant}/{task}/{kb-version}/l2.json
s3://enterprise-knowledge/manifests/{tenant}/{task}/{kb-version}.json

清单文件至少记录输入文档哈希、压缩提示词版本、模型版本、生成时间和产物位置。部署后还需要重点验证：

  • 增量更新：一份文档变化时，是否能只重算受影响的任务表示；
  • 权限继承：压缩结果不能把用户无权访问的文档事实混入上下文；
  • 来源可追溯：摘要中的每条关键结论能否定位到原始对象和版本；
  • 路由降级：分类置信度不足时，应提升保真度或回退到检索，而不是强行回答；
  • 质量评估：除了答案正确率，还要测量证据覆盖率、压缩遗漏率、延迟和单次查询成本；
  • 提示词注入防护：文档内容是数据，不应被当成压缩流程的系统指令执行。

采用 TAKC 的判断标准

TAKC 不是对所有 RAG 系统的替代。知识库规模较小、问题高度局部、文档频繁变化且任务难以预先定义时，传统检索通常更简单。TAKC 更适合知识库相对稳定、分析任务可枚举、查询量较高，并且反复执行跨文档综合的场景。

落地时可以从一个成本高、重复率高的任务开始，只建立两档表示，并保留原有 RAG 作为证据回退路径。对比相同测试集上的覆盖率、引用准确率、P95 延迟与 token 成本后，再决定是否扩展到更多任务。真正的目标不是制造更短的摘要，而是把昂贵的跨文档理解变成可版本化、可缓存、可审计的知识资产。

# AI # AWS # LLM # 云平台 # 机器学习

相关推荐