目錄

RAG的工作流程

python實現(xiàn)RAG

1.引入相關(guān)庫及相關(guān)準(zhǔn)備工作

函數(shù)

1.?加載并讀取文檔

2. 文檔分割

3. embedding

4. 向集合中添加文檔

5. 用戶輸入內(nèi)容

6. 查詢集合中的文檔

7. 構(gòu)建Prompt并生成答案

主流程

附錄

函數(shù)解釋

1. open() 函數(shù)語法

2.client.embeddings.create()

3.collection.query()

完整代碼

---

RAG的工作流程

流程描述

加載，讀取文檔
文檔分割
文檔向量化
用戶輸入內(nèi)容
內(nèi)容向量化
文本向量中匹配出與問句向量相似的 top_k 個
匹配出的文本作為上下文和問題一起添加到 prompt 中
提交給 LLM 生成答案

Indexing過程

Retrieval過程

python實現(xiàn)RAG

1.引入相關(guān)庫及相關(guān)準(zhǔn)備工作

import chromadb  # 導(dǎo)入 chromadb 庫
from openai import OpenAI  # 導(dǎo)入 OpenAI 庫client = OpenAI()  # 創(chuàng)建一個 OpenAI 客戶端實例file_path = "./巴黎奧運會金牌信息.txt"# 創(chuàng)建一個 chroma 客戶端實例
chroma_client = chromadb.Client()# 創(chuàng)建一個名為 "my_collection" 的集合
collection = chroma_client.create_collection(name="my_collection")

函數(shù)

1.?加載并讀取文檔

# 1. 加載并讀取文檔
def load_document(filepath):with open(filepath, 'r', encoding='utf-8') as file:document = file.read()return document

def load_document(filepath):

定義一個名為 load_document 的函數(shù)，并接收一個參數(shù) filepath，表示要讀取的文件路徑。

    with open(filepath, 'r', encoding='utf-8') as file:

使用 open() 函數(shù)打開 filepath 指定的文件。

'r' 表示以只讀模式打開文件。

encoding='utf-8' 指定UTF-8 編碼，以確保能夠正確處理包含中文或其他特殊字符的文件。

with 語句用于上下文管理，可以在讀取文件后自動關(guān)閉文件，防止資源泄露。

        document = file.read()

2. 文檔分割

# 2. 文檔分割
def split_document(document):# 使用兩個換行符來分割段落chunks = document.strip().split('\n\n')return chunks  # 返回包含所有文本塊的列表

1. def split_document(document):

• 這是一個函數(shù)定義，document 是輸入的文本（字符串類型）。
• 該函數(shù)的作用是按照段落分割文本。

2. document.strip()

• strip() 方法用于去掉字符串開頭和結(jié)尾的空白字符（包括空格、換行符 \n、制表符 \t 等）。
• 這樣可以避免因為文本頭尾的換行符導(dǎo)致分割時出現(xiàn)空字符串。

3. .split('\n\n')

• split('\n\n') 按照兩個連續(xù)的換行符分割文本。
• \n 代表換行，而 \n\n 代表兩個換行符，通常用于分隔不同的段落。
• 這個方法會返回一個列表，其中每個元素是一個段落。

4. return chunks

• chunks 是分割后的文本塊列表，返回它供后續(xù)使用。

3. embedding

# 3. embedding
def get_embedding(texts, model="text-embedding-3-large"):result = client.embeddings.create(input=texts,model=model)return [x.embedding for x in result.data]

封裝embedding模型。

4. 向集合中添加文檔

# 4. 向集合中添加文檔
def add_documents_to_collection(chunks):embeddings = get_embedding(chunks)  # 獲取文檔塊的嵌入collection.add(documents=chunks,  # 文檔內(nèi)容embeddings=embeddings,  # 文檔對應(yīng)的嵌入向量ids=[f"id{i+1}" for i in range(len(chunks))]  # 生成文檔 ID)

這段代碼定義了 add_documents_to_collection 函數(shù)，用于將文檔（chunks）添加到一個集合（collection）中，并為每個文檔計算嵌入（embedding）。這個過程通常用于向量數(shù)據(jù)庫（如 FAISS、ChromaDB 或 Pinecone），以支持向量搜索、相似性檢索和信息檢索。

    embeddings = get_embedding(chunks)  # 獲取文檔塊的嵌入

• 調(diào)用 get_embedding(chunks)，為 chunks 中的每個文本計算嵌入（embedding）。
• embeddings 是一個嵌入向量列表，每個向量對應(yīng) chunks 里的一個文本片段。

    collection.add(

• collection 是一個數(shù)據(jù)庫或向量存儲集合，可以是 ChromaDB、FAISS、Pinecone 等向量數(shù)據(jù)庫對象。
• .add() 方法用于向集合中添加數(shù)據(jù)，包括原始文檔、嵌入向量和唯一 ID。

        ids=[f"id{i+1}" for i in range(len(chunks))]  # 生成文檔 ID

• ids 是文檔唯一標(biāo)識符，用于在數(shù)據(jù)庫中區(qū)分不同文檔。
• f"id{i+1}" 生成 "id1", "id2", "id3" 這樣的字符串 ID。
• for i in range(len(chunks)) 依次編號，確保每個文檔有唯一 ID。

示例

輸入

假設(shè) chunks 是：

chunks = ["文本片段1", "文本片段2", "文本片段3"]

執(zhí)行：

add_documents_to_collection(chunks)

執(zhí)行過程

1.計算 chunks 的嵌入：

embeddings = get_embedding(chunks)  

假設(shè)返回：

[[0.1, 0.2, 0.3],  # 文本片段1的嵌入[0.4, 0.5, 0.6],  # 文本片段2的嵌入[0.7, 0.8, 0.9]   # 文本片段3的嵌入
]

2.添加到 collection：

collection.add(documents=["文本片段1", "文本片段2", "文本片段3"],embeddings=[[0.1, 0.2, 0.3],[0.4, 0.5, 0.6],[0.7, 0.8, 0.9]],ids=["id1", "id2", "id3"]
)

5. 用戶輸入內(nèi)容

def get_user_input():return input("請輸入您的問題: ")

6. 查詢集合中的文檔

# 6. 查詢集合中的文檔
def query_collection(query_embeddings, n_results=1):results = collection.query(query_embeddings=[query_embeddings],  # 查詢文本的嵌入n_results=n_results  # 返回的結(jié)果數(shù)量)return results['documents']

• collection.query(...)：對 collection 進行查詢，基于嵌入向量 執(zhí)行相似度檢索。
• query_embeddings=[query_embeddings]：將單個查詢嵌入封裝在列表中，確保符合 API 要求。
• n_results=n_results：指定要返回的最相似的 n_results 個文檔。

return results['documents']

• results 是 collection.query(...) 的返回結(jié)果，它應(yīng)該是一個 字典，其中包含多個字段（如 documents,embeddings , ids?等）。
• results['documents'] 獲取查詢返回的文檔列表并返回。

7. 構(gòu)建Prompt并生成答案

# 7. 構(gòu)建Prompt并生成答案
def get_completion(prompt, model='gpt-3.5-turbo'):message = [{"role": "user", "content": prompt}]result = client.chat.completions.create(model=model,messages=message)return result.choices[0].message.content

主流程

if __name__ == "__main__":# 步驟1 => 加載文檔document = load_document(file_path)# 步驟2 => 文檔分割chunks = split_document(document)# 步驟3 => embeddingadd_documents_to_collection(chunks)  # 在分割后立即添加文檔# 步驟4 => 用戶輸入內(nèi)容user_input = get_user_input()# 步驟5 => 將用戶輸入的問題進行embeddinginput_embedding = get_embedding(user_input)[0]  # 獲取用戶問題的嵌入# 步驟6 => 查詢集合中的文檔context_texts = query_collection(input_embedding, n_results=1)  # 查詢相關(guān)文檔print(context_texts)# 步驟7 => 構(gòu)建Prompt并生成答案prompt = f"上下文: {context_texts}\n\n問題: {user_input}\n\n請?zhí)峁┐鸢?"answer = get_completion(prompt)print(answer)

附錄

函數(shù)解釋

1. open() 函數(shù)語法

open(file, mode='r', buffering=-1, encoding=None, errors=None, newline=None, closefd=True, opener=None)

參數(shù)說明

常見 mode（模式）

組合模式示例：

• 'rb' ：以 二進制模式 讀取文件。
• 'wt' ：以 文本模式 寫入文件。
• 'a+' ：以 讀寫模式 打開文件，并在文件末尾追加內(nèi)容。

2.client.embeddings.create()

client.embeddings.create() 是 OpenAI API 提供的一個函數(shù)，用于生成文本的嵌入（embedding）向量。它將文本轉(zhuǎn)換為高維數(shù)值向量，通常用于相似性計算、文本分類、搜索、推薦系統(tǒng)等 NLP 任務(wù)。

基本使用

import openai# 創(chuàng)建 OpenAI 客戶端（需要提供 API Key）
client = openai.OpenAI(api_key="your-api-key")# 生成文本嵌入
response = client.embeddings.create(input=["Hello world", "How are you?"],  # 輸入文本，可以是單個字符串或字符串列表model="text-embedding-3-large"  # 選擇的嵌入模型
)

參數(shù)說明

3.collection.query()

collection.query() 是一個用于查詢向量數(shù)據(jù)庫的函數(shù)，主要用于 基于向量嵌入（embeddings）的相似性搜索。它通常用于檢索與查詢向量最接近的文檔或數(shù)據(jù)點。

results = collection.query(query_embeddings=[query_vector],  # 查詢向量（必須是列表）n_results=3  # 需要返回的最相似的結(jié)果數(shù)量
)

collection.query() 的返回結(jié)果

查詢后的 results 變量通常是一個 字典，常見的字段包括：

{"documents": [["文檔1內(nèi)容"], ["文檔2內(nèi)容"], ["文檔3內(nèi)容"]],"distances": [[0.12], [0.15], [0.18]],"metadatas": [{"id": "doc1"}, {"id": "doc2"}, {"id": "doc3"}]
}

完整代碼

import chromadb  # 導(dǎo)入 chromadb 庫
from openai import OpenAI  # 導(dǎo)入 OpenAI 庫
client = OpenAI()  # 創(chuàng)建一個 OpenAI 客戶端實例file_path = "./巴黎奧運會金牌信息.txt"# 創(chuàng)建一個 chroma 客戶端實例
chroma_client = chromadb.Client()# 創(chuàng)建一個名為 "my_collection" 的集合
collection = chroma_client.create_collection(name="my_collection")# 1. 加載并讀取文檔
def load_document(filepath):with open(filepath, 'r', encoding='utf-8') as file:document = file.read()return document# 2. 文檔分割
def split_document(document):# 使用兩個換行符來分割段落chunks = document.strip().split('\n\n')return chunks  # 返回包含所有文本塊的列表# 3. embedding
def get_embedding(texts, model="text-embedding-3-large"):result = client.embeddings.create(input=texts,model=model)return [x.embedding for x in result.data]# 4. 向集合中添加文檔
def add_documents_to_collection(chunks):embeddings = get_embedding(chunks)  # 獲取文檔塊的嵌入collection.add(documents=chunks,  # 文檔內(nèi)容embeddings=embeddings,  # 文檔對應(yīng)的嵌入向量ids=[f"id{i+1}" for i in range(len(chunks))]  # 生成文檔 ID)# 5. 用戶輸入內(nèi)容
def get_user_input():return input("請輸入您的問題: ")# 6. 查詢集合中的文檔
def query_collection(query_embeddings, n_results=1):results = collection.query(query_embeddings=[query_embeddings],  # 查詢文本的嵌入n_results=n_results  # 返回的結(jié)果數(shù)量)return results['documents']# 7. 構(gòu)建Prompt并生成答案
def get_completion(prompt, model='gpt-3.5-turbo'):message = [{"role": "user", "content": prompt}]result = client.chat.completions.create(model=model,messages=message)return result.choices[0].message.content# 主流程
if __name__ == "__main__":# 步驟1 => 加載文檔document = load_document(file_path)# 步驟2 => 文檔分割chunks = split_document(document)# 步驟3 => embeddingadd_documents_to_collection(chunks)  # 在分割后立即添加文檔# 步驟4 => 用戶輸入內(nèi)容user_input = get_user_input()# 步驟5 => 將用戶輸入的問題進行embeddinginput_embedding = get_embedding(user_input)[0]  # 獲取用戶問題的嵌入# 步驟6 => 查詢集合中的文檔context_texts = query_collection(input_embedding, n_results=1)  # 查詢相關(guān)文檔print(context_texts)# 步驟7 => 構(gòu)建Prompt并生成答案prompt = f"上下文: {context_texts}\n\n問題: {user_input}\n\n請?zhí)峁┐鸢?"answer = get_completion(prompt)print(answer)