Caching-Cost-aware-RAG

高并发下大量近重复问题反复走嵌入和 LLM，费用与 P95 一起坏。根因是每次请求都当冷启动。本方法多层缓存加预算路由。TTL 与新鲜度冲突，命中率不能单独当 KPI。

本文属于 RAG 工程框架中的「5 在线运营与成本治理」环节，聚焦「Caching Cost aware RAG」方法。

定位

维度	内容
角色	多级缓存与预算路由
输入 → 输出	query → 缓存命中值或主链结果
默认组合	Redis；语义层可用 GPTCache
何时不用	查询几乎不重复

核心机制

$$
\mathbb{E}[\mathrm{cost}]=p_{\mathrm{miss}}\cdot c_{\mathrm{infer}}+c_{\mathrm{cache}}
$$

实现路径与心智：先查精确/语义缓存，命中直接返；未命中走检索+生成并按 TTL 写入。底层心智：期望费用是未命中率乘推理成本。TTL 过长等于用过期方法回答新预印本。

优缺点

优点：P95 与费用可降 30%–70%（视访问分布）。
缺点：命中策略与新鲜度难两全。

契约与走通样例

输入

1	{"query": "GAPDH siRNA 处理多久"}

中间量

语义缓存命中近邻问句，返 48 h 答案。文档刚勘误为 24 h 且 TTL=24 h → 过期命中。应缩短 TTL 或按 doc version 作 cache key。

输出

1	{"hit": true, "ttl_s": 3600, "key": "v3:gapdh-sirna-duration"}

社区实现

Redis；GPTCache。风险：只按 query 文本做 key，忽略索引版本。

工程落地

最小可运行示例

复制为 .py 后直接运行（仅标准库）。生产缓存换 Redis；key 必须含 index_version。

"""带索引版本的查询缓存，避免脏读旧向量。"""
from __future__ import annotations

import hashlib
import time
from dataclasses import dataclass, field


@dataclass
class LocalTtlCache:
    """进程内 TTL 缓存。生产对应 Redis SETEX。"""

    store: dict[str, tuple[float, str]] = field(default_factory=dict)

    def key(self, index_version: str, query: str) -> str:
        raw = f"{index_version}\t{query.strip().lower()}"
        return hashlib.sha256(raw.encode()).hexdigest()

    def get(self, key: str) -> str | None:
        hit = self.store.get(key)
        if not hit:
            return None
        expire_at, value = hit
        if expire_at < time.time():
            self.store.pop(key, None)
            return None
        return value

    def set(self, key: str, value: str, ttl_s: int) -> None:
        self.store[key] = (time.time() + ttl_s, value)


def cost_aware_answer(query: str, cache: LocalTtlCache, index_version: str, ttl_s: int, pipeline) -> tuple[str, bool]:
    """输出 (答案, 是否命中)。"""
    key = cache.key(index_version, query)
    hit = cache.get(key)
    if hit is not None:
        return hit, True
    ans = pipeline(query)
    cache.set(key, ans, ttl_s)
    return ans, False


if __name__ == "__main__":
    cache = LocalTtlCache()
    calls = {"n": 0}

    def pipeline(q: str) -> str:
        calls["n"] += 1
        return "20 nM for 48 h"

    print(cost_aware_answer("siRNA 浓度？", cache, "idx-v1", 3600, pipeline))
    print(cost_aware_answer("siRNA 浓度？", cache, "idx-v1", 3600, pipeline), "pipeline_calls", calls["n"])
    print(cost_aware_answer("siRNA 浓度？", cache, "idx-v2", 3600, pipeline), "pipeline_calls", calls["n"])

参数

参数	起点	影响
TTL	小时级 FAQ；预印本分钟级	过长过期
key	含 index_version	不含则脏读

失效—信号—螺丝

命中率高但投诉勘误：螺丝：TTL 跟 TTS，key 加 version。
几乎不命中：螺丝：语义缓存或规范化问句。
预算路由把贵链打满：螺丝：先缓存再精排。

规模（100 篇生物学 PDF）

Redis 4–32 GB（视 QPS）。与篇数无固定线性关系，取决于访问分布。