SERP API + Redis 缓存层:4 种方案对比与选型

背景调 serpbase 直查 SERP,1000 次调用成本 $0.30,看起来不多。但 LLM 应用 1 个对话可能查 3-5 次 SERP(扩展 query 验证 重写),月调用量是用户数的 10-50 倍。加 Redis 缓存能降 30-60% 调用,但缓存策略错了反而坑多。4 种方案对比方案 1:纯 cache(简单)importredis,hashlib,json rredis.Redis()defsearch(query):keyfserp:{hashlib.md5(query.encode()).hexdigest()}cachedr.get(key)ifcached:returnjson.loads(cached)datacall_serp(query)r.setex(key,300,json.dumps(data))# 5 分钟 TTLreturndata优点:简单,5 行代码缺点:TTL 选错 → 数据 stale;Cache 击穿 → 数据库雪崩方案 2:cache fallback(推荐)defsearch(query):keyfserp:{hash(query)}cachedr.get(key)ifcached:datajson.loads(cached)ifnotdata.get(stale,True):# 检查 stale 标记returndatatry:freshcall_serp(query)fresh[stale]Falser.setex(key,300,json.dumps(fresh))returnfreshexceptException:ifcached:returndata# fallback 用 stalereturn{organic:[],_stale:True}优点:故障兜底;缺点:代码增加 5 行方案 3:多层 cache(高 QPS)importtime L1{}# 进程内(1 秒 TTL)L2redis.Redis()# 共享(5 分钟 TTL)defsearch(query):ifqueryinL1andtime.time()-L1[query][ts]1:returnL1[query][data]cachedL2.get(fserp:{query})ifcached:datajson.loads(cached)L1[query]{data:data,ts:time.time()}returndata freshcall_serp(query)L1[query]{data:fresh,ts:time.time()}L2.setex(fserp:{query},300,json.dumps(fresh))returnfresh优点:L1 命中 50% 减少 Redis 往返,L2 共享给多进程缺点:进程内 L1 占用内存方案 4:stream cache(高实时)# 同时缓存 query ai_overview 多个 variantimportasynciofromcollectionsimportdefaultdictclassStreamCache:def__init__(self):self.cachedefaultdict(dict)# {query: {variant: data}}asyncdefget(self,query,varianten-US):ifqueryinself.cacheandvariantinself.cache[query]:returnself.cache[query][variant]dataawaitasync_call_serp(query,variantvariant)self.cache[query][variant]datareturndata cacheStreamCache()# 同一 query 不同 variant 缓存awaitcache.get(SERP API 价格,en-US)awaitcache.get(SERP API 价格,zh-CN)优点:同一 query 不同语言 / 地区共用缺点:内存占用大实测数据(我项目 30 天)方案命中率SerpBase 调用月成本实现复杂度1 纯 cache35%650 次 / 天$5.85★2 cache fallback38%620 次 / 天$5.58★★3 多层 cache60%400 次 / 天$3.60★★★4 stream cache55%450 次 / 天$4.05★★★★方案 3 多层 cache 命中率最高,月成本最低。4 个选型建议场景推荐中小流量 / MVP方案 1 纯 cache生产高可用方案 2 cache fallback高 QPS / 多 worker方案 3 多层 cache多语言 / 多 region方案 4 stream cache5 个工程细节1. Cache key 设计# 错:太短(碰撞)keyfserp:{query}# 对:含所有影响结果的参数importhashlib keyserp:hashlib.md5(f{query}|{gl}|{hl}|{num}.encode()).hexdigest()2. TTL 不要一刀切# 错:统一 TTLr.setex(key,300,data)# 对:不同类型不同 TTLifnewsinqueryortodayinquery:ttl60# 新闻类 1 分钟elifrankinqueryorpositioninquery:ttl3600# 排名类 1 小时else:ttl300# 普通 5 分钟3. 击穿防护(防 cache miss 雪崩)# 单个 key 失效时,只允许 1 个请求回源locksredis.Redis()defsearch_with_lock(query):ifdata:r.get(query):returndata lock_keyflock:{query}ifnotlocks.set(lock_key,1,ex10,nxTrue):# 10 秒锁# 其他请求在回源,稍等time.sleep(0.1)returnr.get(query)orsearch_with_lock(query)# 递归重试try:freshcall_serp(query)r.setex(query,300,json.dumps(fresh))returnfreshfinally:locks.delete(lock_key)4. 预热热点数据HOT_QUERIES[SERP API,SERP API 价格,SERP API vs,...]defwarmup():forqinHOT_QUERIES:ifnotr.exists(fserp:{q}):r.setex(fserp:{q},3600,json.dumps(call_serp(q)))schedule.every().day.at(03:00).do(warmup)# 凌晨跑5. 监控命中率hit_count0miss_count0defsearch_tracked(query):ifr.get(fserp:{query}):hit_count1else:miss_count1# ...returnr.get(fserp:{query})orcall_serp(query)# Prometheus metriccache_hit_ratehit_count/(hit_countmiss_count)ifcache_hit_rate0.30:# 命中率低alert(SERP cache hit rate too low, check key design)与 SerpBase 集成serpbase 失败 auto-refund 100% 触发,缓存层可以放心设长 TTL:# 不需要担心 stale 太久,serpbase 失败会退r.setex(key,3600,data)# 1 小时 TTL OK我项目跑 30 天:命中率 60%(方案 3)失败 auto-refund 0.3%月成本 $3.60实现复杂度中等小结SERP API Redis 缓存选型:90% 场景用方案 2(cache fallback):简单 兜底高 QPS 用方案 3(多层 cache):50% 额外命中多语言用方案 4(stream cache):不同 variant 共享方案 1 纯 cache 不推荐,fallback 必加(API 故障时退化)。我项目 30 天跑下来,多层 cache 5 分钟 TTL 预热 监控 4 件套,SerpBase 月成本从 $5.85 降到 $3.60,降 38%,且 0 故障。