查询改写与扩展：让RAG听懂人话

这篇文章解决什么问题

用户问”电脑开不了机怎么办”，但知识库里写的是”计算机无法启动”——同一个意思两种说法，传统检索就抓瞎了。查询改写就是来解决这个”语言鸿沟”问题的。

前言：语言的千变万化

先说个让我头疼的问题。

做客服知识库的时候，遇到这种情况：

用户问：“电脑黑屏了咋整” 知识库写：“显示器无信号排查步骤”

你猜传统向量检索能找到吗？

有时候能，有时候不能。原因很玄学——取决于Embedding模型有没有见过类似的表达方式。

后来我学会了查询改写，效果立竿见影。

一、为什么需要查询改写

1.1 语言的多样性

同一个意思，人类有数不清的表达方式：

用户问法	知识库写法
电脑黑屏	显示器黑屏/屏幕不亮
开不了机	无法启动/启动失败
网速慢	网络延迟高/卡顿
密码忘了	账户无法登录/登录失败

1.2 检索的尴尬

传统检索有两种极端：

太死板：用户问”黑屏”，只匹配包含”黑屏”的文档，其他意思相近的都被忽略了。

太宽泛：用Embedding强行语义匹配，把”屏幕碎了”这种不相关内容也捞出来了。

1.3 查询改写的价值

查询改写就是在检索之前，先把用户的问题”翻译”成更适合检索的形式：

用户原始问题 → 改写优化 → 检索 → 返回结果

核心价值：

弥合语言鸿沟
提高召回率
减少无关结果

二、查询改写的核心技术

2.1 查询扩展

原理：把一个查询变成多个相关查询，一起检索

async def expand_query(query, llm, num_expansions=3):
    """
    查询扩展
    """
    
    prompt = f"""
请为以下查询生成{num_expansions}个语义相近的变体：
 
原始查询：{query}
 
要求：
1. 每个变体要表达相同的意思
2. 用不同的词汇和句式
3. 涵盖同义词、近义词
 
请以JSON格式返回：
{{
    "expansions": ["变体1", "变体2", "变体3"]
}}
"""
    
    result = await llm.generate(prompt)
    expansions = json.loads(result)["expansions"]
    
    return [query] + expansions  # 原始查询 + 扩展查询

使用示例：

# 原始查询
query = "电脑黑屏了咋整"
 
# 扩展后
expansions = await expand_query(query, llm)
# ['电脑黑屏了咋整', '计算机显示器无信号', '电脑屏幕不亮怎么办', '主机正常运行屏幕黑屏']
 
# 用所有查询检索
all_results = []
for q in expansions:
    results = await retriever.search(q)
    all_results.extend(results)
 
# 去重和融合
final_results = deduplicate_and_merge(all_results)

2.2 同义词替换

原理：用同义词替换查询中的关键词

# 简单的同义词词典
SYNONYM_DICT = {
    "电脑": ["计算机", "PC", "主机"],
    "黑屏": ["无显示", "屏幕不亮", "显示器不工作"],
    "手机": ["移动电话", "智能手机"],
    "慢": ["卡顿", "延迟高", "速度慢"],
    "密码": ["口令", "登录密码", "账户密码"],
}
 
def synonym_expansion(query):
    """
    同义词扩展
    """
    
    words = query.split()
    expansions = []
    
    # 对每个词进行同义词替换
    for i, word in enumerate(words):
        if word in SYNONYM_DICT:
            for synonym in SYNONYM_DICT[word]:
                new_query = words.copy()
                new_query[i] = synonym
                expansions.append(" ".join(new_query))
    
    return expansions if expansions else [query]

效果：

query = "电脑黑屏"
expansions = synonym_expansion(query)
# ['计算机黑屏', 'PC黑屏', '主机黑屏', '电脑无显示', '电脑屏幕不亮', ...]

2.3 HyDE：生成假答案

HyDE（Hypothetical Document Embeddings）是一种很有创意的技术：

让LLM根据问题生成一个”假答案”
用这个假答案去检索
返回真正相关的文档

async def hyde_retrieval(query, llm, retriever):
    """
    HyDE检索
    """
    
    # 1. 生成假答案
    hyde_prompt = f"""
基于以下问题，生成一段假设性的答案内容。
答案不需要准确，但应该包含相关的术语和概念。
 
问题：{query}
 
请生成一段100字左右的假设性答案：
"""
    
    hypothetical_doc = await llm.generate(hyde_prompt)
    
    # 2. 用假答案做检索
    results = await retriever.search(hypothetical_doc)
    
    return results
 
# 使用示例
results = await hyde_retrieval("电脑黑屏怎么办", llm, retriever)

为什么HyDE有效：

假答案包含了”什么是黑屏”、“怎么处理”、“相关概念”等丰富信息
比单纯的关键词更容易匹配到相关文档
LLM生成的文本风格更接近知识库

2.4 意图识别

原理：先理解用户真正想问什么，再针对性地改写

async def classify_intent(query):
    """
    识别查询意图
    """
    
    prompt = f"""
请分析以下查询的意图类型：
 
查询：{query}
 
可能的意图类型：
1. 故障排查：用户遇到了问题，需要解决方案
2. 概念了解：用户想了解某个概念或术语
3. 操作指导：用户想知道如何完成某个操作
4. 对比分析：用户想比较两个或多个事物
5. 寻求建议：用户需要推荐或建议
 
请返回JSON格式：
{{
    "intent": "意图类型",
    "confidence": 0.0-1.0,
    "key_elements": ["关键要素"]
}}
"""
    
    result = await llm.generate(prompt)
    return json.loads(result)

根据意图改写：

async def intent_based_rewrite(query):
    """
    基于意图的查询改写
    """
    
    intent_info = await classify_intent(query)
    intent = intent_info["intent"]
    
    if intent == "故障排查":
        # 故障排查类查询，扩展更多问题描述
        rewrite_prompt = f"""
问题：{query}
意图：故障排查
 
请生成更精确的故障排查查询：
1. 列出可能的故障现象
2. 列出可能的解决步骤
3. 列出相关的技术术语
"""
    
    elif intent == "概念了解":
        # 概念了解类查询，更注重定义和解释
        rewrite_prompt = f"""
问题：{query}
意图：概念了解
 
请生成更学术化的查询：
1. 列出标准的定义表达
2. 列出相关的理论框架
3. 列出经典的应用场景
"""
    
    # ... 其他意图类似处理
    
    rewritten = await llm.generate(rewrite_prompt)
    return rewritten

三、查询纠错

3.1 错别字纠正

用户打字经常有错别字：

# 简单的拼写检查
import re
 
COMMON_MISSPELLINGS = {
    "电脑": ["电恼", "电恼", "电脑"],
    "网络": ["网罗", "两络"],
    "密码": ["密码", "登陆密码"],
}
 
def correct_spelling(query):
    """
    纠正拼写错误
    """
    
    for correct_word, misspellings in COMMON_MISSPELLINGS.items():
        for misspelled in misspellings:
            if misspelled in query:
                query = query.replace(misspelled, correct_word)
    
    return query

更智能的做法：用LLM纠正

async def llm_spell_check(query):
    """
    用LLM纠正拼写错误
    """
    
    prompt = f"""
请检查以下文本中的拼写错误，并返回纠正后的版本：
 
原文：{query}
 
注意：
1. 只修改明显的拼写错误
2. 保持原意不变
3. 对于不确定的词语，保持原样
 
请只返回纠正后的文本，不要其他解释：
"""
    
    corrected = await llm.generate(prompt)
    return corrected.strip()

3.2 歧义消解

有时候用户说的话有歧义：

用户问：“苹果怎么吃？”

可能是水果苹果的吃法
可能是苹果公司（Apple）的相关问题

async def disambiguate(query, context=None):
    """
    歧义消解
    """
    
    prompt = f"""
请分析以下查询是否存在歧义：
 
查询：{query}
 
如果存在歧义：
1. 列出可能的解释
2. 根据上下文（如果有）判断最可能的解释
3. 为每种解释生成改写后的查询
 
上下文：{context if context else "无"}
 
请以JSON格式返回：
{{
    "is_ambiguous": true/false,
    "interpretations": [
        {{"meaning": "解释1", "rewritten_query": "改写后查询1"}},
        {{"meaning": "解释2", "rewritten_query": "改写后查询2"}}
    ],
    "selected_interpretation": "选择的解释"
}}
"""
    
    result = await llm.generate(prompt)
    return json.loads(result)

3.3 伪相关反馈（PRF）

原理：利用初始检索结果中的好词来优化查询

async def pseudo_relevance_feedback(query, retriever, top_k=5, num_terms=10):
    """
    伪相关反馈
    """
    
    # 1. 初始检索
    initial_results = await retriever.search(query, top_k=top_k)
    
    # 2. 从好结果中提取关键词
    all_text = " ".join([r.content for r in initial_results])
    keywords = extract_keywords(all_text, top_n=num_terms)
    
    # 3. 把好词加入查询
    enhanced_query = query + " " + " ".join(keywords)
    
    # 4. 重新检索
    final_results = await retriever.search(enhanced_query)
    
    return final_results
 
def extract_keywords(text, top_n=10):
    """
    提取关键词（简单实现）
    """
    # 实际应该用TF-IDF、KeyBERT等
    words = text.split()
    
    # 简单统计词频
    word_freq = {}
    for word in words:
        if len(word) > 2:  # 过滤太短的词
            word_freq[word] = word_freq.get(word, 0) + 1
    
    # 返回高频词
    sorted_words = sorted(word_freq.items(), key=lambda x: x[1], reverse=True)
    return [w[0] for w in sorted_words[:top_n]]

四、查询路由

4.1 什么是查询路由

不同问题适合不同的检索方式，查询路由就是决定”用什么方式查”。

def route_query(query):
    """
    查询路由
    """
    
    # 1. 检测查询特征
    is_precise = detect_precise_query(query)  # 精确查询
    is_semantic = detect_semantic_query(query)  # 语义查询
    needs_kg = detect_kg_query(query)  # 需要知识图谱
    needs_code = detect_code_query(query)  # 需要代码检索
    
    # 2. 路由决策
    if is_precise:
        return "bm25"  # 精确查询用BM25
    elif is_semantic:
        return "vector"  # 语义查询用向量检索
    elif needs_kg:
        return "knowledge_graph"  # 关系查询用知识图谱
    elif needs_code:
        return "code_search"  # 代码查询用专门的代码检索
    else:
        return "hybrid"  # 默认混合检索

4.2 特征检测

def detect_precise_query(query):
    """检测是否是精确查询"""
    
    precise_indicators = [
        "第", "条", "条款", "规定",  # 法律条款
        "版本", "型号", "型号",      # 特定版本
        "2021", "2022", "2023",     # 年份
        "1.0", "2.0", "v1",         # 版本号
    ]
    
    return any(ind in query for ind in precise_indicators)
 
def detect_semantic_query(query):
    """检测是否是语义查询"""
    
    semantic_indicators = [
        "是什么", "为什么", "如何",
        "什么意思", "解释", "理解"
    ]
    
    return any(ind in query for ind in semantic_indicators)
 
def detect_kg_query(query):
    """检测是否需要知识图谱"""
    
    kg_indicators = [
        "和", "与", "或者",          # 关系连接词
        "区别", "对比", "关系",
        "谁", "哪个公司", "创始人"
    ]
    
    return any(ind in query for ind in kg_indicators)
 
def detect_code_query(query):
    """检测是否需要代码检索"""
    
    code_indicators = [
        "代码", "函数", "API",
        "import", "def ", "class",
        "怎么写", "代码示例"
    ]
    
    return any(ind.lower() in query.lower() for ind in code_indicators)

4.3 动态路由

async def dynamic_route(query, retrievers, llm):
    """
    动态路由：根据查询内容自动选择检索器
    """
    
    # 1. LLM判断查询类型
    routing_decision = await llm.generate(f"""
请分析以下查询应该使用哪种检索方式：
 
查询：{query}
 
检索方式选项：
1. semantic_vector：语义向量检索，适合理解性的问题
2. keyword_bm25：关键词BM25检索，适合精确匹配
3. knowledge_graph：知识图谱检索，适合关系类问题
4. hybrid：混合检索，适合复杂问题
 
请返回JSON：
{{"strategy": "选择的检索方式", "reason": "选择原因"}}
""")
    
    # 2. 选择对应检索器
    strategy = json.loads(routing_decision)["strategy"]
    
    if strategy == "semantic_vector":
        return await retrievers["vector"].search(query)
    elif strategy == "keyword_bm25":
        return await retrievers["bm25"].search(query)
    elif strategy == "knowledge_graph":
        return await retrievers["kg"].search(query)
    else:
        return await retrievers["hybrid"].search(query)

五、实战代码

5.1 完整的查询改写管道

from dataclasses import dataclass
from typing import List, Optional
import asyncio
 
@dataclass
class QueryRewriteConfig:
    enable_expansion: bool = True
    enable_spellcheck: bool = True
    enable_hyde: bool = False
    expansion_count: int = 3
    max_retries: int = 2
 
class QueryRewriter:
    """查询改写管道"""
    
    def __init__(self, llm, config: QueryRewriteConfig = None):
        self.llm = llm
        self.config = config or QueryRewriteConfig()
    
    async def rewrite(self, query: str) -> dict:
        """
        执行完整的查询改写流程
        """
        
        results = {
            "original": query,
            "corrected": query,
            "rewritten": query,
            "expansions": []
        }
        
        # 1. 拼写检查
        if self.config.enable_spellcheck:
            results["corrected"] = await self._spellcheck(query)
        
        # 2. 歧义消解
        disambig_result = await self._disambiguate(results["corrected"])
        if disambig_result["is_ambiguous"]:
            results["rewritten"] = disambig_result["selected_query"]
            results["interpretations"] = disambig_result["interpretations"]
        else:
            results["rewritten"] = results["corrected"]
        
        # 3. 查询扩展
        if self.config.enable_expansion:
            results["expansions"] = await self._expand_query(
                results["rewritten"], 
                self.config.expansion_count
            )
        
        return results
    
    async def _spellcheck(self, query: str) -> str:
        """拼写检查"""
        prompt = f"请检查并纠正以下文本的拼写错误，只返回纠正后的文本：\n{query}"
        return (await self.llm.generate(prompt)).strip()
    
    async def _disambiguate(self, query: str) -> dict:
        """歧义消解"""
        prompt = f"""
分析以下查询是否存在歧义，返回JSON：
{{"is_ambiguous": bool, "selected_query": str, "interpretations": []}}
 
查询：{query}
"""
        result = await self.llm.generate(prompt)
        try:
            return json.loads(result)
        except:
            return {"is_ambiguous": False, "selected_query": query}
    
    async def _expand_query(self, query: str, count: int) -> List[str]:
        """查询扩展"""
        prompt = f"""
为以下查询生成{count}个语义相近的变体，返回JSON：
{{"expansions": []}}
 
查询：{query}
"""
        result = await self.llm.generate(prompt)
        try:
            return json.loads(result).get("expansions", [])
        except:
            return []

5.2 改写后的检索

class RewriteRetrievePipeline:
    """改写+检索管道"""
    
    def __init__(self, rewriter: QueryRewriter, retriever):
        self.rewriter = rewriter
        self.retriever = retriever
    
    async def search(self, query: str, top_k: int = 10):
        # 1. 改写查询
        rewrite_result = await self.rewriter.rewrite(query)
        
        # 2. 收集所有查询
        all_queries = [rewrite_result["rewritten"]] + rewrite_result["expansions"]
        
        # 3. 并行检索
        all_results = []
        tasks = [self.retriever.search(q, top_k=top_k) for q in all_queries]
        results_list = await asyncio.gather(*tasks)
        
        for results in results_list:
            all_results.extend(results)
        
        # 4. 去重和排序
        unique_results = self._deduplicate(all_results)
        
        return {
            "query_info": rewrite_result,
            "results": unique_results[:top_k]
        }
    
    def _deduplicate(self, results):
        """简单去重"""
        seen_ids = set()
        unique = []
        
        for r in results:
            if r.doc_id not in seen_ids:
                seen_ids.add(r.doc_id)
                unique.append(r)
        
        return unique

5.3 使用示例

# 初始化
rewriter = QueryRewriter(llm=your_llm)
pipeline = RewriteRetrievePipeline(rewriter, retriever)
 
# 使用
result = await pipeline.search("电脑黑屏了怎么办", top_k=10)
 
print("原始查询：", result["query_info"]["original"])
print("纠正后：", result["query_info"]["corrected"])
print("扩展查询：", result["query_info"]["expansions"])
print(f"找到 {len(result['results'])} 条结果")

六、进阶技巧

6.1 查询改写的陷阱

陷阱1：过度改写 改写得太多，反而偏离原意。

# 原始：苹果公司创始人是谁
# 过度改写后：Apple Inc. founding members
# 结果：检索到Tim Cook变成创始人的错误内容

陷阱2：引入噪声 扩展的查询带入了不相关的概念。

# 原始：iPhone怎么截图
# 扩展后：iPhone screenshot capture Apple device
# 结果：可能匹配到iPad截图的内容

解决方案：设置改写上限

MAX_EXPANSION = 5
MIN_SIMILARITY = 0.7  # 扩展查询与原查询的最小相似度

6.2 缓存改写结果

频繁查询的改写结果可以缓存：

from functools import lru_cache
 
@lru_cache(maxsize=1000)
def get_cached_expansion(query: str):
    """缓存查询扩展结果"""
    return asyncio.run(expand_query(query))

6.3 A/B测试改写策略

async def test_rewrite_strategies(query, retriever):
    """测试不同改写策略的效果"""
    
    strategies = {
        "no_rewrite": lambda q: [q],
        "simple_expansion": lambda q: expand_query_simple(q),
        "llm_expansion": lambda q: expand_query(q, llm),
        "hyde": lambda q: hyde_query(q, llm),
    }
    
    results = {}
    
    for name, strategy in strategies.items():
        queries = await strategy(query)
        all_results = [await retriever.search(q) for q in queries]
        
        # 评估效果
        results[name] = {
            "num_results": len(all_results),
            "avg_score": sum(r.score for r in all_results) / len(all_results) if all_results else 0
        }
    
    return results

七、常见问题

7.1 改写后效果反而变差

可能的原因：

扩展的查询偏离原意
引入了噪声概念
改写次数太多

解决方案：

限制扩展数量
人工审核改写结果
设置相似度阈值

7.2 延迟太高

改写管道增加了LLM调用次数。

解决方案：

缓存常用查询的改写结果
并行执行独立的改写步骤
用更小的模型做改写

7.3 改写质量不稳定

LLM改写的结果可能每次不一样。

解决方案：

多次改写取交集
人工设定改写模板
用更稳定的模型

八、总结

查询改写的核心：

弥合用户语言和知识库语言的鸿沟
提高召回率，减少漏检
但不要过度改写，适得其反

常用技术：

查询扩展：生成多个相关查询
同义词替换：用同义词扩展查询
HyDE：生成假答案辅助检索
意图识别：理解用户真正想问什么
查询路由：根据问题类型选择检索方式

实战建议：

先用简单的改写（如同义词扩展）
效果不够再加LLM改写
始终监控改写带来的副作用

人工智能知识库

探索

查询改写与扩展

查询改写与扩展：让RAG听懂人话

前言：语言的千变万化

一、为什么需要查询改写

1.1 语言的多样性

1.2 检索的尴尬

1.3 查询改写的价值

二、查询改写的核心技术

2.1 查询扩展

2.2 同义词替换

2.3 HyDE：生成假答案

2.4 意图识别

三、查询纠错

3.1 错别字纠正

3.2 歧义消解

3.3 伪相关反馈（PRF）

四、查询路由

4.1 什么是查询路由

4.2 特征检测

4.3 动态路由

五、实战代码

5.1 完整的查询改写管道

5.2 改写后的检索

5.3 使用示例

六、进阶技巧

6.1 查询改写的陷阱

6.2 缓存改写结果

6.3 A/B测试改写策略

七、常见问题

7.1 改写后效果反而变差

7.2 延迟太高

7.3 改写质量不稳定

八、总结

相关主题

关系图谱

目录

反向链接