RAG高级优化：基于问题生成的文档检索增强

在基于向量的文档检索系统中，如何让系统更精准地定位到包含答案的那一小段文本，始终是个核心挑战。今天要聊的是一种非常实用的增强策略——通过额外生成问题来“喂养”向量数据库，从而大幅提升检索召回率。简单来说，就是在每个文本片段上附加上可能被问到的问题，这样当用户实际提问时，系统能更快地匹配到对应的上下文。这不是什么黑科技，但效果立竿见影。

实现步骤

整个方案并不复杂，分四步走：

• 文档解析和文本分块：
  把PDF等原始文档拆成可控长度的文本片段，这是最基础的一步。

• 问题增强：
  借助语言模型，在文档级别或片段级别自动生成与内容高度相关的问题。

• 矢量存储创建：
  用向量模型计算所有文本片段（包括原始片段和生成的问题）的嵌入，存入FAISS索引。

• 检索和答案生成：
  用户查询时，FAISS找出最相关的文档片段，再将这些片段作为上下文交给生成模型输出答案。

这里有一个灵活的地方：你可以控制问题生成是在整个文档层面做，还是细化到每个片段层面。两种粒度各有适用场景，后面会看到代码实现。先看这个枚举类的定义：

class QuestionGeneration(Enum):
    """
    Enum class to specify the level of question generation for document processing.

    Attributes:
        DOCUMENT_LEVEL (int): Represents question generation at the entire document level.
        FRAGMENT_LEVEL (int): Represents question generation at the individual text fragment level.
    """
    DOCUMENT_LEVEL = 1
    FRAGMENT_LEVEL = 2

方案实现

问题生成

核心是调用大模型根据一段文本生成若干可回答的问题。下面这个函数展示了具体做法：它使用GPT-4o-mini，通过一个Prompt要求模型从上下文中提取出至少指定数量的问题，并且要求问题必须直接基于文本回答、不包含答案或标题。生成后还会做去重和过滤。

def generate_questions(text: str) -> List[str]:
    """
    Generates a list of questions based on the provided text using OpenAI.

    Args:
        text (str): The context data from which questions are generated.

    Returns:
        List[str]: A list of unique, filtered questions.
    """
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
    prompt = PromptTemplate(
        input_variables=["context", "num_questions"],
        template="Using the context data: {context}\n\nGenerate a list of at least {num_questions} "
                 "possible questions that can be asked about this context. Ensure the questions are "
                 "directly answerable within the context and do not include any answers or headers. "
                 "Separate the questions with a new line character."
    )
    chain = prompt | llm.with_structured_output(QuestionList)
    input_data = {"context": text, "num_questions": QUESTIONS_PER_DOCUMENT}
    result = chain.invoke(input_data)

    # Extract the list of questions from the QuestionList object
    questions = result.question_list

    filtered_questions = clean_and_filter_questions(questions)
    return list(set(filtered_questions))

处理主流程

整个文档处理的主函数将上述步骤串起来：先切分文档为多个子文档，每个子文档再切成更小的片段。然后根据配置的粒度（文档级或片段级），为每个片段或文档生成问题，并将原始片段和生成的问题一起作为Document对象放入列表。最后用FAISS创建向量存储，并返回一个检索器（这里设置k=1，即每次只返回最相关的一个文档）。代码逻辑很清晰：

def process_documents(content: str, embedding_model: OpenAIEmbeddings):
    """
    Process the document content, split it into fragments, generate questions,
    create a FAISS vector store, and return a retriever.

    Args:
        content (str): The content of the document to process.
        embedding_model (OpenAIEmbeddings): The embedding model to use for vectorization.

    Returns:
        VectorStoreRetriever: A retriever for the most relevant FAISS document.
    """
    # Split the whole text content into text documents
    text_documents = split_document(content, DOCUMENT_MAX_TOKENS, DOCUMENT_OVERLAP_TOKENS)
    print(f'Text content split into: {len(text_documents)} documents')

    documents = []
    counter = 0
    for i, text_document in enumerate(text_documents):
        text_fragments = split_document(text_document, FRAGMENT_MAX_TOKENS, FRAGMENT_OVERLAP_TOKENS)
        print(f'Text document {i} - split into: {len(text_fragments)} fragments')

        for j, text_fragment in enumerate(text_fragments):
            documents.append(Document(
                page_content=text_fragment,
                metadata={"type": "ORIGINAL", "index": counter, "text": text_document}
            ))
            counter += 1

            if QUESTION_GENERATION == QuestionGeneration.FRAGMENT_LEVEL:
                questions = generate_questions(text_fragment)
                documents.extend([
                    Document(page_content=question, metadata={"type": "AUGMENTED", "index": counter + idx, "text": text_document})
                    for idx, question in enumerate(questions)
                ])
                counter += len(questions)
                print(f'Text document {i} Text fragment {j} - generated: {len(questions)} questions')

        if QUESTION_GENERATION == QuestionGeneration.DOCUMENT_LEVEL:
            questions = generate_questions(text_document)
            documents.extend([
                Document(page_content=question, metadata={"type": "AUGMENTED", "index": counter + idx, "text": text_document})
                for idx, question in enumerate(questions)
            ])
            counter += len(questions)
            print(f'Text document {i} - generated: {len(questions)} questions')

    for document in documents:
        print_document("Dataset", document)

    print(f'Creating store, calculating embeddings for {len(documents)} FAISS documents')
    vectorstore = FAISS.from_documents(documents, embedding_model)

    print("Creating retriever returning the most relevant FAISS document")
    return vectorstore.as_retriever(search_kwargs={"k": 1})

归根结底，这项技术的核心思路就是：与其让用户查询直接匹配原始文本片段，不如先用问题把“可能被问到的问题”提前塞进向量空间，这样查询与问题匹配的成功率自然就高了。当然，这里用了大模型API生成问题，每调用一次都涉及token消耗，实际落地时需要根据业务量估算成本。但考虑到检索精度提升带来的收益，这点投入往往物有所值。