AI Agent开发实战⑯Query改写与扩展让检索更懂用户意图用户问Python性能优化但文档里写的是Python性能调优技巧——语义相同但词汇不同向量检索可能漏掉。Query改写和扩展就是解决这个问题的用多种表达方式检索提升召回率。一、用户意图的真实问题用户输入往往存在三个问题问题1表达不准确 用户输入怎么让Python跑得更快 文档表达Python性能优化方法 问题词汇不匹配 问题2意图模糊 用户输入Python性能 文档表达可能关于性能测试、性能优化、性能监控 问题无法确定用户要什么 问题3长尾词缺失 用户输入Python性能优化 文档表达Python 3.11性能提升30%的原因分析 问题没提到版本号、具体数字Query改写的目标原始Query → 扩展Query 1, 2, 3... → 分别检索 → 结果融合二、Query改写的四种策略2.1 同义词扩展classSynonymExpander:同义词扩展def__init__(self):# 领域同义词词典self.synonyms{性能优化:[性能调优,性能提升,运行速度优化,执行效率提升],配置:[设置,参数配置,环境配置],错误:[异常,报错,Error,Exception],部署:[上线,发布,Deploy],}defexpand(self,query:str)-list[str]:扩展同义词expanded[query]# 保留原始查询forterm,synonymsinself.synonyms.items():ifterminquery:forsyninsynonyms:expanded.append(query.replace(term,syn))returnexpanded# 使用示例expanderSynonymExpander()queriesexpander.expand(Python性能优化方法)# 输出# [Python性能优化方法,# Python性能调优方法,# Python性能提升方法,# Python运行速度优化方法]2.2 LLM改写classLLMQueryRewriter:LLM改写Querydef__init__(self,llm):self.llmllmdefrewrite(self,query:str,num_rewrites:int3)-list[str]:用LLM生成多个改写版本promptf 用户查询{query}请生成{num_rewrites}个意思相同但表达不同的查询改写版本。 要求 1. 保持原意不变 2. 使用不同的词汇和句式 3. 包含可能的同义词和专业术语 4. 每行一个不要编号 示例 原查询Python性能优化 改写 Python性能调优技巧 如何提升Python运行速度 Python执行效率优化方法 responseself.llm.invoke(prompt)rewrites[line.strip()forlineinresponse.content.split(\n)ifline.strip()]return[query]rewrites[:num_rewrites]# 使用示例rewriterLLMQueryRewriter(llm)queriesrewriter.rewrite(Python性能优化,num_rewrites3)# 输出# [Python性能优化,# Python性能调优技巧,# 如何提升Python运行速度,# Python执行效率优化方法]2.3 HyDE假设文档嵌入HyDE的核心思想先让LLM生成一个假设的答案用答案去检索而不是用问题。classHyDERewriter:Hypothetical Document Embeddingsdef__init__(self,llm,embedder):self.llmllm self.embedderembedderdefgenerate_hypothetical_doc(self,query:str)-str:生成假设文档promptf 用户查询{query}请生成一段可能回答这个问题的文档内容。 要求 1. 长度200-300字 2. 包含可能的关键信息 3. 使用专业术语 直接输出文档内容不要解释。 responseself.llm.invoke(prompt)returnresponse.contentdefretrieve_with_hyde(self,query:str,vector_store,k:int5)-list:HyDE检索# 生成假设文档hypo_docself.generate_hypothetical_doc(query)# 用假设文档的向量检索hypo_embeddingself.embedder.embed(hypo_doc)resultsvector_store.search(hypo_embedding,kk)returnresults,hypo_doc# 使用示例hydeHyDERewriter(llm,embedder)queryPython性能优化results,hypo_dochyde.retrieve_with_hyde(query,vector_store)print(假设文档)print(hypo_doc)print(\n检索结果)forrinresults:print(r)HyDE的优势答案比问题更像答案用答案检索更精准。2.4 Query2DocQuery转文档Query2Doc是HyDE的简化版把Query扩展成一个简短的文档描述。classQuery2Doc:Query转文档def__init__(self,llm):self.llmllmdefexpand(self,query:str)-str:将Query扩展为文档描述promptf 用户查询{query}请用一段简短的文字描述用户可能想要查找的内容。 格式用户可能想要了解关于XXX的内容包括AAA、BBB、CCC等方面。 只输出描述不要解释。 responseself.llm.invoke(prompt)returnresponse.contentdefretrieve_with_q2d(self,query:str,embedder,vector_store,k:int5):Query2Doc检索# 扩展Queryexpandedself.expand(query)# 合并原始Query和扩展文档combinedf{query}\n{expanded}# 检索embeddingembedder.embed(combined)resultsvector_store.search(embedding,kk)returnresults,expanded三、多Query检索融合无论用哪种策略最终都是生成多个Query版本。如何融合检索结果3.1 Reciprocal Rank FusionRRFfromcollectionsimportdefaultdictdefreciprocal_rank_fusion(results_list:list[list],k:int60)-list: RRF融合算法 results_list: 多个Query的检索结果列表 k: RRF参数默认60 公式RRF(d) Σ 1/(k rank(d)) scoresdefaultdict(float)forresultsinresults_list:forrank,docinenumerate(results):# RRF分数scores[doc[id]]1/(krank1)# 按分数排序sorted_docssorted(scores.items(),keylambdax:x[1],reverseTrue)return[{id:doc_id,rrf_score:score}fordoc_id,scoreinsorted_docs]3.2 加权分数融合defweighted_fusion(results_list:list[list],weights:list[float]None)-list:加权分数融合ifweightsisNone:weights[1.0/len(results_list)]*len(results_list)scoresdefaultdict(float)forresults,weightinzip(results_list,weights):fordocinresults:scores[doc[id]]doc.get(score,1.0)*weight sorted_docssorted(scores.items(),keylambdax:x[1],reverseTrue)return[{id:doc_id,weighted_score:score}fordoc_id,scoreinsorted_docs]四、实测对比4.1 测试设置测试数据-文档10000篇中文技术文档-查询100个测试查询-评估Recall10,NDCG104.2 单策略效果策略Recall10NDCG10耗时原始Query71.2%0.6812ms同义词扩展76.3%0.7215msLLM改写79.8%0.76350msHyDE82.1%0.79520msQuery2Doc80.4%0.77280ms关键发现HyDE效果最好但耗时最长LLM改写效果不错性价比高同义词扩展最简单效果也不错4.3 组合策略效果组合Recall10NDCG10耗时原始 同义词78.5%0.7420ms原始 LLM改写82.3%0.78380ms原始 HyDE84.2%0.81550ms原始 同义词 LLM改写83.1%0.79390ms组合策略提升有限2-3%建议选单一策略即可。五、选型决策第一步场景评估 │ ├── 专业领域有明确术语体系 │ → 【同义词扩展】 │ 理由术语明确同义词词典效果好 │ ├── 通用领域 │ → 【LLM改写】 │ 理由通用性强效果稳定 │ └── 追求最优效果 → 【HyDE】 理由效果最好 第二步延迟要求 │ ├── 要求50ms │ → 【同义词扩展】 │ ├── 要求500ms │ → 【LLM改写】或【Query2Doc】 │ └── 对延迟不敏感 → 【HyDE】六、完整代码智能Query改写器classSmartQueryRewriter:智能Query改写器自动选择策略def__init__(self,llm,embedder):self.llmllm self.embedderembedder self.synonym_expanderSynonymExpander()defanalyze_query(self,query:str)-dict:分析Query特征# 检测是否包含专业术语tech_terms[API,SDK,GPU,CPU,内存,配置,优化]has_tech_termany(terminqueryfortermintech_terms)# 检测Query长度is_shortlen(query)10# 检测是否是问句is_questionany(kwinqueryforkwin[如何,怎么,为什么,什么是])return{has_tech_term:has_tech_term,is_short:is_short,is_question:is_question}defrewrite(self,query:str,strategy:strauto)-list[str]:改写Queryifstrategyauto:strategyself._select_strategy(query)ifstrategysynonym:returnself.synonym_expander.expand(query)elifstrategyllm:returnLLMQueryRewriter(self.llm).rewrite(query)elifstrategyhyde:hypoHyDERewriter(self.llm,self.embedder).generate_hypothetical_doc(query)return[query,hypo]else:return[query]def_select_strategy(self,query:str)-str:自动选择策略analysisself.analyze_query(query)ifanalysis[has_tech_term]:returnsynonym# 专业术语用同义词elifanalysis[is_question]:returnllm# 问句用LLM改写elifanalysis[is_short]:returnhyde# 短Query用HyDE扩展else:returnllm# 默认LLM改写# 使用示例rewriterSmartQueryRewriter(llm,embedder)queriesrewriter.rewrite(Python性能优化)print(f改写结果{queries})七、总结策略适用场景Recall提升耗时同义词扩展专业领域5%20msLLM改写通用场景8%300-400msHyDE追求最优11%500-600msQuery2Doc平衡效果和速度9%200-300msQuery改写是低成本高回报的优化手段建议所有RAG系统都接入。下篇预告「多跳检索与知识图谱让RAG突破单跳限制」——复杂问题需要多次检索如何设计多跳检索架构需要完整Query改写代码的同学可以看我主页的付费资源专栏。有问题欢迎评论区留言大家一起讨论