新的一年开始了,也许您正打算通过构建自己的第一个RAG系统进入RAG领域。或者,您可能已经构建了基本的RAG系统,现在希望将它们改进为更高级的系统,以更好地处理用户的查询和数据结构。
无论哪种情况,了解从何处或如何开始可能本身就是一个挑战!如果是这样,希望这篇博客文章能为您指明下一步的正确方向,更重要的是,在构建高级RAG系统时为您提供一个思维模型,以便在做决策时有所依据。
上面分享的 RAG 备忘单受到了最近一篇RAG调查论文的启发(Retrieval-Augmented Generation for Large Language Models: A Survey” Gao, Yunfan, et al. 2023
)。
技术要学会分享、交流,不建议闭门造车。一个人可以走的很快、一堆人可以走的更远。
成立了大模型技术交流群,本文完整代码、相关资料、技术交流&答疑,均可加我们的交流群获取,群友已超过2000人,添加时最好的备注方式为:来源+兴趣方向,方便找到志同道合的朋友。
方式①、微信搜索公众号:机器学习社区,后台回复:加群
方式②、添加微信号:mlc2060,备注:来自CSDN + 技术交流
如今,主流的RAG定义涉及从外部知识数据库中检索文档,并将这些文档与用户的查询一起传递给LLM进行响应生成。换句话说,RAG包含检索组件、外部知识数据库和生成组件。
LlamaIndex 基础RAG配方:
from llama_index import SimpleDirectoryReader, VectorStoreIndex
# 加载数据
documents = SimpleDirectoryReader(input_dir="...").load_data()
# 构建VectorStoreIndex,负责对文档进行分块
# 并为将来的检索编码块以进行嵌入
index = VectorStoreIndex.from_documents(documents=documents)
# QueryEngine类配备了生成器
# 并促进了检索和生成步骤
query_engine = index.as_query_engine()
# 使用默认的RAG
response = query_engine.query("用户的查询")
为了使RAG系统被认为是成功的(以提供有用且相关的答案给用户问题为标准),实际上只有两个高层次的要求:
有了成功的要求定义,我们可以说构建高级RAG实际上是关于应用更复杂的技术和策略(对检索或生成组件)以确保最终满足这些要求。
此外,我们可以将复杂的技术分类为解决两个高层次成功要求之一的技术(或多或少独立于另一个),或者是同时解决这两个要求的技术。
检索的高级技术必须能够找到与用户查询最相关的文档
以下简要描述了几种更复杂的技术,以帮助实现第一个成功要求。
LlamaIndex 块大小优化配方:
from llama_index import ServiceContext
from llama_index.param_tuner.base import ParamTuner, RunResult
from llama_index.evaluation import SemanticSimilarityEvaluator, BatchEvalRunner
# 执行超参数调整,就像在传统ML中通过网格搜索一样
# 1. 定义一个排名不同参数组合的目标函数
# 2. 构建ParamTuner对象
# 3. 使用ParamTuner.tune()执行超参数调整
# 1. 定义目标函数
def objective_function(params_dict):
chunk_size = params_dict["chunk_size"]
docs = params_dict["docs"]
top_k = params_dict["top_k"]
eval_qs = params_dict["eval_qs"]
ref_response_strs = params_dict["ref_response_strs"]
# 构建RAG流水线
index = _build_index(chunk_size, docs) # 此处未显示辅助函数
query_engine = index.as_query_engine(similarity_top_k=top_k)
# 对提供的问题`eval_qs`执行RAG流水线的推断
pred_response_objs = get_responses(
eval_qs, query_engine, show_progress=True
)
# 通过将其与参考响应`ref_response_strs`进行比较来执行预测的评估
evaluator = SemanticSimilarityEvaluator(...)
eval_batch_runner = BatchEvalRunner(
{"semantic_similarity": evaluator}, workers=2, show_progress=True
)
eval_results = eval_batch_runner.evaluate_responses(
eval_qs, responses=pred_response_objs, reference=ref_response_strs
)
# 获取语义相似性指标
mean_score = np.array(
[r.score for r in eval_results["semantic_similarity"]]
).mean()
return RunResult(score=mean_score, params=params_dict)
# 2. 构建ParamTuner对象
param_dict = {"chunk_size": [256, 512, 1024]} # 要搜索的参数/值
fixed_param_dict = { # 固定的超参数
"top_k": 2,
"docs": docs,
"eval_qs": eval_qs[:10],
"ref_response_strs": ref_response_strs[:10],
}
param_tuner = ParamTuner(
param_fn=objective_function,
param_dict=param_dict,
fixed_param_dict=fixed_param_dict,
show_progress=True,
)
# 3. 执行超参数搜索
results = param_tuner.tune()
best_result = results.best_run_result
best_chunk_size = results.best_run_result.params["chunk_size"]
LlamaIndex递归检索配方:
from llama_index import SimpleDirectoryReader, VectorStoreIndex
from llama_index.node_parser import SentenceSplitter
from llama_index.schema import IndexNode
### 配方
### 构建一个递归检索器,使用小块进行检索
### 但将相关的较大块传递到生成阶段
# 加载数据
documents = SimpleDirectoryReader(
input_file="some_data_path/llama2.pdf"
).load_data()
# 通过NodeParser构建父块
node_parser = SentenceSplitter(chunk_size=1024)
base_nodes = node_parser.get_nodes_from_documents(documents)
# 定义较小的子块
sub_chunk_sizes = [256, 512]
sub_node_parsers = [
SentenceSplitter(chunk_size=c, chunk_overlap=20) for c in sub_chunk_sizes
]
all_nodes = []
for base_node in base_nodes:
for n in sub_node_parsers:
sub_nodes = n.get_nodes_from_documents([base_node])
sub_inodes = [
IndexNode.from_text_node(sn, base_node.node_id) for sn in sub_nodes
]
all_nodes.extend(sub_inodes)
# 还要添加原始节点到节点
original_node = IndexNode.from_text_node(base_node, base_node.node_id)
all_nodes.append(original_node)
# 使用所有节点定义一个VectorStoreIndex
vector_index_chunk = VectorStoreIndex(
all_nodes, service_context=service_context
)
vector_retriever_chunk = vector_index_chunk.as_retriever(similarity_top_k=2)
# 构建RecursiveRetriever
all_nodes_dict = {n.node_id: n for n in all_nodes}
retriever_chunk = RecursiveRetriever(
"vector",
retriever_dict={"vector": vector_retriever_chunk},
node_dict=all_nodes_dict,
verbose=True,
)
# 使用recursive_retriever构建RetrieverQueryEngine
query_engine_chunk = RetrieverQueryEngine.from_args(
retriever_chunk, service_context=service_context
)
# 使用高级RAG执行推理(即查询引擎)
response = query_engine_chunk.query(
"你能告诉我有关安全微调的关键概念吗?"
)
其他有用的链接
我们有几个指南演示了在复杂情况下应用其他高级技术以确保准确检索。以下是其中一些的链接:
与前一节类似,我们提供了这一类别下的一些复杂技术的示例,这可以被描述为确保检索到的文档与生成器的LLM很好地对齐。
LlamaIndex信息压缩配方:
from llama_index import SimpleDirectoryReader, VectorStoreIndex
from llama_index.query_engine import RetrieverQueryEngine
from llama_index.postprocessor import LongLLMLinguaPostprocessor
### 配方
### 定义一个Postprocessor对象,这里是LongLLMLinguaPostprocessor
### 构建使用此Postprocessor处理检索到的文档的QueryEngine
# 定义Postprocessor
node_postprocessor = LongLLMLinguaPostprocessor(
instruction_str="在给定的上下文中,请回答最后的问题",
target_token=300,
rank_method="longllmlingua",
additional_compress_kwargs={
"condition_compare": True,
"condition_in_question": "after",
"context_budget": "+100",
"reorder_context": "sort", # 启用文档重新排序
},
)
# 定义VectorStoreIndex
documents = SimpleDirectoryReader(input_dir="...").load_data()
index = VectorStoreIndex.from_documents(documents)
# 定义QueryEngine
retriever = index.as_retriever(similarity_top_k=2)
retriever_query_engine = RetrieverQueryEngine.from_args(
retriever, node_postprocessors=[node_postprocessor]
)
# 使用你的高级RAG
response = retriever_query_engine.query("用户的查询")
LlamaIndex重新排名以获得更好生成配方:
import os
from llama_index import SimpleDirectoryReader, VectorStoreIndex
from llama_index.postprocessor.cohere_rerank import CohereRerank
from llama_index.postprocessor import LongLLMLinguaPostprocessor
### 配方
### 定义一个Postprocessor对象,这里是CohereRerank
### 构建使用此Postprocessor处理检索到的文档的QueryEngine
# 构建CohereRerank检索后处理器
api_key = os.environ["COHERE_API_KEY"]
cohere_rerank = CohereRerank(api_key=api_key, top_n=2)
# 构建QueryEngine(RAG),使用后处理器
documents = SimpleDirectoryReader("./data/paul_graham/").load_data()
index = VectorStoreIndex.from_documents(documents=documents)
query_engine = index.as_query_engine(
similarity_top_k=10,
node_postprocessors=[cohere_rerank],
)
# 使用你的高级RAG
response = query_engine.query(
"在这篇文章中,Sam Altman做了什么?"
)
在这个子节中,我们考虑使用检索和生成的协同作用来实现更好的检索以及对用户查询更准确生成响应的复杂方法。
LlamaIndex生成器增强检索配方:
from llama_index.llms import OpenAI
from llama_index.query_engine import FLAREInstructQueryEngine
from llama_index import (
VectorStoreIndex,
SimpleDirectoryReader,
ServiceContext,
)
### 配方
### 构建一个FLAREInstructQueryEngine,其中生成器LLM通过提示
### 在检索中发挥更积极的作用,以引导它提取检索
### 关于如何回答用户查询的指示。
# 构建FLAREInstructQueryEngine
documents = SimpleDirectoryReader("./data/paul_graham").load_data()
index = VectorStoreIndex.from_documents(documents)
index_query_engine = index.as_query_engine(similarity_top_k=2)
service_context = ServiceContext.from_defaults(llm=OpenAI(model="gpt-4"))
flare_query_engine = FLAREInstructQueryEngine(
query_engine=index_query_engine,
service_context=service_context,
max_iterations=7,
verbose=True,
)
# 使用你的高级RAG
response = flare_query_engine.query(
"你能告诉我作者在创业界的轨迹吗?"
)
LlamaIndex迭代检索生成配方:
from llama_index.query_engine import RetryQueryEngine
from llama_index.evaluation import RelevancyEvaluator
### 配方
### 构建RetryQueryEngine,执行检索生成周期
### 直到它达到合格的评估或达到最大次数
### 周期已达到
# 构建RetryQueryEngine
documents = SimpleDirectoryReader("./data/paul_graham").load_data()
index = VectorStoreIndex.from_documents(documents)
base_query_engine = index.as_query_engine()
query_response_evaluator = RelevancyEvaluator() # 评估器来批评
# 检索生成周期
retry_query_engine = RetryQueryEngine(
base_query_engine, query_response_evaluator
)
# 使用你的高级RAG
retry_response = retry_query_engine.query("一个用户查询")
评估RAG系统当然是至关重要的。在他们的调查论文中,Gao,Yunfan等人在附上的RAG备忘单的右上部分指出了7个测量方面。llama-index库包括多个评估抽象以及与RAGAs的集成,以帮助构建者通过这些测量方面的视角了解他们的RAG系统达到成功要求的水平。下面,我们列举了一些评估笔记本指南。
希望通过阅读本博客文章,您感到更有能力和信心应用一些这些复杂的技术来构建高级RAG系统!
参考文献
https://blog.llamaindex.ai/a-cheat-sheet-and-some-recipes-for-building-advanced-rag-803a9d94c41b