RAG 系统的 Query 改写与扩展：检索效果差，很多时候不是向量库的问题

用户 query → LLM 生成假设答案 → 对假设答案做 embedding → 向量检索

def hyde_retrieve(query: str, llm, embedder, vector_store, top_k: int = 5):
    # 1. 生成假设文档
    prompt = f"""请根据以下问题，生成一段可能包含答案的详细文档片段。
要求：内容具体、信息丰富，不要出现"根据...""如上所述"等引用性表述。

问题：{query}

假设文档："""
    hypothetical_doc = llm.generate(prompt)
    
    # 2. 用假设文档做检索
    query_vector = embedder.embed(hypothetical_doc)
    results = vector_store.search(query_vector, top_k=top_k)
    return results

用户 query → LLM 生成 3-5 个变体 query → 分别检索 → 合并去重 → 重排序

def multi_query_retrieve(query: str, llm, embedder, vector_store, 
                         num_queries: int = 3, top_k_per_query: int = 5):
    # 1. 生成多个 query 变体
    prompt = f"""你是一个检索专家。用户的问题是：{query}

请从 {num_queries} 个不同角度改写这个问题，使检索更全面。
每个变体用一行表示，不要编号。

要求：改写后的问题应该语义完整、信息丰富，比原问题更具体。"""
    variants = llm.generate(prompt).strip().split("\n")[:num_queries]
    variants.append(query)  # 保留原始 query
    
    # 2. 分别检索
    all_results = []
    seen_docs = set()
    for q in variants:
        vector = embedder.embed(q)
        results = vector_store.search(vector, top_k=top_k_per_query)
        for doc in results:
            if doc.id not in seen_docs:
                all_results.append(doc)
                seen_docs.add(doc.id)
    
    # 3. 合并后重排序（RRF 或 Reranker）
    return rerank(all_results, query)

def decompose_query(query: str, llm) -> list[str]:
    prompt = f"""请将以下复杂问题拆解为 2-4 个独立的子问题。
每个子问题应该能够独立检索答案。
用 JSON 数组格式输出，每个元素包含 "question" 和 "depends_on" 字段。

问题：{query}

输出格式：
[
  {{"question": "子问题1", "depends_on": []}},
  {{"question": "子问题2", "depends_on": [0]}}
]"""
    response = llm.generate(prompt)
    sub_questions = json.loads(response)
    return sub_questions

def expand_query_with_context(query: str, chat_history: list[dict], llm) -> str:
    prompt = f"""对话历史：
{format_history(chat_history)}

用户最新问题：{query}

请结合对话历史，将用户的最新问题改写为一个独立的、完整的查询。
只输出改写后的查询，不要额外解释。"""
    expanded_query = llm.generate(prompt)
    return expanded_query

def should_rewrite(query: str) -> bool:
    """判断是否需要改写 query"""
    # 1. query 太短
    if len(query) < 10:
        return True
    # 2. 包含指代性词汇
    if any(w in query for w in ["它", "这个", "那个", "刚才", "它", "they", "it", "this", "that"]):
        return True
    # 3. query 是疑问词开头但信息不足
    if any(w in query for w in ["怎么", "如何", "what", "how"]) and len(query) < 20:
        return True
    return False

# 改写质量的离线评估指标
def evaluate_rewrite_quality(original_query, rewritten_query, retrieved_docs, ground_truth_docs):
    score = 0
    # 1. 改写后的 recall 是否提升
    recall_original = recall_at_k(retrieved_docs.original, ground_truth_docs, k=5)
    recall_rewritten = recall_at_k(retrieved_docs.rewritten, ground_truth_docs, k=5)
    score += (recall_rewritten - recall_original) * 0.5
    
    # 2. 改写后的 precision 是否下降（防止引入噪声）
    precision_original = precision_at_k(retrieved_docs.original, ground_truth_docs, k=5)
    precision_rewritten = precision_at_k(retrieved_docs.rewritten, ground_truth_docs, k=5)
    score += (precision_rewritten - precision_original) * 0.3
    
    # 3. 改写前后的语义相似度（防止改写偏离）
    similarity = cosine_similarity(embed(original_query), embed(rewritten_query))
    score += similarity * 0.2
    
    return score

Day 1-3: 5% 流量，只监控不改结果
Day 4-7: 20% 流量，A/B 对比改写 vs 不改写
Day 8-14: 50% 流量，逐步放开
Day 15+: 全量，持续监控