RAG查询改写实战 意图识别、子问题分解与查询路由

一、检索效果差，问题往往出在查询端

很多团队把RAG调优的重点放在分块和向量库上，却忽略了一个事实：用户很少会用"文档的语言"提问。真实查询往往短、口语化、带指代，甚至一个问题里藏着多个意图。查询端不处理，召回率天花板就在那里。

二、意图识别：先判断"该不该检索"

不是所有问题都需要检索。闲聊、问候、直接计算类问题走检索只会引入噪声。用规则兜底 + LLM分类的混合方案，兼顾准确率和成本：

import re

INTENT_KEYWORDS = {
    "greeting": ["你好", "hi", "hello", "在吗"],
    "calculation": ["计算", "等于", "多少加", "换算"],
    "chitchat": ["谢谢", "再见", "你是谁"],
}

def rule_intent(query: str) -> str | None:
    """规则层：快速识别无需检索的意图"""
    for intent, words in INTENT_KEYWORDS.items():
        if any(w in query for w in words):
            return intent
    return None

def classify_intent(query: str, llm) -> str:
    """LLM层：兜底分类，返回 retrieval / greeting / calculation / chitchat"""
    rule_result = rule_intent(query)
    if rule_result:
        return rule_result
    prompt = f"""判断以下用户问题的意图，只输出一个词：
- retrieval：需要查资料才能回答
- greeting：打招呼
- calculation：直接计算即可
- chitchat：闲聊

问题：{query}
意图："""
    resp = llm.chat(prompt, max_tokens=10, temperature=0)
    intent = resp.strip().lower()
    return intent if intent in ("retrieval", "greeting", "calculation", "chitchat") else "retrieval"

规则层零成本秒回，只有规则覆盖不到时才调用LLM，约70%的流量不需要过LLM分类器。

三、查询改写：把口语翻译成"文档的语言"

识别为 retrieval 后，下一步是改写。三种高频改写策略：

def rewrite_query(query: str, history: list, llm) -> str:
    """查询改写：补全指代 + 术语归一 + 必要扩展"""
    prompt = f"""你是一名检索专家。请将用户的模糊问题改写为适合向量检索的清晰查询。
要求：
1. 结合对话历史补全指代（它/这个/上个月）
2. 将口语表达转换为书面术语
3. 保留核心实体，不添加原文没有的事实
4. 只输出改写后的查询，不要解释

对话历史：
{chr(10).join(history[-4:])}

用户问题：{query}

改写结果："""
    rewritten = llm.chat(prompt, max_tokens=100, temperature=0.2).strip()
    # 兜底：改写失败或返回空则用原查询
    return rewritten if len(rewritten) >= 2 else query

改写质量的验收标准：改写前后向量相似度应保持在0.7以上——如果改得面目全非，说明LLM在编造信息，宁可退回原查询。

四、子问题分解：复杂问题拆着问

对复合意图查询，先用LLM拆解成子问题，分别检索，再让LLM基于多路证据综合回答：

def decompose_query(query: str, llm) -> list[str]:
    prompt = f"""把下面的复杂问题拆解为2-4个独立的子问题。
要求：每个子问题只问一件事，且都能独立检索资料回答。

问题：{query}

子问题（每行一个）："""
    lines = [l.strip("- ").strip() for l in llm.chat(prompt, max_tokens=200).splitlines()]
    return [l for l in lines if l and len(l) > 3][:4]

def multi_hop_retrieve(query: str, llm, retriever, top_k=3):
    """多跳检索：分解 -> 分别召回 -> 汇总证据"""
    sub_questions = decompose_query(query, llm)
    evidence = []
    for sq in sub_questions:
        # 每个子问题独立改写后检索
        rq = rewrite_query(sq, [], llm)
        evidence.extend(retriever.search(rq, k=top_k))
    return deduplicate(evidence), sub_questions

子问题分解适合对比类、方案类、流程类问题；简单事实类问题直接检索即可，不要过度设计。

五、查询路由：不同查询走不同通道

路由是查询改写的最后一环：把改写好、分解好的查询分发到最合适的检索通道。

class QueryRouter:
    def __init__(self, vector_store, keyword_index, sql_engine=None):
        self.vector_store = vector_store      # 语义检索
        self.keyword_index = keyword_index    # BM25 关键词检索
        self.sql_engine = sql_engine          # 结构化查询

    def route(self, query: str) -> str:
        # 规则：含明确数值/筛选词 -> 结构化查询
        if re.search(r"\d{4}年|\d+月|大于|小于|排名|TOP", query):
            return "sql"
        # 含专有名词/型号/编号 -> 关键词检索优先
        if re.search(r"[A-Z]{2,}|型号|编号|版本", query):
            return "keyword"
        # 默认走混合检索
        return "hybrid"

    def search(self, query: str, top_k=5):
        channel = self.route(query)
        if channel == "sql" and self.sql_engine:
            return self.sql_engine.query(query)
        if channel == "keyword":
            return self.keyword_index.search(query, k=top_k)
        # hybrid: 向量 + BM25 结果融合（RRF）
        return rrf_fusion(
            self.vector_store.search(query, k=top_k * 2),
            self.keyword_index.search(query, k=top_k * 2),
            top_k,
        )

路由策略要与你的知识库形态匹配：文档型知识库以 hybrid 为主；库表型数据一定要接 SQL 通道，别硬塞进向量库。

六、生产落地建议

**关键原则**：查询改写链路每一环都要有"兜底回退"。任何一环失败或超时，直接使用原始查询继续检索，绝不让改写链路本身成为可用性瓶颈。建议对改写质量做A/B抽样评估，用真实用户查询日志持续迭代。