RAG系统优化：Query Expansion 与 Query Rewrite 完整迁移指南

在构建 RAG（检索增强生成）系统时，查询处理是决定答案质量的关键环节。许多团队发现，简单地将用户问题直接传入向量数据库往往无法获得理想的检索结果。这正是 Query Expansion（查询扩展）与 Query Rewrite（查询重写）技术发挥作用的地方。

本文将从工程实践角度，详细讲解如何使用 HolySheep AI 提供的 API 来实现这两种查询优化技术，并完成从传统方案到 HolySheep 的完整迁移。作为专注于 AI 推理的云服务提供商，HolySheep 提供低于 50ms 的响应延迟，同时支持微信、支付宝付款，并提供注册免费额度，让团队可以零成本开始优化。

为什么需要 Query Expansion 与 Query Rewrite

用户的自然语言问题往往存在以下问题：表达模糊、缺少上下文、包含口语化表述、或者涉及隐含假设。直接使用原始查询检索向量数据库，会因为语义匹配不精准而返回低相关性的文档。

Query Expansion 的核心思想是：将一个查询扩展为多个语义相近的查询，从而提高召回率。例如，用户问“深度学习框架”，系统可以自动扩展出 PyTorch、TensorFlow、JAX 等相关框架的查询。

Query Rewrite 则侧重于理解和修正查询本身：纠正拼写错误、分解复杂问题、补充缺失实体、统一表达风格等。例如，用户输入“怎么训练”，Rewrite 后可能变成“如何训练深度学习模型”。

迁移前的准备工作

在开始迁移之前，请确保完成以下准备工作。首先，你需要注册 HolySheep 账号并获取 API Key。HolySheep 的定价极具竞争力：GPT-4.1 为每百万 Token 8 美元，Claude Sonnet 4.5 为 15 美元，Gemini 2.5 Flash 仅为 2.50 美元，而 DeepSeek V3.2 更是低至 0.42 美元，相比其他主流服务商可节省 85% 以上的成本。注册后即可获得免费试用额度。

其次，确保你的开发环境已安装必要的依赖库。最后，备份现有配置文件和向量数据库索引，以便在迁移出现问题时能够快速回滚。

# 安装必要的依赖库
pip install openai requests python-dotenv

创建 .env 文件存储 API Key
echo "HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY" > .env

验证依赖安装成功
python -c "import openai; print('Dependencies installed successfully')"

方案一：使用 Query Expansion 提升召回率

Query Expansion 的实现思路是：先用 LLM 生成多个语义等价的查询，然后并行检索这些查询的结果，最后合并去重。下面是基于 HolySheep API 的完整实现代码：

import os
import json
from openai import OpenAI
from dotenv import load_dotenv

加载环境变量
load_dotenv()

初始化 HolySheep API 客户端
重要：base_url 必须使用 https://api.holysheep.ai/v1
client = OpenAI(
    api_key=os.getenv("HOLYSHEEP_API_KEY"),
    base_url="https://api.holysheep.ai/v1"
)

def generate_expanded_queries(original_query: str, num_expansions: int = 3) -> list:
    """
    使用 LLM 生成查询扩展
    生成多个语义相近但表达不同的查询，提高召回率
    """
    prompt = f"""你是一个查询优化专家。请将以下用户查询扩展为 {num_expansions} 个语义等价的查询。

要求：
1. 每个查询应该用不同的表达方式
2. 可以补充相关的实体和概念
3. 保持查询的原始意图
4. 输出格式为 JSON 数组

用户查询：{original_query}

扩展后的查询："""
    
    response = client.chat.completions.create(
        model="gpt-4.1",  # $8/MTok - 高性价比选择
        messages=[
            {"role": "system", "content": "你是一个查询优化专家。"},
            {"role": "user", "content": prompt}
        ],
        temperature=0.7,
        max_tokens=500
    )
    
    result = response.choices[0].message.content.strip()
    # 解析 JSON 响应
    try:
        expanded_queries = json.loads(result)
    except json.JSONDecodeError:
        # 如果 JSON 解析失败，手动提取
        expanded_queries = [line.strip() for line in result.split('\n') if line.strip()]
    
    return expanded_queries if isinstance(expanded_queries, list) else [original_query]


def expand_and_retrieve(query: str, vector_store, num_expansions: int = 3):
    """
    执行查询扩展并检索的完整流程
    """
    # 第一步：生成扩展查询
    expanded_queries = generate_expanded_queries(query, num_expansions)
    print(f"原始查询: {query}")
    print(f"扩展查询: {expanded_queries}")
    
    # 第二步：并行检索所有扩展查询
    all_results = []
    seen_ids = set()
    
    for expanded_query in expanded_queries:
        results = vector_store.search(expanded_query, top_k=5)
        for result in results:
            if result['id'] not in seen_ids:
                seen_ids.add(result['id'])
                all_results.append(result)
    
    # 第三步：按相关性分数排序
    all_results.sort(key=lambda x: x['score'], reverse=True)
    
    return all_results[:10]  # 返回前10个最相关结果


使用示例
if __name__ == "__main__":
    original_query = "深度学习框架比较"
    expanded = generate_expanded_queries(original_query)
    print(f"成功生成 {len(expanded)} 个扩展查询")

方案二：使用 Query Rewrite 提升精确度

Query Rewrite 与 Expansion 不同，它更关注于理解和修正原始查询，而不是生成多个变体。Rewrite 的典型应用场景包括：纠正拼写错误、分解复合问题、补充缺失的上下文、转换口语为正式表达等。

import os
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()

client = OpenAI(
    api_key=os.getenv("HOLYSHEEP_API_KEY"),
    base_url="https://api.holysheep.ai/v1"
)


class QueryRewriter:
    """
    查询重写器：优化用户查询以提高检索质量
    """
    
    def __init__(self, model: str = "gpt-4.1"):
        self.model = model
    
    def rewrite(self, query: str, rewrite_type: str = "standard") -> dict:
        """
        执行查询重写
        
        rewrite_type 选项:
        - "standard": 标准重写，纠正错误并清晰化表达
        - "decompose": 分解复杂问题为多个子问题
        - "expand": 补充上下文和实体信息
        - "formal": 将口语转换为正式表达
        """
        
        prompts = {
            "standard": f"""请重写以下查询，纠正任何拼写错误，清晰化模糊表达，
并确保查询可以被向量数据库准确检索。

原始查询：{query}

重写后的查询：""",
            
            "decompose": f"""请将以下复杂查询分解为3-5个简单的子查询。
每个子查询应该聚焦于一个具体方面。

原始查询：{query}

分解后的子查询（JSON数组格式）：""",
            
            "expand": f"""请扩展以下查询，补充相关的背景信息、实体名称和上下文。
扩展后的查询应该更详细、更具体。

原始查询：{query}

扩展后的查询：""",
            
            "formal": f"""请将以下口语化表达转换为正式的技术查询表述。
保持查询的核心意图，但使用更规范的技术术语。

原始查询：{query}

正式化后的查询："""
        }
        
        response = client.chat.completions.create(
            model=self.model,
            messages=[
                {"role": "system", "content": "你是一个专业的查询优化助手。"},
                {"role": "user", "content": prompts.get(rewrite_type, prompts["standard"])}
            ],
            temperature=0.3,  # 低温度确保输出稳定
            max_tokens=300
        )
        
        rewritten = response.choices[0].message.content.strip()
        
        return {
            "original": query,
            "rewritten": rewritten,
            "type": rewrite_type,
            "model": self.model,
            "latency_ms": response.response_headers.get("x-latency", "N/A")
        }


def intelligent_rewrite_pipeline(query: str, context: dict = None):
    """
    智能重写管道：根据查询特征自动选择重写策略
    """
    rewriter = QueryRewriter()
    
    # 自动检测查询特征并选择策略
    if len(query) < 10:
        # 短查询需要扩展
        result = rewriter.rewrite(query, "expand")
    elif "怎么" in query or "如何" in query:
        # 问句需要分解
        result = rewriter.rewrite(query, "decompose")
    elif any(word in query for word in ["搞定", "弄", "搞"]):
        # 口语化表达需要正式化
        result = rewriter.rewrite(query, "formal")
    else:
        # 默认标准重写
        result = rewriter.rewrite(query, "standard")
    
    return result


使用示例
if __name__ == "__main__":
    test_queries = [
        "深度学习",  # 短查询
        "怎么用 PyTorch 训练一个图像分类模型",  # 问句
        "GPU 显存不够了怎么搞",  # 口语
        "RAG系统的评估指标有哪些"  # 标准查询
    ]
    
    rewriter = QueryRewriter()
    for q in test_queries:
        result = intelligent_rewrite_pipeline(q)
        print(f"原始: {q}")
        print(f"重写: {result['rewritten']}")
        print(f"策略: {result['type']}\n")

进阶方案：Query Expansion + Rewrite 组合使用

在实际生产环境中，将 Expansion 和 Rewrite 组合使用往往能获得最佳效果。推荐的工作流程是：首先对原始查询进行 Rewrite 以理解用户意图，然后对 Rewrite 后的查询进行 Expansion 以提高召回率，最后合并所有检索结果。

import os
from typing import List, Dict
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()

client = OpenAI(
    api_key=os.getenv("HOLYSHEEP_API_KEY"),
    base_url="https://api.holysheep.ai/v1"
)


class HybridQueryOptimizer:
    """
    混合查询优化器：结合 Query Rewrite 和 Query Expansion
    """
    
    def __init__(self, model: str = "gpt-4.1"):
        self.model = model
        self.expansion_prompt = """你是一个信息检索专家。请为以下查询生成 {num} 个语义等价的变体查询，
用于从向量数据库检索相关文档。每个变体应该：
1. 使用不同的词汇表达相同概念
2. 可以包含相关的上下位词
3. 保持原始查询的核心意图

查询：{query}

变体查询（JSON数组）："""
    
    def rewrite_query(self, query: str) -> str:
        """第一步：重写查询以理解用户真实意图"""
        prompt = f"""分析以下查询，识别用户的真实信息需求，并生成一个清晰、具体的优化查询。

要求：
1. 纠正拼写和语法错误
2. 补充缺失的实体信息
3. 将模糊表达具体化
4. 保持原意不变

查询：{query}

优化后的查询："""
        
        response = client.chat.completions.create(
            model=self.model,
            messages=[
                {"role": "system", "content": "你是一个专业的查询分析助手。"},
                {"role": "user", "content": prompt}
            ],
            temperature=0.3,
            max_tokens=200
        )
        
        return response.choices[0].message.content.strip()
    
    def expand_query(self, query: str, num_variants: int = 3) -> List[str]:
        """第二步：扩展查询以提高召回率"""
        prompt = self.expansion_prompt.format(query=query, num=num_variants)
        
        response = client.chat.completions.create(
            model=self.model,
            messages=[
                {"role": "system", "content": "你是一个信息检索专家。"},
                {"role": "user", "content": prompt}
            ],
            temperature=0.7,
            max_tokens=500
        )
        
        result = response.choices[0].message.content.strip()
        try:
            variants = eval(result)  # 安全考虑：在生产环境中应使用 json.loads
            return variants
        except:
            return [query]
    
    def optimize(self, query: str, num_variants: int = 3) -> Dict:
        """
        完整的查询优化流程
        返回重写后的查询和所有扩展变体
        """
        # 阶段一：理解意图
        rewritten = self.rewrite_query(query)
        
        # 阶段二：扩展查询
        expanded = self.expand_query(rewritten, num_variants)
        
        # 合并所有查询用于检索
        all_queries = [rewritten] + expanded
        
        return {
            "original_query": query,
            "rewritten_query": rewritten,
            "expanded_queries": expanded,
            "all_queries_for_retrieval": all_queries,
            "total_variants": len(all_queries)
        }


def run_hybrid_search(query: str, vector_store, optimizer: HybridQueryOptimizer):
    """
    使用混合优化策略执行搜索
    """
    # 优化查询
    optimization_result = optimizer.optimize(query, num_variants=3)
    
    print(f"原始查询: {optimization_result['original_query']}")
    print(f"重写后: {optimization_result['rewritten_query']}")
    print(f"扩展查询: {optimization_result['expanded_queries']}")
    
    # 执行多查询检索
    all_documents = []
    seen_ids = set()
    
    for search_query in optimization_result['all_queries_for_retrieval']:
        results = vector_store.search(search_query, top_k=5)
        
        for doc in results:
            if doc['id'] not in seen_ids:
                seen_ids.add(doc['id'])
                all_documents.append({
                    **doc,
                    'source_query': search_query
                })
    
    # 按相关性排序
    all_documents.sort(key=lambda x: x['score'], reverse=True)
    
    return {
        'optimization': optimization_result,
        'retrieved_documents': all_documents,
        'unique_doc_count': len(all_documents)
    }


使用示例
if __name__ == "__main__":
    optimizer = HybridQueryOptimizer(model="gpt-4.1")
    
    result = optimizer.optimize(
        "怎么优化 Transformer 的 attention 计算",
        num_variants=3
    )
    
    print(f"查询优化完成，共生成 {result['total_variants']} 个检索变体")
    print(f"重写结果: {result['rewritten_query']}")
    print(f"扩展变体: {result['expanded_queries']}")

迁移风险评估与回滚方案

任何系统迁移都存在风险。在将 Query Expansion 和 Query Rewrite 功能迁移到 HolySheep 时，需要注意以下几点风险评估。

首先是延迟风险。HolySheep 承诺的延迟低于 50ms，但在高峰期可能会略有波动。建议在迁移初期设置监控告警，当响应时间超过 100ms 时自动降级到本地模型。其次是成本风险。虽然 HolySheep 的价格极具竞争力，但 Query Expansion 会产生额外的 API 调用。需要计算好单位成本，避免预算超支。最后是质量风险。不同模型的输出质量可能有所差异，建议在迁移后进行 A/B 测试对比。

对于回滚方案，建议保留原有系统的完整代码和配置，并设置功能开关。当 HolySheep 服务出现异常时，可以快速切换回原有方案。同时，定期备份检索结果，以便在必要时进行对比分析。

ROI 分析与性能对比

我们通过实际测试对比了迁移前后的关键指标。测试场景为：日均处理 10 万次查询，平均每次查询扩展为 3 个变体。使用 HolySheep 的成本分析如下。

如果使用 GPT-4.1（8 美元/百万 Token），假设每次扩展消耗 500 Token，则日均成本约为 10 万次 × 3 变体 × 500 Token × 8 美元 / 100 万 = 12 美元，月均约 360 美元。如果改用 DeepSeek V3.2（0.42 美元/百万 Token），成本将降至月均约 19 美元，节省幅度高达 95%。

相比使用 OpenAI 官方 API（GPT-4o 价格为 15 美元/百万 Token 输入），HolySheep 的 GPT-4.1 可节省约 47%；而 DeepSeek V3.2 的性价比更是无与伦比，特别适合大规模生产环境使用。

ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข

错误一：API Key 配置错误导致 401 Unauthorized

# 错误配置示例
client = OpenAI(
    api_key="YOUR_HOLYSHEEP_API_KEY",  # 这是占位符，没有替换
    base_url="https://api.holysheep.ai/v1"
)

正确配置方式
import os
from dotenv import load_dotenv
load_dotenv()  # 确保加载 .env 文件

client = OpenAI(
    api_key=os.getenv("HOLYSHEEP_API_KEY"),  # 从环境变量读取
    base_url="https://api.holysheep.ai/v1"
)

验证配置是否正确
print(f"API Key 已配置: {bool(os.getenv('HOLYSHEEP_API_KEY'))}")
print(f"Base URL: {client.base_url}")

错误二：JSON 解析失败导致扩展查询为空

import json
import re

def safe_parse_json(response_text: str, fallback_query: str) -> list:
    """
    安全解析 LLM 返回的 JSON，包含多个降级策略
    """
    # 策略一：直接解析
    try:
        return json.loads(response_text)
    except json.JSONDecodeError:
        pass
    
    # 策略二：提取 JSON 数组
    json_array_pattern = r'\[.*\]'
    match = re.search(json_array_pattern, response_text, re.DOTALL)
    if match:
        try:
            return json.loads(match.group(0))
        except json.JSONDecodeError:
            pass
    
    # 策略三：按行分割并清理
    lines = response_text.strip().split('\n')
    queries = []
    for line in lines:
        line = line.strip().strip('-').strip('*').strip()
        if line and len(line) > 5:
            queries.append(line)
    
    if queries:
        return queries
    
    # 策略四：返回原始查询作为降级方案
    print(f"警告：JSON 解析失败，使用原始查询作为降级方案")
    return [fallback_query]

使用示例
response_text = """以下是扩展查询：
1. 深度学习框架比较
2. 主流深度学习框架
3. 深度学习框架对比分析"""
result = safe_parse_json(response_text, "深度学习框架")
print(f"解析结果: {result}")

错误三：温度参数设置不当导致输出不稳定

# 问题：Expansion 使用高温度可能导致查询变体过于发散
例如：原始查询是"Python教程"，高温度可能生成"Java教程"

解决方案：针对不同任务设置合理的温度值
def create_optimized_client():
    """
    根据不同任务创建专用的 LLM 配置
    """
    return {
        "rewrite": {
            "temperature": 0.2,  # 低温度，确保意图理解准确
            "model": "gpt-4.1",
            "max_tokens": 200
        },
        "expand": {
            "temperature": 0.5,  # 中等温度，平衡多样性和相关性
            "model": "gpt-4.1",
            "max_tokens": 500
        },
        "decompose": {
            "temperature": 0.3,  # 低温度，确保子问题逻辑清晰
            "model": "gpt-4.1",
            "max_tokens": 400
        }
    }

def get_llm_response(query: str, task_type: str, client: OpenAI) -> str:
    """
    根据任务类型使用最优配置
    """
    configs = create_optimized_client()
    config = configs.get(task_type, configs["rewrite"])
    
    response = client.chat.completions.create(
        model=config["model"],
        messages=[
            {"role": "system", "content": "你是一个专业的查询优化助手。"},
            {"role": "user", "content": query}
        ],
        temperature=config["temperature"],
        max_tokens=config["max_tokens"]
    )
    
    return response.choices[0].message.content

验证不同温度的效果差异
print("Rewrite (T=0.2): ", get_llm_response("Python咋学", "rewrite", client))
print("Expand (T=0.5): ", get_llm_response("Python教程", "expand", client))

常见问题 FAQ

Q1：Query Expansion 是否会增加延迟？

A1：会，但 HolySheep 的低延迟特性（低于 50ms）可以有效控制额外延迟。Expansion 阶段大约增加 100-200ms，但由于检索结果质量更高，整体 RAG 系统的用户体验通常会更好。

Q2：应该使用多少个扩展查询？

A2：这取决于查询的复杂度和检索库的大小。对于简单查询，使用 2-3 个扩展；对于复杂查询，可以使用 5 个以上。建议通过实验找到最佳平衡点。

Q3：DeepSeek V3.2 是否适合用于 Query Rewrite？

A3：DeepSeek V3.2 价格仅为 0.42 美元/百万 Token，性价比极高，特别适合对成本敏感的场景。但对于需要高精度意图理解的任务，建议仍使用 GPT-4.1。

总结

Query Expansion 与 Query Rewrite 是提升 RAG 系统质量的关键技术。通过本文介绍的迁移方案，你可以充分利用 HolySheep AI 的高性价比 API 来实现这些功能。

HolySheep 的核心优势包括：延迟低于 50ms、支持微信和支付宝付款、价格相比官方 API 节省 85% 以上（DeepSeek V3.2 仅为 0.42 美元/百万 Token）、注册即送免费额度。无论你是初创团队还是大型企业，HolySheep 都能提供适合你需求的 AI 推理解决方案。

建议从本文提供的示例代码开始，逐步将 Query Expansion 和 Query Rewrite 功能集成到你的 RAG 系统中。同时建立完善的监控和回滚机制，确保迁移过程平稳可控。

👉 สมัคร HolySheep AI — รับเครดิตฟรีเมื่อลงทะเบียน