高级 RAG：HyDE、查询扩展与自反思检索

应用查询重写、假设性文档嵌入（hypothetical document embeddings）与自我反思式检索。

🎯 你将学到

• 理解并应用本课涵盖的核心概念

优秀与卓越之间的质量鸿沟

一个基线 RAG 系统（递归分块 + 稠密检索 + GPT-4o-mini）通常能达到如下 RAGAS 分数：

• Faithfulness（忠实度）：0.82
• Answer Relevancy（答案相关性）：0.78
• Context Precision（上下文精度）：0.61

这还行——比没有好，对许多内部工具有用。但它有明显的失败模式：

• 简短查询： 用户敲了「SSO setup?」——过于模糊，不利于检索
• 非对称问题： 「How do I configure SSO?」这个问题，与标题为「Single Sign-On Configuration Walkthrough」的文档段落，具有不同的嵌入签名
• 听起来对、实则错： 答案听起来很自信，却缺乏依据

本教程覆盖三种能把分数推到 0.90+ 的技术。每种都增加一些复杂度；在决定是否采用前，都值得理解。

---

技术一：HyDE——假设性文档嵌入

非对称问题

标准 RAG 有个微妙的问题：问题与答案处在嵌入空间的不同位置。

当用户问「How do I configure SSO?」，那是个问题。相关的文档块写着「To configure SSO, navigate to Settings > Security and enter your Identity Provider metadata.」，那是个答案。

问题与答案具有不同的语言结构。「How do I…?」与「To configure…」有不同的嵌入签名。这意味着，即便在讨论同一主题，查询向量与相关块向量也不如它们本可以的那么接近。

HyDE 的解法： 与其嵌入问题，不如生成一个_假设性答案_并嵌入它。

直觉

想象一个图书管理员的调研技巧。你进来问：「我在找关于配置 SSO 的信息。」图书管理员不会只搜索「configuring SSO」。相反，他们会想：「一篇讲这个的文章大概会这么说——『单点登录（SSO）配置需要一个身份提供方 URL、一张证书和属性映射……』让我去搜听起来像这样的文章。」

图书管理员构建的这个假设性摘要，在嵌入空间里比原始问题更接近真实文档。

实现

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# Setup
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.3)  # slight temperature for variety
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma(persist_directory="./chroma_db", embedding_function=embeddings)

# Step 1: Generate a hypothetical answer
hyde_prompt = ChatPromptTemplate.from_template("""
Generate a concise, informative paragraph that would appear in technical documentation
and directly answer the following question.
Write as if you are the documentation — authoritative and specific.
Do not say "the documentation says" — just write the content directly.

Question: {question}

Documentation excerpt:""")

hyde_chain = hyde_prompt | llm | StrOutputParser()

def hyde_retrieve(question: str, k: int = 5) -> list:
    """Use HyDE: embed a hypothetical answer instead of the question."""

    # Generate hypothetical answer
    hypothetical_answer = hyde_chain.invoke({"question": question})
    print(f"Hypothetical answer: {hypothetical_answer[:200]}...")

    # Embed the hypothetical answer (not the question!)
    results = vectorstore.similarity_search(
        hypothetical_answer,  # key difference from standard RAG
        k=k
    )

    return results

# Compare standard vs HyDE retrieval
question = "What happens when I exceed rate limits?"

print("=== Standard Retrieval ===")
standard_results = vectorstore.similarity_search(question, k=3)
for doc in standard_results:
    print(f"  {doc.page_content[:150]}")

print("\n=== HyDE Retrieval ===")
hyde_results = hyde_retrieve(question, k=3)
for doc in hyde_results:
    print(f"  {doc.page_content[:150]}")

HyDE 帮助最大的场景

HyDE 带来的提升最大，当：

• 查询简短而简略（「SSO config?」「rate limits?」）
• 领域有专门的行话——假设性答案使用与文档相同的术语
• 你的文档像文档一样结构化（陈述性的、第三人称散文）

HyDE 每次查询只增加一次 LLM 调用（约 50ms，按 GPT-4o-mini 价格约 $0.0001）。这几乎总是值得。

---

技术二：查询扩展

歧义问题

用户敲了「can't connect」。连什么？数据库？API？网络？他们的 VPN？

一个「can't connect」的单一查询向量会落在所有这些含义的中间某处，并检索出对每一种都平庸的块。

查询扩展生成查询的多个改写版本，对每个版本检索，然后合并结果。你是在跨多个语义角度拓宽搜索网。

并行检索实现

import asyncio
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

expansion_prompt = ChatPromptTemplate.from_template("""
Generate {n} different phrasings of the following question.
Each rephrasing should capture a different interpretation or emphasis.
Return only the questions, one per line, no numbering.

Original question: {question}

Alternative phrasings:""")

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.5)
expansion_chain = expansion_prompt | llm | StrOutputParser()

def expand_query(question: str, n: int = 4) -> list[str]:
    """Generate multiple query reformulations."""
    response = expansion_chain.invoke({"question": question, "n": n})
    alternatives = [q.strip() for q in response.strip().split('\n') if q.strip()]
    return [question] + alternatives[:n]  # include original + n alternatives

def deduplicate_docs(doc_lists: list[list]) -> list:
    """Remove duplicate documents across multiple retrieval results."""
    seen_content = set()
    unique_docs = []

    for doc_list in doc_lists:
        for doc in doc_list:
            content_key = doc.page_content[:100]  # use first 100 chars as fingerprint
            if content_key not in seen_content:
                seen_content.add(content_key)
                unique_docs.append(doc)

    return unique_docs

def query_expansion_retrieve(question: str, k: int = 5) -> list:
    """
    Retrieve with query expansion:
    1. Generate multiple query variants
    2. Retrieve for each variant
    3. Deduplicate and return top results
    """
    # Generate query variants
    queries = expand_query(question, n=3)
    print(f"Expanded queries:")
    for q in queries:
        print(f"  - {q}")

    # Retrieve for each query
    all_results = []
    for query in queries:
        results = vectorstore.similarity_search(query, k=k)
        all_results.append(results)

    # Deduplicate
    unique_results = deduplicate_docs(all_results)

    # Return top k unique results (they're roughly ordered by first-retrieval rank)
    return unique_results[:k * 2]  # return more candidates for re-ranking

# Example
question = "can't connect"
results = query_expansion_retrieve(question, k=5)

# Expanded queries will include:
# - "can't connect"
# - "connection failure troubleshooting"
# - "how to resolve connection errors"
# - "troubleshooting network connectivity issues"
# - "why is my connection being refused"

异步版本（用于生产吞吐量）

顺序运行多个检索查询会增加延迟。生产环境应并行运行：

async def query_expansion_retrieve_async(question: str, k: int = 5) -> list:
    """Parallel query expansion retrieval."""
    queries = expand_query(question, n=3)

    async def retrieve_single(query: str) -> list:
        # Note: most vector stores are synchronous; run in thread pool
        loop = asyncio.get_event_loop()
        return await loop.run_in_executor(
            None,
            lambda: vectorstore.similarity_search(query, k=k)
        )

    # Run all retrievals concurrently
    all_results = await asyncio.gather(*[retrieve_single(q) for q in queries])

    unique_results = deduplicate_docs(list(all_results))
    return unique_results[:k * 2]

# Usage
results = asyncio.run(query_expansion_retrieve_async("can't connect"))

并行检索把延迟从 n×T 降到约 T（单次检索的时间），这使查询扩展在墙上时钟时间上几乎免费。

---

技术三：Self-RAG

验证问题

标准 RAG 回答一次问题就停。但如果检索到的上下文其实不包含答案呢？LLM 要么说「我不知道」（好），要么编造点东西（坏）。

Self-RAG 增加了一个验证循环：生成答案后，让模型评估它自己的答案。如果答案在上下文中依据不足，就用更精细的查询重新检索。

Self-RAG 循环

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_community.vectorstores import Chroma

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# Grounding check prompt
grounding_check_prompt = ChatPromptTemplate.from_template("""
You are a fact-checking assistant. Your job is to determine if an answer is well-grounded
in the provided context.

Context:
{context}

Question: {question}

Generated Answer: {answer}

Evaluation:
1. Does the answer rely only on information from the context? (Yes/No)
2. Is every specific claim in the answer supported by the context? (Yes/No)
3. If no to either: what information is the answer claiming that isn't in the context?

Verdict (respond with exactly "GROUNDED" or "NOT_GROUNDED"):""")

grounding_chain = grounding_check_prompt | llm | StrOutputParser()

# Query refinement prompt (used when answer is not grounded)
refinement_prompt = ChatPromptTemplate.from_template("""
The following answer to a question was not well-supported by the retrieved context.

Original question: {question}
Poor answer: {answer}
Missing information: {missing_info}

Write a more specific search query that would retrieve the missing information:""")

refinement_chain = refinement_prompt | llm | StrOutputParser()

def self_rag(
    question: str,
    vectorstore: Chroma,
    max_iterations: int = 3
) -> dict:
    """
    Self-RAG loop: generate → verify → refine → repeat if needed.
    Returns the final answer with iteration count.
    """
    current_query = question

    for iteration in range(max_iterations):
        print(f"\nIteration {iteration + 1}: Query = '{current_query}'")

        # Retrieve
        docs = vectorstore.similarity_search(current_query, k=5)
        context = "\n\n".join(doc.page_content for doc in docs)

        # Generate answer
        answer_prompt = ChatPromptTemplate.from_template("""
Answer based ONLY on the context. Be specific and cite details from the context.
If the context doesn't contain the answer, say "The context does not contain this information."

Context: {context}
Question: {question}
Answer:""")

        answer = (answer_prompt | llm | StrOutputParser()).invoke({
            "context": context,
            "question": question
        })

        print(f"Answer: {answer[:200]}...")

        # Check grounding
        grounding_result = grounding_chain.invoke({
            "context": context,
            "question": question,
            "answer": answer
        })

        print(f"Grounding check: {grounding_result[:100]}")

        if "GROUNDED" in grounding_result:
            print(f"Answer grounded on iteration {iteration + 1}")
            return {
                "answer": answer,
                "iterations": iteration + 1,
                "final_query": current_query,
                "sources": [doc.metadata.get('source') for doc in docs]
            }

        # Answer not grounded — refine the query
        # Extract what's missing from the grounding check
        missing_info = grounding_result.split("NOT_GROUNDED")[0].strip()

        current_query = refinement_chain.invoke({
            "question": question,
            "answer": answer,
            "missing_info": missing_info
        })

        print(f"Refined query: {current_query}")

    # Return best answer after max iterations
    print(f"Reached max iterations ({max_iterations})")
    return {
        "answer": answer,
        "iterations": max_iterations,
        "final_query": current_query,
        "warning": "Max iterations reached; answer may not be fully grounded"
    }

# Usage
result = self_rag(
    question="What is the SLA for the Enterprise tier?",
    vectorstore=vectorstore
)

print(f"\nFinal answer: {result['answer']}")
print(f"Iterations needed: {result['iterations']}")

Self-RAG 的权衡

维度	标准 RAG	Self-RAG
延迟	~300ms	~600-1500ms（每次迭代）
LLM 调用	1	2-6
忠实度	0.82	~0.91
成本	低	2-5 倍更高
最佳用例	实时聊天	高风险查询

Self-RAG 在错误答案有真实后果的高风险应用中价值最大：医疗信息系统、法律研究工具、金融建议平台。对于延迟更重要的随意问答，坚持用标准 RAG 或 HyDE。

---

三者结合：高级管道

下面是一个结合全部三种技术的管道：

async def advanced_rag_pipeline(
    question: str,
    vectorstore,
    use_hyde: bool = True,
    use_expansion: bool = True,
    use_self_rag: bool = False  # enable for high-stakes queries
) -> str:
    """Full advanced RAG pipeline with configurable techniques."""

    # Step 1: HyDE — Generate hypothetical answer for better query embedding
    if use_hyde:
        hypothetical = hyde_chain.invoke({"question": question})
        retrieval_query = hypothetical
    else:
        retrieval_query = question

    # Step 2: Query expansion — multiple retrieval angles
    if use_expansion:
        queries = expand_query(retrieval_query, n=3)
        all_docs = []
        for q in queries:
            docs = vectorstore.similarity_search(q, k=8)
            all_docs.extend(docs)
        candidates = deduplicate_docs([all_docs])[:15]
    else:
        candidates = vectorstore.similarity_search(retrieval_query, k=15)

    # Step 3: Re-rank (from previous lesson)
    reranked = reranker.rerank(question, candidates, top_k=5)
    top_docs = [doc for doc, score in reranked]

    # Step 4: Generate answer
    context = "\n\n".join(doc.page_content for doc in top_docs)
    answer_prompt = ChatPromptTemplate.from_template("""
Answer based ONLY on the context. Be specific and cite document sections.

Context:
{context}

Question: {question}
Answer:""")
    answer = (answer_prompt | llm | StrOutputParser()).invoke({
        "context": context, "question": question
    })

    # Step 5: Self-RAG verification (optional)
    if use_self_rag:
        grounding = grounding_chain.invoke({
            "context": context, "question": question, "answer": answer
        })
        if "NOT_GROUNDED" in grounding:
            result = self_rag(question, vectorstore, max_iterations=2)
            return result["answer"]

    return answer

# Usage
answer = asyncio.run(advanced_rag_pipeline(
    "What are the retry policies for API calls?",
    vectorstore=vectorstore,
    use_hyde=True,
    use_expansion=True,
    use_self_rag=False
))

---

何时实现每种技术

技术	RAGAS 提升	延迟成本	复杂度	实现时机
HyDE	忠实度 +0.05-0.10	+50ms	低	始终——轻松的胜手
查询扩展	上下文精度 +0.05-0.08	+0ms（并行）	中	查询常简略时
Self-RAG	忠实度 +0.08-0.15	+300-1000ms	高	需要高风险准确性时

Capstone 项目把所有这些整合成一个连贯的系统。我们开始构建吧。