RAG正在烧钱——我构建了一层成本控制机制来修复它

一言以蔽之

一个完全用 Python 实现的完整工作版本，以及从本地设置中得出的基准测试结果。

RAG 系统不仅在质量方面存在问题，还可能在成本方面变得低效，通常以不立即显现的方式。

每个额外检索的标记都有成本。在我的系统中，上下文过取的范围是实际查询所需标记的 3-8 倍。

在许多基础实现中，重复查询被独立处理，没有重用先前的结果。

在单模型设置中，大量简单的查询可能由高成本模型处理，即使较低成本的模型已经足够。

通过语义缓存（在预种子、预热缓存基准测试中命中率高达 98.5%）、查询路由（在基准测试混合中，大约 81% 的请求被重新路由到较低成本的模型）和带有熔断器的标记预算层，系统在每天 10,000 次请求时实现了最多 85.8% 的成本 reduction，同时在评估设置中保持响应质量。

这些结果基于在以下基础配置描述的本地基准测试运行。

运行良好的系统——悄然烧钱

我建立了一个 RAG 系统，它完美地工作，我在相同的管道中运行相同的查询，每次得到相同的输出。在测试中，没有什么看起来不对，延迟稳定，答案正确。

然后我查看了标记日志。

在我的设置中，即使是简单的问题，如“RAG 是什么？”或“定义语义搜索。”也会击中最昂贵的模型。每次重复查询都会按全额计费，即使在十分钟前我已经回答了完全相同的问题。每个请求都会检索十个片段，而实际上两个片段就能完成工作。

系统没有 broken。它只是财务盲目。但在大规模应用中，这种区别不再重要。

在本地笔记本电脑上运行 RAG 管道很容易。但标准蓝图：检索、提示、调用留下了巨大的操作差距。生产成本行为在许多 RAG 实施指南中往往不是主要关注点。在现实世界中，你需要关注计算和标记效率。你是否在处理 exact 相同的查询时重复支付预算？一个简单的事实查询是否真的需要通过与多跳推理查询相同的重负模型路径？

我已经为我之前的系统 [7] 建立了一个上下文工程层，控制进入上下文窗口的内容，以确保质量。但质量和成本是不同的失败领域。你可以拥有完美的上下文控制，仍然支付 8 倍 unnecessary cost。

这就是我建立的成本控制层——带有真实数字和你可以运行的代码。

下面的所有结果都是从实际运行系统得出的（Python 3.12.6，Windows 11，CPU-only，无 GPU），除非明确指出为计算得出。

RAG 为什么在财务上设计盲目

RAG 是为了解决检索质量问题而设计的 [1]。它从未设计过解决成本问题。这不是批评——这只是堆栈的不同层面。

但在生产中，这两个层面相撞。而碰撞是昂贵的。

有三种具体的失败模式。

失败模式 1：上下文窗口过取

大多数实现默认检索前 10 个片段。“只是为了安全起见。”

问题：实际上，2-3 个片段包含答案。其他 7-8 个是噪声——多余的上下文，增加了标记但没有增加信息。你每次都在为这些标记付费。

在每次查询 500 个标记的情况下，使用前 10 个检索，其中 7 个是不必要的：

每次查询不必要的标记数：   ~350
每天 10,000 次请求：         3,500,000 每天不必要的标记
每 1000 标记 $0.015：            $52.50/天 纯浪费
每月：                        $1,575 不必要的上下文

这个数字是根据所述假设计算得出的，而不是端到端测量的。

失败模式 2：没有缓存层

两个用户在十分钟内分别问“RAG 是什么？”系统产生相同的嵌入、检索相同的片段并返回相同的答案。

你为大型语言模型的费用支付两次。

标准 RAG 管道中没有语义记忆。每个查询都 treated 仿佛它从未被问过。在 30% 的重复查询率下，基于我自己的领域特定流量的保守估计——你为 30% 的流量支付两次费用。

失败模式 3：没有模型路由

一些管道默认使用一个高能力模型处理所有查询，无论复杂度如何。

即使查询是：“LLM 是什么意思？”

这个问题不需要 GPT-4.5 或 Claude Opus。它不需要多跳推理。它不需要 200,000 上下文窗口。它需要一个快速、便宜的模型，并且需要在 200ms 内完成。

根据本设置中的定价假设，最高层模型的每标记费用是最低层的 ~90 倍 [2]。考虑到基准查询中 81% 是简单事实查询，未能适当路由它们会导致 substantial 和不必要的服务成本增加。

这些模式可以在更简单的 RAG 设置中出现，特别是在不包括成本感知优化的情况下。

完整代码：[https://github.com/Emmimal/rag-cost-control-layer/](https://github.com/Emmimal/rag-cost-control-layer/)

规模上的成本现实

在建立任何东西之前，我想如实看到数字。

基础 RAG 设置通常对每个请求运行检索，并不使用缓存或路由层。在更简单的实现中，它还依赖于单个高能力模型，如 GPT-4.5 级模型，处理所有查询。

Scale            Naive cost/day    Optimized cost/day    Saving
100 req/day          1.20美元             0.18美元            84.6%
1,000 req/day        12.00美元            1.71美元            85.7%
10,000 req/day       120.00美元           17.00美元           85.8%

传统RAG迅速消耗预算。成本控制层将LLM支出最多减少85%——在不牺牲答案质量的情况下。图片作者提供

每月10,000 req/day：3,600美元传统 vs 510美元优化。每月节省3,090美元。

（所有数据基于所声明的定价假设计算，未从实时API调用中测量。）

在大规模应用中，这些差异对系统是否能有效运营有显著影响。

系统架构：四层，一个系统

成本控制层由四个组件组成，每个组件都针对系统中的不同故障模式。

系统架构图详细描述了一个成本有效的LLM路由管道，包括语义缓存、动态模型选择和自动预算保护。图片作者提供

每层都有一个任务。它们共同确保系统在每个决策点都具有成本意识。

组件1：语义缓存

整个系统中最简单的成本节省方式。为已经回答过的问题不再支付LLM费用。

工作原理

语义缓存是LLM pipeline中的一个成熟模式——工具如GPTCache [8]证明，通过语义相似性而不是精确字符串匹配进行缓存可以消除大量LLM调用。本实现使用一个纯Python的TF-IDF嵌入器，没有外部依赖。

每个传入的查询都会使用TF-IDF向量化器 [3]进行嵌入。缓存中存储着之前的查询-回答对，每个对都与其嵌入。当新的查询传入时：

1. 嵌入查询
2. 计算与所有缓存嵌入的余弦相似性
3. 如果最佳相似性 ≥ 阈值（默认0.75）：返回缓存响应
4. 如果未命中：调用LLM，存储结果

class SemanticCache:
    def get(self, query: str) -> Optional[str]:
        query = self._validate(query)
        if query is None:
            return None

        with self._lock:
            self.stats.total_requests += 1
            if not self._entries:
                self.stats.cache_misses += 1
                return None

            q_vec = self._embedder.embed(query)
            best, best_sim = self._find_best(q_vec)

            if best is not None and best_sim >= self.threshold:
                best.hit_count += 1
                self.stats.cache_hits += 1
                self.stats.total_cost_saved_usd += self.cost_per_llm_call_usd
                return best.response

            self.stats.cache_misses += 1
            return None

缓存使用RLock以确保线程安全。每个查询的嵌入在词汇表变化时才重新计算，因此即使缓存大小增加，查找时间保持稳定。

阈值调优

默认的0.75阈值适用于TF-IDF相似性。Sentence-transformer嵌入通常会产生更高的相似性分数，因此使用OpenAI的text-embedding-3-small时，阈值通常在0.92-0.95之间。

阈值越低 → 命中率越高 → 有风险在边缘情况给出错误答案
阈值越高 → 命中率越低 → 更保守但更准确

合适的阈值取决于领域。窄系统（如单产品支持机器人或内部知识库）可以激进地设置在0.70-0.75。宽系统通常需要更高的阈值，通常是0.90或更高。

真实基准数据

运行200个查询，使用现实的混合（60%简单，30%标准，10%复杂，20%重复）：

命中率:             98.5%
平均命中延迟:       ~4 ms
平均未命中延迟:      ~4-5 ms
p95命中延迟:       ~5-7 ms
成本节省（200查询）: $0.788

基准达到98.5%命中率，因为40%的查询预种子到缓存中，模拟生产系统在初始流量积累后的状态。

延迟差距更重要：~4ms的缓存命中延迟与~700ms的LLM调用延迟相比，大约是175倍的改善，belum考虑成本节省。

生产注意事项

• 默认max_size=1000，使用LRU淘汰。高流量系统可调高。
• 推荐ttl_seconds=3600用于 domains where facts change。稳定知识库设为None。
• TF-IDF嵌入器无需外部依赖。生产中使用真实语义相似性，可替换为API嵌入器——一个接口方法，代码中有文档。

组件2：查询路由器

并非所有查询都值得相同模型。路由器将传入的查询根据复杂度分类，并自动在0.025ms内将其路由到合适的层级。

长度得分（权重：0.20） 正规化后的单词数量。5个单词的查询和50个单词的查询是不同问题。超过80个单词时达到饱和。

def _length_score(self, query: str) -> float:
    return min(len(query.split()) / 80.0, 1.0)

实体密度（权重：0.30） 大写字母、数字和专业标点符号与总单词数的比率。实体密度高的查询往往更具体且更复杂。

def _entity_score(self, query: str) -> float:
    tokens = query.split()
    if not tokens:
        return 0.0
    hits = sum(
        1 for t in tokens
        if (t[0].isupper() and len(t) > 1)
        or re.search(r"\d", t)
        or re.search(r"[:>/%]", t)
    )
    return min(hits / len(tokens), 1.0)

推理深度拥有最高的权重（0.50）。它由推理相关关键词计算，如“比较”、“对比”、“分析”、“为什么”、“权衡”、“设计”和“架构”。两个匹配即可达到满分。

REASONING_KEYWORDS: frozenset[str] = frozenset({
    "compare", "contrast", "analyze", "why", "trade-off",
    "design", "architecture", "failure mode", "evaluate",
    "relationship between", "when should", "how should", ...
})

def _reasoning_score(self, query: str) -> float:
    q_lower = query.lower()
    hits = sum(1 for kw in REASONING_KEYWORDS if kw in q_lower)
    return min(hits / 2.0, 1.0)

快速路径：事实oid检测

在评分之前，路由器检测事实oid模式，如“X是什么”，“定义X”和“列出X”。这些直接路由为SIMPLE，固定得分为0.10，跳过完整评分。

FACTOID_PATTERNS = [
    re.compile(r"^(what is|what are|who is|where is)\b", re.I),
    re.compile(r"^(define|definition of|meaning of)\b", re.I),
    re.compile(r"^(list|name|give me)\b.{0,40}$", re.I),
]

路由实践

从我的演示输出：

[查询 01] RAG是什么？
  等级：简单  (得分：0.10)  → gpt-4o-mini

[查询 04] 混合检索与纯向量搜索有何不同？
  等级：标准  (得分：0.306)  → gpt-4o

[查询 06] 比较代理RAG与标准RAG的成本和延迟权衡
  等级：标准  (得分：0.611)  → gpt-4o

“RAG是什么？”是一个典型的事实oid。它走快速路径并立即路由到廉价模型。“比较成本和延迟权衡…”仅从推理关键词得分0.611 — 它是一个多维度分析问题，确实需要更强的模型。

基准：大规模分布

在500个查询的现实混合中运行：

简单：81.0%  → gpt-4o-mini  ($0.000165/1K tokens)
标准：16.4%  → gpt-4o      ($0.005/1K tokens)
复杂：2.6%   → gpt-4.5     ($0.015/1K tokens)

总节省与全价相比：$3.41 (500 queries)
平均路由延迟：<0.025 ms

在基准查询混合中，81%的流量路由到低成本模型。路由器开销<0.025 ms/决策，在实际应用中可以忽略不计。

缺失的模型等级 — 生产安全

一个关键的生产修复：如果model_map中缺少等级，路由器不会因KeyError而崩溃。它会安全地退回到STANDARD等级：

# 合并提供的映射与默认值 — 缺失的键安全退回到默认值
self.model_map = {**DEFAULT_MODEL_MAP, **(model_map or {})}

这在部署到某些模型可用的环境时非常重要。系统会优雅地降级而不是崩溃。

组件3：Token预算层

缓存和路由器减少了LLM调用的数量和成本。Token预算层处理单次调用的Token分配，防止无声溢出，并记录Token使用情况。

这直接建立在我的上下文工程系统[7]的概念上，但通过显式成本跟踪每个槽位进行了扩展。

槽位分配

每个请求按照固定优先级顺序预留Token：

# 按优先顺序预留：固定 → 历史 → 文档 → 输出
ctx.budget.reserve("system_prompt", 200)        # 1. 永不谈判
ctx.budget.reserve_text("history", history)     # 2. 保持多轮连贯
ctx.budget.reserve_text("retrieved_docs", docs) # 3. 固定成本后的可压缩层
ctx.budget.reserve("output", min(512, ctx.budget.remaining()))  # 4. 生成空间

分配顺序是固定的。系统提示被视为开销，历史保持连贯，检索的文档在空间紧张时是可压缩层。文本槽位的Token数量估算为1 Token ≈ 4字符（英文散文）[6]。

如果顺序不正确，文档会在历史记录之前被丢弃。预算强加器会明确执行此行为。

槽位成本跟踪

每次预留都会记录其成本：

self._slots[slot_name] = SlotUsage(
    name=slot_name,
    reserved_tokens=granted,
    cost_usd=granted * self._cost_per_token,
)

生成后，记录实际值：

ctx.record_actual(actual_tokens=620, cost_usd=0.0031) record_actual是幂等的。重复调用在警告后被忽略，防止在支出账本中双重计数。

负Token保护

一个看似微不足道但很重要的生产修复：

def reserve(self, slot_name: str, tokens: int) -> int:
    if tokens <= 0:
        logger.debug("reserve(%s, %d) — 非正Token数被拒绝", slot_name, tokens)
        return 0

如果上游某处计算错误并传递负Token数，预算不会变为负数并破坏所有后续计算。它记录并返回0。

组件4：CostLedger和CircuitBreaker

这是保护系统免受生产噩梦的缺失层： runaway cost。

生产盲点

你向RAG代理添加工具使用。代理进入重试循环 — 工具调用失败，代理重试，重试失败，再重试。每个循环是一个完整的LLM调用，成本高昂。循环在夜间运行6小时，你在睡觉。

没有断路器，你就会在账单上 woke up。

有了断路器，系统在达到你的每小时阈值后会自动 throttling 或者 block。

CostLedger: 滚动支出可见性

class CostLedger:
    def record(self, cost_usd, tokens, model_tier, request_id=""):
        event = SpendEvent(timestamp=time.time(), cost_usd=cost_usd, ...)
        with self._lock:
            self._events.append(event)
            self._total_lifetime_usd += cost_usd
            self._prune()  # 移除 24 小时前的事件

    def hourly_spend(self) -> float:
        return self._window_spend(3600)

    def daily_spend(self) -> float:
        return self._window_spend(86400)

账本维护一个滑动窗口的支出事件。_prune() 移除 24 小时前的事件，保持内存bounded。通过 RLock 实现线程安全。

CircuitBreaker: 三种状态 [4, 5]

RAG 的断路器——停止 runaway 成本，安全恢复，并在压力下保持 LLM 系统稳定。作者提供图像

CLOSED    → 正常操作。所有请求通过。
OPEN      → 阈值被突破。请求被阻止或降级。
HALF_OPEN → 冷却时间结束。允许一个探针请求测试恢复。

def _check_and_trip(self) -> None:
    if self.ledger.hourly_breach() or self.ledger.daily_breach():
        self.breaker.trip()

这在每个请求后自动运行。当每小时或每天的支出超过你的限制时，断路器打开。在 cooldown_seconds 后，它过渡到 HALF_OPEN 并允许一个探针。如果探针成功，它关闭。如果失败，它重新打开。

降级 vs 阻止

两种生产模式：

enforcer = BudgetEnforcer(
    hourly_limit_usd=5.0,
    daily_limit_usd=50.0,
    downgrade_on_breach=True,   # 渐进式降级
)

downgrade_on_breach=True：当断路器打开时，请求被路由到便宜的模型而不是被阻止。用户得到降级的质量，而不是错误。对于大多数生产系统，这是正确的选择。

downgrade_on_breach=False：请求被完全阻止，带有备用消息。对于成本关键系统，错误答案比没有答案更糟糕，使用这种方法。

假阳性风险——一个诚实的警告

这是文章必须解决的边缘情况。从我的基准测试：

严格阈值（每小时阈值=$0.001）:
  → {'allowed': 0, 'downgraded': 0, 'blocked': 10}
  → 10/10 合法请求被阻止

合理的阈值（每小时阈值=$5.00）:
  → {'allowed': 10, 'downgraded': 0, 'blocked': 10}
  → 等等：那不对。

合理的阈值（每小时阈值=$5.00）:
  → {'allowed': 10, 'downgraded': 0, 'blocked': 0}
  → 10/10 请求正确服务

一行配置。灾难性的差异。

将 hourly_limit 设置得太低，你会阻止自己的生产流量。规则：将你的限制设置为预期峰值的 2-3 倍，而不是平均值。 平均支出是当一切正常的时候 things cost。限制保护 against 波动。

从基准输出：将 hourly_limit 设置为预期峰值的 2-3 倍 —— 不是平均值。使用 downgrade_on_breach=True 以渐进式降级而不是阻止用户。

全 Pipeline 连接在一起

class ProductionRAGPipeline:
    def __init__(self):
        self.cache = SemanticCache(threshold=0.75, ttl_seconds=3600)
        self.router = QueryRouter(simple_threshold=0.25, complex_threshold=0.65)
        self.enforcer = BudgetEnforcer(
            hourly_limit_usd=5.0,
            daily_limit_usd=50.0,
            per_request_limit_usd=0.10,
            downgrade_on_breach=True,
        )

    def query(self, user_query: str, retrieved_context: str = "") -> dict:
        # 步骤 1：缓存查找
        cached = self.cache.get(user_query)
        if cached is not None:
            return {"response": cached, "source": "CACHE HIT", "cost_usd": 0.0}

        # 步骤 2：路由到模型层
        routing = self.router.route(user_query)

        # 步骤 3：标记预算 + 成本执行
        with self.enforcer.request(
            model_tier=routing.tier.value,
            estimated_tokens=500,
        ) as ctx:
            if not ctx.allowed:
                return {"response": ctx.fallback_response, "source": "BLOCKED"}

            ctx.budget.reserve("system_prompt", 200)
            ctx.budget.reserve_text("history", "...")
            ctx.budget.reserve_text("retrieved_docs", retrieved_context)
            ctx.budget.reserve("output", min(512, ctx.budget.remaining()))

            response, tokens, cost = call_llm(user_query, ctx.model_tier)
            ctx.record_actual(actual_tokens=tokens, cost_usd=cost)

        # 步骤 4：缓存以供将来使用
        self.cache.set(user_query, response)
        return {"response": response, "cost_usd": cost, "tier": routing.tier.value}

流程是：首先缓存。如果有命中，其他操作都不运行。然后路由选择最便宜的模型来处理查询。预算层跟踪标记，执行限制，并在需要时触发断路器。最后，结果被缓存，以便相同的查询成本为零。

• * *

演示实际展示了什么

运行整个 Pipeline 对 8 个演示查询（从我的实际输出）：

[查询 01] 什么是 RAG？
  来源：LLM 调用  |  层级：simple  |  模型：gpt-4o-mini
  成本：$0.000015    |  节省：$0.007417 vs 昂贵的模型

[查询 02] 什么是向量数据库？
  来源：CACHE HIT  |  节省：$0.0040  (LLM 调用避免)

[查询 06] 比较代理 RAG 的成本和延迟权衡... 源：LLM CALL | 等级：标准 | 模型：gpt-4o 分数：0.611 | 成本：$0.000790

[查询 07] 什么是 RAG？（重复） 源：CACHE HIT | 节省：$0.0040

运行总结： 总成本（8 个查询）： $0.001389 与天真方法相比节省的总成本： $0.047668 电路断路器： 关闭

查询 01 和查询 07 是同一个问题问了两次。在第二次出现时，缓存以 0.5 毫秒返回，不产生任何成本。这就是系统按设计工作的方式。

查询 06 是一个真正复杂的问題——它包含“比较”、“权衡”，并参考了两种架构。它得分为 0.611，路由到 gpt-4o，成本为 $0.000790。路由决策是正确的。

延迟免责声明： 所有延迟数字都是通过模拟 LLM 调用测量的。实际世界的延迟根据提供商和负载在 200-800 毫秒之间。缓存命中的延迟保持在约 4 毫秒左右。

基准测试：实际节省的数字

下面的所有数字都是在我的机器上实际基准测试的结果（Python 3.12.6，Windows 11，CPU-only）。

语义缓存性能

运行的查询：           200
命中率：              98.5%
平均命中延迟：        约 4 毫秒
平均未命中延迟：       约 4-5 毫秒
p95 命中延迟：        约 5-7 毫秒
节省的成本（200 个查询）：    $0.788

98.5% 的命中率来自经过几个小时流量训练的温暖缓存。冷启动命中率通常从约 20-30% 开始，并随着缓存的填充而改善。

查询路由器分布

运行的查询：           500
简单：                81.0%  → gpt-4o-mini
标准：                16.4%  → gpt-4o
复杂：                2.6%  → gpt-4.5
总节省：           $3.41
平均路由延迟：   <0.025 毫秒

81% 的查询路由到便宜的模型。路由步骤每请求添加的延迟不到 0.025 毫秒，并在大规模上产生可测量的成本节省。

规模比较：天真 vs 优化

对于成本模型， baseline 架构假设最坏情况，完全依赖 GPT-4.5 级模型，平均每个请求 800 个标记。在大规模上，优化的系统假设保守的 28% 语义缓存命中率，并将大约 62% 的传入请求路由到简单且低成本的模型。

规模            天真/天   优化/天    节省    每月节省
100 个请求/天       $1.20      $0.18      84.6%         $30
1,000 个请求/天     $12.00     $1.71      85.7%         $309
10,000 个请求/天   $120.00    $17.00      85.8%        $3,090

节省的百分比在 1,000 个请求/天以上稳定在约 85.8%。低于这个数字，管道的固定开销（嵌入生成、路由计算）开始相对于节省成本变得重要。

真实的设计决策

TF-IDF vs 句子变换器

缓存使用纯 Python TF-IDF 嵌入器 — 没有 PyTorch，没有 sentence-transformers，也没有 Windows 上的后台线程。TF-IDF 匹配共享的标记，而不是语义意义。

对于用不同词语表达的同一查询（“什么是 RAG？” vs “定义检索增强生成”），TF-IDF 相似度将低于句子变换器相似度。如果您的用户倾向于重新phrase而不是重复，命中率将低于基准测试所示。

要交换到真正的语义嵌入器 — 一个接口方法：

class OpenAIEmbedder:
    def fit(self, texts): pass
    def embed(self, text):
        import openai
        r = openai.embeddings.create(model="text-embedding-3-small", input=text)
        return r.data[0].embedding

将其传递给 SemanticCache，其他一切不变。

路由阈值是经验数据

simple_threshold=0.25 和 complex_threshold=0.65 的默认值是基于 RAG 领域查询集校准的。不同的领域，如法律、医疗或客户服务，需要不同的阈值。

路由分布（81/16/2.6）反映了 RAG 定向的查询组合。客户服务系统强烈倾向于 SIMPLE 查询，而研究型助手则有更高的 COMPLEX 查询比例。

CostLedger 没有持久化

CostLedger 是严格内存化的。如果进程重启，您的支出历史将重置。在实践中，这意味着每小时和每天的速率限制只在单个进程的生命周期内提供保护。

如果您正在迁移到生产环境，使用多个工作进程或频繁的容器重启，您将需要将此分类账支持 Redis 或轻量级数据库。接口本身——record()、hourly_spend() 和 daily_spend()——是专门解耦的，您可以交换存储层而无需重写应用程序逻辑。

延迟数字是模拟的

对数字进行快速现实检查：演示显示的延迟为 0.09-1.05 毫秒。这些反映了具有模拟 LLM 调用的核心管道开销，而不是实际的 API 延迟。在生产中，实际的 LLM 调用将根据您的提供商、模型选择和当前网络负载添加 200-800 毫秒。其余指标完全是真实的。缓存命中延迟（约 4 毫秒）是真实的。路由决策延迟（不到 0.025 毫秒）是真实的。预算执行开销是微不足道的。模拟的唯一部分是到 LLM 提供商的实际往返。

这 NOT 是

这 NOT 是检索质量改进。 如果您的基础 RAG 系统正在检索错误的文档，这一层不会修复。对于检索质量、重新排名和上下文压缩，请参阅前一篇文章中讨论的上下文工程层。

这 NOT 是延迟优化层。 虽然缓存在命中时大大减少了延迟，但整体管道在缓存未命中时添加了虽小但不可忽视的开销。

这 NOT 是 LLM 可观察性的替代方案。 CostLedger 作为 guardrail 用以跟踪和控制支出，但您仍然需要生产环境中的强大日志、跟踪和监控工具。这一层提供了成本可见性，而不是全面的可观察性。

综合成本感知生产层

RAG 系统在质量方面存在问题。 already 有很多工作针对这个问题进行了研究。检索召回、重新排名和上下文质量都已得到广泛研究。

但 RAG 系统在成本方面也存在问题。大多数面向生产的写作重点在于检索质量。这个成本问题往往被忽视——当它发生时，是无声的。没有错误、没有警告，也没有报警。系统仍然完美地工作。只是账单一直在增长。

为了解决这个问题，我在本文中描述的架构在您的检索管道和 LLM 调用之间插入了四个独立的防御层：

• 语义缓存 — 在不到 4ms 内返回已知答案，$0 LLM 成本
• 查询路由器 — 将基准流量的 81% 路由到最多可节省 90 倍成本的模型
• 标记预算 — 跟踪每个标记，防止无声溢出
• 电路断路器 — 在重试循环变为账单之前自动限制

Bottom line: 每天 10,000 次请求时，成本总共减少了 85.8%。在本评估设置中，这对应于每月估计节省的 3,090 美元，实现无需修改底层 baseline 模型且未 measurable 降响应质量。

最好的部分？系统用纯 Python 运行。没有重量级框架，没有 sentence-transformers，也没有庞大的外部依赖。它在所有平台上提供即时启动和清洁退出。

完整代码：[https://github.com/Emmimal/rag-cost-control-layer/](https://github.com/Emmimal/rag-cost-control-layer/)

RAG 给你正确的答案。

这给你正确的账单。

参考文献

[1] Lewis, P., Perez, E., Piktus, A., Petroni, F., Karpukhin, V., Goyal, N., Küttler, H., Lewis, M., Yih, W., Rocktäschel, T., Riedel, S., & Kiela, D. (2020). Retrieval-Augmented Generation for Knowledge-Intensive NLP Tasks. _Advances in Neural Information Processing Systems_, 33, 9459–9474. https://arxiv.org/abs/2005.11401

[2] OpenAI. (2026). OpenAI API 定价。https://openai.com/api/pricing/_(定价可能会更改；在实施时验证当前费率。)

[3] Pedregosa, F., Varoquaux, G., Gramfort, A., Michel, V., Thirion, B., Grisel, O., Blondel, M., Prettenhofer, P., Weiss, R., Dubourg, V., Vanderplas, J., Passos, A., Cournapeau, D., Brucher, M., Perrot, M., & Duchesnay, E. (2011). Scikit-learn: Python 中的机器学习。_Journal of Machine Learning Research_, 12, 2825–2830. https://jmlr.org/papers/v12/pedregosa11a.html_(TF-IDF 实现参考。)

[4] Fowler, M. (2002). _企业应用程序架构模式_. Addison-Wesley. _(电路断路器模式.)_

[5] Nygard, M. (2007). _发布它！设计和部署生产就绪软件_. Pragmatic Bookshelf. _(电路断路器设计；本实现中使用的原始公式。)

[6] OpenAI. (2023). 使用 tiktoken 计数标记。https://github.com/openai/tiktoken_(标记估算参考：1 标记 ≈ 4 个字符（英文散文）。)

[7] Alexander, E. P. (2026). RAG 不够 — 我构建了使 LLM 系统工作的缺失上下文层。_数据科学世界_. https://towardsdatascience.com/rag-isnt-enough-i-built-the-missing-context-layer-that-makes-llm-systems-work/_(交叉引用：上下文质量层；本文解决的是成本层。)

[8] Bang, Z., et al. (2023). GPTCache：一个用于 LLM 应用的开源语义缓存，实现更快的答案和成本节省。https://github.com/zilliztech/GPTCache

披露

本文中的所有代码都是我原创的作品，是原创工作，使用 Python 3.12.6，在 Windows 11 上开发和测试，仅使用 CPU，没有使用 GPU。系统不使用任何外部 ML 库——没有 PyTorch，没有 sentence-transformers，没有 numpy。所有组件仅使用 Python 标准库运行。

基准数字是我在本地机器上实际运行系统的结果，通过克隆存储库并运行 demo/demo.py 和 benchmarks/run_benchmarks.py 完全可再现。演示使用了一个模拟的 LLM 调用——LLM 响应的延迟 (0.09ms–1.05ms) 反映了模拟管道；实际世界的 LLM API 延迟根据提供商和负载在 200-800ms 之间。缓存命中延迟 (~4ms) 和路由延迟（小于 0.025ms）是从实际 Python 实现中测量的。

所有计算中使用的每 1K 标记成本：gpt-4o-mini ($0.000165)，gpt-4o ($0.005)，gpt-4.5 ($0.015)。这些反映了撰写本文时的公开定价，可能会更改。在使用这些数字进行预算规划之前，请在 https://openai.com/api/pricing/ 验证当前费率。

我与 OpenAI、Anthropic 或本文提到的任何其他公司或工具没有财务关系。