第8章：查询增强技术

用户的问题往往不够精确。查询增强技术可以"理解"用户意图，将复杂问题的检索质量提升15-20%。

---

📚 学习目标

学完本章后，你将能够：

• 理解查询增强的原理和价值
• 掌握HyDE技术的实现
• 实现查询重写
• 应用多查询策略
• 提升复杂问题的准确率

预计学习时间：3小时 难度等级：⭐⭐⭐☆☆

---

前置知识

• 完成模块1的基础RAG实现
• 理解向量检索原理
• 熟悉OpenAI API使用

环境要求： - OpenAI API（或类似） - 向量数据库

---

8.1 为什么需要查询增强？

问题：用户查询不精确

场景1：模糊查询

用户问题："那个修电脑的东西怎么用？"

实际意图："如何使用螺丝刀？"

问题：直接检索"修电脑的东西"无法找到相关文档

场景2：信息缺失

用户问题："提高代码性能"

缺失信息：什么语言？什么场景？
优化查询："如何提高Python代码的性能？"

场景3：复杂多跳

用户问题："比较Python和JavaScript在Web开发中的差异"

需要分解：
1. Python在Web开发中的特点
2. JavaScript在Web开发中的特点
3. 两者的对比

解决方案：查询增强

原始查询
  ↓
查询增强
  ├─ 重写：表达更清晰
  ├─ 扩展：增加相关信息
  ├─ 分解：拆分成子查询
  └─ 生成：假设答案
  ↓
增强查询
  ↓
检索质量提升

---

8.2 HyDE（假设文档嵌入）

原理

HyDE (Hypothetical Document Embeddings)：先让LLM生成假设答案，然后用假设答案进行检索。

用户问题："Python的性能优势是什么？"

步骤1：生成假设答案
  LLM生成："Python具有简洁的语法、动态类型系统、丰富的库..."

步骤2：嵌入假设答案
  假设向量 = embed("Python具有简洁的语法...")

步骤3：检索
  用假设向量检索，找到类似的文档

步骤4：生成最终答案
  基于检索到的文档生成准确答案

为什么HyDE有效？

核心思想：问题和答案在语义空间中更接近

语义空间中：

问题向量：[查询方向]
  ⬆
  ⬆
答案向量：[答案方向]

直接检索问题 ⚠️
  距离可能较远

HyDE：用答案向量检索 ✅
  距离更近，检索更准确

实现代码

# 文件名：08_01_hyde_implementation.py
"""
HyDE实现
"""

from openai import OpenAI
import numpy as np
from typing import List

class HyDEQueryEngine:
    """HyDE查询引擎"""

    def __init__(self, llm_model="gpt-3.5-turbo"):
        self.client = OpenAI()
        self.llm_model = llm_model

    def generate_hypothetical_answer(self, query: str) -> str:
        """
        生成假设答案

        Args:
            query: 原始查询

        Returns:
            假设答案
        """
        prompt = f"""
请基于以下问题，提供一个详细的假设答案。

问题：{query}

要求：
1. 提供详细、具体的答案
2. 假设答案是事实性的
3. 包含相关技术细节

假设答案：
"""

        response = self.client.chat.completions.create(
            model=self.llm_model,
            messages=[{"role": "user", "content": prompt}],
            temperature=0.7,
            max_tokens=200
        )

        hypothetical_doc = response.choices[0].message.content
        return hypothetical_doc

    def hyde_retrieve(self, query: str, vector_store, top_k: int = 3):
        """
        使用HyDE检索

        Args:
            query: 用户查询
            vector_store: 向量数据库
            top_k: 返回文档数

        Returns:
            检索结果
        """
        print("="*60)
        print("HyDE检索演示")
        print("="*60 + "\n")

        # 步骤1：生成假设答案
        print("步骤1: 生成假设答案")
        hypothetical_answer = self.generate_hypothetical_answer(query)
        print(f"原始问题: {query}")
        print(f"假设答案: {hypothetical_answer[:150]}...\n")

        # 步骤2：嵌入假设答案
        print("步骤2: 嵌入假设答案")
        hypothetical_embedding = vector_store.get_embedding(hypothetical_answer)
        print(f"嵌入维度: {len(hypothetical_embedding)}\n")

        # 步骤3：检索
        print("步骤3: 基于假设答案检索")
        results = vector_store.query(hypothetical_embedding, top_k=top_k)

        print("检索结果:")
        for i, (doc, score) in enumerate(zip(results['documents'], results['scores']), 1):
            print(f"  {i}. 相似度 {score:.3f}: {doc[:80]}...")

        return results

# 使用示例
if __name__ == "__main__":
    # 假设有一个简单的向量存储
    class SimpleVectorStore:
        def __init__(self):
            self.docs = [
                "Python是一种高级编程语言，具有简洁的语法和强大的功能。",
                "Python的性能优势包括：简洁的语法减少开发时间，丰富的库支持快速开发。",
                "JavaScript主要用于Web前端开发，可以创建动态的网页内容。"
            ]

        def get_embedding(self, text):
            # 简化：返回随机向量
            return np.random.rand(768)

        def query(self, embedding, top_k=3):
            # 简化：返回所有文档
            return {
                'documents': self.docs,
                'scores': [0.9, 0.7, 0.5]
            }

    store = SimpleVectorStore()
    hyde = HyDEQueryEngine()

    query = "Python有什么优势？"
    results = hyde.hyde_retrieve(query, store)

HyDE vs 直接检索对比

# 文件名：08_02_hyde_comparison.py
"""
HyDE vs 直接检索对比
"""

def compare_retrieval_methods(query, vector_store):
    """
    对比HyDE和直接检索

    Args:
        query: 查询
        vector_store: 向量数据库

    Returns:
        对比结果
    """
    print("="*70)
    print("HyDE vs 直接检索对比")
    print("="*70 + "\n")

    # 方法1：直接检索
    print("方法1: 直接检索")
    query_emb = vector_store.get_embedding(query)
    results_direct = vector_store.query(query_emb, top_k=3)

    print("检索结果:")
    for i, doc in enumerate(results_direct['documents'][:3], 1):
        print(f"  {i}. {doc[:80]}...")
    print()

    # 方法2：HyDE
    print("方法2: HyDE检索")
    hyde = HyDEQueryEngine()
    results_hyde = hyde.hyde_retrieve(query, vector_store)

    # 对比
    print("\n" + "="*70)
    print("对比总结")
    print("="*70)
    print("""
直接检索:
  优点: 简单快速，只需一次嵌入和检索
  缺点: 对模糊查询效果差

HyDE检索:
  优点: 对模糊查询、复杂问题效果好
  缺点: 需要额外LLM调用，增加延迟和成本
    """)

    return results_direct, results_hyde

---

8.3 查询重写

原理

将用户的模糊查询重写为更清晰、更具体的查询。

原始查询："怎么用？"

重写查询："如何使用[产品名称]进行[具体操作]？"

实现代码

# 文件名：08_03_query_rewriting.py
"""
查询重写实现
"""

class QueryRewriter:
    """查询重写器"""

    def __init__(self, llm_model="gpt-3.5-turbo"):
        self.client = OpenAI()
        self.llm_model = llm_model

    def rewrite_query(self, query: str, context: str = "") -> str:
        """
        重写查询

        Args:
            query: 原始查询
            context: 上下文信息

        Returns:
            重写后的查询
        """
        prompt = f"""
你是一个查询优化专家。请将用户的模糊查询重写为更清晰、更具体的查询。

原始查询：{query}

{f"上下文：{context}" if context else ""}

要求：
1. 保持原意
2. 使其更具体
3. 添加相关细节
4. 保持简洁

重写后的查询：
"""

        response = self.client.chat.completions.create(
            model=self.llm_model,
            messages=[{"role": "user", "content": prompt}],
            temperature=0.3,  # 降低随机性
            max_tokens=100
        )

        rewritten_query = response.choices[0].message.content.strip()
        return rewritten_query

# 使用示例
if __name__ == "__main__":
    rewriter = QueryRewriter()

    # 测试查询
    test_queries = [
        "怎么用？",
        "性能好不好？",
        "支持中文吗？"
    ]

    print("="*60)
    print("查询重写演示")
    print("="*60 + "\n")

    for query in test_queries:
        print(f"原始查询: {query}")
        rewritten = rewriter.rewrite_query(query)
        print(f"重写查询: {rewritten}")
        print("-" * 60)

多轮重写

# 文件名：08_04_multi_turn_rewriting.py
"""
多轮查询重写
"""

class MultiTurnQueryRewriter:
    """多轮查询重写器"""

    def __init__(self):
        self.client = OpenAI()

    def expand_query(self, query: str, max_turns: int = 3) -> List[str]:
        """
        扩展查询（多轮对话）

        Args:
            query: 原始查询
            max_turns: 最大轮数

        Returns:
            查询变体列表
        """
        print("="*60)
        print("多轮查询扩展")
        print("="*60 + "\n")

        queries = [query]

        for turn in range(max_turns):
            print(f"第{turn+1}轮扩展")

            # 确定是否需要继续扩展
            prompt = f"""
原始查询：{query}

当前扩展：
{chr(10).join(f'{i+1}. {q}' for i, q in enumerate(queries))}

是否还需要扩展查询？
- 如果当前扩展已经足够清晰，回答"停止"
- 如果需要更多信息，提供一个新的查询变体
- 回答"停止"或提供新的变体

判断：
"""

            response = self.client.chat.completions.create(
                model="gpt-3.5-turbo",
                messages=[{"role": "user", "content": prompt}],
                temperature=0.3,
                max_tokens=50
            )

            decision = response.choices[0].message.content.strip()

            if decision.lower() in ["停止", "stop", "足够"]:
                print("  → 停止扩展")
                break
            else:
                queries.append(decision)
                print(f"  → 新增: {decision}")

        print(f"\n最终生成 {len(queries)} 个查询变体")
        return queries

---

8.4 多查询策略

原理

将一个查询扩展为多个相关查询，检索更多样化的结果。

原始查询："Python的性能如何？"

扩展为多个查询：
  1. "Python的执行速度如何？"
  2. "Python有哪些性能优化技巧？"
  3. "Python相比其他语言性能如何？"

分别检索 → 融合结果

实现代码

# 文件名：08_05_multi_query.py
"""
多查询策略实现
"""

from typing import List
import numpy as np

class MultiQueryEngine:
    """多查询引擎"""

    def __init__(self, vector_store):
        self.vector_store = vector_store

    def expand_query(self, query: str) -> List[str]:
        """
        扩展查询

        Args:
            query: 原始查询

        Returns:
            查询列表
        """
        # 这里可以使用LLM生成
        # 或使用规则扩展

        # 简化版规则扩展
        expansions = []

        # 原始查询
        expansions.append(query)

        # 添加同义词
        if "性能" in query:
            expansions.append(query.replace("性能", "速度"))
            expansions.append(query.replace("性能", "效率"))

        if "如何" in query:
            expansions.append(query.replace("如何", "怎么"))
            expansions.append(query.replace("如何", "方法"))

        # 去重
        return list(set(expansions))

    def multi_query_retrieve(self, query: str, top_k: int = 3):
        """
        多查询检索

        Args:
            query: 原始查询
            top_k: 每个查询返回的文档数

        Returns:
            融合后的检索结果
        """
        print("="*60)
        print("多查询检索演示")
        print("="*60 + "\n")

        # 步骤1：扩展查询
        queries = self.expand_query(query)
        print(f"扩展为 {len(queries)} 个查询:")
        for i, q in enumerate(queries, 1):
            print(f"  {i}. {q}")
        print()

        # 步骤2：分别检索
        all_results = {}
        for i, q in enumerate(queries, 1):
            print(f"检索查询 {i}: {q}")
            emb = self.vector_store.get_embedding(q)
            results = self.vector_store.query(emb, top_k=top_k)
            all_results[q] = results
            print(f"  找到 {len(results['documents'])} 个文档\n")

        # 步骤3：融合结果
        print("融合结果:")
        fused_results = self.reciprocal_rank_fusion(all_results)

        for i, (doc, score) in enumerate(zip(
            fused_results['documents'][:5],
            fused_results['scores'][:5]
        ), 1):
            print(f"  {i}. 相似度 {score:.3f}: {doc[:80]}...")

        return fused_results

    def reciprocal_rank_fusion(self, all_results: dict):
        """
        RRF融合算法

        Args:
            all_results: 所有查询的检索结果

        Returns:
            融合后的结果
        """
        # 收集所有文档
        all_docs = set()
        for results in all_results.values():
            all_docs.update(results['documents'])

        # 计算RRF分数
        scores = {}
        k = 60  # RRF常数

        for doc in all_docs:
            rrf_score = 0

            for results in all_results.values():
                # 找到该文档在这次检索中的排名
                try:
                    rank = results['documents'].index(doc) + 1
                except ValueError:
                    rank = len(results['documents']) + 1

                rrf_score += 1 / (k + rank)

            scores[doc] = rrf_score

        # 排序
        sorted_docs = sorted(scores.items(), key=lambda x: x[1], reverse=True)

        return {
            'documents': [doc for doc, _ in sorted_docs],
            'scores': [score for _, score in sorted_docs]
        }

# 使用示例
if __name__ == "__main__":
    class SimpleVectorStore:
        def __init__(self):
            self.docs = [
                "Python性能优化包括使用内置函数、避免全局变量等技巧。",
                "Python的执行速度相对较慢，但开发效率高。",
                "JavaScript是Web开发的主要语言。"
            ]

        def get_embedding(self, text):
            return np.random.rand(768)

        def query(self, emb, top_k):
            return {
                'documents': self.docs,
                'scores': [0.9, 0.7, 0.5]
            }

    store = SimpleVectorStore()
    engine = MultiQueryEngine(store)

    query = "Python的性能如何？"
    results = engine.multi_query_retrieve(query)

---

8.5 查询分解

原理

将复杂问题分解为多个子问题，分别检索后再综合。

复杂问题："比较Python和JavaScript在Web开发中的差异"

分解：
  子问题1：Python在Web开发中的应用
  子问题2：JavaScript在Web开发中的应用
  子问题3：两者的优缺点对比

分别检索 → 综合答案

实现代码

# 文件名：08_06_query_decomposition.py
"""
查询分解实现
"""

class QueryDecomposer:
    """查询分解器"""

    def __init__(self, llm_model="gpt-3.5-turbo"):
        self.client = OpenAI()

    def decompose_query(self, query: str) -> List[str]:
        """
        分解查询

        Args:
            query: 复杂查询

        Returns:
            子查询列表
        """
        prompt = f"""
将以下复杂问题分解为2-4个子问题。

原始问题：{query}

要求：
1. 每个子问题应该独立可回答
2. 子问题合起来应该能完整回答原问题
3. 子问题之间应该有逻辑关系

子问题（用JSON数组格式）：
"""

        response = self.client.chat.completions.create(
            model="gpt-3.5-turbo",
            messages=[{"role": "user", "content": prompt}],
            temperature=0.3,
            response_format={"type": "json_object"}
        )

        import json
        try:
            result = json.loads(response.choices[0].message.content)
            sub_queries = result.get("sub_queries", [])
        except:
            # 简化处理：按行分割
            sub_queries = response.choices[0].message.content.split("\n")

        return sub_queries

    def multi_step_query(self, query: str):
        """
        多步查询处理

        Args:
            query: 复杂查询

        Returns:
            综合答案
        """
        print("="*60)
        print("多步查询演示")
        print("="*60 + "\n")

        # 步骤1：分解
        print("步骤1: 分解查询")
        sub_queries = self.decompose_query(query)

        print(f"生成了 {len(sub_queries)} 个子查询:")
        for i, sub_q in enumerate(sub_queries, 1):
            print(f"  {i}. {sub_q}")
        print()

        # 步骤2：分别检索和回答
        print("步骤2: 分别检索和回答")
        answers = []

        for i, sub_q in enumerate(sub_queries, 1):
            print(f"\n子问题 {i}: {sub_q}")

            # 检索
            # （这里简化，实际应该调用检索器）
            retrieved_docs = [f"与{sub_q}相关的文档"]

            # 生成答案
            answer = f"这是关于{sub_q}的答案"
            answers.append(answer)

            print(f"  答案: {answer[:100]}...")

        # 步骤3：综合
        print("\n步骤3: 综合答案")
        final_answer = self.synthesize_answers(query, sub_queries, answers)

        return final_answer

    def synthesize_answers(self, original_query, sub_queries, answers):
        """
        综合多个答案

        Args:
            original_query: 原始问题
            sub_queries: 子问题列表
            answers: 答案列表

        Returns:
            综合答案
        """
        prompt = f"""
原始问题：{original_query}

子问题和答案：
{chr(10).join(f'Q{i+1}: {q}{chr(10)}A: {a}' for i, (q, a) in enumerate(zip(sub_queries, answers)))}

请基于以上信息，给出一个综合的、结构化的答案。
"""

        response = self.client.chat.completions.create(
            model="gpt-3.5-turbo",
            messages=[{"role": "user", "content": prompt}],
            temperature=0.7,
            max_tokens=500
        )

        return response.choices[0].message.content

# 使用示例
if __name__ == "__main__":
    decomposer = QueryDecomposer()

    complex_query = "比较Python和JavaScript在Web开发中的差异，并说明各自的适用场景。"

    final_answer = decomposer.multi_step_query(complex_query)

    print(f"\n最终答案:\n{final_answer}")

---

总结

本章要点回顾

1. 查询增强价值
2. 处理模糊查询
3. 补充缺失信息

4. 分解复杂问题

5. 核心技术

6. HyDE：生成假设答案检索
7. 查询重写：表达更清晰
8. 多查询：融合多个检索

9. 查询分解：分步处理

10. 效果提升

11. 简单问题：+5-10%
12. 复杂问题：+15-20%

学习检查清单

• 理解查询增强的必要性
• 掌握HyDE实现
• 能够重写查询
• 实现多查询策略
• 掌握查询分解

下一步学习

• 下一章：第9章：混合检索与重排序
• 目标：结合多种检索策略

实践练习

1. 基础练习
2. 实现HyDE检索
3. 测试查询重写效果

4. 实现多查询融合

5. 进阶练习

6. 组合HyDE和多查询
7. 优化查询分解策略

8. 测量性能提升

9. 挑战项目

10. 实现自适应查询增强
11. 处理多轮对话查询
12. 达到复杂问题准确率>75%

---

返回目录 | 上一章：高级分块策略 | 下一章：混合检索与重排序

---

本章结束

查询增强是处理复杂和模糊查询的关键技术。通过HyDE等技术，即使是不完美的问题也能找到相关的文档，这是让RAG系统更"聪明"的重要一步！