L2-3 基于LangChain的RAG系统优化实践
·
关键知识点
一、RAG商业化痛点分析
1、Index Process（文本向量化构建索引的过程）
- Missing Content（内容缺失）：原本的文本中就没有问题的答案。
- 增加相应知识库：将相应的知识文本加入到向量知识库中。
- 数据清洗与增强：输入垃圾，那也必定输出垃圾。任何RAG工作流程想要获得优良表现，都必须先清洁数据。
- 更好的Prompt设计：通过Prompts，比如让大模型在找不到答案的情况下，输出“根据当前知识库，无法回答该问题”等提示。
- 文档加载准确性和效率：比如pdf文件的加载，如何提取其中的有用文字信息和图片信息等。
- 优化文档读取器：一般知识库中的文档格式都不尽相同，针对每一类文档，涉及一个专门的读取器。
- 数据清洗与增强
- 文档切分的粒度：文本切分的大小和位置会影响后面检索出来的上下文完整性和与大模型交互的token数量，怎么控制好文档切分的度，是个难题。
- 内容重叠分块：为了保持文本块之间语义上下文的连贯性，在分块时，保持文本块之间有一定的内容重叠。
- 基于结构的分块：基于结构的分块方法利用文档的固有结构，如HTML或Markdown中的标题和段落，以保持内容的逻辑性和完整性。
- 基于递归的分块：重复的利用分块规则不断细分文本块。比如先通过段落换行符（\n\n）进行分割。然后，检查这些块的大小。如果大小不超过一定阈值，则该块被保留。对于大小超过标准的块，使用单换行符（\n）再次分割。以此类推，不断根据块大小更新更小的分块规则（如空格，句号）。
- 分块大小的选择：不同的嵌入模型有其最佳输入大小。比如Openai的text-embedding-ada-002的模型在256 或 512大小的块上效果更好。文档的类型和用户查询的长度及复杂性也是决定分块大小的重要因素。
2、Query Process（检索增强回答的过程中）
- Missed Top Ranked：错过排名靠前的文档。
- 外挂知识库中存在回答问题所需的知识，但是可能这个知识块与问题的向量相似度排名并不是靠前的，导致无法召回该知识块传给大模型，导致大模型始终无法得到正确的答案。
- 增加召回数量：增加召回的 topK 数量，也就是说，例如原来召回前3个知识块，修改为召回前5个知识块。不过此种方法，因为知识块多了，不光会增加token消耗，也会增加大模型回答问题的干扰。
- 重排（Reranking）
- Not in Context：提取上下文与答案无关。
- 内容缺失 或 错过排名靠前的文档 的具体体现。
- Wrong Format（格式错误）：例如需要Json，给了字符串。
- Prompt调优优化Prompt逐渐让大模型返回正确的格式。
- 进行结果格式验证，例如使用LangChain中的PydanticOutputParser类来校验输出格式。
- Auto-Fixing自修复：对不符合要求的格式进行自动修复。
- Incomplete（答案不完整）：答案只回答了问题的一部分。
- 将问题分开提问一方面引导用户精简问题，一次只提问一个问题。 另一方面，针对用户的问题进行内部拆分处理，拆分成数个子问题，等子问题答案都找到后，再总结起来回复给用户。
- Not Extracted（未提取到答案）：提取的上下文中有答案，但大模型没有提取出来。
- 提示压缩技术
- Incorrect Specificity：答案不够具体或过于具体。
- 提示词改善 或者 提升基座大模型能力
二、Advanced RAG概述
- Advanced RAG重点聚焦在检索增强，即优化Retrieval阶段。增加了Pre-Retrieval预检索和Post-Retrieval后检索阶段，同时对检索本身也有优化。
- Pre-Retrieval 预检索过程优化/检索前优化：高级RAG着重优化了索引结构和查询的方式。优化索引旨在提高被索引内容的质量，包括增强数据颗粒度、优化索引结构、添加元数据、对齐优化等策略。查询优化的目标则是明确用户的原始问题，使其更适合检索任务，使用了查询重写、查询转换、查询扩展等技术。
- Retrieval 检索优化：检索阶段的目标是确定最相关的上下文。通常，检索基于向量搜索，它计算查询与索引数据之间的语义相似性。因此，大多数检索优化技术都围绕嵌入模型展开，比如微调嵌入模型，将嵌入模型定制为特定领域的上下文，特别是对于术语不断演化或罕见的领域。还有其他检索技术，例如混合搜索，通常是指将向量搜索与基于关键字的搜索相结合的概念。
- Post-Retrieval 后检索过程优化/检索后优化：对于由问题检索得到的一系列上下文，后检索策略关注如何优化它们与查询问题的集成。这一过程主要包括重新排序和压缩上下文。重新排列检索到的信息，将最相关的内容予以定位标记，这种策略已经在LlamaIndex2、LangChain等框架中得以实施。有时直接将所有相关文档输入到大型语言模型（LLMs）可能导致信息过载，为了缓解这一点，后检索工作集中选择必要的信息，强调关键部分，并限制了相应的上下文长度。
三、Pre-Retrieval预检索-优化索引
0、models.py
# models.py
# 可用模型列表，以及获得访问模型的客户端
# 实际使用时可以根据自己的实际情况调整
ALI_TONGYI_API_KEY_OS_VAR_NAME = "DASHSCOPE_API_KEY"
ALI_TONGYI_URL = "https://dashscope.aliyuncs.com/compatible-mode/v1"
ALI_TONGYI_MAX_MODEL = "qwen-max-latest"
ALI_TONGYI_DEEPSEEK_R1 = "deepseek-r1"
ALI_TONGYI_DEEPSEEK_V3 = "deepseek-v3"
ALI_TONGYI_3 = "qwen3-235b-a22b"
ALI_TONGYI_REASONER_MODEL = "qwq-plus"
ALI_TONGYI_EMBEDDING_MODEL = "text-embedding-v3"
ALI_TONGYI_EMBEDDING_MODEL_V4 = "text-embedding-v4"
ALI_TONGYI_RERANK_MODEL = "gte-rerank-v2"
DEEPSEEK_API_KEY_OS_VAR_NAME = "Deepseek_Key"
DEEPSEEK_URL = "https://api.deepseek.com/v1"
DEEPSEEK_CHAT_MODEL = "deepseek-chat"
DEEPSEEK_REASONER_MODEL = "deepseek-reasoner"
TENCENT_HUNYUAN_API_KEY_OS_VAR_NAME = "HUNYUAN_API_KEY"
TENCENT_HUNYUAN_URL = "https://api.hunyuan.cloud.tencent.com/v1"
TENCENT_HUNYUAN_TURBO_MODEL = "hunyuan-turbos-latest"
TENCENT_HUNYUAN_REASONER_MODEL = "hunyuan-t1-latest"
TENCENT_HUNYUAN_LONGCONTEXT_MODEL = "hunyuan-large-longcontext"
TENCENT_HUNYUAN_EMBEDDING_MODEL = "hunyuan-embedding"
TENCENT_SECRET_ID_OS_VAR_NAME = "Tencent_SecretId"
TENCENT_SECRET_KEY_OS_VAR_NAME = "Tencent_SecretKey"
BAICHUAN_API_KEY_OS_VAR_NAME = "Baichuan_API_Key"
BAICHUAN_EMBEDDING_URL = "https://api.baichuan-ai.com/v1/embeddings"
BAICHUAN_EMBEDDING_MODEL = "Baichuan-Text-Embedding"
import os
import inspect
from langchain_openai import ChatOpenAI
from langchain_community.embeddings import BaichuanTextEmbeddings, DashScopeEmbeddings, HunyuanEmbeddings
from langchain_community.document_compressors.dashscope_rerank import DashScopeRerank
def get_lc_o_model_client(api_key=os.getenv(ALI_TONGYI_API_KEY_OS_VAR_NAME), base_url=TENCENT_HUNYUAN_URL,
model=ALI_TONGYI_3, temperature=0.7, verbose=False, debug=False):
'''
以OpenAI兼容的方式，通过LangChain获得指定平台和模型的客户端
可以通过传入api_key，base_url，model，temperature四个参数来覆盖默认值
verbose，debug两个参数，分别控制是否输出详细调试信息，是否输出调试信息，默认不打印
:return: 指定平台和模型的客户端，默认平台和模型为阿里百炼qwen3-235b-a22b，温度=0.7
'''
function_name = inspect.currentframe().f_code.co_name
if (verbose):
print(f"{function_name}-平台：{base_url},模型：{model},温度：{temperature}")
if (debug):
print(f"{function_name}-平台：{base_url},模型：{model},温度：{temperature},key：{api_key}")
return ChatOpenAI(api_key=api_key, base_url=base_url, model=model, temperature=temperature)
def get_lc_o_ali_model_client(model=ALI_TONGYI_MAX_MODEL, temperature=0.7, verbose=False, debug=False):
'''
以OpenAI兼容的方式，通过LangChain获得阿里大模型的客户端
可以通过传入model，temperature 两个参数来覆盖默认值
verbose，debug两个参数，分别控制是否输出调试信息，是否输出详细调试信息，默认不打印
:return: 指定平台和模型的客户端，默认模型为阿里百炼里的qwen-max-latest，温度=0.7
'''
return get_lc_o_model_client(api_key=os.getenv(ALI_TONGYI_API_KEY_OS_VAR_NAME), base_url=ALI_TONGYI_URL,
model=model, temperature=temperature, verbose=verbose, debug=debug)
def get_lc_o_tencent_model_client(model=TENCENT_HUNYUAN_TURBO_MODEL, temperature=0.7, verbose=False, debug=False):
"""通过LangChain使用腾讯大模型 hunyuan-turbos-latest"""
return get_lc_o_model_client(api_key=os.getenv(TENCENT_HUNYUAN_API_KEY_OS_VAR_NAME), base_url=TENCENT_HUNYUAN_URL,
model=model, temperature=temperature, verbose=verbose, debug=debug)
def get_lc_o_ds_model_client(model=DEEPSEEK_CHAT_MODEL, temperature=0.7, verbose=False, debug=False):
'''
以OpenAI兼容的方式，通过LangChain获得DeepSeek大模型的客户端
可以通过传入model，temperature 两个参数来覆盖默认值
verbose，debug两个参数，分别控制是否输出调试信息，是否输出详细调试信息，默认不打印
:return: 指定平台和模型的客户端，默认模型为DeepSeek的deepseek-chat，温度=0.7
'''
return get_lc_o_model_client(api_key=os.getenv(DEEPSEEK_API_KEY_OS_VAR_NAME), base_url=DEEPSEEK_URL,
model=model, temperature=temperature, verbose=verbose, debug=debug)
def get_lc_baichuan_embeddings():
'''
通过LangChain获得一个百川嵌入模型的实例，百川嵌入模型服务限流严重，
大概有10~20%的概率访问报错
:return: 百川嵌入模型的实例
'''
return BaichuanTextEmbeddings(
api_key=os.getenv(BAICHUAN_API_KEY_OS_VAR_NAME)
)
def get_lc_ali_embeddings():
'''
通过LangChain获得一个阿里通义千问嵌入模型的实例
:return: 阿里通义千问嵌入模型的实例，目前为text-embedding-v4
'''
return DashScopeEmbeddings(
model=ALI_TONGYI_EMBEDDING_MODEL_V4, dashscope_api_key=os.getenv(ALI_TONGYI_API_KEY_OS_VAR_NAME)
)
def get_lc_tencent_embeddings():
'''
通过LangChain获得一个腾讯嵌入模型的实例
腾讯模型需要单独在腾讯云上再申请hunyuan_secret_id和hunyuan_secret_key
:return: 腾讯嵌入模型的实例
'''
return HunyuanEmbeddings(
hunyuan_secret_id=os.getenv(TENCENT_SECRET_ID_OS_VAR_NAME),
hunyuan_secret_key=os.getenv(TENCENT_SECRET_KEY_OS_VAR_NAME),
region="ap-guangzhou",
)
def get_lc_ali_all_clients():
'''
以OpenAI兼容的方式，同时产生阿里大模型客户端和嵌入模型的客户端
:return: 阿里大模型客户端和嵌入模型的客户端
'''
return get_lc_o_ali_model_client(), get_lc_ali_embeddings()
def get_lc_a_t_mix_clients():
'''
以OpenAI兼容的方式，混合产生阿里的大模型客户端和腾讯嵌入模型的客户端
:return: 阿里的大模型客户端和腾讯嵌入模型的客户端
'''
return get_lc_o_ali_model_client(), get_lc_tencent_embeddings()
def get_lc_ali_rerank(top_n=3):
'''
通过LangChain获得一个阿里重排序模型的实例
:return: 阿里通义千问重排序模型的实例 gte-rerank-v2
'''
return DashScopeRerank(
model=ALI_TONGYI_RERANK_MODEL, dashscope_api_key=os.getenv(ALI_TONGYI_API_KEY_OS_VAR_NAME),
top_n=top_n
)
1、摘要索引
-
让LLM为每个块生成summary，并作为embedding存到summary database中。
-
在检索时，通过summary database找到最相关的summary，再回溯到原始文档中去。
-
将原始文本块作为上下文发送给LLM以获取答案。
import uuid from models import get_lc_ali_all_clients from langchain.storage import InMemoryByteStore from langchain_chroma import Chroma from langchain_community.document_loaders import TextLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain.retrievers import MultiVectorRetriever from langchain_core.documents import Document from langchain_core.output_parsers import StrOutputParser from langchain_core.prompts import ChatPromptTemplate from langchain_core.runnables import RunnableMap # 获得访问大模型和嵌入模型客户端 client, embeddings_model = get_lc_ali_all_clients() # 初始化文档加载器 loader = TextLoader("a.txt", encoding="utf-8") # 加载文档 docs = loader.load() print(docs) # [Document()] # 初始化递归文本分割器（设置块大小和重叠） text_splitter = RecursiveCharacterTextSplitter(chunk_size=1024, chunk_overlap=100) docs = text_splitter.split_documents(docs) print(docs) # [Document(), Document()] # 初始化Chroma实例（用于存储摘要向量） vectorstore = Chroma( collection_name="summaries", 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] # 借助大模型，把文本做相关的摘要 chain = ( {"doc": lambda x: x.page_content} | ChatPromptTemplate.from_template("总结下面的文档:\n\n{doc}") | client | StrOutputParser() ) print("准备生成文档摘要，时间稍长，请耐心等待...") # invoke 一次调用，stream 流式调用，batch批量调用 summaries = chain.batch(docs, {"max_concurrency": 5, }) # print(summaries) # 大模型的答复本质上是个字符串，再重新包装为Document对象 summary_docs = [ Document(page_content=s, metadata={id_key: doc_ids[i]}) for i, s in enumerate(summaries) ] # 将摘要添加到向量数据库 print("准备将摘要添加到向量数据库...") retriever.vectorstore.add_documents(summary_docs) print("准备将原始文档存储到字节存储...") 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("-------------对应的原始文档--------------") # 通过ID获取原始文档 # original_doc = retriever.docstore.mget([matched_id]) # print(original_doc) # 执行相似性搜索测试---完成 prompt = ChatPromptTemplate.from_template("根据下面的文档回答问题:\n\n{doc}\n\n问题: {question}") chain = RunnableMap({ "doc": lambda x: retriever.invoke(x["question"]), "question": lambda x: x["question"] }) | prompt | client | StrOutputParser() # 生成问题回答 answer = chain.invoke({"question": query}) print("-------------回答--------------") print(answer) # retriever.invoke将"对摘要进行检索，但是通过关联ID获得原始文档，最终返回原始文档"的过程全部都包含完成了 retrieved_docs = retriever.invoke(query) print("-------------检索到的文档--------------") print(retrieved_docs)
2、父子索引
-
文档被分割成一个层级化的块结构，随后用最小的叶子块进行索引。
-
在检索过程中检索出top k个叶子块。
-
如果存在n个叶子块都指向同一个更大的父块，那么我们就用这个父块来替换这些子块，并将其送入大模型用于生成答案。
import os from models import get_lc_ali_all_clients from langchain_community.document_loaders import WebBaseLoader, TextLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_chroma import Chroma from langchain_community.embeddings.dashscope import DashScopeEmbeddings from langchain.retrievers import ParentDocumentRetriever from langchain_core.stores import InMemoryStore from langchain_openai.chat_models import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.runnables import RunnableMap from langchain_core.output_parsers import StrOutputParser # 获得访问大模型和嵌入模型客户端 client, embeddings_model = get_lc_ali_all_clients() # 加载数据 loader = TextLoader("a.txt", encoding="utf-8") docs = loader.load() print(docs) # 查看长度 print(f"文章的长度：{len(docs[0].page_content)}") parent_splitter = RecursiveCharacterTextSplitter(chunk_size=1024) child_splitter = RecursiveCharacterTextSplitter(chunk_size=256) # 创建向量数据库对象 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": 1} ) # 添加文档集 retriever.add_documents(docs) print(f"主文块的数量：{len(list(store.yield_keys()))}") # 测试 - 相似性搜索 print("------------similarity_search------------------------") sub_docs = vectorstore.similarity_search("deepseek的应用场景") print(sub_docs[0].page_content) print("------------get_relevant_documents------------------------") retrieved_docs = retriever.invoke("deepseek的应用场景") print(retrieved_docs[0].page_content) # 创建prompt模板 template = """请根据下面给出的上下文来回答问题: {context} 问题: {question} """ # 由模板生成prompt prompt = ChatPromptTemplate.from_template(template) # 创建chain chain = RunnableMap({ "context": lambda x: retriever.invoke(x["question"]), "question": lambda x: x["question"] }) | prompt | client | StrOutputParser() print("------------模型回复------------------------") response = chain.invoke({"question": "deepseek的应用场景"}) print(response)
3、假设性问题索引
-
让LLM为每个块生成3个假设性问题，并将这些问题以向量形式嵌入。
-
在运行时，针对这个问题向量的索引进行查询搜索（用问题向量替换文档的块向量）。
-
检索后将原始文本块作为上下文发送给LLM以获取答案。
import uuid import langchain from typing import List from langchain.storage import InMemoryByteStore from langchain_chroma import Chroma from langchain_community.document_loaders import TextLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain.retrievers import MultiVectorRetriever from langchain_core.documents import Document from langchain_core.output_parsers import StrOutputParser from langchain_core.prompts import ChatPromptTemplate from langchain_core.runnables import RunnableMap from pydantic import BaseModel, Field from models import get_lc_ali_all_clients # 获得访问大模型和嵌入模型客户端 client, embeddings_model = get_lc_ali_all_clients() # 初始化文档加载器列表 loader = TextLoader("a.txt", encoding="utf-8") docs = loader.load() # 初始化递归文本分割器（设置块大小和重叠） text_splitter = RecursiveCharacterTextSplitter(chunk_size=1024, chunk_overlap=100) docs = text_splitter.split_documents(docs) print(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] # 以下开始用大模型生成假设性问题 class HypotheticalQuestions(BaseModel): """生成假设性问题""" questions: List[str] = Field(..., description="List of 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输出构建为字符串列表 | client.with_structured_output(HypotheticalQuestions) # 提取问题列表 | (lambda x: x.questions) ) # 测试-在单个文档上调用链，链的最终输出是大模型答复的假设性问题列表 print("测试：", docs[0]) print("测试生成假设性问题：", chain.invoke(docs[0])) # 批量处理所有文档生成假设性问题（最大并行数5） hypothetical_questions = chain.batch(docs, {"max_concurrency": 5}) 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] ) 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("测试-执行相似性搜索：", sub_docs) prompt1 = ChatPromptTemplate.from_template("根据下面的文档回答问题:\n\n{doc}\n\n问题: {question}") chain = RunnableMap({ "doc": lambda x: retriever.invoke(x["question"]), "question": lambda x: x["question"] }) | prompt1 | client | StrOutputParser() # 生成问题回答 answer = chain.invoke({"question": query}) print("-------------回答--------------") print(answer) retrieved_docs = retriever.invoke(query) print("-------------检索到的文档--------------") print(retrieved_docs)
4、元数据索引
-
定义元数据标签，如果文档本身没有，可以利用大模型推理出输入问题的元数据。
-
通过标签先对文档进行过滤。
-
结合向量检索进一步定位到最相关的前 K 个知识块。
from models import get_lc_ali_all_clients from langchain_chroma import Chroma from langchain_core.documents import Document from langchain.retrievers.self_query.base import SelfQueryRetriever from langchain.chains.query_constructor.schema import AttributeInfo from langchain.chains.query_constructor.base import ( StructuredQueryOutputParser, get_query_constructor_prompt, ) # 获得访问大模型和嵌入模型客户端 llm, embeddings_model = get_lc_ali_all_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) 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 = "技术文章简述" retriever = SelfQueryRetriever.from_llm( llm, vectorstore, document_content_description, metadata_field_info, ) print(retriever.invoke("我想了解评分在9分以上的文章")) print(retriever.invoke("作者B在2023年发布的文章")) # 构建查询解析器（看工作原理用） prompt = get_query_constructor_prompt( document_content_description, metadata_field_info, ) output_parser = StructuredQueryOutputParser.from_components() query_constructor = prompt | llm | output_parser print("提示词：", prompt.format(query="我想了解评分在9分以上的文章")) print("提示词显示结束-------------------------------") print("结构化查询结果：", query_constructor.invoke( { "query": "作者B在2023年发布的文章" } )) # 看工作原理-结束 # retriever = SelfQueryRetriever.from_llm( # llm, # vectorstore, # document_content_description, # metadata_field_info, # enable_limit=True, # 让用户通过自然语言控制返回结果数量时启用 # ) # # print(retriever.invoke("我想了解一篇评分在9分以上的文章"))
5、总结
| 索引优化 | 适用场景 | 案例 |
|---|---|---|
| 摘要索引 | 适用于需要快速检索和生成简洁上下文的场景。 | 在新闻资讯平台中，系统需要快速从海量新闻中提取关键信息，通过摘要索引可以迅速生成简洁的上下文，帮助用户快速了解新闻的核心内容。 |
| 父子索引 | 适用于需要确保语义完整性和层次化检索的场景。 | 在法律检索系统中，用户查询法律条款时，父子索引通过分层检索精准查找相关内容，并召回对应大文档块确保上下文的完整性，避免因分块过细导致语义丢失。 |
| 假设性问题索引 | 适用于需要处理复杂查询和多样化表达的场景。 | 在药品咨询系统中，用户查询症状时可能会问：“感冒了吃什么药？”。假设性问题索引通过为每种药品生成一系列假设性问题，帮助用户更准确地检索到相关信息。 |
| 元数据索引 | 适用于需要快速筛选和分类的场景。 | 在电商推荐系统中，系统通过元数据索引快速筛选出符合用户偏好的商品信息，提高推荐效率和准确性。 |
四、Pre-Retrieval预检索-查询优化
1、Enrich 完善问题
-
在口语表达需求和意图时，人们往往会遇到一些问题。例如，表达过于简略或含糊，容易引发语义歧义，导致大模型产生误解；用户的问题可能包含许多隐含要素，但表达的信息却不足，只能通过多轮对话逐步补全。
-
理想情况：通过大模型多次主动与用户沟通，不断收集信息，完善对用户真实意图的理解，补全执行用户需求所需的各项参数。
import os import json from models import get_lc_o_ali_model_client from langchain.retrievers import RePhraseQueryRetriever from langchain_community.embeddings.dashscope import DashScopeEmbeddings from langchain_openai import ChatOpenAI from langchain.prompts import PromptTemplate, ChatPromptTemplate from langchain_core.runnables.history import RunnableWithMessageHistory from langchain_community.chat_message_histories import ChatMessageHistory from langchain_core.output_parsers import JsonOutputParser # 获得访问大模型客户端 llm = get_lc_o_ali_model_client() user_input = "我想订一张长沙去北京的机票" templates = { "订机票": ["起点", "终点", "时间", "座位等级", "座位偏好"], "订酒店": ["城市", "入住日期", "退房日期", "房型", "人数"], } intent_prompt = PromptTemplate( input_variables=["user_input", "templates"], template="根据用户输入 '{user_input}'，选择最合适的业务模板。可用模板如下：{templates}。请返回模板名称。" ) intent_chain = intent_prompt | llm intent = intent_chain.invoke({"user_input": user_input, "templates": str(list(templates.keys()))}).content print("意图：", intent) # 获取对应模板 selected_template = templates.get(intent) print("模板：", selected_template) # 补充信息提示模板 info_prompt = f""" 请根据用户原始问题和模板，判断原始问题是否完善。如果问题缺乏需要的信息，请生成一个友好的请求，明确指出需要补充的信息。若问题完善后，返回包含所有信息的完整问题。 ### 原始问题 {user_input} ### 模板 {",".join(selected_template)} ### 输出示例 {{ "isComplete": true, "content": "`完整问题`" }} {{ "isComplete": false, "content": "`友好的引导到需要补充信息`" }} """ # 历史记录 chat_history = ChatMessageHistory() # 聊天模版 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", ) # 判断问题是否完整，如果不完整，则生成追问请求 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: user_answer = input(f"\033[1;33m{json_data['content']}\033[0m\n你的回复：") info_request = with_message_history.invoke( input={"input": user_answer}, config={"configurable": {"session_id": "unused"}} ).content json_data = parser.parse(info_request) except json.JSONDecodeError: print("\033[1;31m[错误] AI返回了无效的JSON格式，请重试\033[0m") continue except KeyError: print("\033[1;31m[错误] 响应格式异常，正在终止流程\033[0m") break # 输出最终结果 print(f"\033[1;32m[最终查询] {info_request}\033[0m")
2、Multi-Query 多路召回
-
当用户没有正确书写查询语句，或者LLM不能够正确理解用户查询语句的含义时，此时LLM生成的答案可能就不够完整和全面。
-
当用户输入查询语句（自然语言）时，我们让大模型(LLM)基于用户的问题再生成多个查询语句，这些生成的查询语句是对用户查询语句的补充，它们是从不同的视角来补充用户的查询语句，然后每条查询语句都会从向量数据库中检索到一批相关文档，最后所有的相关文档都会被喂给LLM，这样LLM就会生成比较完整和全面的答案。这样就可以避免因为查询语句的差异而导致结果不正确。
-
利用 LLM 生成 N 个与原始查询相关的问题。将所有问题（加上原始查询）发送给检索系统。通过这种方法，可以从向量库中检索到更多文档。
# Multi-Query 多路召回 from operator import itemgetter from models import get_lc_ali_all_clients from langchain.load import dumps, loads from langchain.prompts import ChatPromptTemplate from langchain.retrievers import MultiQueryRetriever from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_community.document_loaders import TextLoader from langchain_community.vectorstores import Chroma from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnableMap # 获得访问大模型和嵌入模型客户端 llm, embeddings_model = get_lc_ali_all_clients() # 加载文档 loader = TextLoader("../a.txt", encoding="utf-8") docs = loader.load() # 创建文档分割器，并分割文档 text_splitter = RecursiveCharacterTextSplitter(chunk_size=600, chunk_overlap=100) 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) chain = RunnableMap({ "context": lambda x: relevant_docs, "question": lambda x: x["question"] }) | prompt | llm | StrOutputParser() print("--------------优化前-------------------") response = chain.invoke({"question": "deepseek的应用场景"}) print(response) print("--------------开始优化-------------------") # 方法一：使用langchain的MultiQueryRetriever import logging logging.basicConfig() logging.getLogger("langchain.retrievers.multi_query").setLevel(logging.INFO) retrieval_from_llm = MultiQueryRetriever.from_llm( retriever=retriever, llm=llm ) # unique_docs = retrieval_from_llm.invoke({"question": 'deepseek的应用场景'}) unique_docs = retrieval_from_llm.invoke('deepseek的应用场景') print(unique_docs) print(len(unique_docs)) # 方法二：自定义prompt # 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) def get_unique_union(documents: list[list]): """ 获取检索文档的唯一并集 """ # 将列表中的Document展开，并将每个 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] # 进行检索 ''' 假设 generate_queries 生成了以下查询列表：["deepseek的应用场景", "deepseek的使用方法", "deepseek的优势"] ''' question = "deepseek的应用场景" retrieval_chain = generate_queries | retriever.map() | get_unique_union docs = retrieval_chain.invoke({"question": question}) print(len(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)
3、Decomposition 问题分解
-
如果用户的问题很复杂，大模型需要推理分解多个步骤才能完成，但是大模型不具备推理能力怎么办？
-
可以用提示词工程中的CoT策略，把用户的问题拆成一个一个小问题来理解，接下来可以使用并行与串行两个策略来执行子任务。
-
并行执行是将每个子任务抛出去获得一个答案，然后再让大模型把所有子任务的答案汇总起来。
-
串行是依次执行子任务，然后将前一个任务生成的答案作为后一个任务的提示词的一部分。
# 查询优化，问题分解 from typing import List from models import get_lc_ali_all_clients from langchain_core.documents import Document from langchain_chroma import Chroma from langchain_openai import ChatOpenAI from langchain_community.embeddings.dashscope import DashScopeEmbeddings from langchain.retrievers.multi_query import LineListOutputParser from langchain_core.prompts import ChatPromptTemplate from langchain_core.retrievers import BaseRetriever from langchain_core.language_models import BaseLanguageModel from langchain_core.prompts import PromptTemplate, BasePromptTemplate from langchain.retrievers.multi_query import LineListOutputParser from langchain_core.callbacks import CallbackManagerForRetrieverRun from langchain_core.documents import Document from langchain_core.runnables import Runnable, RunnableLambda # 获得访问大模型和嵌入模型客户端 llm, embeddings_model = get_lc_ali_all_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") retriever = vectorstore.as_retriever(search_kwargs={"k": 1}) print("-------------检索到的文档（拆解前）--------------") pretty_print_docs(retriever.invoke("新手如何制作番茄炒蛋？")) print("-------------问题拆解--------------") template = """你是一名AI语言模型助理。你的任务是将输入问题分解成3个子问题，通过一个个解决这些子问题从而解决完整的问题。 子问题需要在矢量数据库中检索相关文档。通过分解用户问题生成子问题，你的目标是帮助用户克服基于距离的相似性搜索的一些局限性。 请提供这些用换行符分隔的子问题本身，不需要额外内容。 原始问题: {question}""" DEFAULT_QUERY_PROMPT = PromptTemplate( input_variables=["question"], template=template, ) print("-------------测试大模型对问题的拆解，实际业务中可不用--------------") chain = DEFAULT_QUERY_PROMPT | llm | LineListOutputParser() result = chain.invoke({"question": "番茄炒蛋怎么制作？"}) print(result) print("-------------完成测试大模型对问题的拆解--------------") DEFAULT_SUB_QUESTION_PROMPT = PromptTemplate( input_variables=["question", "sub_question", "documents"], template="""要解决主要问题{question}，需要先解决子问题{sub_question}。以下是为支持您的推理而提供的参考文档：{documents}。请直接给出当前子问题的答案。不需要额外内容。""", ) class DecompositionQueryRetriever(BaseRetriever): # 向量数据库检索器 retriever: BaseRetriever # 生成子问题链 llm_chain: Runnable # 解决子问题链 sub_llm_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() llm_chain = prompt | llm | output_parser sub_llm_chain = sub_prompt | llm return cls( retriever=retriever, llm_chain=llm_chain, sub_llm_chain=sub_llm_chain ) 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 def generate_queries(self, question: str) -> List[str]: response = self.llm_chain.invoke({"question": question}) lines = response print(f"Generated queries: {lines}") return lines def retrieve_documents(self, query: str, sub_queries: List[str]) -> List[Document]: sub_llm_chain = RunnableLambda( # 传入子问题，检索文档并回答 lambda sub_query: self.sub_llm_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 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)
4、总结：查询优化的目标是提升用户意图理解的准确性。
- 完善问题（Enrich）：通过大模型引导完善用户问题，产生一个更利于系统理解的完善后的用户问题。
- 多路召回（Multi-Query）：针对用户问题生成多个相关问题，分别检索后汇总结果。
- 问题分解（Decomposition）：将复杂问题拆分为多个子问题，依次或同步解决所有子问题从而获取最终答案。
五、检索优化-混合检索
-
混合检索的核心价值：取长补短，动态适配混合检索的本质是根据数据特性、查询需求和场景约束，动态组合多种检索技术：向量检索擅长捕捉语义相似性，但可能受限于向量空间的表示能力；关键词 / 全文检索适合精确匹配，但对自然语言表达不友好；SQL检索利用数据库，却难以应对非结构化文本。
-
适用场景：异构数据场景：处理多类型、多格式数据。复杂查询场景：兼顾精确匹配与语义理解。
-
适用场景：动态知识与实时性需求：融合静态知识与实时数据。高准确率与召回率要求：医疗、法律等关键领域。
from models import get_lc_ali_all_clients from langchain_community.document_loaders import TextLoader from langchain_chroma import Chroma from langchain_community.retrievers import BM25Retriever from langchain.retrievers import EnsembleRetriever from langchain_core.output_parsers import StrOutputParser from langchain_core.prompts import ChatPromptTemplate from langchain_core.runnables import RunnableMap from langchain.text_splitter import RecursiveCharacterTextSplitter # 获得访问大模型和嵌入模型客户端 llm, embeddings_model = get_lc_ali_all_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("../a.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提供的一个集合多个检索器的检索器 # 注意：EnsembleRetrieve没有使用归一化，使用的其他的算法融合两者的结果 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}))
六、Post-Retrieval后检索优化
与检索前处理相对应，这是在完成检索后对检索出的相关知识块做必要补充处理的阶段。比如，对检索的结果借助更专业的排序模型与算法进行重排序或者过滤掉一些不符合条件的知识块等，使得最需要、最合规的知识块处于上下文的最前端，这有助于提高大模型的输出质量。
1、重排序
（1）使用排序模型
from langchain_core.documents import Document
from models import get_lc_ali_rerank
reranker = get_lc_ali_rerank()
query = "孕妇感冒了怎么办"
documents = [
"感冒应该吃999感冒灵",
"高血压患者感冒了吃什么",
"感冒了可以吃感康，但是孕妇禁用"
]
scores = reranker.rerank(documents, query)
print(scores) # 2,0,1
documents = [
Document(
page_content="感冒应该吃999感冒灵",
metadata={"source": "999感冒灵"},
),
Document(
page_content="高血压患者感冒了吃什么",
metadata={"source": "高血压患者"},
),
Document(
page_content="感冒了可以吃感康，但是孕妇禁用",
metadata={"source": "感康"},
),
]
scores = reranker.compress_documents(documents, query)
print(scores) # 2,0,1
（2）使用算法
from langchain_community.document_transformers import LongContextReorder
# 5,4,3,2,1
# 倒排：1,2,3,4,5
# 前一个，后一个：5,3,1,2,4
# 按相关性排序5，4，3，2，1，5是最相关的，相关性依次递减
documents = [
"相关性:5",
"相关性:4",
"相关性:3",
"相关性:2",
"相关性:1",
]
reordering = LongContextReorder()
reordered_docs = reordering.transform_documents(documents)
print(reordered_docs)
# ['相关性:5', '相关性:3', '相关性:1', '相关性:2', '相关性:4']
2、RAG-Fusion
-
痛点分析：在多个查询检索后，会检索到大量的上下文，但并非所有上下文都与问题相关，有的不相关文档可能出现在文档前面，影响答案生成的准确性。
-
RAG-Fusion 是一种搜索方法，通过使用多重查询生成和互惠排名融合（Reciprocal Rank Fusion）对搜索结果进行重新排序。在Multi Query的基础上，对其检索结果进行重新排序(即reranking)后输出Top K个最相关文档，最后将这top k个文档喂给LLM并生成最终的答案。
-
Reciprocal Rank Fusion（倒数排名融合，RRF）：1/(k+rank_1(d)) + 1/(k+rank_2(d)) + … + 1/(k+rank_i(d))
- N：参与融合的检索列表数量（例如BM25和向量检索，则N=2；multi−query生成了3个问题，则N=3）
- rank_i(d)：文档d在第i个检索系统的排名（从1开始计数）
- k：平滑常数，通常设置为60
# Post-Retrieval后检索-重排序RAG-Fusion
from langchain import hub
from langchain_chroma import Chroma
from langchain_core.output_parsers import StrOutputParser
from langchain.load import dumps, loads
from models import get_lc_ali_all_clients
# 获得访问大模型和嵌入模型客户端
llm, embeddings_model = get_lc_ali_all_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}")
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, "最相关的文档及其得分：")
for item in combined_tuples:
print(item)
print("--" * 15, "分析一下这些分数是如何统计出来的：")
# 分析一下这些分数是如何统计出来的
chain1 = generate_queries | retriever.map()
chain1_result = chain1.invoke({"original_query": original_query})
# 原始输出是一个二维列表，每个元素是由4个query生成的Document列表
print(chain1_result)
# 处理输出格式
for group in chain1_result: # 遍历外层列表
contents = [doc.page_content for doc in group] # 遍历内层文档列表
print(contents)
# 原始查询：人工智能的应用, 生成的查询：['1. 人工智能在医疗领域的应用 ', '2. 人工智能如何改变教育行业 ', '3. 人工智能在金融行业的实际应用案例 ', '4. 人工智能技术在日常生活中的使用场景']
# ------------------------------ 最相关的文档及其得分：
# ('人工智能在医疗诊断中的应用。', 0.06559979182930002)
# ('人工智能在制造业的应用。', 0.06530936012691697)
# ('人工智能如何影响未来就业市场。', 0.06454173067076292)
# ('人工智能在金融风险管理中的应用。', 0.04866871479774705)
# ('人工智能如何提升供应链效率。', 0.016129032258064516)
# ------------------------------ 分析一下这些分数是如何统计出来的：
# ['人工智能在医疗诊断中的应用。', '人工智能在制造业的应用。', '人工智能在金融风险管理中的应用。', '人工智能如何影响未来就业市场。']
# ['人工智能如何影响未来就业市场。', '人工智能在制造业的应用。', '人工智能如何提升供应链效率。', '人工智能在医疗诊断中的应用。']
# ['人工智能在金融风险管理中的应用。', '人工智能在医疗诊断中的应用。', '人工智能在制造业的应用。', '人工智能如何影响未来就业市场。']
# ['人工智能在医疗诊断中的应用。', '人工智能在制造业的应用。', '人工智能如何影响未来就业市场。', '人工智能在金融风险管理中的应用。']
3、上下文压缩和过滤
# Post-Retrieval后检索-上下文压缩
from langchain_community.vectorstores import Chroma
from langchain_community.document_loaders import TextLoader
from langchain_community.document_transformers import EmbeddingsRedundantFilter
from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor
from langchain.retrievers.document_compressors import LLMChainFilter
from langchain.retrievers.document_compressors import EmbeddingsFilter
from langchain.retrievers.document_compressors import DocumentCompressorPipeline
from langchain_text_splitters import CharacterTextSplitter
from langchain_text_splitters import RecursiveCharacterTextSplitter
from models import get_lc_ali_all_clients
# 获得访问大模型和嵌入模型客户端
llm, embeddings_model = get_lc_ali_all_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("../a.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 通过嵌入文档和查询并仅返回那些与查询具有足够相似嵌入的文档来提供更便宜且更快的选项
embeddings_filter = EmbeddingsFilter(embeddings=embeddings_model, similarity_threshold=0.66)
compression_retriever = ContextualCompressionRetriever(
base_compressor=embeddings_filter, base_retriever=retriever
)
compressed_docs = compression_retriever.invoke(
"deepseek的发展历程"
)
pretty_print_docs(compressed_docs)
print("-------------------第四种：组合压缩后--------------------------")
# DocumentCompressorPipeline轻松地按顺序组合多个压缩器
splitter = CharacterTextSplitter(chunk_size=300, chunk_overlap=0, separator=". ")
# 根据文档的相似度来过滤冗余文档
redundant_filter = EmbeddingsRedundantFilter(embeddings=embeddings_model)
# 根据查询的问题和文档的相似性来过滤不相关文档
relevant_filter = EmbeddingsFilter(embeddings=embeddings_model, similarity_threshold=0.66)
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)
七、Advanced RAG实战——金融助手
基本步骤：
-
首先用 Unstructured 来提取文档 （PDF） 中的文本和表格，并进行分块
Unstructured 使用：注意：本程序financial_assistant.py的运行需要科学上网
-
tesseract ：用于光学字符识别 （OCR）
-
poppler ：用于 PDF 渲染和处理
-
-
然后用LLM分别对每个文本和表格创建摘要，将其嵌入向量数据库
-
最后通过摘要使用MultiVectorRetriever过滤出相关文档，喂给LLM当作上下文
import os
import uuid
import time
from typing import Any
from pydantic import BaseModel
from unstructured.partition.pdf import partition_pdf
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain.retrievers.multi_vector import MultiVectorRetriever
from langchain.storage import InMemoryStore
from langchain_chroma import Chroma
from langchain_core.documents import Document
from langchain_core.runnables import RunnablePassthrough
from models import get_lc_ali_all_clients
print("\033[1;31m本程序的运行需要科学上网！\033[0m")
# 获得访问大模型和嵌入模型客户端
llm, embeddings_model = get_lc_ali_all_clients()
print("\033[1;34m请根据自己的实际情况修改代码中TESSDATA_PREFIX环境变量、Poppler 的 bin 目录、"
"tesseract安装目录这三者的具体值！\033[0m")
# 设置 TESSDATA_PREFIX 环境变量
os.environ["TESSDATA_PREFIX"] = r"C:\Program Files\Tesseract-OCR\tessdata"
# 添加 Poppler 的 bin 目录到系统 PATH : 用于 PDF 渲染和处理
os.environ["PATH"] += os.pathsep + r"D:\developTools\pythonDevelop\poppler-24.08.0\Library\bin"
# 添加 tesseract 安装目录到系统 PATH : 用于光学字符识别 （OCR）
os.environ["PATH"] += os.pathsep + r"C:\Program Files\Tesseract-OCR"
# 定义文件路径（示例路径，实际使用时需修改）
path = "./2020-03-17__厦门灿坤实业股份有限公司__200512__闽灿坤__2019年__年度报告.pdf"
# ------------------------ 第一阶段：PDF解析处理 ------------------------
print("正在解析pdf文件，请耐心等候!")
start = time.time_ns()
# 使用unstructured库解析PDF文档
raw_pdf_elements = partition_pdf(
filename=path,
extract_images_in_pdf=False, # 不提取PDF中的图片
infer_table_structure=True, # 启用表格结构识别
chunking_strategy="by_title", # 按标题分块策略
max_characters=4000, # 每个文本块最大字符数
new_after_n_chars=3800, # 达到3800字符后分新块
combine_text_under_n_chars=2000, # 合并小于2000字符的碎片文本
image_output_dir_path=path, # 图片输出目录（本例未使用）
)
# 统计各类元素数量
category_counts = {}
for element in raw_pdf_elements:
category = str(type(element))
category_counts[category] = category_counts.get(category, 0) + 1
print("解析完成，元素类型统计:", category_counts)
print(f"解析运行时间：{(time.time_ns() - start) / 1_000_000:.2f} ms")
# ------------------------ 第二阶段：元素分类处理 ------------------------
start = time.time_ns()
# 定义文档元素模型
class Element(BaseModel):
"""文档元素基类"""
type: str # 元素类型（table/text）
text: Any # 元素内容
# 分类处理PDF元素，并提取表格和文本元素
table_elements = []
text_elements = []
for element in raw_pdf_elements:
if "unstructured.documents.elements.Table" in str(type(element)):
table_elements.append(Element(type="table", text=str(element)))
elif "unstructured.documents.elements.CompositeElement" in str(type(element)):
text_elements.append(Element(type="text", text=str(element)))
print(f"识别到表格数量: {len(table_elements)}, 文本块数量: {len(text_elements)}")
print("表格内容示例:", table_elements[0:10])
print(f"元素分类运行时间：{(time.time_ns() - start) / 1_000_000:.2f} ms")
# ------------------------ 第三阶段：内容摘要生成 ------------------------
start = time.time_ns()
# 定义摘要生成提示模板
prompt_text = """您是一个专业的内容摘要助手，请对以下表格或文本块进行简洁的总结：
{element}"""
prompt = ChatPromptTemplate.from_template(prompt_text)
# 构建摘要生成链
summarize_chain = {"element": lambda x: x} | prompt | llm | StrOutputParser()
print("准备提取表格摘要......")
# 批量生成表格摘要
tables = [i.text for i in table_elements]
table_summaries = summarize_chain.batch(tables, {"max_concurrency": 5}) # 并发处理
print("----表格摘要示例:", table_summaries[0:10])
print("准备提取文本摘要......")
# 批量生成文本摘要
texts = [i.text for i in text_elements]
text_summaries = summarize_chain.batch(texts, {"max_concurrency": 1})
print("----文本摘要示例:", text_summaries[0:10])
print(f"内容摘要生成运行时间：{(time.time_ns() - start) / 1_000_000:.2f} ms")
# ------------------------ 第四阶段：构建多向量检索器 ------------------------
start = time.time_ns()
# 创建向量数据库（用于存储摘要）
vectorstore = Chroma(
collection_name="summaries",
embedding_function=embeddings_model
)
# 创建内存存储（用于存储原始内容）
store = InMemoryStore()
id_key = "doc_id" # 文档标识键
# 初始化多向量检索器
retriever = MultiVectorRetriever(
vectorstore=vectorstore,
docstore=store,
id_key=id_key,
)
# 添加文本数据到向量数据库
text_ids = [str(uuid.uuid4()) for _ in texts]
summary_texts = [
Document(page_content=s, metadata={id_key: text_ids[i]})
for i, s in enumerate(text_summaries)
]
retriever.vectorstore.add_documents(summary_texts)
retriever.docstore.mset(list(zip(text_ids, texts)))
# 添加表格数据到向量数据库
table_ids = [str(uuid.uuid4()) for _ in tables]
summary_tables = [
Document(page_content=s, metadata={id_key: table_ids[i]})
for i, s in enumerate(table_summaries)
]
retriever.vectorstore.add_documents(summary_tables)
retriever.docstore.mset(list(zip(table_ids, tables)))
# ------------------------ 第五阶段：构建问答链 ------------------------
# 定义问答提示模板
template = """请仅根据以下上下文（包含文本和表格）回答问题：
{context}
问题：{question}
"""
prompt = ChatPromptTemplate.from_template(template)
# 构建问答链
chain = (
{"context": retriever, "question": RunnablePassthrough()}
| prompt
| llm
| StrOutputParser()
)
# 示例问答测试
question = "2019年比去年的营业收入增长了百分之多少"
print("回答：", chain.invoke(question))
print("检索结果：", retriever.invoke(question))
print(f"构建知识库及问答运行时间：{(time.time_ns() - start) / 1_000_000:.2f} ms")
惟楚有才，于斯为盛。欢迎来到长沙！！！ 茶颜悦色、臭豆腐、CSDN和你一个都不能少~
更多推荐
4
0- 0
已为社区贡献1条内容
所有评论(0)