构造自己的RAG（增强检索生成）系统（一）
作者： Archieli 来源： Archieli
上一篇文章介绍了RAG技术的演进及技术要点，本文拿一个具体的项目来动手实现，走一遍RAG的流程。
项目的github链接：
https://github.com/archieli/rag_project
用到的第三方框架LlamaIndex，它是一个基于大语言模型构建上下文增强应用的数据框架。把上文提到的技术和理论封装成模块和函数，能够极大地增加开发效率。https://docs.llamaindex.ai/en/stable/
RAG的整个流程（pipline）一般包括以下的步骤，加载（Loading）、索引（Indexing）、存储（Storing）、查询（Querying）、评估（Evaluating）。
数据处理（Loading）
这个阶段的工作是对各种类型的数据源（PDF、网页、数据库、API等）收集和预处理，形成文档和节点。
在LlamaIndex中，文档(Document) 指的是封装任意数据源的容器，可以认为是一个PDF文档、一个页面的数据等。节点（Node） 是LlamaIndex的基本单位，是数据源/文档按照某种规则切分的数据块，可以是一块文本、图片等。
把一个目录下的所有文件加载并转换成数据块，在LlamaIndex中实现起来非常简单，
def directory_to_nodes(self):
''' Loads the data from the directory and returns the nodes
'''
#加载指定目录下的数据，生成 Document对象列表
documents = SimpleDirectoryReader("../../data/essay").load_data()
#加载数据块划分模块
parser = SentenceSplitter()
#将Document对象列表按划分，转换为Node对象列表
nodes = parser.get_nodes_from_documents(documents)
return nodes
-
SimpleDirectoryReader读取指定目录下的所有文件（PDF、TXT、HTML等）并转行成Document对象的列表。
-
SentenceSplitter是在尊重句子完整性的条件下分割文本、生成数据节点。它有两个参数：chunk_size和chunk_overlap，分别表示数据块的字符数及数据块间的重叠字符数。还有其他的数据节点生成的方法，如SentenceWindowNodeParser（把所有文档分割成单独的句子）、SemanticSplitterNodeParser（根据语义相似度选择分割节点）、HierarchicalNodeParser（把输入分割成多层的数据节点）
-
get_nodes_from_documents，将文档按照指定的分割方法生成数据块
看看生成的数据块（TextNode）长啥样：
{
"id_": "763ef92c-09b5-4ca5-94be-4311a368b2e7",
"embedding": null,
"metadata": {
"file_name": "paul_graham_essay.txt",
"file_type": "text/plain",
"file_size": 75033,
"creation_date": "2024-04-25",
"last_modified_date": "2024-04-25"
},
"excluded_embed_metadata_keys":["file_name", "file_type", "file_size", "creation_date", "last_modified_date", "last_accessed_date"],
"excluded_llm_metadata_keys": ["file_name", "file_type", "file_size", "creation_date", "last_modified_date", "last_accessed_date"],
"relationships": {
"1": {
"node_id": "3b7c41e1-313f-4af3-95f5-bca65395568a",
"node_type": "4",
"metadata": {
"file_name": "paul_graham_essay.txt",
"file_type": "text/plain",
"file_size": 75033,
"creation_date": "2024-04-25",
"last_modified_date": "2024-04-25"
},
"hash": "546429f31386324e2213e56cc133bb94aae3c0c5527c768d6d5f9f060949605b",
"class_name": "RelatedNodeInfo"
},
"3": {
"node_id": "96997954-05d9-4bf1-853b-273af2dd8b1c",
"node_type": "1",
"metadata": {
},
"hash": "284f12df0915dfbf354e5dd0d63efcff4bc1af50700bcb5b2493bafd192a0ec8",
"class_name": "RelatedNodeInfo"
}
},
"text": "What I Worked On\n\nFebruary 2021\n\nBefore college the two main things I worked on, outside of school, were writing and programming.",
"start_char_idx": 2,
"end_char_idx": 131,
"text_template": "{metadata_str}\n\n{content}",
"metadata_template": "{key}: {value}",
"metadata_seperator": "\n",
"class_name": "TextNode"
}
数据块的信息非常丰富，每个块都有唯一的_id，metadata包含了数据块的元信息，可以自行指定内容。
excluded_embed_metadata_keys 和 excluded_llm_metadata_keys是设置排除向量化或者与LLM交互的元数据；
text是数据块的实际内容，start_char_idx 和 end_char_idx表示数据划分的起始和终止位置。
relationships表示数据块和其他块的关系，在召回的时候如果满足某关系便能一起返回，提高数据的准确度。
构建数据块是一个非常精细的工作，需要考虑文档的结构、语句顺序、各页关系等，综合起来选择或者自定义数据块的分割方法，让用户的问题能够尽量全、准确地召回相关的数据块。
索引（Indexing）和存储（Storing）
索引是用于支持快速查询到相关数据块的数据结构，通常涉及到向量化Embedding，通过数字化的方法来表示数据块的核心内容。可以在构建索引和查询阶段添加元数据信息等策略来提升数据块的召回效果。**
存储阶段是保存创建的索引及相关元数据的阶段，确保一旦构建完成，无需重复构建，节省调用API向量化的消耗、提升响应时间。
Qdrant(https://qdrant.tech/)是一个向量数据库，从官方发布的镜像拉到本地，用Docker快速部署起来。
docker pull qdrant/qdrant
docker run -p 6333:6333 -p 6334:6334 \
-v $(pwd)/qdrant_storage:/qdrant/storage:z \
qdrant/qdrant
启动之后，就可以通过客户端访问了，接下来构建索引并存储数据。
def index_node(folder_path: str):
''' 将指定目录下的所有文件，解析成数据块后构建索引
'''
collection_name = f'{folder_path}_index'
data_nodes = DataNode(folder_path, collection_name) #生成数据块
client = QdrantClient("localhost", port=6333) #连接向量数据库
try:
count = client.count(collection_name).count
except UnexpectedResponse:
count = 0
embed_model = OpenAIEmbedding(model='text-embedding-ada-002')
if count == len(data_nodes.nodes): #已经构建过索引，直接返回
logger.info(f"Found {count} existing nodes. Using the existing collection.")
vector_store = QdrantVectorStore(
collection_name=collection_name,
client=client,
)
return VectorStoreIndex.from_vector_store(
vector_store,
embed_model=embed_model,
)
logger.info(f"Found {count} existing nodes. Creating a new index with {len(data_nodes.nodes)} nodes. This may take a while.")
if count > 0: #删除已有的索引，准备重构索引
client.delete_collection(collection_name)
vector_store = QdrantVectorStore(
collection_name=collection_name,
client=client,
)
storage_context = StorageContext.from_defaults(vector_store=vector_store)
index = VectorStoreIndex(
data_nodes.nodes,
storage_context=storage_context,
embed_model=embed_model,
)
return index
-
首先，用DataNode把指定目录下的文件分块，并准备好QdrantClient本地客户端，同时指定OpenAIEmbedding使用的模型text-embedding-ada-002
-
然后根据名称collection_name检查本地向量数据库存储的节点个数，如果一致则返回已有的索引
vector_store = QdrantVectorStore(collection_name, client,)
return VectorStoreIndex.from_vector_store(
vector_store,
embed_model=embed_model,
)
否则创建一个新的索引
vector_store = QdrantVectorStore(collection_name, client)
storage_context = StorageContext.from_defaults(vector_store=vector_store)
index = VectorStoreIndex(
data_nodes.nodes,
storage_context=storage_context,
embed_model=embed_model,
)
查询（Querying）
对任意的查询请求，计算查询语句的向量和索引节点向量的相似度，找出相似度最高的前K个数据块，再把这些数据块和用户的问题合成一个上下文提交给LLM，生成答案。
下面是根据索引创建查询引擎的代码
def create_query_engine(
folder_path: str = "essay",
embedding_model: str = "text-embedding-ada-002",
similarity_top_k: int = 5,
) -> BaseQueryEngine:
index = index_nodes(folder_path=folder_path, embedding_model=embedding_model)
query_engine = index.as_query_engine(similarity_top_k=similarity_top_k)
query_engine = update_prompts_for_query_engine(query_engine)
return query_engine
其中，update_prompts_for_query_engine是把用户的查询语句和检索召回的数据块合成一个上下文的功能的函数**
def update_prompts_for_query_engine(query_engine: BaseQueryEngine) -> BaseQueryEngine:
new_tmpl_str = (
"Below is the specific context required to answer the upcoming query. You must base your response solely on this context, strictly avoiding the use of external knowledge or assumptions..\n"
"---------------------\n"
"{context_str}\n"
"---------------------\n"
"Given this context, please formulate your response to the following query. Ensure your response adheres to these instructions to maintain accuracy and relevance."
"Furthermore, it is crucial to respond in the same language in which the query is presented. This requirement is to ensure the response is directly applicable and understandable in the context of the query provided."
"Query: {query_str}\n"
"Answer: "
)
new_tmpl = PromptTemplate(new_tmpl_str)
query_engine.update_prompts({"response_synthesizer:text_qa_template": new_tmpl})
return query_engine
其中，new_tmpl_str把用户的问题和数据块合成一个上下文，该prompt提示LLM只使用提供的数据块、同时可以指定它按照用户提问的语言进行回复。
小结
以上的流程完成了数据处理、索引、存储、查询，在LlamaIndex的例子中有一个实现RAG系统的最简方案，只需要5行代码即可实现自己的RAG。https://docs.llamaindex.ai/en/stable/getting_started/starter_example/
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
documents = SimpleDirectoryReader("data").load_data()
index = VectorStoreIndex.from_documents(documents)
query_engine = index.as_query_engine()
response = query_engine.query("What did the author do growing up?")
print(response)
不过，如果要深入理解每个组件的原理、使用及优化技巧，还得一步一步动手实现才能有更深的理解。
下一步聊聊效果评估以及优化技巧：数据块窗口化、查询改写、混合检索等。这些实用技巧都非常值得研究。
欢迎关注&探讨
参考：
https://docs.llamaindex.ai/en/stable/getting_started/concepts/
更多AI工具，参考Github-AiBard123，国内AiBard123
