ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

生产级RAG知识库与Agent网关:混合检索、RRF融合与缓存限流实战

生产级RAG知识库与Agent网关:混合检索、RRF融合与缓存限流实战 1. 生产级知识库与 Agent 网关的整体设计思路1.1 为什么单机 RAG Demo 一上生产就崩我最早做知识库是从一个本地脚本开始的把 PDF 丢进目录切块灌进向量库接一个 OpenAI 的接口跑起来效果惊艳。但真把它放到生产环境问题一个接一个冒出来。最典型的是三类召回不稳定、成本失控、多 Agent 抢资源。召回不稳定的根源在于纯向量检索对精确匹配这件事天然不擅长。用户问config.toml 里 model provider openai not found 怎么修向量检索会把语义相近但完全无关的段落排到前面因为 embedding 抓的是语义氛围不是字面命中。这时候就需要BM25这种基于词频和逆文档频率的经典检索算法来兜底——它对专有名词、报错信息、代码标识符的命中能力是向量检索比不了的。成本失控则来自两个地方一是 embedding 调用量随文档量线性增长二是 Agent 网关如果没有做请求合并和缓存同一个问题会被不同 Agent 重复打到大模型上。我实测过一个中等规模的知识库没做缓存前每月调用费用是做完之后的 3.7 倍。多 Agent 抢资源的问题更隐蔽。当你有多个 Agent问答、摘要、检索增强、工具调用共用一个网关时如果没有优先级和限流一个批量摘要任务能把问答接口的响应时间从 800ms 拖到 6 秒以上。1.2 混合检索 网关分层我最终选定的架构踩了几轮坑之后我把架构定成了两层检索层做混合召回网关层做统一调度。检索层用 BM25 向量双路召回再用 RRFReciprocal Rank Fusion做融合排序。选 RRF 而不是加权求和是因为加权求和需要调权重而不同查询类型精确报错 vs 模糊概念最优权重完全不同调参成本极高。RRF 只依赖排名鲁棒性好得多公式也简单RRF_score(d) Σ 1 / (k rank_i(d))其中 k 一般取 60rank_i(d) 是文档 d 在第 i 路召回中的排名。这个 k 值不是拍脑袋来的原始论文里验证过 60 在多数场景下表现稳定我实测下来 40 到 80 之间差异不大直接用 60 就行。网关层则负责请求路由、模型选择、限流、缓存、可观测性。这一层的关键设计原则是把 Agent 和具体模型解耦。Agent 只说我要一个能处理长上下文的模型网关决定用哪个。这样换模型、加模型、降级都不需要动 Agent 代码。1.3 关键组件选型与取舍组件候选方案我的选择理由向量库FAISS / Milvus / Qdrant / pgvectorQdrant支持过滤向量混合查询运维比 Milvus 轻BM25rank_bm25 / Elasticsearch / TantivyTantivy纯 Rust无 JVM 依赖内存占用低网关自研 / LiteLLM / One-API自研 LiteLLM 参考需要深度定制限流和缓存策略切块固定长度 / 语义切块 / 递归切块递归 语义边界代码和文档混排场景下效果最好选 Qdrant 而不是 pgvector是因为我需要先按 metadata 过滤再向量检索的能力pgvector 在过滤选择性高的时候性能下降明显。选 Tantivy 而不是 Elasticsearch是因为我不想为了一个 BM25 引入一整套 JVM 生态Tantivy 单机就能扛住千万级文档。2. 混合检索的核心细节与实操要点2.1 BM25 到底在算什么为什么它对报错信息这么灵BM25 的核心思想是一个词在当前文档里出现得越多文档越相关但这个词如果在所有文档里都很常见它的区分度就低权重应该降低。公式长这样score(D, Q) Σ IDF(qi) * (f(qi, D) * (k1 1)) / (f(qi, D) k1 * (1 - b b * |D| / avgdl))k1 控制词频饱和速度一般取 1.2 到 2.0b 控制文档长度归一化取 0.75。这两个值我基本没动过默认就够用。关键在于IDF 部分。像 openai 这种词如果知识库里到处都是IDF 就低但 model provider not found 这种组合在正常文档里几乎不出现IDF 极高一旦命中就能把相关文档顶到最前面。这就是为什么 BM25 对报错信息、配置项、函数名特别灵。实操中有一个坑分词。中文和代码混排的文档用默认的英文分词器会把中文整段当成一个 tokenBM25 直接失效。我的做法是中文用 jieba 分词代码标识符用正则单独抽出来作为独立 token两者合并成最终的 token 流。import jieba import re def tokenize_mixed(text): # 先抽出代码标识符含下划线、点、驼峰 code_tokens re.findall(r[A-Za-z_][A-Za-z0-9_.]*, text) # 去掉代码部分后再做中文分词 text_no_code re.sub(r[A-Za-z_][A-Za-z0-9_.]*, , text) cn_tokens list(jieba.cut(text_no_code)) # 代码 token 统一小写中文保持原样 return [t.lower() for t in code_tokens] [t for t in cn_tokens if t.strip()]注意代码 token 一定要小写化否则 OpenAI 和 openai 会被当成两个词召回率直接砍半。2.2 向量检索的切块策略递归 语义边界切块是 RAG 里最容易被低估的环节。我见过太多人直接用固定 512 token 切结果一个完整的函数被切成两半检索出来的片段根本没法用。我的策略是递归切块 语义边界优先。具体来说先按文档结构切Markdown 标题、代码块边界、段落如果某块超过 800 token再按句子边界切如果还超才硬切块大小我定在 400 到 800 token 之间重叠 80 token。为什么是这个范围太小了上下文不够模型答不出来太大了检索精度下降而且浪费上下文窗口。400 到 800 是我在多个知识库上试出来的甜点区。重叠 80 token 是为了防止关键信息正好落在切块边界上。这个值不用太大80 足够覆盖大多数句子。def recursive_chunk(text, max_tokens800, overlap80): # 第一层按 Markdown 标题切 sections split_by_heading(text) chunks [] for sec in sections: if count_tokens(sec) max_tokens: chunks.append(sec) else: # 第二层按段落切 for para in split_by_paragraph(sec): if count_tokens(para) max_tokens: chunks.append(para) else: # 第三层按句子切带重叠 chunks.extend(split_by_sentence(para, max_tokens, overlap)) return chunks2.3 RRF 融合排序的实现与调参双路召回之后两边的分数尺度完全不同——BM25 分数可能是 0 到 30向量相似度是 0 到 1。直接相加没有意义。RRF 的好处就是只看排名不看分数。def rrf_fusion(bm25_results, vector_results, k60): scores {} for rank, doc_id in enumerate(bm25_results): scores[doc_id] scores.get(doc_id, 0) 1 / (k rank 1) for rank, doc_id in enumerate(vector_results): scores[doc_id] scores.get(doc_id, 0) 1 / (k rank 1) return sorted(scores.items(), keylambda x: -x[1])这里有个细节两路召回的数量要匹配。如果 BM25 召回 50 条向量只召回 10 条那向量路的排名权重就被稀释了。我的做法是两路都召回 50 条融合后取前 20 条送给重排模型。重排这一步可选但强烈建议加。我用的是一个小型的 cross-encoder 模型对 top 20 做精排能把最终 top 5 的准确率再提升 15% 左右。如果不想引入额外模型至少也要做一次基于关键词覆盖度的简单重排。2.4 实操心得三个容易翻车的地方第一BM25 索引要定期重建。知识库文档更新后如果只更新向量库不更新 BM25 索引会出现向量能召回新文档但 BM25 召回旧文档的诡异现象。我的做法是文档变更时同时打两个索引用一个版本号保证一致性。第二向量维度和模型要绑定。换 embedding 模型必须重建整个向量库不能混用。我踩过一次坑部分文档用旧模型部分用新模型检索结果乱七八糟排查了半天才发现是维度不一致导致的静默错误。第三metadata 过滤要前置。如果知识库有权限隔离或多租户需求过滤条件必须在检索时就带上不能检索完再过滤。否则一是浪费算力二是可能把不该出现的文档排名信息泄露出去。3. Agent 网关的实操过程与核心环节实现3.1 网关的职责边界它该做什么不该做什么Agent 网关最容易犯的错是什么都管。我一开始把 prompt 模板、工具注册、会话管理全塞进网关结果网关变成了一个巨型单体改一处牵动全身。后来我把职责重新划清网关只管通信和调度不管业务逻辑。具体来说管请求路由、模型选择、限流、缓存、重试、日志、鉴权不管prompt 构造、工具调用逻辑、会话状态、业务规则这样划分之后Agent 可以独立演进网关保持稳定。加一个新 Agent 只需要在网关注册一个路由不需要改网关核心代码。3.2 请求路由与模型选择策略网关收到请求后第一件事是决定用哪个模型。我的策略是基于任务类型 成本预算 当前负载三维决策。任务类型由 Agent 在请求头里声明比如task: qa、task: summarize、task: tool_call。不同任务对模型能力要求不同问答需要强推理摘要需要长上下文工具调用需要结构化输出能力。成本预算是一个滑动窗口内的累计花费上限。超过阈值就自动降级到更便宜的模型。当前负载则决定是否排队。如果某个模型的后端已经打满网关会把请求路由到备用模型而不是让用户干等。def select_model(task_type, budget_remaining, current_load): candidates MODEL_REGISTRY[task_type] # 先按负载过滤 available [m for m in candidates if current_load[m.name] m.max_qps * 0.8] if not available: available candidates # 都满了就选最便宜的排队 # 再按预算过滤 affordable [m for m in available if m.cost_per_1k budget_remaining] if not affordable: affordable [min(available, keylambda m: m.cost_per_1k)] # 最后选能力最强的 return max(affordable, keylambda m: m.capability_score)3.3 缓存设计省钱的真正大头缓存是网关里性价比最高的功能。我做了三层缓存第一层精确匹配缓存。请求的 hash 直接作为 key命中就返回。这一层能挡掉大约 20% 的重复请求。第二层语义缓存。把请求的 embedding 存下来新请求来了先算相似度超过阈值就返回缓存结果。这一层能再挡掉 15% 左右。阈值我定在 0.95太低会返回不相关的答案。第三层前缀缓存。很多请求的 system prompt 是一样的只有用户问题不同。把 system prompt 的处理结果缓存起来能显著降低首 token 延迟。class SemanticCache: def __init__(self, threshold0.95): self.threshold threshold self.entries [] # (embedding, response) def get(self, query_embedding): for emb, resp in self.entries: if cosine_sim(query_embedding, emb) self.threshold: return resp return None def put(self, query_embedding, response): self.entries.append((query_embedding, response)) # 控制缓存大小LRU 淘汰 if len(self.entries) MAX_CACHE_SIZE: self.entries.pop(0)注意语义缓存对事实性问答效果好但对需要实时数据的请求必须绕过。我在请求头里加了一个no_cache: true标志让 Agent 自己决定。3.4 限流与优先级多 Agent 共存的关键限流我用的是令牌桶 优先级队列的组合。每个 Agent 分配一个令牌桶桶的大小和填充速率根据 Agent 的重要程度设定。问答 Agent 优先级最高摘要 Agent 最低。当总负载超过阈值时低优先级的请求会被排队高优先级的照常处理。排队超过一定时间就返回降级结果比如当前繁忙请稍后重试而不是无限等待。class PriorityRateLimiter: def __init__(self): self.buckets {} # agent_id - TokenBucket self.queue PriorityQueue() def acquire(self, agent_id, priority): bucket self.buckets.get(agent_id) if bucket and bucket.try_consume(): return True # 桶空了进优先级队列 self.queue.put((priority, agent_id)) return False实测下来这套机制能把高优先级请求的 P99 延迟稳定在 1.2 秒以内而低优先级请求在高峰期会被延迟到 5 秒以上但不会失败。3.5 可观测性没有日志的网关等于没有网关网关必须记录每一次请求的完整链路请求 ID、Agent ID、任务类型、选择的模型、检索耗时、模型耗时、缓存命中情况、token 消耗、最终状态。这些数据我全部打到结构化日志里然后用一个简单的看板展示。关键指标包括指标含义告警阈值P99 延迟99% 请求的响应时间 3s缓存命中率缓存命中请求占比 25%降级率被降级处理的请求占比 5%单请求平均成本每次请求的平均花费 预算 1.5 倍检索空召回率检索结果为空的比例 10%检索空召回率这个指标特别重要。它高说明要么知识库覆盖不够要么切块/检索策略有问题。我一开始没关注这个指标后来发现某些领域的查询空召回率高达 30%补了文档之后才降下来。4. 常见问题与排查技巧实录4.1 检索相关问题的排查路径检索问题是最常见的表现也最杂。我整理了一个排查顺序第一步确认是召回问题还是排序问题。把 top 20 的结果打出来看如果正确答案根本不在里面是召回问题如果在里面但排名靠后是排序问题。第二步召回问题查分词和索引。最常见的是分词把关键词切碎了或者 BM25 索引没更新。用explain接口看每个 token 的 IDF 和命中情况。第三步排序问题查融合权重和重排。如果 BM25 和向量两路都召回了正确答案但融合后排名下降说明 RRF 的 k 值或者两路召回数量不匹配。第四步检查 metadata 过滤。有时候答案被过滤条件误杀了尤其是多租户场景下租户 ID 匹配错误。4.2 网关相关问题的排查路径网关问题通常表现为延迟飙升或错误率上升。我的排查顺序是先看限流。是不是某个 Agent 把令牌桶打空了导致其他 Agent 排队。再看缓存。缓存命中率突然下降可能是请求分布变了或者缓存被清空了。然后看模型后端。某个模型后端响应变慢会拖累整个网关需要检查后端健康状态。最后看日志。如果以上都正常就要看具体请求的链路日志找出耗时最长的环节。4.3 常见问题速查表现象可能原因排查方法解决方案检索结果不相关分词错误打印 token 流调整分词器代码 token 单独处理检索结果不相关切块过大检查块大小分布缩小块大小增加重叠检索结果不相关向量模型不匹配检查向量维度统一 embedding 模型重建索引空召回率高知识库覆盖不足统计空召回查询补充文档或放宽检索阈值延迟飙升限流配置不当查看令牌桶状态调整桶大小和优先级延迟飙升模型后端慢检查后端健康切换备用模型或降级成本超预算缓存命中率低查看缓存统计调整缓存阈值扩大缓存容量成本超预算模型选择不当查看模型分布调整路由策略多用便宜模型答案不一致缓存污染检查缓存内容清理缓存加 no_cache 标志答案不一致检索结果波动对比多次检索固定随机种子稳定排序4.4 独家避坑技巧技巧一给检索结果加来源标注。每个片段带上文档 ID、标题、位置。这样排查问题时能快速定位是哪个文档出了问题也方便用户验证答案。技巧二网关的降级策略要分级。不要只有正常和失败两种状态。我设计了三级正常、降级用便宜模型、兜底返回缓存或默认答案。这样在高峰期用户体验不会断崖式下跌。技巧三定期做检索质量回归测试。准备一批标准问答对每次改检索策略后跑一遍看准确率有没有下降。这个习惯帮我避免了好几次改一处崩一片的事故。技巧四BM25 和向量的召回数量要动态调整。对于短查询少于 5 个词BM25 权重应该更高对于长查询向量权重更高。我根据查询长度动态调整两路召回数量短查询 BM25 召回 60 条、向量 30 条长查询反过来。技巧五日志里一定要记录最终送给模型的上下文。很多时候答案不对不是检索的问题而是上下文拼接时把关键信息截断了。把最终上下文打出来一眼就能看出问题。5. 从 Demo 到生产的几个关键认知5.1 知识库的质量比检索算法更重要我花了大量时间调检索算法最后发现最大的提升来自把文档写好。结构清晰、标题明确、代码块规范的文档检索准确率天然就高。反过来一堆格式混乱的 PDF 扫描件再好的算法也救不回来。所以我现在做知识库的第一步不是搭系统而是整理文档。统一格式、补全标题、拆分长文档、给关键段落加摘要。这一步做完检索效果能提升 30% 以上比调任何参数都管用。5.2 网关的稳定性来自简单我一开始把网关设计得很复杂支持各种插件、钩子、中间件。结果每次出问题都要排查半天因为调用链太深了。后来我砍掉了大部分扩展点只保留最核心的路由、限流、缓存、日志。网关代码从 8000 行降到 2000 行故障率下降了 70%。生产系统的稳定性往往来自少做事而不是多做事。5.3 成本控制要从第一天就做很多人做 RAG 是先跑通再优化成本结果跑通之后发现成本高得离谱回头改架构代价很大。我的建议是第一天就把缓存和模型路由做进去哪怕一开始很简单。因为这两个东西是架构级的后期加会很痛苦。5.4 监控要覆盖业务指标而不只是技术指标技术指标延迟、错误率、QPS只能告诉你系统是否正常不能告诉你系统是否有用。我额外监控了几个业务指标检索命中率、答案采纳率、用户追问率。追问率高说明答案质量不行这时候要去查检索和 prompt而不是查服务器。6. 后续可以继续深挖的方向这套系统跑了大半年目前支撑着几个内部知识库和十来个 Agent。接下来我打算在几个方向上继续优化。一是引入 ontology 做知识结构化。现在检索还是基于文本片段如果能抽出实体和关系构建一个轻量知识图谱对于多跳推理类问题会有明显提升。比如某个配置项在哪个版本的文档里被废弃了这种问题纯文本检索很难答好。二是做 Agentic RAG。让 Agent 自己决定检索几次、检索什么、要不要改写查询。现在的 RAG 是一次检索定生死Agentic RAG 可以多轮检索、逐步逼近答案。代价是延迟和成本上升需要网关配合做更精细的调度。三是把网关做成模型无关的抽象层。现在虽然做了模型路由但不同模型的接口还是有差异。我想再抽象一层让 Agent 完全感知不到底层是哪个模型这样换模型、加模型、做 A/B 测试都会更方便。四是完善评测体系。现在评测还是靠人工抽查效率低且不全面。我想搭一套自动评测流水线用 LLM 做裁判对检索质量和答案质量做持续评估。这样每次改动都能快速知道是变好还是变坏。这套东西没有银弹都是一点点磨出来的。我踩过的坑、试过的方案、最后留下的设计基本都写在上面的内容里了。如果你也在做类似的事情希望这些经验能帮你少走点弯路。
返回列表