从雪崩到 99.9%：RAG 向量检索限流缓存降级实战

问题背景：RAG 链路为什么比普通 CRUD 更脆弱

一次典型 RAG 请求要经过鉴权、查询改写、Embedding 向量化、向量库 ANN 检索、重排与 Prompt 拼装。真正容易拖垮系统的并不是大模型本身，而是向量检索这一段：Embedding 服务按 token 计费且有并发上限，向量库在索引合并、分片迁移时 P99 会突然抖动，而热门问题（如“退款流程是什么”）往往在同一时刻被成百上千个用户重复请求。

向量检索的稳定性风险有三个鲜明特征。第一，请求成本极不均匀，长 query 的 Embedding 耗时可能是短 query 的十倍；第二，失败具有传导性，检索超时会把 goroutine、连接和上游请求一起占满；第三，结果可缓存但又不能错缓存，租户隔离、模型版本、过滤条件任意一个维度漏掉，都会造成知识越权。

因此，RAG 服务端的稳定性设计不能只靠加机器，而要在入口处削峰、在热点路径上缓存、在依赖异常时优雅降级，形成三道防线。

方案对比：四种保护手段的取舍

手段	解决的问题	适用位置	主要代价
令牌桶限流	挡住突发流量，保护下游配额	API 入口、Embedding 客户端	超限请求被拒绝，需配合重试提示
并发信号量	限制同时飞行的昂贵调用	向量库与 Embedding 调用前	配置过小会牺牲吞吐
多级缓存	消除热点重复计算	query 向量、检索结果	键设计复杂，有一致性与越权风险
降级兜底	依赖故障时仍返回可用答案	检索失败、超时、熔断打开	召回质量下降，必须显式标记

实际经验是：限流决定“系统不被打死”，缓存决定“平时有多省钱”，降级决定“故障时用户看到什么”。三者缺一不可，且顺序上应先做缓存与超时，再做限流，否则上线初期会误杀大量正常请求。

稳定性架构：三道防线如何协同

实操步骤：用 Go 实现可运行的检索保护层

下面用 Go 标准库加 golang.org/x/sync/singleflight 实现一个生产可用的核心骨架，包含本地缓存、singleflight 热点合并、信号量限流和超时降级。为便于直接运行，向量检索与 Embedding 用接口模拟，真实项目中替换成 Redis、Milvus 或 pgvector 客户端即可。

先准备依赖：go mod init ragdemo，然后执行 go get golang.org/x/sync/singleflight。

package main

import (
	"context"
	"errors"
	"fmt"
	"sync"
	"time"

	"golang.org/x/sync/singleflight"
)

// ===== 核心领域模型 =====

type SearchRequest struct {
	TenantID string   // 租户，必须参与缓存键，防止知识越权
	Query    string   // 用户原始问题
	TopK     int      // 召回条数
	Filter   string   // 文档权限/分类过滤条件
	ModelVer string   // Embedding 模型版本，升级后避免命中旧向量
}

type Doc struct {
	ID      string
	Content string
	Score   float64
}

// Vectorizer 与 VectorStore 是外部依赖，真实项目接 Embedding API 与向量库
type Vectorizer interface {
	Embed(ctx context.Context, text string) ([]float32, error)
}
type VectorStore interface {
	Search(ctx context.Context, tenant string, vec []float32, topK int) ([]Doc, error)
}

// ===== 带 TTL 的本地缓存 =====

type cacheItem struct {
	docs      []Doc
	expireAt  time.Time
}

type RetrievalCache struct {
	mu    sync.RWMutex
	items map[string]cacheItem
	ttl   time.Duration
}

func NewRetrievalCache(ttl time.Duration) *RetrievalCache {
	c := &RetrievalCache{items: map[string]cacheItem{}, ttl: ttl}
	go c.gcLoop() // 生产环境也可改用 Redis 承载多实例共享缓存
	return c
}

// cacheKey 把所有影响结果的维度拼进键，缺一个都可能串租户
func cacheKey(req SearchRequest) string {
	return fmt.Sprintf("%s|%s|%s|%d|%s",
		req.TenantID, req.ModelVer, req.Filter, req.TopK, req.Query)
}

func (c *RetrievalCache) Get(key string) ([]Doc, bool) {
	c.mu.RLock()
	item, ok := c.items[key]
	c.mu.RUnlock()
	if !ok || time.Now().After(item.expireAt) {
		return nil, false
	}
	return item.docs, true
}

func (c *RetrievalCache) Set(key string, docs []Doc) {
	c.mu.Lock()
	c.items[key] = cacheItem{docs: docs, expireAt: time.Now().Add(c.ttl)}
	c.mu.Unlock()
}

func (c *RetrievalCache) gcLoop() {
	for range time.Tick(time.Minute) {
		now := time.Now()
		c.mu.Lock()
		for k, v := range c.items {
			if now.After(v.expireAt) { delete(c.items, k) }
		}
		c.mu.Unlock()
	}
}

// ===== 稳定性保护层 =====

type ProtectedRetriever struct {
	vectorizer Vectorizer
	store      VectorStore
	cache      *RetrievalCache
	sem        chan struct{} // 并发信号量，限制同时打到下游的请求数
	g          singleflight.Group
	fallback   []Doc
}

func NewProtectedRetriever(v Vectorizer, s VectorStore, maxConcurrent int) *ProtectedRetriever {
	return &ProtectedRetriever{
		vectorizer: v,
		store:      s,
		cache:      NewRetrievalCache(5 * time.Minute),
		sem:        make(chan struct{}, maxConcurrent),
		fallback: []Doc{{
			ID: "fallback-hotline",
			Content: "智能检索暂时繁忙，已为你转接常见问题入口或人工客服。",
		}}, 
	}
}

func (r *ProtectedRetriever) Search(ctx context.Context, req SearchRequest) ([]Doc, bool, error) {
	key := cacheKey(req)

	// 第一道：本地缓存，命中则完全不碰昂贵下游
	if docs, ok := r.cache.Get(key); ok {
		return docs, true, nil
	}

	// 第二道：singleflight 合并同一时刻的相同请求，防缓存击穿
	v, err, _ := r.g.Do(key, func() (any, error) {
		// double check，避免排队期间其他 goroutine 已回填
		if docs, ok := r.cache.Get(key); ok {
			return docs, nil
		}
		return r.callDownstream(ctx, req)
	})
	if err != nil {
		// 第三道：降级。返回兜底内容并标记 degraded，便于上层提示与监控
		return r.fallback, false, err
	}
	docs := v.([]Doc)
	r.cache.Set(key, docs)
	return docs, false, nil
}

func (r *ProtectedRetriever) callDownstream(ctx context.Context, req SearchRequest) ([]Doc, error) {
	select {
	case r.sem <- struct{}{}:
		defer func() { <-r.sem }()
	case <-ctx.Done():
		return nil, errors.New("获取检索令牌超时，已触发限流")
	}

	// 给外部依赖独立的硬超时，绝不允许无限等待拖垮整体
	callCtx, cancel := context.WithTimeout(ctx, 600*time.Millisecond)
	defer cancel()

	vec, err := r.vectorizer.Embed(callCtx, req.Query)
	if err != nil {
		return nil, fmt.Errorf("embedding 失败: %w", err)
	}
	docs, err := r.store.Search(callCtx, req.TenantID, vec, req.TopK)
	if err != nil {
		return nil, fmt.Errorf("向量检索失败: %w", err)
	}
	if len(docs) == 0 {
		return nil, errors.New("召回为空，转入降级话术")
	}
	return docs, nil
}

// ===== 模拟依赖与演示 =====

type fakeVectorizer struct{}
func (fakeVectorizer) Embed(ctx context.Context, text string) ([]float32, error) {
	select {
	case <-time.After(120 * time.Millisecond):
		return []float32{0.1, 0.2, 0.3}, nil
	case <-ctx.Done():
		return nil, ctx.Err()
	}
}

type fakeStore struct{}
func (fakeStore) Search(ctx context.Context, tenant string, vec []float32, topK int) ([]Doc, error) {
	return []Doc{{ID: "d1", Content: tenant + " 的退款将在 1-3 个工作日到账", Score: 0.93}}, nil
}

func main() {
	r := NewProtectedRetriever(fakeVectorizer{}, fakeStore{}, 50)
	req := SearchRequest{
		TenantID: "tenant-a", Query: "退款多久到账",
		TopK: 5, Filter: "public", ModelVer: "bge-m3-v1",
	}
	for i := 0; i < 2; i++ {
		docs, fromCache, err := r.Search(context.Background(), req)
		fmt.Printf("第%d次: 缓存=%v 降级=%v 内容=%s\n",
			i+1, fromCache, err != nil, docs[0].Content)
	}
}

运行后第一次请求会调用模拟的 Embedding 与向量库，第二次直接命中缓存。把 fakeVectorizer 的延迟改成 800ms，即可观察到 600ms 硬超时后返回兜底文档，而不是把调用方挂死。

限流参数应该怎么定

令牌桶或信号量的容量不要拍脑袋，推荐用下游公式反推：并发上限 = 下游允许 QPS × 平均耗时秒数 × 安全系数(0.7)。例如 Embedding 配额为 200 QPS、平均耗时 150ms，则信号量约为 21。入口全局限流按租户再分一层，避免单个大客户耗尽全局配额。被限流的请求应返回可重试的 429，并带上抖动退避，而不是让客户端立刻重放形成二次风暴。

缓存放在哪一层最划算

单机进程内缓存适合 Top 热点，延迟最低但多实例不共享；Redis 适合跨实例共享与容量扩展，但多一次网络往返。常见组合是“本地 LRU 防热点 + Redis 做二级缓存 + singleflight 防击穿”。缓存对象优先选择 Embedding 向量和最终检索结果：前者省掉按 token 计费的调用，后者直接省掉向量库查询。结果缓存 TTL 建议 3 到 10 分钟，并在知识库更新时按 tenant + docID 主动失效。

避坑总结：三个生产环境真实教训

第一，缓存键漏掉租户或过滤条件，导致知识越权。 现象是 A 公司用户能检索到 B 公司的制度文档，根因常是只拿 query 做 key，或把 filter 放在缓存之后才生效。修正方法是把租户、模型版本、TopK、权限过滤全部纳入键，并在代码评审中把 cacheKey 当作安全边界，增加跨租户命中的单测。

第二，只配置超时却不限制并发，故障时照样雪崩。 典型现象是向量库 P99 从 50ms 飙到 2s，虽然单次 1s 后超时，但 goroutine 和连接仍在堆积，最终 OOM。超时只能控制单请求寿命，必须叠加信号量或令牌桶控制在途请求数，并对“获取令牌超时”单独打点，这往往是容量告警的最早信号。

第三，降级静默生效，用户和监控都不知情。 有些服务故障时返回兜底话术却记录为成功，导致召回质量长期下降无人发现。降级结果必须携带 degraded=true 标记，响应中适度提示“当前为简化结果”，同时上报降级率、缓存命中率、限流拒绝数和下游 P99。建议把降级率超过 5% 设为告警阈值，而不是只盯错误率。

此外还要避免两个小坑：不要缓存空结果太久，否则新文档入库后用户持续搜不到，空结果 TTL 应缩短到 30 秒以内；不要在缓存击穿时让所有实例同时回源，singleflight 只能合并单实例热点，多实例还需 Redis 锁或请求合并。把限流、缓存、降级与可观测性一起交付，RAG 向量检索才能真正从“能跑”走向“敢上量”。

免责声明

本工具与文章内容仅用于学习测试场景，优化建议仅供参考，请勿直接用于生产环境，由此产生的风险自行承担。