簡單使用LlamaIndex實現RAG

1 介紹

LlamaIndex是一個專門為大語言模型（LLM）設計的開源數據管理工具，旨在簡化和優化LLM在外部數據源中的查詢過程。適合在數據索引上構建RAG。

參考的地址

# 官網地址
https://docs.llamaindex.ai/en/stable/# 模塊介紹
https://docs.llamaindex.ai/en/stable/module_guides/# Github地址
https://github.com/run-llama/llama_index

使用的組件

# Openai like
https://docs.llamaindex.ai/en/stable/api_reference/llms/openai_like/
# OpenLLM，我沒測試組件，它繼承了OpenAILike
https://docs.llamaindex.ai/en/stable/api_reference/llms/openllm/# 自定義嵌入模型
https://docs.llamaindex.ai/en/stable/module_guides/models/embeddings/#custom-embedding-model# 自定義LLM模型
https://docs.llamaindex.ai/en/stable/module_guides/models/llms/usage_custom/# ChromaVectorStore向量存儲
https://docs.llamaindex.ai/en/stable/api_reference/storage/vector_store/chroma/#llama_index.vector_stores.chroma.ChromaVectorStore# Chroma數據庫文檔
https://docs.trychroma.com/docs/overview/introduction

需要安裝的包

?? 我使用的llma-index的版本：0.12.26

pip install llama-index
pip install llama-index-llms-openai-like
pip install llama-index-vector-stores-chroma
pip install chromadb

2 使用官網構建RAG

?? 國內基本無法直接使用，因為需要OpenAI的模型，所以無法直接使用。那就需要根據自己的需求定制。

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader# 加載數據
documents = SimpleDirectoryReader("E:/data").load_data()
# 轉化文檔
index = VectorStoreIndex.from_documents(documents)
# 構建查詢引擎
query_engine = index.as_query_engine()
# 使用
response = query_engine.query("Some question about the data should go here")
print(response)

3 自定義構建RAG

3.1 RAG構建思路

使用LlamaIndex構建RAG的思路如下圖，LlamaIndex需要自定義向量模型和類大模型組件。

graph TDA[（1）構建Documet對象列表，讀數據文檔] --> BB[（2）構建Node對象列表，使用分割器分割Document，其中分割器有SentenceSplitter、TextSplitter等] --> CC[（3）向量化和存儲，自定義嵌入模型和存儲到數據庫中，可以使用SimpleVectorStore、ChromaVectorStore等] --> DD[（4）構建向量索引庫，使用VectorStoreIndex構建向量索引] --> EE[（5）構建檢索器，用于用戶檢索輸入的prompt] --> FF[（6）構建響應生成器，自定義大模型生成用戶輸入的prompt] --> GG[（7）構建查詢引擎，組合檢索器和響應生成器構建查詢引擎] --> HH[（8）使用prompt查詢和生成數據]

3.2 自定義RAG

（1） my_document_custorm_engine.py

import chromadb
from llama_index.core import Document, VectorStoreIndex, get_response_synthesizer
from llama_index.core.node_parser import SentenceSplitter
from llama_index.core.query_engine import RetrieverQueryEngine
from llama_index.core.response_synthesizers import ResponseMode
from llama_index.vector_stores.chroma import ChromaVectorStorefrom my_custom_rag.custom_like_openai import CustomLikeOpenAI# 需要安裝的組件
"""
pip install llama-index
pip install llama-index-llms-openai-like
pip install llama-index-vector-stores-chroma
"""from my_custom_rag.custom_embedding import CustomEmbeddings# 自定嵌入模型
my_embedding = CustomEmbeddings()# 構建文檔列表，可根據自己的需求設置文本文檔列表
document_text = ["河南大學（Henan University），簡稱“河大”（HENU），位于中國河南省鄭州市、開封市，是河南省人民政府與中華人民共和國教育部共建公辦高校 [392]、國家“雙一流”建設高校 [69]，入選國家“111計劃” [2]、中西部高校基礎能力建設工程 [313]、中國政府獎學金來華留學生接收院校 [388]。","河南大學創立于1912年，始名河南留學歐美預備學校。后歷經中州大學、國立開封中山大學、省立河南大學等階段，1942年升格為國立河南大學 [153]。1952年院系調整 ，校本部更名為河南師范學院。后經開封師范學院、河南師范大學等階段，1984年恢復河南大學校名 [153]。2000年6月，原河南大學、開封醫學高等專科學校、開封師范高等專科學校合并組建新的河南大學 [154]。2012年，河南大學入選第一批卓越醫生教育培養計劃項目試點高校 [130]；入選國家級卓越法律人才教育培養基地 [390]；入選第一批國家卓越醫生教育培養計劃項目試點高校 [391]。","截至2024年6月，學校設有40個學院、93個本科招生專業 、47個碩士學位授權一級學科 、39種碩士專業學位授權類別 、2種博士專業學位授權類別、24個博士學位授權一級學科 、20個博士后科研流動站、13個學科進入ESI世界排名前1% ；有全日制在校生5萬余人、教職工4700余人，教師中有院士、學部委員6人，長江學者、國家杰青、“萬人計劃”領軍人才等國家級領軍人才26人，國家級青年人才15人；擁有3個國家重點實驗室 ，1個國家野外科學觀測研究站 ，3個國家地方聯合工程研究中心 ，4個河南省實驗室 ， 5個教育部和農業部重點實驗室 [448]。","河南大學軟件學院（Henan University Software College）是全國較早、河南省最早成立的軟件學院之一，位于中國河南省開封市。學院是國家示范性軟件學院聯盟成員單位，信息技術新工科產學研聯盟首批會員單位，2020 年度獲批河南省特色化示范性軟件學院，河南省鯤鵬產業學院建設高校。學院設有軟件工程系、網絡工程系和公共計算機教學中心，擁有“河南省智能數據處理工程研究中心”、“河南省現代網絡技術實驗教學示范中心”、“河南省智能網絡理論與關鍵技術國際聯合實驗室”等省級科研教學平臺、“河南省高等學校學科（軟件工程）引智基地”、“河南省本科高校大學生校外實踐教育基地”和“河南省高校優秀基層教學組織”稱號。學院獨立承建軟件工程和網絡工程兩個本科專業，軟件工程專業為國家一流本科專業， 網絡工程專業為河南省一流本科專業。同時，擁有“軟件工程技術”二級博士學位授權點，電子信息專業具有碩士學位授予權，建有軟件工程博士后科研流動站。"
]# 構建LlamaIndex的對象列表
documents = list()
for text in document_text:documents.append(Document(text=text))
# print(documents)# 將LlamaIndex的Document拆分Nodes
parser = SentenceSplitter()
nodes = parser.get_nodes_from_documents(documents)# 使用嵌入模型對node進行嵌入
for node in nodes:node.embedding = my_embedding.get_text_embedding(node.get_content())# 創建Chroma客戶端
client = chromadb.Client()# 創建集合
collection = client.create_collection("my-documents")
chroma_vector_store = ChromaVectorStore(chroma_collection=collection)# 存入向量庫中
chroma_vector_store.add(nodes)# 使用索引對象
# 注意：VectorStoreIndex必須有向量庫支持，否則會報下面的錯誤
# Cannot initialize from a vector store that does not store text.
vector_index = VectorStoreIndex.from_vector_store(chroma_vector_store, embed_model=my_embedding)# Llama index中的OpenAILike調用千問和Kimi不能使用，一直報參數錯誤，所以只能自定義了
llm = CustomLikeOpenAI(model="qwen2.5-14b-instruct",api_base="https://dashscope.aliyuncs.com/compatible-mode/v1",api_key = "sk-XXX"
)# 下面的是LlamaIndex中的OpenAILike
# llm = OpenAILike(
#     model="qwen2.5-14b-instruct",
#     base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
#     api_key = "sk-XXX"
# )"""
# 顯示構建查詢引擎=檢索器+響應生成器# 創建檢索器，similarity_top_k設置返回值的數量
query_retriever = vector_index.as_retriever(similarity_top_k=1)# 響應生成器有compact（默認模式）、refine、simple_summarize等
response_synthesizer = get_response_synthesizer(llm=llm,response_mode=ResponseMode.COMPACT,# streaming=True
)query_engine = RetrieverQueryEngine(retriever=query_retriever,response_synthesizer=response_synthesizer
)"""
# 隱式構建查詢引擎，上面兩步可以使用1行構建
query_engine = vector_index.as_query_engine(llm=llm,# streaming=True
)# 如果不使用流式輸出，直接打印即可
query_data = query_engine.query("用30字介紹一下河南大學")
print(query_data)# # 使用流式輸出
# for text in query_data.response_gen:
#     print(text)
#     pass

（2）custom_embedding.py

自定義嵌入模型

from llama_index.core.base.embeddings.base import Embedding
from llama_index.core.embeddings import BaseEmbedding
from sentence_transformers import SentenceTransformer# 構建向量模型，可根據自己的需求，自定義調用互聯網和本地模型等
embedder = SentenceTransformer("E:/model/sentencetransformers/distiluse-base-multilingual-cased-v1")class CustomEmbeddings(BaseEmbedding):def _get_query_embedding(self, query: str) -> Embedding:# 生成嵌入列表return embedder.encode(query).tolist()async def _aget_query_embedding(self, query: str) -> Embedding:# 生成嵌入列表return embedder.encode(query).tolist()def _get_text_embedding(self, text: str) -> Embedding:# 生成嵌入列表return embedder.encode(text).tolist()

（3）custom_like_openai.py

自定義大模型

from typing import Anyfrom llama_index.core.base.llms.types import CompletionResponseGen, LLMMetadata, CompletionResponse
from llama_index.core.llms import CustomLLM
from openai import OpenAI
from pydantic import Fieldclass CustomLikeOpenAI(CustomLLM):model: str = Field(description="自定義模型名稱")api_key: str = Field(description="自定義API Key")api_base: str = Field(description="自定義API地址")context_window: int = Field(default=32768, description="上下文窗口大小")temperature: float = Field(ge=0, le=1, default=0.3, description="設置溫度，值域須為 [0, 1]")num_output: int = Field(default=8192, description="設置max_tokens")def __init__(self, **data):# 必須調用父類初始化super().__init__(**data)# 創建對象self._client = OpenAI(api_key=self.api_key,base_url=self.api_base)@propertydef metadata(self) -> LLMMetadata:"""Get LLM metadata."""return LLMMetadata(context_window=self.context_window,num_output=self.num_output,model_name=self.model)def complete(self, prompt: str, **kwargs: Any) -> CompletionResponse:"""生成文本:param prompt: 添加提示詞:param kwargs: 其他相關參數:return: CompletionResponse"""# 構建生成completion = self._client.chat.completions.create(model=self.model,messages=[{"role": "user", "content": prompt}],temperature=self.temperature,max_tokens=self.num_output)# 返回值return CompletionResponse(text=completion.choices[0].message.content)def stream_complete(self, prompt: str, **kwargs: Any) -> CompletionResponseGen:"""生成流式文本:param prompt: 提示詞:param kwargs: 其他參數:return: CompletionResponseGen迭代器"""# 根據需要可以不實現，如果不想實現使用下面代碼即可# raise NotImplementedError("Streaming not supported")# 構建數據流stream = self._client.chat.completions.create(model=self.model,messages=[{"role": "user", "content": prompt}],temperature=self.temperature,max_tokens=self.num_output,stream=True)# 遍歷數據流for chunk in stream:# 獲取新文本delta = chunk.choices[0].delta# 判斷數據是否存在if delta.content:yield CompletionResponse(text=delta.content, delta=delta.content)

3.3 其他學習

（1）使用SimpleVectorStore存儲

from llama_index.core import Document
from llama_index.core.node_parser import SentenceSplitter
from llama_index.core.vector_stores import SimpleVectorStore, VectorStoreQuery
from sentence_transformers import SentenceTransformer# 構建向量模型
embedder = SentenceTransformer("E:/model/sentencetransformers/distiluse-base-multilingual-cased-v1")# 構建文檔列表，可根據自己的需求設置文本文檔列表
document_text = ["A man is eating food.","A man is eating a piece of bread.","The girl is carrying a baby.","A man is riding a horse.","A woman is playing violin.","Two men pushed carts through the woods.","A man is riding a white horse on an enclosed ground.","A monkey is playing drums.","A cheetah is running behind its prey.",
]# 構建LlamaIndex的對象列表
documents = list()
for text in document_text:documents.append(Document(text=text))
# print(documents)# 將LlamaIndex的Document拆分Nodes
parser = SentenceSplitter()
nodes = parser.get_nodes_from_documents(documents)# 使用嵌入模型對node進行嵌入
for node in nodes:node.embedding = embedder.encode(node.get_content())
# print(nodes)# 下面一般不會用于生產環境，生產環境一般用向量庫
# 基于內存的方式存儲向量
simple_vector_store = SimpleVectorStore()
simple_vector_store.add(nodes)# 持久化nodes，默認存儲在”./storage“
# simple_vector_store.persist()
# 獲取持久化數據
# simple_vector_store = SimpleVectorStore.from_persist_path("./storage/vector_store.json")# # 對查詢的文本進行嵌入
query_embed = embedder.encode("A man is eating pasta").tolist()
# 查詢到的目標數據
target_data_embed = simple_vector_store.query(VectorStoreQuery(query_embedding=query_embed, similarity_top_k=2))print(target_data_embed)

（2）使用索引VectorStoreIndex

import chromadb
from llama_index.core import Document, VectorStoreIndex
from llama_index.core.node_parser import SentenceSplitter
from llama_index.vector_stores.chroma import ChromaVectorStorefrom my_custom_rag.custom_embedding import CustomEmbeddings# 自定嵌入模型
my_embedding = CustomEmbeddings()# 構建文檔列表，可根據自己的需求設置文本文檔列表
document_text = ["A man is eating food.","A man is eating a piece of bread.","The girl is carrying a baby.","A man is riding a horse.","A woman is playing violin.","Two men pushed carts through the woods.","A man is riding a white horse on an enclosed ground.","A monkey is playing drums.","A cheetah is running behind its prey.",
]# 構建LlamaIndex的對象列表
documents = list()
for text in document_text:documents.append(Document(text=text))
# print(documents)# 將LlamaIndex的Document拆分Nodes
parser = SentenceSplitter()
nodes = parser.get_nodes_from_documents(documents)# 使用嵌入模型對node進行嵌入
for node in nodes:node.embedding = my_embedding.get_text_embedding(node.get_content())# 創建Chroma客戶端
client = chromadb.Client()# 創建集合
collection = client.create_collection("my-documents")
chroma_vector_store = ChromaVectorStore(chroma_collection=collection)# 存入向量庫中
chroma_vector_store.add(nodes)# 使用索引對象
# 注意：VectorStoreIndex必須有向量庫支持，否則會報下面的錯誤
# Cannot initialize from a vector store that does not store text.
vector_index = VectorStoreIndex.from_vector_store(chroma_vector_store, embed_model=my_embedding)# 創建檢索器，similarity_top_k設置返回值的數量
query_retriever = vector_index.as_retriever(similarity_top_k=1)# 查詢數據
query_text = "A man is eating pasta"
retrieve_data = query_retriever.retrieve(query_text)print(retrieve_data)