Query Rewrite 的核心作用：把“人类表达”变成“检索表达”

一、为什么 RAG 需要 Query Rewrite？

很多 RAG 项目上线后，最大的问题不是模型不会回答，而是检索阶段就没有把正确资料找回来。用户的问题通常很口语、很短、带指代，甚至只说“这个怎么弄”。而企业知识库里的文档，往往是制度标题、字段名、流程名、产品术语。Query Rewrite 的价值，就是在检索之前做一次“翻译”：把用户的自然表达，改写成更容易命中文档的检索表达。

二、在线 RAG 链路里，Query Rewrite 放在哪里？

在在线 RAG 中，Query Rewrite 一般放在检索之前。它不是最终答案生成，而是让后面的检索、融合、精排、Prompt 组装更稳定。

三、方法一：直接改写，把问题补完整

直接改写最常见，也最容易落地。它适合处理口语化、省略、指代、多轮上下文缺失等问题。核心要求是：只补全，不篡改。

代码示例（Python，可复制）：

from pydantic import BaseModel, Field
from typing import List

class RewriteResult(BaseModel):
rewritten_query: str = Field(description="补全后的检索问题")
keywords: List[str] = Field(description="可用于 BM25 的关键词")
keep_original: bool = Field(default=True, description="是否保留原问题一起检索")
risk: str = Field(description="low / medium / high，表示改写漂移风险")

SYSTEM_PROMPT = """
你是 RAG 查询改写器。
要求：
1. 只能补全用户原意，不能新增用户没有表达的业务诉求。
2. 结合最近 3 轮对话，消解“它、这个、那个、上次”等指代。
3. 输出必须是 JSON，不要回答问题本身。
"""

def build_rewrite_prompt(history: list[str], user_query: str) -> str:
return f"""
最近对话：
{chr(10).join(history[-3:])}

用户当前问题：{user_query}

请输出 RewriteResult JSON。
"""

四、方法二：HyDE，先生成“假文档”再检索

HyDE 的思路是：原始 Query 太短，直接 Embedding 不稳定；那就先让模型生成一段“假设性的答案/文档”，再用这段更像知识库文本的内容做向量检索。注意，HyDE 不是相信假答案，而是用它改善召回。

代码示例（Python，可复制）：

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
from llama_index.core.indices.query.query_transform import HyDEQueryTransform
from llama_index.core.query_engine import TransformQueryEngine

# 1. 构建索引
documents = SimpleDirectoryReader("./docs").load_data()
index = VectorStoreIndex.from_documents(documents)

# 2. 普通查询引擎
base_engine = index.as_query_engine(similarity_top_k=8)

# 3. 使用 HyDE 改写：生成 hypothetical document 后再检索
hyde = HyDEQueryTransform(include_original=True)
query_engine = TransformQueryEngine(base_engine, query_transform=hyde)

response = query_engine.query("新员工怎么申请电脑？")
print(response)

五、方法三：Step-back，先退一步问上位问题

有些知识库不是按具体问题组织，而是按制度、原则、流程组织。用户问得越具体，越可能检索不到上位规则。Step-back 会先把问题抽象成更高层问题，再和原问题一起检索。

代码示例（Python，可复制）：

def build_step_back_queries(user_query: str) -> list[str]:
"""
真实生产中这里通常由 LLM 生成。
输出两路查询：一路保留具体问题，一路抽象成上位规则。
"""
return [
user_query,
"与该问题相关的制度依据、处理流程、常见原因和审批规则是什么？",
]

queries = build_step_back_queries("报销发票被退回怎么办？")
for q in queries:
docs = retriever.search(q, top_k=8)
# 后续进入融合排序

六、方法四：Multi-Query + RRF，多角度召回再融合

Multi-Query 会把一个问题改写成多个等价或互补表达，让检索从多个角度去找资料。多个召回列表不能简单拼接，一般要用 RRF 或 Rerank 做融合。

代码示例（Python，可复制）：

from collections import defaultdict

def reciprocal_rank_fusion(result_lists: list[list[str]], k: int = 60) -> list[tuple[str, float]]:
"""
result_lists: 多路召回结果，每一路是按相关性排序的 doc_id 列表。
RRF 思路：同一篇文档在多个列表靠前，最终分数就更高。
"""
scores = defaultdict(float)

for results in result_lists:
for rank, doc_id in enumerate(results, start=1):
scores[doc_id] += 1.0 / (k + rank)

return sorted(scores.items(), key=lambda x: x[1], reverse=True)

result_lists = [
["doc_1", "doc_2", "doc_5"],
["doc_2", "doc_4", "doc_1"],
["doc_3", "doc_2", "doc_6"],
]
print(reciprocal_rank_fusion(result_lists))

代码示例（Python，可复制）：

# LangChain MultiQueryRetriever 示例：让 LLM 生成多条检索 Query
from langchain.retrievers.multi_query import MultiQueryRetriever

retriever = vectorstore.as_retriever(search_kwargs={"k": 6})

multi_query_retriever = MultiQueryRetriever.from_llm(
retriever=retriever,
llm=llm,
)

docs = multi_query_retriever.invoke("员工出差费用怎么报销？")

七、方法五：Self-Query，从自然语言里抽取过滤条件

如果你的文档有 metadata，例如部门、年份、类型、权限、产品线，那么 Self-Query 会非常有用。它把用户问题拆成“语义检索词 + 结构化过滤条件”。

代码示例（Python，可复制）：

from pydantic import BaseModel, Field
from typing import Optional, Literal

class StructuredQuery(BaseModel):
query: str = Field(description="用于向量/BM25 检索的文本")
doc_type: Optional[Literal["制度", "流程", "FAQ", "公告"]] = None
year_gte: Optional[int] = None
department: Optional[str] = None

# 用户问题：查一下 2025 年以后跟医保报销有关的制度
structured = StructuredQuery(
query="医保报销",
doc_type="制度",
year_gte=2025,
)

# 注意：生产中必须把 filter 映射到白名单字段，不能让模型直接拼 SQL。
filter_expr = {
"doc_type": structured.doc_type,
"year": {"$gte": structured.year_gte},
}

八、方法六：Router Query，先判断去哪套知识库查

当知识库很多时，先路由再检索，比全库乱搜更稳定。路由可以按业务域、语言、权限、数据源类型来做。

代码示例（Python，可复制）：

ROUTES = {
"finance": "财务制度库：发票、报销、付款、预算",
"hr": "人事制度库：入离职、薪酬、考勤、福利",
"it": "IT 帮助库：账号、权限、系统故障、设备申请",
}

def route_query(user_query: str) -> str:
prompt = f"""
你是查询路由器，只能从 finance/hr/it 中选择一个。
路由说明：{ROUTES}
用户问题：{user_query}
只输出路由 key。
"""
route = llm.invoke(prompt).strip()
return route if route in ROUTES else "finance"

route = route_query("发票抬头错了怎么处理？")
docs = retrievers[route].search("发票抬头错误处理流程", top_k=8)

九、到底用哪种改写方式？

Query Rewrite 没有银弹。直接改写成本最低，HyDE 对表达差异大的问题有帮助，Multi-Query 更适合复杂问题，Self-Query 依赖高质量元数据，Router Query 适合多知识库系统。

十、工程上怎么评估 Query Rewrite？

评估时不要只看“改写得是否通顺”。真正要看：召回是否更准、排序是否更靠前、答案是否更忠实、延迟是否能接受、有没有改写漂移。

代码示例（Python，可复制）：

def evaluate_rewrite(original_query: str, rewritten_queries: list[str], gold_doc_ids: set[str]) -> dict:
"""
一个最小化离线评测示例。
真实项目里可以扩展 Recall@K、MRR、nDCG、Faithfulness、Latency 等指标。
"""
origin_docs = retriever.search(original_query, top_k=10)
rewrite_docs = []

for q in rewritten_queries:
rewrite_docs.extend(retriever.search(q, top_k=10))

origin_ids = {d.id for d in origin_docs}
rewrite_ids = {d.id for d in rewrite_docs}

return {
"origin_recall": len(origin_ids & gold_doc_ids) / max(len(gold_doc_ids), 1),
"rewrite_recall": len(rewrite_ids & gold_doc_ids) / max(len(gold_doc_ids), 1),
"new_hits": list((rewrite_ids - origin_ids) & gold_doc_ids),
}

十一、生产级护栏：改写可以增强，但不能乱改

Query Rewrite 最大的风险是“漂移”：用户本来问 A，改写器为了检索更顺手，改成了 B。生产环境里必须保留原 Query、输出结构化 JSON、控制扩展数量、记录日志，并对改写前后召回结果做对比。

代码示例（Python，可复制）：

class RewriteGuardrail:
def __init__(self, max_queries: int = 5, max_drift_score: float = 0.35):
self.max_queries = max_queries
self.max_drift_score = max_drift_score

def validate(self, original: str, rewrites: list[str]) -> list[str]:
# 1. 控制数量
rewrites = rewrites[: self.max_queries]

# 2. 保留原问题，避免改写失败时完全跑偏
candidates = [original] + rewrites

# 3. 漂移检测：实际可以用 embedding 相似度、LLM 判别或规则
safe_queries = []
for q in candidates:
drift = estimate_semantic_drift(original, q)
if drift <= self.max_drift_score:
safe_queries.append(q)

return safe_queries or [original]

十二、可以直接记住的面试回答

Query Rewrite 的目的，是把用户原始问题改写成更适合检索系统理解和命中的查询表达。它主要解决口语化、上下文缺失、表达不匹配、单一路径召回不足、结构化过滤不足等问题。

常见方式包括：直接改写、多轮指代消解、HyDE、Step-back、Multi-Query、Self-Query 和 Router Query。工程上不能只追求改写通顺，而要通过 Recall@K、MRR、nDCG、答案忠实度、延迟、成本和漂移率来评估。

生产环境里最稳的做法是：保留原 Query + 多路改写 + 混合检索 + RRF 融合 + Rerank 精排 + 引用校验 + 日志回放。