AI Agent开发实战⑯｜Query改写与扩展：让检索更懂用户意图

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# 使用示例expander=SynonymExpander()queries=expander.expand("Python性能优化方法")# 输出：# ["Python性能优化方法",# "Python性能调优方法",# "Python性能提升方法",# "Python运行速度优化方法"]

2.2 LLM改写

classLLMQueryRewriter:"""LLM改写Query"""def__init__(self,llm):self.llm=llmdefrewrite(self,query:str,num_rewrites:int=3)->list[str]:"""用LLM生成多个改写版本"""prompt=f""" 用户查询：{query}请生成{num_rewrites}个意思相同但表达不同的查询改写版本。 要求： 1. 保持原意不变 2. 使用不同的词汇和句式 3. 包含可能的同义词和专业术语 4. 每行一个，不要编号 示例： 原查询：Python性能优化 改写： Python性能调优技巧 如何提升Python运行速度 Python执行效率优化方法 """response=self.llm.invoke(prompt)rewrites=[line.strip()forlineinresponse.content.split('\n')ifline.strip()]return[query]+rewrites[:num_rewrites]# 使用示例rewriter=LLMQueryRewriter(llm)queries=rewriter.rewrite("Python性能优化",num_rewrites=3)# 输出：# ["Python性能优化",# "Python性能调优技巧",# "如何提升Python运行速度",# "Python执行效率优化方法"]

2.3 HyDE（假设文档嵌入）

HyDE的核心思想：先让LLM生成一个假设的答案，用答案去检索，而不是用问题。

classHyDERewriter:"""Hypothetical Document Embeddings"""def__init__(self,llm,embedder):self.llm=llm self.embedder=embedderdefgenerate_hypothetical_doc(self,query:str)->str:"""生成假设文档"""prompt=f""" 用户查询：{query}请生成一段可能回答这个问题的文档内容。 要求： 1. 长度200-300字 2. 包含可能的关键信息 3. 使用专业术语 直接输出文档内容，不要解释。 """response=self.llm.invoke(prompt)returnresponse.contentdefretrieve_with_hyde(self,query:str,vector_store,k:int=5)->list:"""HyDE检索"""# 生成假设文档hypo_doc=self.generate_hypothetical_doc(query)# 用假设文档的向量检索hypo_embedding=self.embedder.embed(hypo_doc)results=vector_store.search(hypo_embedding,k=k)returnresults,hypo_doc# 使用示例hyde=HyDERewriter(llm,embedder)query="Python性能优化"results,hypo_doc=hyde.retrieve_with_hyde(query,vector_store)print("假设文档：")print(hypo_doc)print("\n检索结果：")forrinresults:print(r)

HyDE的优势：答案比问题更像答案，用答案检索更精准。

2.4 Query2Doc（Query转文档）

Query2Doc是HyDE的简化版：把Query扩展成一个简短的文档描述。

classQuery2Doc:"""Query转文档"""def__init__(self,llm):self.llm=llmdefexpand(self,query:str)->str:"""将Query扩展为文档描述"""prompt=f""" 用户查询：{query}请用一段简短的文字描述用户可能想要查找的内容。 格式：用户可能想要了解关于XXX的内容，包括AAA、BBB、CCC等方面。 只输出描述，不要解释。 """response=self.llm.invoke(prompt)returnresponse.contentdefretrieve_with_q2d(self,query:str,embedder,vector_store,k:int=5):"""Query2Doc检索"""# 扩展Queryexpanded=self.expand(query)# 合并原始Query和扩展文档combined=f"{query}\n{expanded}"# 检索embedding=embedder.embed(combined)results=vector_store.search(embedding,k=k)returnresults,expanded

三、多Query检索融合

无论用哪种策略，最终都是生成多个Query版本。如何融合检索结果？

3.1 Reciprocal Rank Fusion（RRF）

fromcollectionsimportdefaultdictdefreciprocal_rank_fusion(results_list:list[list],k:int=60)->list:""" RRF融合算法 results_list: 多个Query的检索结果列表 k: RRF参数（默认60） 公式：RRF(d) = Σ 1/(k + rank(d)) """scores=defaultdict(float)forresultsinresults_list:forrank,docinenumerate(results):# RRF分数scores[doc["id"]]+=1/(k+rank+1)# 按分数排序sorted_docs=sorted(scores.items(),key=lambdax:x[1],reverse=True)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)scores=defaultdict(float)forresults,weightinzip(results_list,weights):fordocinresults:scores[doc["id"]]+=doc.get("score",1.0)*weight sorted_docs=sorted(scores.items(),key=lambdax:x[1],reverse=True)return[{"id":doc_id,"weighted_score":score}fordoc_id,scoreinsorted_docs]

四、实测对比

4.1 测试设置

测试数据：-文档：10000篇中文技术文档-查询：100个测试查询-评估：Recall@10,NDCG@10

4.2 单策略效果

关键发现：

• HyDE效果最好，但耗时最长
• LLM改写效果不错，性价比高
• 同义词扩展最简单，效果也不错

4.3 组合策略效果

组合策略提升有限（2-3%），建议选单一策略即可。

五、选型决策

第一步：场景评估 │ ├── 专业领域（有明确术语体系） │ → 【同义词扩展】 │ 理由：术语明确，同义词词典效果好 │ ├── 通用领域 │ → 【LLM改写】 │ 理由：通用性强，效果稳定 │ └── 追求最优效果 → 【HyDE】 理由：效果最好 第二步：延迟要求 │ ├── 要求<50ms │ → 【同义词扩展】 │ ├── 要求<500ms │ → 【LLM改写】或【Query2Doc】 │ └── 对延迟不敏感 → 【HyDE】

六、完整代码：智能Query改写器

classSmartQueryRewriter:"""智能Query改写器：自动选择策略"""def__init__(self,llm,embedder):self.llm=llm self.embedder=embedder self.synonym_expander=SynonymExpander()defanalyze_query(self,query:str)->dict:"""分析Query特征"""# 检测是否包含专业术语tech_terms=["API","SDK","GPU","CPU","内存","配置","优化"]has_tech_term=any(terminqueryfortermintech_terms)# 检测Query长度is_short=len(query)<10# 检测是否是问句is_question=any(kwinqueryforkwin["如何","怎么","为什么","什么是"])return{"has_tech_term":has_tech_term,"is_short":is_short,"is_question":is_question}defrewrite(self,query:str,strategy:str="auto")->list[str]:"""改写Query"""ifstrategy=="auto":strategy=self._select_strategy(query)ifstrategy=="synonym":returnself.synonym_expander.expand(query)elifstrategy=="llm":returnLLMQueryRewriter(self.llm).rewrite(query)elifstrategy=="hyde":hypo=HyDERewriter(self.llm,self.embedder).generate_hypothetical_doc(query)return[query,hypo]else:return[query]def_select_strategy(self,query:str)->str:"""自动选择策略"""analysis=self.analyze_query(query)ifanalysis["has_tech_term"]:return"synonym"# 专业术语用同义词elifanalysis["is_question"]:return"llm"# 问句用LLM改写elifanalysis["is_short"]:return"hyde"# 短Query用HyDE扩展else:return"llm"# 默认LLM改写# 使用示例rewriter=SmartQueryRewriter(llm,embedder)queries=rewriter.rewrite("Python性能优化")print(f"改写结果：{queries}")

七、总结

Query改写是低成本高回报的优化手段，建议所有RAG系统都接入。

下篇预告：「多跳检索与知识图谱：让RAG突破单跳限制」——复杂问题需要多次检索，如何设计多跳检索架构？

需要完整Query改写代码的同学，可以看我主页的付费资源专栏。

有问题欢迎评论区留言，大家一起讨论！