9.2 RAG检索优化：问题分解

这是一个 Multi-Query RAG（多查询RAG） 的实现，通过将复杂问题分解为多个子问题来提升检索质量。让我详细拆解每一步：

整体流程图

原始问题
    ↓
问题分解（生成3个子问题）
    ↓
并行处理每个子问题
    │
    ├─→ 子问题1 → 检索 → 生成答案1
    ├─→ 子问题2 → 检索 → 生成答案2
    └─→ 子问题3 → 检索 → 生成答案3
    ↓
合并所有Q+A对
    ↓
最终综合回答

第一步：问题分解

代码

template = """You are a helpful assistant that generates multiple sub-questions
related to an input question.
The goal is to break down the input into a set of sub-problems / sub-questions
that can be answered in isolation.
Generate multiple search queries related to: {question}
Output (3 queries):"""

prompt_decomposition = ChatPromptTemplate.from_template(template)

generate_queries_decomposition = (
    prompt_decomposition
    | ChatOpenAI(temperature=0)
    | StrOutputParser()
    | (lambda x: x.split("\n"))
)

输入/输出

输入：{"question": "What are the main components of an LLM-powered autonomous agent system?"}
处理：
1. 模板填充：将问题填入prompt
2. LLM生成：ChatGPT生成3个子问题
3. 解析：将字符串按换行分割成列表

输出：类似这样的列表

[
  "What are the key architectural components of an LLM agent system?",
  "How does memory work in autonomous agents?",
  "What tools and action executors are used in LLM agents?"
]

第二步：并行检索与回答

代码

def retrieve_and_rag(question, prompt_rag, sub_question_generator_chain):
    sub_questions = sub_question_generator_chain.invoke({"question": question})
    rag_results = []

    for sub_question in sub_questions:
        # 检索
        retrieved_docs = retriever.get_relevant_documents(sub_question)
        # 生成答案
        answer = (
            prompt_rag
            | llm
            | StrOutputParser()
        ).invoke({
            "context": retrieved_docs,
            "question": sub_question
        })
        rag_results.append(answer)

    return rag_results, sub_questions

处理每个子问题的流程

对于每个子问题：

子问题
  ↓
检索相关文档 (retriever.get_relevant_documents)
  ↓
RAG提示填充
  ↓
LLM生成答案
  ↓
存储答案

输入/输出

输入：
- 原始问题
- RAG提示模板
- 子问题生成链
中间处理：
- 生成3个子问题
- 对每个子问题：检索 + 用RAG回答

输出：两个列表

1 2	answers = ["答案1", "答案2", "答案3"] questions = ["子问题1", "子问题2", "子问题3"]

第三步：格式化Q+A对

代码

def format_qa_pairs(questions, answers):
    formatted_string = ""
    for i, (q, a) in enumerate(zip(questions, answers), start=1):
        formatted_string += f"Question {i}: {q}\nAnswer {i}: {a}\n\n"
    return formatted_string.strip()

输入/输出

输入：

1 2	questions = ["Q1", "Q2", "Q3"] answers = ["A1", "A2", "A3"]

输出：

Question 1: Q1
Answer 1: A1

Question 2: Q2
Answer 2: A2

Question 3: Q3
Answer 3: A3

第四步：最终综合回答

代码

final_prompt = ChatPromptTemplate.from_template(
    "Here is a set of Q+A pairs:\n{context}\nUse these to synthesize an answer to the question: {question}"
)

final_answer = (
    final_prompt
    | llm
    | StrOutputParser()
).invoke({
    "context": context,  # 格式化后的Q+A对
    "question": question  # 原始问题
})

输入/输出

输入：

{
  "context": "Question 1: ...\nAnswer 1: ...\n\nQuestion 2: ...",
  "question": "原始问题"
}

处理：LLM基于所有子问题的答案，综合回答原始问题
输出：最终的综合答案

完整数据流示例

假设原始问题："如何学习深度学习？"

步骤1输出

sub_questions = [
    "深度学习需要哪些数学基础？",
    "推荐哪些深度学习框架？", 
    "有哪些学习深度学习的实践项目？"
]

步骤2输出

answers = [
    "需要线性代数、概率论、微积分...",
    "PyTorch、TensorFlow、JAX...",
    "MNIST分类、图像生成、机器翻译..."
]
questions = sub_questions  # 同上

步骤3输出

context = """
Question 1: 深度学习需要哪些数学基础？
Answer 1: 需要线性代数、概率论、微积分...

Question 2: 推荐哪些深度学习框架？
Answer 2: PyTorch、TensorFlow、JAX...

Question 3: 有哪些学习深度学习的实践项目？
Answer 3: MNIST分类、图像生成、机器翻译...
"""

步骤4输出

final_answer = """
学习深度学习需要掌握数学基础（线性代数、概率论、微积分），
选择合适的框架（如PyTorch、TensorFlow），
并通过实践项目（如MNIST分类）巩固知识...
"""

这种方法的优势

检索精度：子问题更具体，检索更精准
覆盖全面：多个子问题覆盖问题的不同方面
答案质量：先回答子问题，再综合，减少幻觉
模块化：易于调试和优化每个步骤

可能的问题

子问题质量：依赖第一个LLM生成好的子问题
上下文长度：如果子问题很多，最终prompt可能过长
延迟：串行处理，3个子问题需要4次LLM调用

这是一个经典的RAG优化模式，特别适合复杂、多方面的查询。

write

by DeepSeeK