查询改写与扩展：让RAG系统更懂你的问题

引言

用户输入的查询往往简短、模糊或不完整。查询改写（Query Rewriting） 和 查询扩展（Query Expansion） 技术可以显著提升检索质量，是 RAG 系统中被低估但极其有效的优化手段。

原始查询："怎么部署？"（模糊、简略）
         │
         ▼
    ┌────────────┐
    │ 查询理解与改写 │
    └──────┬─────┘
           │
           ▼
改写后查询："如何使用Docker Compose部署Milvus向量数据库的生产环境配置？"
           （具体、完整、信息丰富）
           │
           ▼
        检索 → 高质量结果

一、为什么需要查询改写？

1.1 用户查询的常见问题

1.2 查询质量对检索的影响

查询清晰度 ──→ 检索质量 ──→ 生成质量
    │              │            │
    ▼              ▼            ▼
  模糊查询    召回率 < 50%   回答不准确
  清晰查询    召回率 > 85%   回答精准

二、查询改写技术

2.1 基于 LLM 的改写

from openai import OpenAI

def rewrite_query(
    query: str,
    conversation_history: list[dict] = None
) -> str:
    """使用 LLM 改写查询"""
    client = OpenAI()

    messages = [
        {"role": "system", "content": """你是查询改写专家。请根据以下规则改写用户查询：

1. 补充缺失的关键词和上下文
2. 将口语转换为书面语
3. 消除歧义和指代不明
4. 保持原意不变
5. 输出仅包含改写后的查询，不要其他内容"""}
    ]

    if conversation_history:
        messages.append({
            "role": "system",
            "content": f"对话历史：{conversation_history}"
        })

    messages.append({"role": "user", "content": query})

    response = client.chat.completions.create(
        model="gpt-4",
        messages=messages,
        temperature=0.1
    )
    return response.choices[0].message.content

# 示例
original = "它的部署方式是什么？"
context = [
    {"role": "user", "content": "我最近在看Milvus向量数据库"},
    {"role": "assistant", "content": "Milvus是一个分布式向量数据库..."}
]
rewritten = rewrite_query(original, context)
# 结果: "Milvus向量数据库的部署方式和配置要求是什么？"

2.2 多查询扩展（Multi-Query）

生成多个不同角度的查询，增加覆盖范围：

def multi_query_expansion(
    query: str,
    num_queries: int = 5
) -> list[str]:
    """生成多个查询变体"""
    client = OpenAI()

    response = client.chat.completions.create(
        model="gpt-4",
        messages=[
            {"role": "system", "content": f"""
生成 {num_queries} 个不同角度的查询，覆盖：
- 同义词替换
- 不同表达方式
- 更具体/更抽象的版本
- 子问题分解

每行一个查询，不要编号。"""",
            {"role": "user", "content": query}
        ],
        temperature=0.7
    )

    queries = response.choices[0].message.content.strip().split('\n')
    return [q.strip() for q in queries if q.strip()]

# 示例
queries = multi_query_expansion(
    "向量数据库怎么选型？"
)
# 输出:
# 1. "向量数据库选型指南"
# 2. "如何选择向量数据库"
# 3. "Pinecone vs Milvus vs Chroma 对比"
# 4. "向量数据库选型考虑因素"
# 5. "最适合RAG系统的向量数据库"

三、HyDE（假设性文档嵌入）

HyDE 是一种特殊的查询扩展技术：先生成一个假设性的理想答案，然后用这个答案去检索真实文档。

用户查询 ──→ LLM生成假设答案 ──→ Embedding假设答案 ──→ 检索真实文档
    │              │                   │                    │
    ▼              ▼                   ▼                    ▼
"Milvus性能"  "Milvus的QPS达到..."  [向量编码]         找到相关文档

from openai import OpenAI
from langchain_chroma import Chroma

class HyDERetriever:
    def __init__(self, vector_store):
        self.vector_store = vector_store
        self.llm = OpenAI()

    def generate_hypothetical_doc(self, query: str) -> str:
        """生成假设文档"""
        response = self.llm.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": """
生成一个详细的假设文档来回答用户的问题。
这个文档应该：
1. 直接回答问题
2. 包含具体细节和数据
3. 使用专业和精确的语言
4. 长度在 100-200 字之间
                """},
                {"role": "user", "content": query}
            ],
            temperature=0.7
        )
        return response.choices[0].message.content

    def retrieve(self, query: str, k: int = 10) -> list[dict]:
        # 1. 生成假设文档
        hypo_doc = self.generate_hypothetical_doc(query)

        # 2. 用假设文档检索（而非原始查询）
        results = self.vector_store.similarity_search_with_score(
            hypo_doc,
            k=k
        )

        return results

# 使用
hyde_retriever = HyDERetriever(vector_store)
results = hyde_retriever.retrieve("Milvus的索引类型有哪些？")

四、查询分解

将复杂查询分解为多个子查询，分别检索后汇总：

def decompose_query(query: str) -> list[str]:
    """将复杂查询分解为子查询"""
    client = OpenAI()

    response = client.chat.completions.create(
        model="gpt-4",
        messages=[
            {"role": "system", "content": """
将复杂查询分解为多个独立的子问题。
每个子问题应该：
1. 语义完整，可以独立检索
2. 覆盖原始问题的不同方面
3. 列出 3-5 个子问题

格式：每行一个子问题，不要编号。"""},
            {"role": "user", "content": query}
        ],
        temperature=0.3
    )

    sub_queries = response.choices[0].message.content.strip().split('\n')
    return [q.strip() for q in sub_queries if q.strip()]


class DecomposedRetriever:
    """分解查询后分别检索"""
    def __init__(self, base_retriever):
        self.base_retriever = base_retriever

    def retrieve(self, query: str, k: int = 5) -> list[dict]:
        # 分解查询
        sub_queries = decompose_query(query)

        # 分别检索
        all_results = []
        for sub_q in sub_queries:
            results = self.base_retriever.invoke(sub_q)
            all_results.extend(results)

        # 去重
        seen_ids = set()
        unique_results = []
        for doc in all_results:
            if doc.id not in seen_ids:
                seen_ids.add(doc.id)
                unique_results.append(doc)

        return unique_results[:k]

# 示例
query = "比较Milvus和Pinecone在性能、成本和部署难度上的差异"
# 分解为:
# - "Milvus的性能指标和基准测试"
# - "Pinecone的定价方案"
# - "Milvus的部署要求"
# - "Pinecone的部署配置"

五、与 RAG 系统集成

5.1 完整的查询优化管线

class QueryOptimizer:
    """查询优化流水线"""
    def __init__(self):
        self.reranker = CrossEncoder('BAAI/bge-reranker-v2-m3')

    def optimize(self, query: str, history: list = None) -> dict:
        """
        对查询进行全方位优化
        返回: 优化后的查询和检索策略
        """
        result = {
            'original': query,
            'rewritten': None,
            'expansions': [],
            'sub_queries': [],
            'use_hypothetical': False
        }

        # 1. 判断查询是否需要改写
        if self._needs_rewrite(query):
            result['rewritten'] = rewrite_query(query, history)

        # 2. 判断是否适合多查询扩展
        if self._is_broad_query(query):
            result['expansions'] = multi_query_expansion(query)

        # 3. 判断是否适合查询分解
        if self._is_complex_query(query):
            result['sub_queries'] = decompose_query(query)

        # 4. 判断是否适合 HyDE
        if self._needs_precision(query):
            result['use_hypothetical'] = True

        return result

    def _needs_rewrite(self, query: str) -> bool:
        """判断是否需要改写"""
        return len(query) < 10 or any(
            word in query for word in ['它', '如何', '那个', '这个']
        )

    def _is_complex_query(self, query: str) -> bool:
        """判断是否为复杂查询"""
        return len(query) > 20 and any(
            word in query for word in ['对比', '比较', '和', '与']
        )

5.2 LangChain 集成

from langchain.chains.query_constructor.base import (
    StructuredQueryOutputParser,
    get_query_constructor_prompt
)

# LangChain 自带查询构造
from langchain.retrievers.self_query.base import SelfQueryRetriever

# 查询转换链
from langchain.chains.query_rewriting import QueryRewritingChain

query_rewriter = QueryRewritingChain.from_llm(
    llm=ChatOpenAI(model="gpt-4", temperature=0),
    rewrite_template="""
    将以下用户查询改写得更加清晰、完整，适合用于文档检索。
    
    原始查询: {query}
    
    改写后的查询:
    """
)

六、查询优化的效果对比

七、总结

查询改写与扩展的核心经验：

1. 简短查询收益最大 — 5个字以下的查询，改写后召回率提升最为显著
2. HyDE 适合知识密集型 — 对”概念对比”、“原因分析”类问题效果突出
3. 多查询扩展更稳 — 比单个改写更可靠，但延迟更高
4. 不要过度优化 — 简单的查询直接用原始语义检索即可
5. 组合使用要权衡 — 查询改写 → 多查询扩展 → HyDE 效果递进，但延迟也递增

相关资源：

• HyDE 论文
• Query Rewriting in RAG
• LangChain Multi-Query Retriever