AnySearch 负责回答“这次查询值不值得真的发出去”，本地缓存负责回答“同一个语义的查询第二次来能不能直接复用结果”。两者的分界线建议放在归一化后的查询键上：键对齐了，重复调用才有地方被拦下来；键没对齐，缓存只是让旧结果更快地返回，同一问题照样会反复打到检索侧。因此判断顺序是先看日志确认重复是否存在，再定键和过期，最后用实时结果与命中结果做比对，而不是一上来就在 AnySearch 前面挂一层缓存。

建议先用调用日志按归一化查询聚合，确认重复触发次数、相邻间隔和重复结果占比，再决定缓存键的字段集与过期时间。本地缓存只承担去重与复用，不承担检索质量；写入前要判断结果有效性，空结果与上游报错分开处理，并用同一查询的实时返回与命中返回逐项比对来验证缓存没有带进旧结构。

在调用日志里统计重复触发

先别急着加缓存，先把“重复”量出来。建议在 AnySearch 的调用出口处落一条结构化日志，每行至少包含：时间戳、原始查询语句、归一化后的查询语句、top_k、过滤条件、是否命中缓存、结果条数、结果内容哈希、上游状态码。这些字段都能从代码里直接打出来，不依赖任何外部工具。

# 每行一条 JSON，字段示意，按实际字段名替换
# {"ts":1710000000,"query_norm":"xxx","top_k":8,"filters":"doc","hit":false,"count":6,"result_hash":"ab12","status":"ok"}

# 1) 同一归一化查询被调用了多少次
jq -r '[.query_norm, .top_k, .filters] | @tsv' calls.log \
  | sort | uniq -c | sort -nr | head -50

# 2) 相邻两次调用的间隔（秒），看重复是否集中在短时间窗
jq -r '[.query_norm, .ts] | @tsv' calls.log \
  | sort -k1,1 -k2,2n \
  | awk '{ if ($1==q) print $1, $2-p; q=$1; p=$2 }' | head -50

# 3) 每个查询产生过几种不同结果，用于估算重复结果占比
jq -r '[.query_norm, .result_hash] | @tsv' calls.log \
  | sort -u | cut -f1 | uniq -c | sort -nr | head -50

看三个数：同一条查询的调用次数、相邻间隔集中在几分钟还是几小时、同一查询的不同结果哈希有多少个。间隔很短而结果哈希只有一个，说明是典型的重复触发，本地缓存可以直接解决；间隔很长或结果哈希每次都不一样，多半是时效性查询，缓存只能设很短的时间，甚至不该缓存。还要留意分页和追问改写：它们可能被归一化规则错误地并成同一条，这类“假重复”会直接导致缓存返回错的结果。

定缓存键的拼法

缓存键要保证“相同语义命中同一条、不同语义不互相污染”。归一化规则建议保守：去首尾空白、把连续空白折叠成一个、统一全角半角、英文转小写、去掉尾部多余的标点。不要做同义词替换、不要删掉否定词、不要去掉“今天”“最新”这类时效限定词——这些一旦删掉，不同语义的查询就撞到同一个键上。

进键的字段建议是：查询归一化结果、检索库或索引标识、语言、top_k、过滤条件规范化后的指纹、缓存结构版本号。结构版本号很重要，返回字段一旦调整，改一个版本号就能整体失效旧数据。不进键的字段：请求 ID、会话 ID、时间戳、分页游标（分页应单独成键或单独走实时），以及用户身份——除非确实要做租户隔离，那就用键前缀区分而不是混进指纹。

# 键的拼法示意，字段按环境替换
key = f"{schema_ver}:{index}:{lang}:{top_k}:{fingerprint}"
# fingerprint = sha1(归一化查询 + 规范化后的 filters)
# schema_ver 变更即整体失效，便于字段结构调整

判定标准很简单：把两条你认为语义相同的查询喂进拼键函数，键必须相等；把两条语义不同的查询喂进去，键必须不等。这两组用例值得在改动归一化规则时固定下来。

写缓存读写与过期判断的骨架

下面是通用骨架，CACHE 可以是 Redis、进程内字典或一张表，函数签名和判断顺序保持即可。

CACHE = {}  # 占位：替换为实际存储

 def get_cached(key):
    item = CACHE.get(key)
    if not item:
        return None, "miss"
    if item["expire_at"] <= now():
        return None, "expired"
    return item["value"], "hit"

def is_writable(result, status):
    if status != "ok":                      # 上游超时或报错，不写
        return False
    if not result or result.get("count", 0) == 0:
        return False                        # 空结果走单独的负缓存分支
    need = ("title", "url", "snippet")
    return all(all(f in it for f in need) for it in result["items"])

def set_cached(key, result, status, ttl, jitter=0):
    if not is_writable(result, status):
        return False
    CACHE[key] = {"value": result,
                  "expire_at": now() + ttl + jitter}
    return True

过期时间建议按查询类型分级：不含时效词的稳定查询可以给较长的 TTL，含“最新”“今天”“最近”这类词的查询给很短的 TTL 或直接不进缓存；再加一点随机抖动，避免一批键在同一秒集体失效后又同时打回检索侧。TTL 的具体数值需要结合自身查询分布确认，先给一个保守值观察日志里的命中与回源比例，再调整。

验证命中结果与实时结果一致

缓存上线前，固定挑几条代表性查询做对照：第一次强制走实时并记录完整返回，把结果写入缓存后再查一次，这次预期命中，把两次返回按字段逐项比对。比对项包括总条数、每条结果的 url 与 title 是否非空、snippet 是否被截断、排序是否一致、过滤条件是否真的生效。

# 比对骨架：同一查询的实时结果 live 与命中结果 cached
def diff_fields(live, cached):
    if live["count"] != cached["count"]:
        print("count 不一致", live["count"], cached["count"])
    for i, (a, b) in enumerate(zip(live["items"], cached["items"])):
        for f in ("title", "url", "snippet"):
            if bool(a.get(f)) != bool(b.get(f)):
                print("字段完整度差异", i, f)
    return True

如果只有条数一致但字段缺失，通常是写入时缓存了降级结果；如果条数不一致，多半是键的粒度过粗，或者过滤条件没参与拼键。另一个可长期使用的检查是回到调用日志，比对同一查询在 hit=false 与 hit=true 两种情况下的 result_hash，两者不同就意味着缓存键把不同语义的结果并到了一起。

处理空结果与缓存穿透

空结果要不要缓存，取决于空的原因。上游明确返回正常但确实没有匹配内容，建议做短时间的负缓存，把这个空结果也写进键里，避免同一个不存在的查询反复打到检索侧；上游超时、限流或返回错误，不要缓存，否则一次抖动会在整个 TTL 内被当成“没有结果”。负缓存的 TTL 通常比正常结果短得多，需要结合自身查询分布确认。

并发穿透也值得处理：同一个键在同一时刻有多个请求进来时，只放一个请求去实时检索，其余请求等待并复用它的结果。降级返回上，如果缓存里有已过期的条目而上游暂时不可用，可以返回旧结果并显式标记 stale=true，让调用方知道这份内容不是最新的，而不是在无提示的情况下把旧结果当成实时结果交出去。