Advanced RAG 三、假设性问题索引

一、假设性问题索引是什么？

假设性问题是一种提问方式，它基于一个或多个假设的情况或前提来提出问题。在对知识库中的文档内容进行切片时，是可以以该切片为假设条件，利用LLM预先设置几个候选的相关性问题的，也就是说，这几个候选的相关性问题是和切片内容强相关的。

二、优点

1. 语义对齐更强
   用户问题与假设性问题属于同一语义空间，更容易匹配
2. 提升召回率
   即使用户措辞与原文差异较大，只要语义相近，仍能匹配到相关问题
3. 支持复杂意图
   LLM可生成覆盖不同角度的问题（如：原因、影响、步骤等）

三、局限性

1. 依赖LLM生成问题的质量
   若生成的问题偏离真实的用户意图，会降低检索效果
2. 领域适配要求高
   在专业领域，通用LLM生成的问题可能不准确（需微调或人工校验）

四、应用场景

• FAQ类知识库
  问题模式相对固定，适合预生成
• 技术文档 / 产品手册
  用户常问：如何使用、为什么报错 等
• 教育 / 客服场景
  问题具有高度重复性和可预测性

五、基本思路

1. 让LLM为每个块生成N个假设性问题，并将这些问题以向量形式嵌入
2. 在运行时，针对这个问题向量的索引进行查询搜索（用问题向量替换文档块的向量）
3. 检索后将原始文本块作为上下文发送给LLM以获取答案

六、示例

'''假设性问题索引'''
import json
import uuid
from typing import List
from langchain_classic.retrievers import MultiVectorRetriever
from langchain_community.document_loaders import TextLoader
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableParallel
from langchain_core.stores import InMemoryByteStore
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_chroma import Chroma
from pydantic import BaseModel, Field
from langchain_core.documents import Document
from langchain_core.globals import set_debug

from Common import get_models
# set_debug(True)
llm,embeddings = get_models()

loader = TextLoader("./deepseek百度百科.txt",encoding="utf-8")
docs = loader.load()

splitter = RecursiveCharacterTextSplitter(chunk_size=1024, chunk_overlap=100)
chunks = splitter.split_documents(docs)

vectorstore = Chroma(
    collection_name="hypo-questions",
    embedding_function=embeddings,
)
store = InMemoryByteStore()

id_key = 'doc_id'

retriever = MultiVectorRetriever(
    vectorstore=vectorstore,
    byte_store=store,
    id_key=id_key
)

doc_ids = [str(uuid.uuid4()) for _ in chunks]

class HypotheticalQuestions(BaseModel):
    """约束生成假设性问题的格式"""
    questions: List[str] = Field(..., description="List of questions")

def parse_hype_questions(raw_output):
    """解析模型输出为HypotheticalQuestions 对象"""
    try:
        output_text = raw_output.content if hasattr(raw_output, 'content') else str(raw_output)
        output_text = output_text.strip().strip("`").strip("json").strip()
        json_data = json.loads(output_text)
        return HypotheticalQuestions(**json_data)
    except json.JSONDecodeError as e:
        print(f"JSON decoding error: {e}")
        return HypotheticalQuestions(questions=[])
    except Exception as e:
        print(f"结构化解析失败:{e}")
        return HypotheticalQuestions(questions=[])

prompt = ChatPromptTemplate.from_template(
    """
    请基于以下文档生成3个假设性问题（必须使用JSON格式）：
    {doc}
    
    要求：
    1. 输出必须为合法JSON格式，包含questions字段
    2. questions字段的值包含3个问题的数组
    3. 使用中文提问
    
    示例：
    {{
        "questions":["问题1","问题2","问题3"]
    }}
    """
)
chain = (
        {"doc":lambda x:x.page_content}
         | prompt
         | llm
         | parse_hype_questions
         | (lambda x:x.questions)
)
# 测试
# print("测试",chunks[0])
# print("生成的问题",chain.invoke(chunks[0]))

# 批量执行生成假设性问题 30 个并行
hypothetical_questions = chain.batch( chunks,{"max_concurrency":30})
# print("生成假设性问题",hypothetical_questions)


# 将生成的问题转换为带元数据文档的对象
question_docs = []
for i,question_list in enumerate(hypothetical_questions):
    question_docs.extend(
        [Document(page_content=s,metadata={id_key:doc_ids[i]}) for s in question_list]
    )

# print(question_docs)

# 问题存入向量数据库
retriever.vectorstore.add_documents(question_docs)
# 原始文档字节存储
retriever.docstore.mset(list(zip(doc_ids,chunks)))

# 测试-执行相似性搜索
query = "deepseek受到哪些攻击？"
sub_docs = retriever.vectorstore.similarity_search(query)
# print("---------------------匹配的假设性问题--------------------------")
# print(sub_docs[0])

prompt = ChatPromptTemplate.from_template("根据下面的文档回答问题：\n\n{doc}\n\n问题：{question}")

# 生成问题回答链
chian = RunnableParallel({
    "doc":lambda x:retriever.invoke(x["question"]),
    "question":lambda x:x["question"]
}) | prompt | llm | StrOutputParser()

# 生成回答
print(chian.invoke({"question":query}))