向量检索 Top-K 的动态调整根据查询复杂度自适应选择检索数量一、深度引言与场景痛点RAG 系统里有一个被问得最多、答得最随意的参数Top-K。设 5 吧有时候答案缺信息设 20 吧好几篇不相关的文档混进来把 LLM 搞晕了。然后你花了半天调参得到结论10 是个不错的平衡。但上线后发现简单问题 5 个就够了复杂问题 20 个也不够。问题的本质是检索不是越多越好而是刚好覆盖答案所需的信息。简单问题产品价格是多少12 篇文档就能覆盖中等问题产品的功能列表和对比需要 510 篇复杂问题本季度的销售策略变更对毛利率的影响分析可能需要 20 篇以上并且分布在不同的数据分区里。动态 Top-K 的思路就是根据查询的复杂度自动调整检索数量在足够信息和不喂噪声之间找到当前查询的最优解。二、底层机制与原理深度剖析四个维度的权重可以根据你的业务场景调整。如果是知识库问答事实类查询为主意图类型权重可以低些实体数量权重要高些。如果是分析报告生成意图类型权重要高因为分析类查询天然需要更多文档。历史反馈是这套系统持续优化的来源。记录每次查询的复杂度分、使用的 Top-K、最终答案的用户满意度点击、阅读时长、点赞/踩建立一个查询特征 → 最优 Top-K的映射。运行一段时间后复杂度评估不再是启发式规则而是数据驱动的。三、生产级代码实现from __future__ import annotations import asyncio import re from dataclasses import dataclass, field from enum import Enum class QueryIntent(Enum): FACT 1 # 事实查询: X 是什么 LIST 2 # 列表查询: X 有哪些 COMPARE 3 # 对比查询: X 和 Y 有什么区别 ANALYZE 4 # 分析查询: 分析 X 的原因/影响 GENERATE 5 # 生成查询: 写一份 X 的报告 dataclass class ComplexityScore: total: float length_score: float entity_score: float intent_score: float recommended_top_k: int class DynamicTopKEngine: 动态 Top-K 引擎: 根据查询复杂度自适应调整检索数量 def __init__(self): self._entity_pattern re.compile( r[A-Z][a-z]|[A-Z]{2,}|[\u4e00-\u9fff]{2,}(?:公司|集团|平台|产品|系统|服务|模型) ) self._intent_keywords { QueryIntent.COMPARE: [对比, 比较, 区别, 不同, 差异, vs], QueryIntent.ANALYZE: [分析, 原因, 影响, 趋势, 评估, 风险], QueryIntent.GENERATE: [生成, 写, 创建, 制作, 总结, 报告], QueryIntent.LIST: [列表, 有哪些, 所有, 哪些, 列举], } # 历史反馈: {query_hash: optimal_top_k} self._history: dict[str, int] {} def assess(self, query: str) - ComplexityScore: # 维度1: 查询长度 word_count len(query.split()) if word_count 3: len_score 1.0 elif word_count 8: len_score 2.0 elif word_count 15: len_score 3.0 else: len_score 4.0 # 维度2: 实体数量 entities self._entity_pattern.findall(query) entity_count len(set(entities)) entity_score min(entity_count * 1.5, 5.0) # 维度3: 意图类型 intent QueryIntent.FACT for intent_type, keywords in self._intent_keywords.items(): if any(kw in query for kw in keywords): intent intent_type break intent_score { QueryIntent.FACT: 1.0, QueryIntent.LIST: 2.0, QueryIntent.COMPARE: 3.5, QueryIntent.ANALYZE: 4.5, QueryIntent.GENERATE: 5.0, }.get(intent, 1.0) # 加权计算权重可调 total len_score * 0.2 entity_score * 0.4 intent_score * 0.4 # 确定 Top-K if total 2.0: top_k 5 elif total 4.0: top_k 12 elif total 6.0: top_k 25 else: top_k 40 # 检查历史反馈同类查询历史上需要多少 Top-K query_hash str(hash(query)) if query_hash in self._history: historical_k self._history[query_hash] # 取历史值和计算值的中位数避免过度偏离 top_k (top_k historical_k) // 2 return ComplexityScore( totaltotal, length_scorelen_score, entity_scoreentity_score, intent_scoreintent_score, recommended_top_ktop_k, ) def record_feedback(self, query: str, used_top_k: int, was_sufficient: bool) - None: 记录检索结果是否足够用于优化后续评估 query_hash str(hash(query)) if was_sufficient: # 结果足够 → 当前 Top-K 可能是最优的 self._history[query_hash] used_top_k else: # 结果不足 → 下次增加 Top-K self._history[query_hash] min(used_top_k 10, 50) class AdaptiveRetriever: 自适应检索器: 集成动态 Top-K 和检索逻辑 def __init__(self, vector_store, top_k_engine: DynamicTopKEngine): self._store vector_store self._engine top_k_engine async def retrieve(self, query: str) - tuple[list[dict], ComplexityScore]: score self._engine.assess(query) top_k score.recommended_top_k try: async with asyncio.timeout(3.0): results await self._store.search(query, top_k) except asyncio.TimeoutError: results [] except Exception: results [] # 如果结果太少追加一次检索 if len(results) 3 and top_k 40: extra_k min(top_k * 2, 50) try: extra_results await self._store.search(query, extra_k) results extra_results except Exception: pass # 记录反馈简化结果数 0 视为足够 was_sufficient len(results) 3 self._engine.record_feedback(query, top_k, was_sufficient) return results, scoreDynamicTopKEngine.assess的三个维度分别关注查询文本的表面特征长度、语义特征实体、目的特征意图。评分权重是可调的目前实体和意图各占 0.4长度占 0.2。这个权重分配基于一个经验长度只是复杂度的一个弱信号实体数量和意图类型才是强信号。历史反馈的引入让系统有了学习能力。record_feedback在每次检索后记录这个 Top-K 是否足够。足够的话记录为最优值不够的话下次自动增加 10。query_hash的粒度是精确匹配——只对完全相同的查询做优化。如果你想扩展到相似查询需要把 hash 换成 embedding 聚类。AdaptiveRetriever.retrieve里有一个二次检索的安全网如果第一次检索结果不到 3 条且还没到最大 Top-K自动扩大检索范围再查一次。这个安全网防止复杂度评估太低导致关键信息缺失。四、边界分析与架构权衡复杂度评估模型可能把复杂查询误判为简单查询导致 Top-K 不足。这时候 LLM 接收到的信息不完整可能给出片面或错误的答案。错误成本在医疗、金融等高敏感领域不可接受但在通用 RAG 里可以通过降级回答来缓解——如果 LLM 发现上下文不足以回答问题直接说信息不足而不是硬猜。另一个代价是Top-K 越大延迟越高。Top-K40 的检索延迟可能是 Top-K5 的 3 倍。如果复杂度评估让一大部分查询都走了大 Top-K省 Token 的收益可能被延迟抵消。所以要关注 Top-K 的分布如果超过 20% 的查询走了 Top-K ≥ 25说明你的文档质量可能有问题大量文档部分相关而不是高度相关该优化的是文档切分和索引策略而不是一味扩大 Top-K。还有一个容易被忽略的点动态 Top-K 让每次查询的 Token 消耗不确定成本预算难以预测。建议设一个每日总 Token 上限接近上限时强制所有查询走最小 Top-K。五、总结动态 Top-K 的核心理念是够用即可不多不少。评估查询复杂度用三个维度查询长度、实体数量、意图类型加权计算后映射到不同等级的 Top-K。历史反馈帮助系统持续优化评估精度。落地建议先上线三维度评估跑 2 周积累反馈数据然后用历史反馈调整权重让评估更准同时设二次检索安全网防止评估偏低导致信息缺失最后监控 Top-K 分布和 Token 消耗找到成本平衡点。