1、RAG商业化痛点分析

1.1、RAG流程

RAG流程图

1、读取文档
2、将文档分块
3、将分块后的文档存储到向量数据库中进行向量化并进行索引
4、结合用户的问题对索引进行相似度计算（问题和文档用同一向量数据库）
5、检索出来的文档即为上下文。
6、上下文+查询的问题构建提示词模版
7、将提示词和相关文档块一并提交给大模型
8、大模型的结果返回给用户

1.2、RAG详细的执行过程中的问题

RAG详细过程说明

1.2.1、索引构建过程（Index Process）中的问题

内容缺失(Missing Content): 原本的文本中就没有问题的答案。
文档加载准确性和效率：比如pdf文件的加载，如何提取其中的有用文字信息和图片信息等。
文档切分的粒度：文本切分的大小和位置会影响后面检索出来的上下文完整性和与大模型交互的token数量，怎么控制好文档切分的度，是个难题。

索引构建过程问题

1.2.2、检索增强过程中（Query Process）的问题

错过排名靠前的文档（Missed Top Ranked）：例如检索top5, 但是第6条也是相关内容。
提取上下文与答案无关（Not in Context）：例如问的是天气,相关上下文是美食。
格式错误(Wrong Format): 例如需要Json，给了字符串。
答案不完整(Incomplete): 答案只回答了问题的一部分。
未提取到答案(Not Extracted)：提取的上下文中有答案，但大模型没有提取出来
答案不够具体或过于具体（Incorrect Specificity）

检索增强过程中的问题

1.3、RAG的优化方案

1.3.1、内容缺失(Missing Content)

增加相应知识库：将相应的知识文本加入到向量知识库中。
数据清洗与增强：输入垃圾，那必定输出垃圾。任何RAG工作流程想要获得优良表现，就需要先清洗数据。
更好的Prompt设计：针对缺失内容，防止大模型胡乱回答。
比如：让大模型在找不到答案的情况下，输出“根据当前知识库，无法回答该问题”等提示。

1.3.2、文档加载准确性和效率

优化文档读取器：一般知识库中的文档格式都不尽相同，针对每一类文档，涉及一个专门的读取器。
数据清洗与增强

1.3.3、文档切分的粒度

基于结构的分块：基于结构的分块方法利用文档的固有结构，如HTML或Markdown中的标题和段落，以保持内容的逻辑性和完整性。
基于递归的分块：重复的利用分块规则不断细分文本块。比如先通过段落换行符（\n\n）进行分割。然后，检查这些块的大小。如果大小不超过一定阈值，则该块被保留。对于大小超过标准的块，使用单换行符（\n）再次分割。以此类推，不断根据块大小更新更小的分块规则（如空格，句号）。
1、分块大小的选择（chunk_size）：
不同的嵌入模型有其最佳输入大小。比如Openai的text-embedding-ada-002的模型在256 或 512大小的块上效果更好。
文档的类型和用户查询的长度及复杂性也是决定分块大小的重要因素。
2、内容重叠分块（over_lapping）：
为了保持文本块之间语义上下文的连贯性，在分块时，保持文本块之间有一定的内容重叠。

1.3.4、错过排名靠前的文档（Missed Top Ranked）

外挂知识库中存在回答问题所需的知识，但是可能这个知识块与问题的向量相似度排名并不是靠前的，导致无法召回该知识块传给大模型，导致大模型始终无法得到正确的答案。

增加召回数量：增加召回的 topK 数量，也就是说，例如原来召回前3个知识块，修改为召回前5个知识块。不过此种方法，因为知识块多了，不光会增加token消耗，也会增加大模型回答问题的干扰。
重排（Reranking）

重排

1.3.5、提取上下文与答案无关（Not in Context）

内容缺失或错过排名靠前的文档的具体体现

1.3.6、格式错误(Wrong Format)

Prompt调优：优化Prompt逐渐让大模型返回正确的格式。
进行结果格式验证：例如使用LangChain中的PydanticOutputParser类来校验输出格式。
Auto-Fixing自修复：对不符合要求的格式进行自动修复

1.3.7、答案不完整(Incomplete)

将问题分开提问：
• 一方面引导用户精简问题，一次只提问一个问题。
• 另一方面，针对用户的问题进行内部拆分处理，拆分成数个子问题，等子问题答案都找到后，再总结起来回复给用户

1.3.8、未提取到答案(Not Extracted)

使用更强的大模型
在Prompt中强调“必须基于上下文回答”或增强上下文聚焦（如关键句子加粗）

1.3.9、答案太具体或太笼统（Incorrect Specificity）

提示词改善
提升基座大模型能力

2、Advanced RAG概述

2.1、概念

Advanced RAG重点聚焦在检索增强，即优化Retrieval阶段。增加了Pre-Retrieval预检索和Post-Retrieval后检索阶段，同时对检索本身也有优化。

预检索过程优化/检索前优化(Pre-Retrieval):
高级RAG着重优化了索引结构和查询的方式。优化索引旨在提高被索引内容的质量，包括增强数据颗粒度、优化索引结构、添加元数据、对齐优化等策略。查询优化的目标则是明确用户的原始问题，使其更适合检索任务，使用了查询重写、查询转换、查询扩展等技术。
检索优化(Retrieval)
检索阶段的目标是确定最相关的上下文。通常，检索基于向量搜索，它计算查询与索引数据之间的语义相似性。因此，大多数检索优化技术都围绕嵌入模型展开，比如微调嵌入模型，将嵌入模型定制为特定领域的上下文，特别是对于术语不断演化或罕见的领域。
还有其他检索技术，例如：混合搜索，通常是指将向量搜索与基于关键字的搜索相结合的概念。
后检索过程优化/检索后优化(Post-Retrieval)
对于由问题检索得到的一系列上下文，后检索策略关注如何优化它们与查询问题的集成。
这一过程主要包括重新排序和压缩上下文。重新排列检索到的信息，将最相关的内容予以定位标记，这种策略已经在LlamaIndex2、LangChain等框架中得以实施。
有时直接将所有相关文档输入到大型语言模型（LLMs）可能导致信息过载，为了缓解这一点，后检索工作集中选择必要的信息，强调关键部分，并限制了相应的上下文长度。

RAG检索对比图

2.2、Pre-Retrieval预检索-索引优化

2.2.1、索引优化

2.2.1.1、摘要索引优化

2.2.1.1.1、痛点分析

在处理大量文档时，如何快速准确地找到所需信息是一个常见挑战。摘要索引可以用来处理半结构化数据，比如许多文档包含多种内容类型，包括文本和表格。这种半结构化数据对于传统 RAG 来说可能具有挑战性，文本拆分可能会分解表，从而损坏检索中的数据；嵌入表可能会给语义相似性搜索带来挑战。
尤其是长文档和长表格的内容
如下图存在的问题

摘要索引

2.2.1.1.2、摘要索引-基本流程

摘要索引流程

让LLM为每个文档块生成summary，并作为embedding存到summary database中
在检索时，通过summary database找到最相关的summary，再回溯到原始文档中去
将原始文档块作为上下文发送给LLM以获取答案

2.2.1.1.3、摘要索引-代码分析

代码使用的是Ollama本地模型进行实验操作

# 摘要索引示例代码

# 1.提取，分割，块（比较大）
# 2.块，生成摘要
# 3.将摘要向量化， 摘要和原始文档要建立关系
# 4.摘要向量存储到向量数据库中，原始文档存储（不是存到向量数据库）
# 5.检索，匹配摘要向量，返回原始文档


# 注意事项：
# 1.摘要质量至关重要
#   "这是一个关于deepseek的介绍文章" ：太过于简单，丢失了关键信息
# 2.存储开销比较大
#   不用摘要：只存储知识块对应的向量。 摘要索引：既要存储摘要的向量还要存储原始文档
# 3.一致性问题，维护一致性
#   通过uuid 将原始文档和摘要一一对应

# 获得访问大模型和嵌入模型客户端
# client, embeddings_model = get_ali_clients()
# 初始化 Ollama 嵌入模型
embeddings = OllamaEmbeddings(
    model="qwen3-embedding:8b",
    base_url="http://localhost:11434" # 默认 Ollama 本地地址
)
# 2. 初始化本地多模态模型
client = ChatOllama(
    model="qwen3-vl:8b",
    temperature=0.1,
    base_url="http://localhost:11434"  # Ollama 默认本地地址
)

# 初始化文档加载器
loader = TextLoader("./deepseek百度百科.txt", encoding="utf-8")

# 加载文档
docs = loader.load()

# 初始化递归文本分割器（设置块大小和重叠）
text_splitter = RecursiveCharacterTextSplitter(chunk_size=1024, chunk_overlap=100)
docs = text_splitter.split_documents(docs)
print(len(docs))
# print(docs[0])  # Document对象
# exit()


# 创建摘要生成链
chain = (
    {"doc": lambda x: x.page_content}
    | ChatPromptTemplate.from_template("总结下面的文档:\n\n{doc}")
    | client
    | StrOutputParser()
)

print("准备生成文档摘要，请耐心等待...")
# 批量生成文档摘要（最大并发数5）
summaries = chain.batch(docs, {"max_concurrency": 30})

# print(summaries)
# print([doc.page_content for doc in docs])
# exit()



# 初始化Chroma实例（用于存储摘要向量）
vectorstore = Chroma(
    collection_name="summaries",
    embedding_function=embeddings
)

# 初始化内存字节存储（用于存储原始文档）
store = InMemoryByteStore()

# 初始化多向量检索器（结合向量存储和文档存储）
id_key = "doc_id"
retriever = MultiVectorRetriever(
    vectorstore=vectorstore,
    byte_store=store,
    id_key=id_key,
)

# 为每个文档生成唯一ID，该ID用于关联原始文档和摘要
doc_ids = [str(uuid.uuid4()) for _ in docs]


# 将文档摘要转换为LangChain中Document
summary_docs = [
    Document(page_content=s, metadata={id_key: doc_ids[i]})
            for i, s in enumerate(summaries)
]
# print(summary_docs)
# exit()

# 将摘要添加到向量数据库
print("准备将摘要添加到向量数据库...")
retriever.vectorstore.add_documents(summary_docs)

# 将原始文档存储到字节存储（使用ID关联）
print("准备将原始文档存储到字节存储...")
# mset：批量设置键值对
# list(zip(doc_ids, docs))：将ID和文档配对
retriever.docstore.mset(list(zip(doc_ids, docs)))

# 手动测试代码 - 相似性搜索
# query = "deepseek的企业事件"
# sub_docs = retriever.vectorstore.similarity_search(query)
# print("-------------匹配的摘要内容--------------")
# print(sub_docs[0])
#
# # 获取第一个匹配摘要的ID
# matched_id = sub_docs[0].metadata[id_key]
#
# print(f"-------------对应的原始文档--------------")
# # 通过ID获取原始文档
# original_doc = retriever.docstore.mget([matched_id])
# print(original_doc)
# # 执行相似性搜索测试---完成
# exit()



prompt =  ChatPromptTemplate.from_template("根据下面的文档回答问题:\n\n{doc}\n\n问题: {question}")
# 生成问题回答链
# retriever.invoke将上面对摘要进行检索，但是通过关联ID获得原始文档，最终返回原始文档的过程全部都包含完成了
chain = RunnableParallel({
    "doc": lambda x: retriever.invoke(x["question"]),
    "question": lambda x: x["question"]
}) | prompt | client | StrOutputParser()

# 生成问题回答
query = "deepseek的企业事件"
answer = chain.invoke({"question": query})
print("-------------回答--------------")
print(answer)

# retriever.invoke(query)  # 1.向量数据库中检索摘要向量   2.匹配对应的原始文档并返回
# retrieved_docs = retriever.invoke(query)
# print("-------------检索到的文档--------------")
# print(retrieved_docs)

2.2.1.1.3、摘要索引使用场景分析

半结构化数据
高价值表格

半结构化数据

高价值表格

注意事项：
1、摘要索引并非万能，需根据数据价值权衡成本
2、摘要质量非常重要，需要包含高价值表格关键信息，太过于简单，会丢失关键信息

2.2.1.2、父子索引

2.2.1.2.1、父子索引-痛点分析

我们在利用大模型进行文档检索的时候，常常会有相互矛盾的需求，比如：
1、你可能希望得到较小的文档块，以便它们Embedding以后能够最准确地反映出文档的含义，如果文档块太大（会被大量无关内容稀释），Embedding就失去了意义。
2、你可能希望得到较大的文档块以保留较多的内容，然后将它们发送给LLM以便得到全面且正确的答案。
3、检索粒度和理解粒度的矛盾：

检索需要细粒度：越细越容易匹配用户问题中的关键词（精度）。
生成需要粗粒度：LLM推理需要完整上下文才能正确理解（完整性）。
传统RAG无法兼顾二者，陷入精度 vs 完整性的两难。
例：一个合同中“违约要赔偿30%”的条款，前提是“非不可抗力情形下”，如果该前提未被召回，AI生成的答案就是错的

2.2.1.2.2、父子索引-流程分析

父子索引-流程图

文档被分割成一个层级化的块结构，随后用最小的叶子块进行索引
在检索过程中检索出top-k个叶子块
如果存在n个叶子块都指向同一个更大的父块，那么我们就用这个父块来替换这些子块，并将父块送入大模型用于生成答案。

2.2.1.2.3、父子索引-代码分析

#获得访问大模型和嵌入模型客户端
client, embeddings_model = get_ali_clients()

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

# 查看长度
print(f"文章的长度：{len(docs[0].page_content)}")

# 子块是父块内容的子集
#创建主文档分割器
parent_splitter = RecursiveCharacterTextSplitter(chunk_size=1024, chunk_overlap=100)

#创建子文档分割器
child_splitter = RecursiveCharacterTextSplitter(chunk_size=256, chunk_overlap=30)

# 创建向量数据库对象
vectorstore = Chroma(
    collection_name="split_parents",
    embedding_function = embeddings_model
)

# 创建内存存储对象
store = InMemoryStore()

#创建父子文档检索器，帮我们通过检索子块，返回父文档块
retriever = ParentDocumentRetriever(
    vectorstore=vectorstore,
    docstore=store, # 文档存储对象
    child_splitter=child_splitter,  # 子文档分割器，子文档存储到向量数据库
    parent_splitter=parent_splitter,  # 主文档分割器，主文档存储到内存中
    search_kwargs={"k": 5},  # topK=1,相似度最高的子文档块
)


#添加文档集
retriever.add_documents(docs)

print(f"主文块的数量：{len(list(store.yield_keys()))}")

# 测试 - 相似性搜索
'''这里我们通过向量数据库的similarity_search方法搜索出来的是与用户问题相关的子文档块的内容，
下面我们使用检索器的get_relevant_documents的方法来对这个问题进行检索，
它会返回该子文档块所属的主文档块的全部内容： '''
# print("------------similarity_search------------------------")
# sub_docs = vectorstore.similarity_search("deepseek的应用场景", k=5)
# # print(sub_docs[0].page_content)
# print([doc.page_content for doc in sub_docs])
#
# print("------------get_relevant_documents-----------通过子找父-------------")
# retrieved_docs = retriever.invoke("deepseek的应用场景")
# # print(retrieved_docs[0].page_content)
# print([doc.page_content for doc in retrieved_docs])
#
# # 测试 - 相似性搜索 - 完成
# exit()

2.2.1.2.4、父子索引-使用总结

父子索引-图解

1、注意事项：
• 文档有清晰的结构层次
• 需要保持上下文的完整性
• 用户查询通常针对具体细节
• 信息碎片化可能造成误解
2、应用场景：
• 技术文档检索（如API文档）
• 书籍内容检索
• 代码库检索

2.2.1.3、假设性问题索引

2.2.1.3.1、假设性问题索引-概念

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

2.2.1.3.2、假设性问题索引-实现步骤

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

假设性问题索引-实现步骤

2.2.1.3.3、假设性问题索引-代码分析

# 假设性问题索引示例代码
# 获得访问大模型和嵌入模型客户端
client,embeddings_model = get_ali_clients()

# 初始化文档加载器列表
loader = TextLoader("./deepseek百度百科.txt",encoding="utf-8")
docs = loader.load()

# 初始化递归文本分割器（设置块大小和重叠）
text_splitter = RecursiveCharacterTextSplitter(chunk_size=1024, chunk_overlap=100)
docs = text_splitter.split_documents(docs)


# 初始化Chroma向量数据库（存储生成的问题向量）
vectorstore = Chroma(
    collection_name="hypo-questions",
    embedding_function=embeddings_model
)
# 初始化内存存储（存储原始文档）
store = InMemoryByteStore()

id_key = "doc_id"  # 文档标识键名

# 配置多向量检索器
retriever = MultiVectorRetriever(
    vectorstore=vectorstore, #  向量数据库，存储生成的问题向量（调用对话模型生成）
    byte_store=store, # 字节存储，存储原始文档
    id_key=id_key,
)

# 为每个原始文档生成唯一ID
doc_ids = [str(uuid.uuid4()) for _ in docs]

# 以下开始用大模型生成假设性问题
'''
# 将LLM输出构建为字符串列表（HypotheticalQuestions来定义格式为字符串列表）

而HypotheticalQuestions要求的格式是：
    定义了一个字段 questions，它具有以下特性：
        类型注解：List[str] 表示 questions 字段应该是一个字符串列表。
        必需性：Field(...) 中的省略号 ... 表示这个字段是必需的。
        描述信息：description="List of questions" 为该字段添加了描述，这对于生成文档或帮助理解模型结构很有用。
'''
class HypotheticalQuestions(BaseModel):
    """约束生成假设性问题的格式"""
    questions: List[str] = Field(..., description="List of questions")

# 此处使用双括号 {{ 和 }} 是为了在字符串中转义出单个 { 和 }，以确保最终输出的 JSON 格式正确。
prompt = ChatPromptTemplate.from_template(
        """请基于以下文档生成3个假设性问题（必须使用JSON格式）:
        {doc}
        
        要求：
        1. 输出必须为合法JSON格式，包含questions字段
        2. questions字段的值是包含3个问题的数组
        3. 使用中文提问
        示例格式：
        {{
            "questions": ["问题1", "问题2", "问题3"]
        }}"""
)

# 创建假设性问题链
'''
其中的client.with_structured_output可以理解为输出解析器的一种更高级用法
将大模型的输出转换为 HypotheticalQuestions 所限定的格式
'''
chain = (
    {"doc": lambda x: x.page_content}
    | prompt
    # 将LLM输出构建为字符串列表
    | client.with_structured_output(
        HypotheticalQuestions
    )
    # 提取问题列表
    | (lambda x: x.questions)
)

# 测试-在单个文档上调用链，链的最终输出是大模型答复的假设性问题列表
# print("测试：",docs[0])
# print("测试生成问题：",chain.invoke(docs[0]))
# exit()


# 批量处理所有文档生成假设性问题（最大并行数5），每个切块后的文档块都对应的生成三个问题
print(len(docs))
hypothetical_questions = chain.batch(docs, {"max_concurrency": 30})
# print("假设性问题列表：",hypothetical_questions)
# exit()


# 将生成的问题转换为带元数据的文档对象
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)
# exit()

retriever.vectorstore.add_documents(question_docs)  # 将问题文档存入向量数据库
retriever.docstore.mset(list(zip(doc_ids, docs)))  # 将原始文档存入字节存储（通过ID关联）
# 以上的过程是可以在构建知识库的时候提前完成的


# 测试-执行相似性搜索
query = "deepseek受到哪些攻击？"
# sub_docs = retriever.vectorstore.similarity_search(query)
# print("-------------相似性：--------------")
# print("测试-执行相似性搜索：", sub_docs)
# exit()


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

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

# 生成问题回答
answer = chain.invoke({"question": query})
print("-------------回答--------------")
print(answer)

#  返回的是知识块
# retrieved_docs = retriever.invoke(query)
# print("-------------检索到的问题--------------")
# print(retrieved_docs)

2.2.1.3.4、假设性问题索引-使用总结

1、假设性问题索引优点：

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

2、局限性：

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

3、应用场景：

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

2.2.1.4、元数据索引

2.2.1.4.1、元数据索引-痛点分析

在企业复杂的知识密集型应用中，可能会面临几百个不同来源与类型的知识文档。如果只是简单地依赖传统的文本分割与top-k检索，就会产生精度不足、知识相互干扰等问题，从而导致效果不佳。想象一下，你想在一个医学文献数据库中查找关于“糖尿病”的资料，但数据库中也充斥着大量关于其他糖尿病并发症的信息。
一个重要的优化方法是在大文档集下“分层”过滤与检索。
元数据是对文档的一种属性描述，假设我们使用一个存储了大量科技博客文章的向量数据库。每篇文章都关联了以下标签：
topic: 人工智能, 区块链, 云计算, 大数据
author: 作者A, 作者B, 作者C
year: 2022, 2023, 2024

2.2.1.4.2、元数据索引-流程分析

定义元数据标签
通过标签先对文档进行过滤
结合向量检索进一步定位到最相关的前 K个知识块

元数据索引-流程图

2.2.1.4.3、元数据索引-代码分析

#获得访问大模型和嵌入模型客户端
llm,embeddings_model = get_ali_clients()

# 加载文档
docs = [
    Document(
        page_content="作者A团队开发出基于人工智能的自动驾驶决策系统，在复杂路况下的响应速度提升300%",
        metadata={"year": 2024, "rating": 9.2, "genre": "AI", "author": "A"},
    ),
    Document(
        page_content="区块链技术成功应用于跨境贸易结算，作者B主导的项目实现交易确认时间从3天缩短至30分钟",
        metadata={"year": 2023, "rating": 9.8, "genre": "区块链", "author": "B"},
    ),
    Document(
        page_content="云计算平台实现量子计算模拟突破，作者C构建的新型混合云架构支持百万级并发计算",
        metadata={"year": 2022, "rating": 8.6, "genre": "云", "author": "C"},
    ),
    Document(
        page_content="大数据分析预测2024年全球经济趋势，作者A团队构建的模型准确率超92%",
        metadata={"year": 2023, "rating": 8.9, "genre": "大数据", "author": "A"},
    ),
    Document(
        page_content="人工智能病理诊断系统在胃癌筛查中达到三甲医院专家水平，作者B获医疗科技创新奖",
        metadata={"year": 2024, "rating": 7.1, "genre": "AI", "author": "B"},
    ),
    Document(
        page_content="基于区块链的数字身份认证系统落地20省市，作者C设计的新型加密协议通过国家级安全认证",
        metadata={"year": 2022, "rating": 8.7, "genre": "区块链", "author": "C"},
    ),
    Document(
        page_content="云计算资源调度算法重大突破，作者A研发的智能调度器使数据中心能效提升40%",
        metadata={"year": 2023, "rating": 8.5, "genre": "云", "author": "A"},
    ),
    Document(
        page_content="大数据驱动城市交通优化系统上线，作者B团队实现早晚高峰通行效率提升25%",
        metadata={"year": 2024, "rating": 7.4, "genre": "大数据", "author": "B"},
    )
]

vectorstore = Chroma.from_documents(docs, embeddings_model)


# 元数据字段定义（指导LLM如何解析查询条件）
metadata_field_info = [
    AttributeInfo(
        name="genre",
        description="文章的技术领域，选项:['AI '，'区块链'，'云'，'大数据']",
        type="string",
    ),
    AttributeInfo(
        name="year",
        description="文章的出版年份",
        type="integer",
    ),
    AttributeInfo(
        name="author",
        description="署名文章的作者姓名",
        type="string",
    ),
    AttributeInfo(
        name="rating",
        description="技术价值评估得分（1-10分）",
        type="float"
    )
]

# 文档内容描述（指导LLM理解文档内容）
document_content_description = "技术文章简述"

# 创建自查询检索器（核心组件）
'''SelfQueryRetriever 是 langchain 库中的一个工具，其主要功能是把自然语言查询转变为结构化查询，
以此提升检索的精准度。它整合了检索器和语言模型，能依据查询内容自动推断出筛选条件，还能识别出相关的元数据字段。'''
retriever = SelfQueryRetriever.from_llm(
    llm,
    vectorstore,
    document_content_description,
    metadata_field_info,
    # enable_limit=True,  # 限定返回结果，和query搭配使用
)

# 检索
print("---------------------评分在9分以上的文章-------------------------------")
#查询条件：查询只约束分数 rating>9
# print(retriever.invoke("我想了解评分在9分以上的文章"))
# print(retriever.invoke("我想了解评分在9分以上的文章,返回1篇文章"))

print("---------------------作者B在2023年发布的文章-------------------------------")
# 第二个查询只约束作者和年份 author="B", year=2023
print(retriever.invoke("作者B在2023年发布的文章"))
# exit()


# 了解
# 原理：构建查询解析器（分析内部工作机制用）
'''构建查询提示模板get_query_constructor_prompt：
    document_content_description：对文档内容的概括性描述，例如 "有关各种主题的文章"。
    metadata_field_info：元数据字段的详细描述，涵盖字段名称、类型以及描述。
此函数会生成一个提示模板，其用途是指导语言模型如何将自然语言查询转换为结构化查询。'''
# prompt = get_query_constructor_prompt(
#     document_content_description,
#     metadata_field_info,
# )
#
# '''StructuredQueryOutputParser解析器的作用是：把语言模型的输出转换为 StructuredQuery 对象，
#     这个对象包含了 query（文本查询）和 filter（元数据筛选条件）
# '''
# output_parser = StructuredQueryOutputParser.from_components()
# # 链式操作，先将用户查询填入提示模板，接着由语言模型生成结构化输出，最后通过解析器得到结构化查询。
# query_constructor = prompt | llm | output_parser
#
# # 打印查询构造提示
# print("提示词：", prompt.format(query="我想了解评分在9分以上的文章"))
# print("提示词显示结束--------------------------提示词显示结束----------------------")
#
# # 打印结构化查询的结果
# print("结构化查询结果：",query_constructor.invoke({"query": "作者B在2023年发布的文章"}))
# 看工作原理-结束

2.2.2、查询优化

2.2.2.1、Enrich完善问题

2.2.2.1.1、痛点分析

我们希望实现让人们通过的口语对话来使用大模型应用。然而，在口语表达需求和意图时，人们往往会遇到一些问题。例如，表自然达过于简略或含糊，容易引发语义歧义，导致大模型产生误解；用户的问题可能包含许多隐含要素，但表达的信息却不足，只能通过多轮对话逐步补全；

理想情况：通过大模型多次主动与用户沟通，不断收集信息，完善对用户真实意图的理解，补全执行用户需求所需的各项参数。

2.2.2.1.2、基本思路

Enrich完善问题-流程图

2.2.2.1.3、代码分析

# 获得访问大模型客户端
llm = get_ali_model_client()

# 用户的要求
user_input = "我想订一张长沙去北京的机票"

# 首先根据用户的要求，进行意图识别，获取对应的模板
# 示例业务模板
templates = {
    "订机票": ["起点", "终点", "时间", "座位等级", "座位偏好"],
    "订酒店": ["城市", "入住日期", "退房日期", "房型", "人数"],
}

# 意图识别提示模板, 自动匹配上面的模版templates
intent_prompt = PromptTemplate(
    input_variables=["user_input", "templates"],
    template="根据用户输入 '{user_input}'，选择最合适的业务模板。可用模板如下：{templates}。请返回模板名称。"
)

# 创建意图识别链
intent_chain = intent_prompt | llm

# 识别意图 str(list(templates.keys())) 返回的内容为用户意图 '订机票'、‘订酒店’
intent = intent_chain.invoke({"user_input": user_input, "templates": str(list(templates.keys()))}).content
print("意图：", intent)

# 获取对应模板、比如用户的意图为‘订机票’，返回的模版为['起点', '终点', '时间', '座位等级', '座位偏好']
selected_template = templates.get(intent)
print("模板：", selected_template)
# exit()

# 根据用户意图已经获取对应的模板，然后判断是否需要补充信息
# 补充信息提示模板
info_prompt = f"""
    请根据用户原始问题和模板，判断原始问题是否完善。如果问题缺乏需要的信息，请生成一个友好的请求，明确指出需要补充的信息。若问题完善后，返回包含所有信息的完整问题。

    ### 原始问题    
    {user_input}

    ### 模板
    {",".join(selected_template)}                                   

    ### 输出示例
    {{
        "isComplete": true,
        "content": "`完整问题`"
    }}
    {{
        "isComplete": false,
        "content": "`友好的引导用户补充需要的信息`"
    }}                                       
"""

# 历史记录
chat_history = ChatMessageHistory()

# 聊天模版，{history} 名称与下面的 history_messages_key 的 "history"相对应
#  {input} 名称与下面的 input_variables 的 "input" 相应
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "你是一个信息补充助手，任务是分析用户问题是否完整。"),
        ("placeholder", "{history}"),  # 历史记录的占位
        ("human", "{input}"),
    ]
)

# 补充信息链
info_chain = prompt | llm

# 自动处理历史记录，将记录注入输入并在每次调用后更新它
with_message_history = RunnableWithMessageHistory(
    info_chain,
    lambda session_id: chat_history,
    input_messages_key="input",
    history_messages_key="history",
)

# 判断问题是否完整，如果不完整，则生成追问请求
# session_id 是会话ID，因为这里只设置了一个会话，因此可以设置为 unused
info_request = with_message_history.invoke(input={"input": info_prompt},
                                           config={"configurable": {"session_id": "unused"}}).content
parser = JsonOutputParser()
json_data = parser.parse(info_request)
print("json_data：",json_data)


# 循环判断是否完整，并提交用户补充信息
while json_data.get('isComplete', False) is False:
    try:
        # 显示引导信息并等待用户输入，用\033[1;33m和\033[0m设置和重置文本颜色及样式（黄色加粗）
        user_answer = input(f"\033[1;33m{json_data['content']}\033[0m\n请补充：")

        # 提交补充信息给AI处理
        info_request = with_message_history.invoke(
            input={"input": user_answer},
            config={"configurable": {"session_id": "unused"}}
        ).content

        # 解析AI响应
        json_data = parser.parse(info_request)

    except json.JSONDecodeError:
        # \033[1;31m 是 ANSI 转义字符，用于设置字体颜色为红色并加粗，
        # \033[0m 用于恢复默认字体样式
        print("\033[1;31m[错误] AI返回了无效的JSON格式，请重试\033[0m")
        continue
    except:
        print("\033[1;31m[错误] 响应格式异常，正在终止流程\033[0m")
        break

# 输出最终结果
print(f"\033[1;32m[最终查询] {info_request}\033[0m")

2.2.2.2、Multi-Query 多路召回

2.2.2.2.1、多路召回-解决的问题

当用户没有正确书写查询语句，或者LLM不能够正确理解用户查询语句的含义时，此时LLM生成的答案可能就不够完整和全面。

当用户输入查询语句(自然语言)时，我们让大模型(LLM)基于用户的问题再生成多个查询语句，这些生成的查询语句是对用户查询语句的补充，它们是从不同的视角来补充用户的查询语句，然后每条查询语句都会从向量数据库中检索到一批相关文档，最后所有的相关文档都会被喂给LLM，这样LLM就会生成比较完整和全面的答案。这样就可以避免因为查询语句的差异而导致结果不正确

实现方式
1、使用langchain提供的API MultiQueryRetriever。
2、使用自定义提示词。

2.2.2.2.2、多路召回-基本思路

1、利用 LLM 生成 N 个与原始查询相关的问题
2、将所有问题（加上原始查询）发送给检索系统。
3、通过这种方法，可以从向量库中检索到更多文档。

多路召回-流程图

2.2.2.2.3、代码分析

# Multi-Query 多路召回
# 获得访问大模型和嵌入模型客户端
llm, embeddings_model = get_ali_clients()

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

# 创建文档分割器，并分割文档
text_splitter = RecursiveCharacterTextSplitter(chunk_size=600, chunk_overlap=60)
splits = text_splitter.split_documents(docs)

# 创建向量数据库
vectorstore = Chroma.from_documents(documents=splits, 
                                    embedding=embeddings_model)
# 创建检索器
retriever = vectorstore.as_retriever()

# 检索测试，召回文档数量
relevant_docs= retriever.invoke('deepseek的应用场景')
print(relevant_docs)
# 查看一下检索到的相关文档的数量：
print("检索器检索的文档数量为：",len(relevant_docs))
# 检索测试-完成

# 创建prompt模板
template = """请根据下面给出的上下文来回答问题:
{context}
问题: {question}
"""

#由模板生成prompt
prompt = ChatPromptTemplate.from_template(template)

# 用来并发的生成N个类似的提问
chain1 = RunnableMap({
    "context": lambda x: retriever.invoke(x["question"]),
    "question": lambda x: x["question"]
}) | prompt | llm | StrOutputParser()

print("--------------优化前-------------------")
response = chain1.invoke({"question": "deepseek的应用场景"})
print("没有多路召回前，大模型生成的回答：",response)
# 以上是没有多路召回时，我们的大模型生成的回答
# exit()


print("--------------开始优化-------------------")

# 方法一：使用langchain的MultiQueryRetriever
# 引入日志组件查看llm在原查询的基础上生成的多个查询
import logging
logging.basicConfig()
logging.getLogger("langchain_classic.retrievers.multi_query").setLevel(logging.INFO)

# MultiQueryRetriever是对查询的优化，默认生成3个问题
retrieval_from_llm = MultiQueryRetriever.from_llm(
    retriever=retriever,
    llm=llm,
)
unique_docs = retrieval_from_llm.invoke({"question":'deepseek的应用场景'})
print(unique_docs)
print(len(unique_docs))
exit()


# 方法二：自定义prompt，修改提示词获取多个不同的 问题（默认是3个问题）
# prompt模版
template = """你是一个AI语言模型助手。你的任务是生成5个给定用户问题的不同版本，以从向量中检索相关文档
数据库。通过对用户问题产生多种观点，你的目标是提供帮助用户克服了基于距离的相似性搜索的一些限制。
提供了这些用换行符隔开的可选问题。原始问题: {question}"""

prompt_perspectives = ChatPromptTemplate.from_template(template)
generate_queries = (
    prompt_perspectives 
    | llm
    | StrOutputParser() 
    | (lambda x: x.split("\n"))
)

response = generate_queries.invoke({"question":'deepseek的应用场景'})
print(response)

'''接下来我们使用所有的查询语句（假设为n个）去检索向量数据库，
理论上每条查询语句都会检索出4个相关文档，
那么总共可以检索出 n* 4 个相关文档，但是由于这些查询语句含义相近，
而已检索出来的相关文档可能出现重复，
因此我们必须过滤掉重复的相关文档,只保留唯一的文档'''
# 定义一个链，用于获取检索文档的唯一并集（对文档进行去重）
@chain
def get_unique_union(documents: list[list]):
    """ 获取检索文档的唯一并集 """
    # 将列表中的列表展开，并将每个 Document 转换为字符串
    flattened_docs = [dumps(doc) for sublist in documents for doc in sublist]
    # 文档去重
    unique_docs = list(set(flattened_docs))
    # 返回去重后的文档列表
    return [loads(doc) for doc in unique_docs]

# 进行检索，‘retriever.map()’生成的5个问题并行计算
'''retriever.map() 的作用是对输入的查询列表中的每个查询分别进行检索操作，
并返回每个查询对应的检索结果列表。
假设 generate_queries 生成了以下查询列表：["deepseek的应用场景", "deepseek的使用方法", "deepseek的优势"]
retriever.map() 对每个查询进行检索，假设检索结果如下：
    对于查询 "deepseek的应用场景"，检索到文档列表 docs1,docs2。
    对于查询 "deepseek的使用方法"，检索到文档列表 docs2。
    对于查询 "deepseek的优势"，检索到文档列表 docs3。
因此，retriever.map() 的输出将是：[docs1,docs2, docs2, docs3]'''
question = "deepseek的应用场景"
retrieval_chain = generate_queries | retriever.map() | get_unique_union
docs = retrieval_chain.invoke({"question":question})
# 去除重复后的知识库数量
print(len(docs))
print(docs)

print("--------------优化后-------------------")
template = """请根据下面给出的上下文来回答问题:
{context}
问题: {question}
"""

prompt = ChatPromptTemplate.from_template(template)

final_rag_chain = (
    {"context": retrieval_chain, 
     "question": itemgetter("question")}
    | prompt
    | llm
    | StrOutputParser()
)

question = "deepseek的应用场景"
response = final_rag_chain.invoke({"question":question})
print(response)

2.2.2.3、Decomposition 问题分解

2.2.2.3.1、问题分解 -解决的问题

如果用户的问题很复杂，大模型需要推理分解多个步骤才能完成，但是大模型不具备推理能力怎么办？

可以用提示词工程中的CoT策略，把用户的问题拆成一个一个小问题来理解接下来可以使用并行与串行两个策略来执行子任务：
并行执行是将每个子任务抛出去获得一个答案，然后再让大模型把所有子任务的答案汇总起来。
串行执行是依次执行子任务，然后将前一个任务生成的答案作为后一个任务的提示词的一部分。

实现方式
1、使用自定义的API DecompositionQueryRetriever仿多路检索API。

2.2.2.3.2、问题分解-流程

问题分解-流程图

2.2.2.3.3、代码分析

#获得访问大模型和嵌入模型客户端
llm, embeddings_model = get_ali_clients()

# 格式化输出内容
def pretty_print_docs(docs):
    print(
        f"\n{'-' * 100}\n".join(
            [f"Document {i+1}:\n\n" + d.page_content for i, d in enumerate(docs)]
        )
    )

documents = [
    Document(page_content="番茄炒蛋的食材：\n\n- 新鲜鸡蛋：3-4个（根据人数调整）\n- 番茄：2-3个中等大小\n- 盐：适量\n- 白糖：一小勺（可选，用于提鲜）\n- 食用油：适量\n- 葱花：少许（可选，用于增香）\n\n这些是最基本的材料，当然也可以根据个人口味添加其他调料或配料。"),
    Document(page_content="番茄炒蛋的步骤：鸡蛋打入碗中，加入少许盐，用筷子或打蛋器充分搅拌均匀；\n   - 番茄洗净后切成小块备用。\n\n3. **炒鸡蛋**：锅内倒入适量食用油加热至温热状态，然后将搅拌好的鸡蛋液缓缓倒入锅中。待鸡蛋凝固时轻轻翻动几下，让其受热均匀直至完全熟透，随后盛出备用。\n\n4. **炒番茄**：在同一锅里留下的底油中放入切好的番茄块，中小火慢慢翻炒至出汁，可根据个人口味加一点点白糖提鲜。\n\n5. **合炒**：当番茄炒至软烂并开始释放大量汤汁时，再把之前炒好的鸡蛋倒回锅里，快速与番茄混合均匀，同时加入适量的盐调味。如果喜欢的话还可以撒上一些葱花增加香气。\n\n6. **完成**：最后检查一下味道是否合适，确认无误后即可关火装盘享用美味的番茄炒蛋啦！"),
    Document(page_content="技巧与注意事项：1. **选材**：选择新鲜的鸡蛋和成熟的番茄。新鲜的食材是做好这道菜的基础。\n2. **打蛋液**：将鸡蛋打入碗中后加入少许盐（根据个人口味调整），然后充分搅拌均匀。这样做可以让蛋更加松软且味道更佳。\n3. **处理番茄**：番茄最好先用开水稍微焯一下皮，然后去皮切块。这样可以去除表皮的硬质部分，让番茄更容易入味，并且口感更好。\n4. **热锅冷油**：先用中小火把锅烧热，再倒入适量食用油，待油温五成热时下蛋液。这样的做法可以使蛋快速凝固形成漂亮的形状而不易粘锅。\n5. **分步烹饪**：通常建议先炒鸡蛋至半熟状态取出备用；接着利用剩下的底油继续翻炒番茄至出汁，最后再将之前炒好的鸡蛋倒回锅里与番茄混合均匀加热即可。\n6. **调味品**：除了基本的盐之外，还可以根据喜好添加少量糖来提鲜或者一点酱油增色添香。注意调味料不宜过多以免掩盖了食材本身的味道。\n7. **出锅前加葱花**：如果喜欢的话，在即将完成时撒上一些葱花不仅能增加菜品色泽还能增添香气。")
]

vectorstore = Chroma.from_documents(documents=documents,
                                    embedding=embeddings_model,
                                    collection_name="decomposition")
# 利用向量数据库检索，设置topk=1
retriever = vectorstore.as_retriever(search_kwargs={"k": 1})

print("-------------检索到的文档（拆解前）--------------")
#从实际来说，该问题的答案应该包括原材料、步骤，以及注意事项才算完整
pretty_print_docs(retriever.invoke("新手如何制作番茄炒蛋？"))
# exit()


print("开始问题拆解的处理=======================>")
template = """你是一名AI语言模型助理。你的任务是将输入问题分解成3个子问题，通过一个个解决这些子问题从而解决完整的问题。
            子问题需要在向量数据库中检索相关文档。通过分解用户问题生成子问题，你的目标是帮助用户克服基于距离的相似性搜索的一些局限性。
            请提供这些用换行符分隔的子问题本身，不需要额外内容。
            原始问题: {question}"""
DEFAULT_QUERY_PROMPT = PromptTemplate(
    input_variables=["question"],
    template=template,
)

print("-------------测试大模型对问题的拆解，实际业务中可不用--------------")
# LineListOutputParser() 的作用是将模型输出的文本按换行符分割成字符串列表。
chain = DEFAULT_QUERY_PROMPT | llm | LineListOutputParser()
result = chain.invoke({"question":"新手如何制作番茄炒蛋？"})
print("问题拆解：", result)
print("-------------完成测试大模型对问题的拆解--------------")
# exit()

DEFAULT_SUB_QUESTION_PROMPT = PromptTemplate(
    input_variables=["question", "sub_question", "documents"],
    template="""要解决主要问题{question}，需要先解决子问题{sub_question}。
    以下是为支持您的推理而提供的参考文档：{documents}。请直接给出当前子问题的答案。不需要额外内容。""",
)

#自定义一个检索器，将对子问题的生成、获得子问题的答案组合起来，通过组合简化使用过程
# 仿写 langchain.retrievers.multi_query.MultiQueryRetriever
class DecompositionQueryRetriever(BaseRetriever):
    # 向量数据库检索器
    retriever: BaseRetriever
    # 生成子问题链
    make_sub_chain: Runnable
    # 解决子问题链
    resolve_sub_chain: Runnable

    @classmethod
    def from_llm(
            cls,
            retriever: BaseRetriever,
            llm: BaseLanguageModel,
            prompt: BasePromptTemplate = DEFAULT_QUERY_PROMPT,
            sub_prompt: BasePromptTemplate = DEFAULT_SUB_QUESTION_PROMPT
    ) -> "DecompositionQueryRetriever":

        output_parser = LineListOutputParser()
        # make_sub_chain = prompt | llm | output_parser
        # resolve_sub_chain = sub_prompt | llm
        return cls(
            retriever=retriever,
            make_sub_chain=prompt | llm | output_parser,
            resolve_sub_chain=sub_prompt | llm
        )

    #生成子问题
    def generate_queries(self, question: str) -> List[str]:
        response = self.make_sub_chain.invoke({"question": question})
        lines = response
        print(f"生成子问题: {lines}")
        return lines

    # 获得子问题答案
    def retrieve_documents(self, query: str, sub_queries: List[str]) -> List[Document]:
        sub_llm_chain = RunnableLambda(
            # 传入子问题，检索文档并回答
            lambda sub_query: self.resolve_sub_chain.invoke(
                {
                    "question": query,
                    "sub_question": sub_query,
                    "documents": [doc.page_content for doc in self.retriever.invoke(sub_query)]
                }
            )
        )
        # 批量执行所有的子问题
        responses = sub_llm_chain.batch(sub_queries)
        # 将子问题和答案合并作为解决主问题的文档
        documents = [
            Document(page_content=sub_query + "\n" + response.content)
            for sub_query, response in zip(sub_queries, responses)
        ]
        return documents

    # 重写_get_relevant_documents，大模型默认会调用该方法
    def _get_relevant_documents(
            self,
            query: str,
            *,
            run_manager: CallbackManagerForRetrieverRun,
    ) -> List[Document]:
        # 生成子问题
        sub_queries = self.generate_queries(query)
        # 解决子问题
        documents = self.retrieve_documents(query, sub_queries)
        return documents

print("使用DecompositionQueryRetriever来分解问题------------>")
decompositionQueryRetriever = DecompositionQueryRetriever.from_llm(llm=llm, retriever=retriever)
decomposition_docs = decompositionQueryRetriever.invoke("番茄炒蛋怎么制作？")
print("-------------检索到的文档（拆解后）--------------")
pretty_print_docs(decomposition_docs)


# 创建prompt模板
template = """请根据以下文档回答问题:
### 文档:
{context}
### 问题:
{question}
"""

# 由模板生成prompt
prompt = ChatPromptTemplate.from_template(template)

chain = prompt | llm

print("-------------回答--------------")
question = "新手如何制作番茄炒蛋？"
response = chain.invoke({"context": [doc.page_content for doc in decomposition_docs], "question": question})
print(response.content)

2.2.2.4、查询优化小结

查询优化的目标是提升用户意图理解的准确性，包括：
1、Enrich完善问题：通过大模型引导完善用户问题，产生一个更利于系统理解的完善后的用户问题。
2、多路召回（Multi-Query）：针对用户问题生成多个相关问题，分别检索后汇总结果。
3、问题分解（Decomposition）：将复杂问题拆分为多个子问题，依次或并行同步解决所有子问题从而获取最终答案。

2.2.3、检索优化-混合检索

2.2.3.1、混合检索特点

混合检索的核心价值：取长补短，动态适配
本质是根据数据特性、查询需求和场景约束，动态组合多种检索技术：
1、向量检索：擅长捕捉语义相似性，但可能受限于向量空间的表示能力；
2、关键词/全文检索：适合精确匹配，但对自然语言表达不友好；
3、SQL检索：利用数据库，却难以应对非结构化文本。

适用场景
1、异构数据场景：处理多类型、多格式数据
2、复杂查询场景：兼顾精确匹配与语义理解

实现方式
1、使用APIEnsembleRetriever

2.2.3.2、混合检索-图示

混合检索-流程图

2.2.3.3、代码实现

#获得访问大模型和嵌入模型客户端
llm,embeddings_model = get_ali_clients()

# 格式化输出内容
def pretty_print_docs(docs):
    print(
        f"\n{'-' * 100}\n".join(
            [f"Document {i+1}:\n\n" + d.page_content for i, d in enumerate(docs)]
        )
    )

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

# 分割文档
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,
    chunk_overlap=50,
)
split_docs = text_splitter.split_documents(docs)

vectorstore = Chroma.from_documents(
    documents=split_docs, embedding=embeddings_model
)

question = "相关评价"

# 向量检索
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 3})
doc_vector_retriever = vector_retriever.invoke(question)
print("-------------------向量检索-------------------------")
pretty_print_docs(doc_vector_retriever)

# 关键词检索
BM25_retriever = BM25Retriever.from_documents(split_docs)
BM25Retriever.k = 3
doc_BM25Retriever = BM25_retriever.invoke(question)
print("-------------------BM25检索-------------------------")
pretty_print_docs(doc_BM25Retriever)

# 混合检索
# EnsembleRetriever 是Langchain集合多个检索器的检索器。
# EnsembleRetriever 归一化内部进行了封装，使用的不是分数归一化
# retrievers 列表，表示检索器列表 归一化RAG Fusion
ensembleRetriever = EnsembleRetriever(retrievers=[BM25_retriever, vector_retriever], weights=[0.5, 0.5])
retriever_doc = ensembleRetriever.invoke(question)
print("-------------------混合检索-------------------------")
print(retriever_doc)

# 创建prompt模板
template = """请根据下面给出的上下文来回答问题:
{context}
问题: {question}
"""

# 由模板生成prompt
prompt = ChatPromptTemplate.from_template(template)

# 创建chain
chain1 = RunnableMap({
    "context": lambda x: ensembleRetriever.invoke(x["question"]),
    "question": lambda x: x["question"]
}) | prompt | llm | StrOutputParser()

chain2 = RunnableMap({
    "context": lambda x: vector_retriever.invoke(x["question"]),
    "question": lambda x: x["question"]
}) | prompt | llm | StrOutputParser()

print("------------模型回复------------------------")
print("------------向量检索+BM25[0.5, 0.5]------------------------")
print(chain1.invoke({"question":question}))
print("------------向量检索------------------------")
print(chain2.invoke({"question":question}))

2.3、Post-Retrieval后检索优化

与检索前处理相对应，这是在完成检索后对检索出的相关知识块做必要补充处理的阶段。
比如：对检索的结果借助更专业的排序模型与算法进行重排序或者过滤掉一些不符合条件的知识块等，使得最需要、最合规的知识块处于上下文的最前端，这有助于提高大模型的输出质量。

重排序：可以使用重排模型或者算法进行
RAG-Fusion：多路召回 + 倒排融合RRF
上下文压缩和过滤: 用压缩器对检索到的信息过滤和处理，只提取对回答问题最有用的信息

2.3.1、重排序

2.3.1.1、重排序-问题分析

知识库中存在回答问题所需的知识，但是可能这个知识块与问题的向量相似度排名并不是靠前的，导致无法召回该知识块传给大模型，导致大模型始终无法得到正确的答案。

2.3.1.2、重排序-基本思路

增加召回的topK数量，也就是说，例如原来召回前5个知识块，修改为召回前10-20个知识块。然后将召回的10-20个知识块交给重排大模型进行重新精确排序，最后再将排序完的前5个知识块提交给大模型。这种方式容易增加Token数量的消耗

2.3.1.3、重排序-流程图

重排序-流程图

2.3.1.4、重排序-相关代码

reranker = get_ali_rerank()
query = "孕妇感冒了怎么办"

documents = [
    Document(
        page_content="感冒应该吃999感冒灵",
        metadata={"source": "999感冒灵"},
    ),
    Document(
        page_content="高血压患者感冒了吃什么",
        metadata={"source": "高血压患者"},
    ),
    Document(
        page_content="感冒了可以吃感康，但是孕妇禁用",
        metadata={"source": "感康"},
    ),
    Document(
        page_content="感冒了可以咨询专业医生",
        metadata={"source": "专业建议"},
    ),
]
# 得到相关得分
scores = reranker.rerank(documents,query)
print(scores)

# 获取相关文档
scores = reranker.compress_documents(documents, query)
print(scores)

2.3.2、RAG-Fusion

2.3.2.1、RAG-Fusion-痛点分析：

在多个查询检索后，会检索到大量的上下文，但并非所有上下文都与问题相关，有的不相关文档可能出现在文档前面，影响答案生成的准确性。

2.3.2.2、RAG-Fusion-基本思路：

RAG-Fusion是一种搜索方法，通过使用多重查询生成和倒数排名融合（Reciprocal Rank Fusion）对搜索结果进行重新排序。在Multi Query的基础上，对其检索结果进行重新排序(即reranking)后输出Top-K个最相关文档，最后将这top-k个文档喂给LLM并生成最终的答案。
Reciprocal Rank Fusion（倒数排名融合，RRF）公式如下：
$RRF(d)=\sum_{i=1}^n \frac{1}{k+rank_i(d)}$

N：参与融合的检索列表数量（例如BM25和向量检索，则N=2；multi−query生成了3个问题，则N=3
rank_i(d)：文档d在第i个检索系统中的排名（从1开始计数）
k：平滑常数，通常设置为60

2.3.2.3、RAG-Fusion-图示：

RAG-Fusion-流程图

2.3.2.4、RAG-Fusion-代码详解：

# 获得访问大模型和嵌入模型客户端
llm,embeddings_model = get_ali_clients()

texts=[
    "人工智能在医疗诊断中的应用。",
    "人工智能如何提升供应链效率。",
    "NBA季后赛最新赛况分析。",
    "传统法式烘焙的五大技巧。",
    "红楼梦人物关系图谱分析。",
    "人工智能在金融风险管理中的应用。",
    "人工智能如何影响未来就业市场。",
    "人工智能在制造业的应用。",
    "今天天气怎么样",
    "人工智能伦理：公平性与透明度。"
]

# 创建向量数据库对象
vectorstore = Chroma.from_texts(
    texts=texts, embedding= embeddings_model
)   

retriever = vectorstore.as_retriever()

#从langchain官网拉取预先定义好的prompt
prompt = hub.pull("langchain-ai/rag-fusion-query-generation")
print(prompt)
#也可以手工定义prompt如下：
# prompt = ChatPromptTemplate.from_messages([
#     ("system", "You are a helpful assistant that generates multiple search queries based on a single input query."),
#     ("user", "Generate multiple search queries related to: {original_query}"),
#     ("user", "OUTPUT (4 queries):")
# ])

# 创建多重查询chain
generate_queries = (
    prompt | llm | StrOutputParser() | (lambda x: x.split("\n"))
)

original_query = "人工智能的应用"
queries = generate_queries.invoke({"original_query": original_query})
print(f"原始查询：{original_query},生成的查询：{queries}")

@chain
def reciprocal_rank_fusion(results: list[list], k=60):
    """互逆排序融合算法，用于合并多个排序文档列表
    Args:
        results: 包含多个排序文档列表的二维列表
        k: 融合公式中的平滑参数（默认60），值越小排名影响越大
    Returns:
        按融合分数降序排列的文档列表，每个元素为(文档对象, 分数)元组
    """
    # 初始化融合分数字典（key=序列化文档，value=累计分数）
    fused_scores = {}

    # 遍历每个检索结果列表（每个查询对应的结果）
    for docs in results:
        # 对当前结果列表中的文档进行遍历（rank从0开始计算）
        for rank, doc in enumerate(docs):
            # 序列化文档对象为字符串（用于唯一标识）
            doc_str = dumps(doc)
            # 初始化文档得分（如果是首次出现）
            if doc_str not in fused_scores:
                fused_scores[doc_str] = 0
            # 计算并累加RRF分数：1 / (当前排名 + k)
            # 排名越靠前（rank值小）的文档获得的分数越高
            fused_scores[doc_str] += 1 / (rank + k)

    # 按融合分数降序排序（分数越高排名越前）
    reranked_results = [
        (loads(doc), score)  # 反序列化还原文档对象
        for doc, score in sorted(fused_scores.items(), 
                               key=lambda x: x[1], 
                               reverse=True)
    ]

    return reranked_results


original_query = "人工智能的应用"
'''
generate_queries会生成4个多角度的query,
retriever.map()的作用是根据generate_queries的结果映射出4个retriever(可以理解为同时复制出4个retriever)
与generate_queries生成的4个query对应，
并为每个query检索出来的一组相关文档集(默认为4个相关文档)，
那么4个query总共可以生成16个相关文档。
最后会经过RRF算法重新排序后输出最相关的文档
'''
chain = generate_queries | retriever.map() | reciprocal_rank_fusion

# 输入结果列表
result_list = chain.invoke({"original_query": original_query})
# 提取文档内容和对应分数
contents = [doc[0].page_content for doc in result_list]
scores = [doc[1] for doc in result_list]

combined_tuples = list(zip(contents, scores))
print("--"*15, "\n最相关的文档及其得分：")
for item in combined_tuples:
    print(item)

print("--"*15, "\n分析一下这些分数是如何统计出来的：")
#分析一下这些分数是如何统计出来的
chain1 = generate_queries | retriever.map() 
chain1_result = chain1.invoke({"original_query": original_query})

#原始输出是一个二维列表，每个元素是由4个query生成的Document列表
print(chain1_result)

# 处理输出格式
contents = [doc.page_content 
            for group in chain1_result  # 遍历外层列表
            for doc in group]        # 遍历内层文档列表
print(contents)

2.3.4、上下文压缩和过滤

2.3.4.1、上下文压缩和过滤 - 痛点分析：

在划分文档块的时候，通常不知道用户的查询，这意味着，与查询最相关的信息可能隐藏在一个包含大量不相关文本的文档中，这样输入给LLM，可能会导致更昂贵的LLM调用和较差的响应。

2.3.4.2、上下文压缩和过滤 - 解决步骤

使用给定查询的上下文来压缩它们，以便只返回相关信息，而不是立即按原样返回检索到的文档。

使用某种基本的检索器来检索不同的信息;
然后将检索到的信息添加到文档压缩器中;
压缩器对这些信息进行过滤和处理，只提取对回答问题有用的信息。

2.3.4.3、上下文压缩和过滤 - 流程图

流程图

2.3.4.4、上下文压缩和过滤-代码示例

# Post-Retrieval后检索-上下文压缩

# 获得访问大模型和嵌入模型客户端
llm, embeddings_model = get_ali_clients()

# 格式化输出内容
def pretty_print_docs(docs):
    print(
        f"\n{'-' * 100}\n".join(
            [f"Document {i+1}:\n\n" + d.page_content for i, d in enumerate(docs)]
        )
    )

documents = TextLoader("./deepseek百度百科.txt",encoding="utf-8").load()
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1024,  
    chunk_overlap=100
)
texts = text_splitter.split_documents(documents)

#使用基础检索器
retriever = Chroma.from_documents(texts, embeddings_model).as_retriever()

# docs = retriever.invoke("deepseek的发展历程")
# print("-------------------压缩前--------------------------")
# pretty_print_docs(docs)
#
# print("-------------------第一种：LLMChainExtractor压缩------------------")
#使用上下文压缩检索器
compressor = LLMChainExtractor.from_llm(llm)
compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor, base_retriever=retriever
)
#
# compressed_docs = compression_retriever.invoke(
#     "deepseek的发展历程"
# )
# print("-------------------压缩后--------------------------")
# pretty_print_docs(compressed_docs)
#
#
# print("-------------------第二种：LLMChainFilter压缩后--------------------------")
# #LLMChainFilter 是稍微简单但更强大的压缩器
# _filter = LLMChainFilter.from_llm(llm)
# compression_retriever = ContextualCompressionRetriever(
#     base_compressor=_filter, base_retriever=retriever
# )
#
# compressed_docs = compression_retriever.invoke(
#     "deepseek的发展历程"
# )
#
# pretty_print_docs(compressed_docs)

print("-------------------第三种：EmbeddingsFilter压缩后--------------------------")
#对每个检索到的文档进行额外的 LLM 调用既昂贵又缓慢。
#EmbeddingsFilter 通过嵌入文档和查询并仅返回那些与查询具有足够相似嵌入的文档来提供更便宜且更快的选项
from langchain_classic.retrievers.document_compressors import EmbeddingsFilter, DocumentCompressorPipeline

embeddings_filter = EmbeddingsFilter(embeddings=embeddings_model, similarity_threshold=0.6)
compression_retriever = ContextualCompressionRetriever(
    base_compressor=embeddings_filter, base_retriever=retriever
)

compressed_docs = compression_retriever.invoke(
    "deepseek的发展历程"
)

pretty_print_docs(compressed_docs)

print("-------------------第四种：组合压缩后--------------------------")
# DocumentCompressorPipeline轻松地按顺序组合多个压缩器
'''
1.首先TextSplitters可以用作文档转换器，将文档分割成更小的块，
2.然后EmbeddingsRedundantFilter 根据文档之间嵌入的相似性来过滤掉冗余文档，
该过滤操作以文本的嵌入向量为依据，也就是借助余弦相似度来衡量文本之间的相似程度，
进而判定是否存在冗余，它会把文本列表转化成对应的嵌入向量，然后计算每对文本之间的余弦相似度。
一旦相似度超出设定的阈值，就会将其中一个文本判定为冗余并过滤掉。
3.最后 EmbeddingsFilter 根据与查询的相关性进行过滤。'''
splitter = CharacterTextSplitter(chunk_size=300, chunk_overlap=0, separator=". ")
# EmbeddingsRedundantFilter 去除重复的文档块
redundant_filter = EmbeddingsRedundantFilter(embeddings=embeddings_model)
# EmbeddingsFilter 过滤掉相似度小于0.6的
relevant_filter = EmbeddingsFilter(embeddings=embeddings_model, similarity_threshold=0.6)
# 组合以上多种方式
pipeline_compressor = DocumentCompressorPipeline(
    transformers=[splitter, redundant_filter, relevant_filter]
)
# 压缩检索器
compression_retriever = ContextualCompressionRetriever(
    base_compressor=pipeline_compressor, base_retriever=retriever
)

compressed_docs = compression_retriever.invoke("deepseek的发展历程")
pretty_print_docs(compressed_docs)

RAG痛点分析-Advanced RAG