在基于向量的文档检索系统中，如何让系统更精准地定位到包含答案的那一小段文本，始终是RAG优化的核心难题。今天要分享的是一种非常实用的增强策略——通过额外生成问题来“喂养”向量数据库，从而大幅提升检索召回率。简单来说，就是在每个文本片段上附加可能被问到的相关提问，这样当用户实际搜索时，系统能更快地匹配到对应的上下文。这不是什么黑科技，但效果立竿见影，能显著改善检索质量。
核心实现步骤
整个RAG优化方案并不复杂，主要分为以下四个步骤：
- 文档解析与文本分块：将PDF等原始文档拆分为长度可控的文本片段，这是整个流程的基础环节。
- 问题生成与增强：借助大语言模型，在文档级别或文本片段级别自动生成与内容高度相关的问题。
- 向量存储创建：使用向量模型计算所有文本片段（包含原始片段和生成的问题）的嵌入向量，并存入FAISS索引库。
- 检索与答案生成：当用户发起查询时，FAISS召回最相关的文档片段，再将这些片段作为上下文交给生成模型输出最终答案。
这里有一个灵活的设计：你可以控制问题生成是在整个文档层面执行，还是细化到每个文本片段层面。两种粒度各有适用场景，下文代码中会体现。先看这个枚举类的定义：
class QuestionGeneration(Enum):
"""
该枚举类用于指定文档处理过程中问题生成的粒度级别。
属性:
DOCUMENT_LEVEL (int): 表示在整个文档级别生成问题。
FRAGMENT_LEVEL (int): 表示在单个文本片段级别生成问题。
"""
DOCUMENT_LEVEL = 1
FRAGMENT_LEVEL = 2
方案详细实现
基于大模型的问题生成
核心思路是调用大语言模型根据一段文本生成若干可回答的问题。下面这个函数展示了具体实现：它使用GPT-4o-mini，通过一个Prompt要求模型从上下文内容中提取出至少指定数量的问题，并且要求问题必须直接基于文本回答、不得包含答案或标题。生成完成后还会进行去重和过滤处理。
def generate_questions(text: str) -> List[str]:
"""
基于提供的文本，使用OpenAI模型生成一系列问题。
参数:
text (str): 用于生成问题的上下文数据。
返回:
List[str]: 去重并过滤后的唯一问题列表。
"""
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
prompt = PromptTemplate(
input_variables=["context", "num_questions"],
template="根据以下上下文：{context}\n\n请生成至少 {num_questions} "
"个与这段上下文相关的问题。请确保问题可以直接在上下文中找到答案，"
"不要包含答案或标题。每个问题用换行符分隔。"
)
chain = prompt | llm.with_structured_output(QuestionList)
input_data = {"context": text, "num_questions": QUESTIONS_PER_DOCUMENT}
result = chain.invoke(input_data)
# 从QuestionList对象中提取问题列表
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):
"""
处理文档内容：将其切分为片段、生成问题、创建FAISS向量存储并返回检索器。
参数:
content (str): 待处理的文档内容。
embedding_model (OpenAIEmbeddings): 用于向量化的嵌入模型。
返回:
VectorStoreRetriever: 返回最相关FAISS文档的检索器。
"""
# 将整个文本内容切分为多个文本文档
text_documents = split_document(content, DOCUMENT_MAX_TOKENS, DOCUMENT_OVERLAP_TOKENS)
print(f'文本内容被切分为: {len(text_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'文本文档 {i} - 切分为: {len(text_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'文本文档 {i} 片段 {j} - 生成了 {len(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'文本文档 {i} - 生成了 {len(questions)} 个问题')
for document in documents:
print_document("数据集", document)
print(f'正在创建存储，计算 {len(documents)} 个FAISS文档的嵌入向量')
vectorstore = FAISS.from_documents(documents, embedding_model)
print("创建检索器，返回最相关的FAISS文档")
return vectorstore.as_retriever(search_kwargs={"k": 1})
归根结底，这项技术的核心思路在于：与其让用户查询直接匹配原始文本片段，不如先用问题将“可能被提问的预测问题”提前注入向量空间，这样用户查询与问题匹配的成功率自然大幅提升。当然，这里调用了大模型API生成问题，每次调用都会消耗token，实际落地时需要根据业务量评估成本。但考虑到检索精度的显著提升，这点投入往往物超所值。