前言

Hello，大家好，我是GISer Liu😁，一名熱愛AI技術的GIS開發(fā)者，這個LLM開發(fā)基礎階段已經進入尾聲了，本文中我們不介紹更多的理論與知識點，而是通過的分析開源項目的解決方案來幫助各位開發(fā)者理清自己開發(fā)的思路；

在本文中作者將通過分析開源項目 個人知識庫助手：學習這個RAG應用的開發(fā)流程，思路以及業(yè)務代碼；幫助讀者能學會如何規(guī)劃自己的LLM的應用開發(fā)思路；

---

一、個人知識庫助手

1.項目介紹

①背景

在當今數(shù)據(jù)量迅速增長的時代，高效管理和檢索信息已成為關鍵技能。為了應對這一挑戰(zhàn)以及LLM技術的發(fā)展，該項目應運而生，旨在構建一個基于 Langchain 的個人知識庫助手。該助手可以通過高效的信息管理系統(tǒng)和強大的檢索功能，為用戶提供了一個可靠的信息獲取平臺。

②目標意義

1. 核心目標：充分利用大型語言模型在處理自然語言查詢方面的優(yōu)勢，并進行定制化開發(fā)以滿足用戶需求，從而實現(xiàn)對復雜信息的智能理解和精確回應。
2. 在項目開發(fā)過程中，團隊深入分析了大型語言模型的潛力和局限，特別是其生成幻覺信息的傾向。
3. 為了解決幻覺信息的問題，項目集成了 RAG 技術，這是一種結合檢索和生成的方法。

③主要功能

• 信息檢索：從大規(guī)模數(shù)據(jù)庫或知識庫中提取相關信息?？焖俣ㄎ缓瞳@取精確的內容。
• 生成式問答：利用檢索到的信息生成自然語言回答。提供詳細和上下文相關的回答，提高用戶體驗。
• 知識更新：不斷更新和擴充知識庫，保持信息的時效性和準確性
• 。支持多種數(shù)據(jù)源的集成和管理。
• 多領域適用：支持在不同領域和主題下的應用，如技術文檔、醫(yī)學知識、法律咨詢等。靈活定制以適應特定行業(yè)需求。
• 用戶交互：提供自然語言界面，便于用戶查詢和獲取信息。提高用戶的搜索效率和滿意度。

2.項目部署

①環(huán)境要求

• CPU: Intel 5代處理器（云CPU方面，建議選擇 2 核以上的云CPU服務）；

阿里云199一年的都可以，也可以自己本機部署；因為是API調用，對電腦性能要求不高；

• 內存（RAM）: 至少 4 GB

• 操作系統(tǒng)：Windows、macOS、Linux均可

②部署流程

這里作者將整個流程打包:

git clone https://github.com/logan-zou/Chat_with_Datawhale_langchain.git
cd Chat_with_Datawhale_langchain
# 創(chuàng)建 Conda 環(huán)境
conda create -n llm-universe python==3.9.0
# 激活 Conda 環(huán)境
conda activate llm-universe
# 安裝依賴項
pip install -r requirements.txt

運行項目:

# Linux 系統(tǒng)
cd serve
uvicorn api:app --reload # Windows 系統(tǒng)
cd serve
python api.py

或者：

python run_gradio.py -model_name='模型名稱' -embedding_model='嵌入模型編號' -db_path='知識庫文件路徑' -persist_path='持久化目錄文件路徑'

這里記得配置自己的API Key：

③核心思想

本項目其實是針對四種大模型 API 實現(xiàn)了底層封裝，基于 Langchain 搭建了可切換模型的檢索問答鏈，并實現(xiàn) API 以及 Gradio 部署的個人輕量大模型應用；

④技術棧

• LLM層：統(tǒng)一封裝了四個大模型，用作底層模型進行調用。
• 數(shù)據(jù)層：通過選擇的Embedding模型API進行向量數(shù)據(jù)庫的創(chuàng)建和向量檢索。源數(shù)據(jù)經過Embedding處理后可以被向量數(shù)據(jù)庫使用。
• 數(shù)據(jù)庫層：基于個人知識庫源數(shù)據(jù)搭建的向量數(shù)據(jù)庫。在本項目中，我們選擇了Chroma，當然Faiss也不錯。
• 應用層：確定我們的應用有哪些？RAG、工具使用、RPA等 。將LLM和具體業(yè)務結合使用更佳。在本項目中，我們僅使用RAG，基于LangChain提供的檢索問答鏈基類進行了進一步封裝，從而支持不同模型切換以及便捷實現(xiàn)基于數(shù)據(jù)庫的檢索問答。
• 服務層：本項目基于FastAPI和Gradio。對于后端，我們使用FastAPI即可，無需改變。前端如果只能使用Python開發(fā)，使用Gradio和Streamlit都是快速不錯的選擇。如果是全棧開發(fā)者或企業(yè)開發(fā)，使用Vue或React可以開發(fā)出更專業(yè)且美觀的應用。

當然，這里是因為我們的應用單一，因此只選擇了向量數(shù)據(jù)庫。如果還具有其他業(yè)務，如列表信息、歷史記錄、檢索信息等，使用結構化數(shù)據(jù)庫也是必要的。向量數(shù)據(jù)庫和結構化數(shù)據(jù)混合使用更合適。

• 本項目支持本地M3E Embedding模型和 API Embedding結合的方式進行向量化；

3.應用詳解

①業(yè)務流程

1、核心架構

llm-universe 個人知識庫助手是一個典型的 RAG 項目，通過 langchain+LLM 實現(xiàn)本地知識庫問答，建立了全流程可使用開源模型實現(xiàn)的本地知識庫對話應用。該項目當前已經支持使用 ChatGPT，星火 Spark 模型，文心大模型，智譜 GLM 等大語言模型的接入。

作者這里繪制了一個流程圖可以看看：😺😺

整個 RAG 過程包括以下操作：

1. 用戶提出問題 (Query)
2. 加載和讀取知識庫文檔
3. 對知識庫文檔進行分割
4. 對分割后的知識庫文本向量化并存入向量庫建立索引
5. 對用戶提問(Query) 向量化
6. 在知識庫文檔向量中匹配出與問句 (Query) 向量最相似的 top k 個
7. 匹配出的知識庫文本作為上下文 (Context) 和問題一起添加到 prompt 中
8. 提交給 LLM 生成回答 (response)；

這里很好理解，之前文章中，LLM基礎學習的第三篇詳細講解了其中的每個步驟；

②索引 Index

三步走戰(zhàn)略：

• 文本數(shù)據(jù)加載和讀取
• 文本數(shù)據(jù)分割
• 文本數(shù)據(jù)向量化

這和我們之前向量數(shù)據(jù)庫搭建的過程差不多；詳細內容如下：

(1)知識庫數(shù)據(jù)加載和讀取

該項目使用的是：
《機器學習公式詳解》PDF版本
《面向開發(fā)者的 LLM 入門教程 第一部分 Prompt Engineering》md版本
《強化學習入門指南》MP4版本
以及datawhale總倉庫所有開源項目的readme：https://github.com/datawhalechina

大家可以根據(jù)自己的實際業(yè)務選擇自己的需要的數(shù)據(jù)，數(shù)據(jù)存放在 ../../data_base/knowledge_db 目錄下，用戶可以將自己的文件存放到這里；

項目官方提供了拉去官網readme的爬蟲：

import json
import requests
import os
import base64
import loguru
from dotenv import load_dotenv
# 加載環(huán)境變量
load_dotenv()
# 從環(huán)境變量中獲取TOKEN
TOKEN = os.getenv('TOKEN')
# 定義獲取組織倉庫的函數(shù)
def get_repos(org_name, token, export_dir):headers = {'Authorization': f'token {token}',}url = f'https://api.github.com/orgs/{org_name}/repos'response = requests.get(url, headers=headers, params={'per_page': 200, 'page': 0})if response.status_code == 200:repos = response.json()loguru.logger.info(f'Fetched {len(repos)} repositories for {org_name}.')# 使用 export_dir 確定保存?zhèn)}庫名的文件路徑repositories_path = os.path.join(export_dir, 'repositories.txt')with open(repositories_path, 'w', encoding='utf-8') as file:for repo in repos:file.write(repo['name'] + '\n')return reposelse:loguru.logger.error(f"Error fetching repositories: {response.status_code}")loguru.logger.error(response.text)return []
# 定義拉取倉庫README文件的函數(shù)
def fetch_repo_readme(org_name, repo_name, token, export_dir):headers = {'Authorization': f'token {token}',}url = f'https://api.github.com/repos/{org_name}/{repo_name}/readme'response = requests.get(url, headers=headers)if response.status_code == 200:readme_content = response.json()['content']# 解碼base64內容readme_content = base64.b64decode(readme_content).decode('utf-8')# 使用 export_dir 確定保存 README 的文件路徑repo_dir = os.path.join(export_dir, repo_name)if not os.path.exists(repo_dir):os.makedirs(repo_dir)readme_path = os.path.join(repo_dir, 'README.md')with open(readme_path, 'w', encoding='utf-8') as file:file.write(readme_content)else:loguru.logger.error(f"Error fetching README for {repo_name}: {response.status_code}")loguru.logger.error(response.text)
# 主函數(shù)
if __name__ == '__main__':# 配置組織名稱org_name = 'datawhalechina'# 配置 export_direxport_dir = "../../database/readme_db"  # 請?zhí)鎿Q為實際的目錄路徑# 獲取倉庫列表repos = get_repos(org_name, TOKEN, export_dir)# 打印倉庫名稱if repos:for repo in repos:repo_name = repo['name']# 拉取每個倉庫的READMEfetch_repo_readme(org_name, repo_name, TOKEN, export_dir)# 清理臨時文件夾# if os.path.exists('temp'):#     shutil.rmtree('temp')

以上默認會把這些readme文件放在同目錄database下的readme_db文件。其中這些readme文件含有不少無關信息;

😏再運行database/text_summary_readme.py文件可以調用大模型生成每個readme文件的摘要并保存到上述知識庫目錄/data_base/knowledge_db /readme_summary文件夾中，。代碼如下：

import os
from dotenv import load_dotenv
import openai
from test_get_all_repo import get_repos
from bs4 import BeautifulSoup
import markdown
import re
import time
# Load environment variables
load_dotenv()
TOKEN = os.getenv('TOKEN')
# Set up the OpenAI API client
openai_api_key = os.environ["OPENAI_API_KEY"]# 過濾文本中鏈接防止大語言模型風控
def remove_urls(text):# 正則表達式模式，用于匹配URLurl_pattern = re.compile(r'https?://[^\s]*')# 替換所有匹配的URL為空字符串text = re.sub(url_pattern, '', text)# 正則表達式模式，用于匹配特定的文本specific_text_pattern = re.compile(r'掃描下方二維碼關注公眾號|提取碼|關注|科學上網|回復關鍵詞|侵權|版權|致謝|引用|LICENSE'r'|組隊打卡|任務打卡|組隊學習的那些事|學習周期|開源內容|打卡|組隊學習|鏈接')# 替換所有匹配的特定文本為空字符串text = re.sub(specific_text_pattern, '', text)return text# 抽取md中的文本
def extract_text_from_md(md_content):# Convert Markdown to HTMLhtml = markdown.markdown(md_content)# Use BeautifulSoup to extract textsoup = BeautifulSoup(html, 'html.parser')return remove_urls(soup.get_text())def generate_llm_summary(repo_name, readme_content,model):prompt = f"1：這個倉庫名是 {repo_name}. 此倉庫的readme全部內容是: {readme_content}\2:請用約200以內的中文概括這個倉庫readme的內容,返回的概括格式要求：這個倉庫名是...,這倉庫內容主要是..."openai.api_key = openai_api_key# 具體調用messages = [{"role": "system", "content": "你是一個人工智能助手"},{"role": "user", "content": prompt}]response = openai.ChatCompletion.create(model=model,messages=messages,)return response.choices[0].message["content"]def main(org_name,export_dir,summary_dir,model):repos = get_repos(org_name, TOKEN, export_dir)# Create a directory to save summariesos.makedirs(summary_dir, exist_ok=True)for id, repo in enumerate(repos):repo_name = repo['name']readme_path = os.path.join(export_dir, repo_name, 'README.md')print(repo_name)if os.path.exists(readme_path):with open(readme_path, 'r', encoding='utf-8') as file:readme_content = file.read()# Extract text from the READMEreadme_text = extract_text_from_md(readme_content)# Generate a summary for the README# 訪問受限，每min一次time.sleep(60)print('第' + str(id) + '條' + 'summary開始')try:summary = generate_llm_summary(repo_name, readme_text,model)print(summary)# Write summary to a Markdown file in the summary directorysummary_file_path = os.path.join(summary_dir, f"{repo_name}_summary.md")with open(summary_file_path, 'w', encoding='utf-8') as summary_file:summary_file.write(f"# {repo_name} Summary\n\n")summary_file.write(summary)except openai.OpenAIError as e:summary_file_path = os.path.join(summary_dir, f"{repo_name}_summary風控.md")with open(summary_file_path, 'w', encoding='utf-8') as summary_file:summary_file.write(f"# {repo_name} Summary風控\n\n")summary_file.write("README內容風控。\n")print(f"Error generating summary for {repo_name}: {e}")# print(readme_text)else:print(f"文件不存在: {readme_path}")# If README doesn't exist, create an empty Markdown filesummary_file_path = os.path.join(summary_dir, f"{repo_name}_summary不存在.md")with open(summary_file_path, 'w', encoding='utf-8') as summary_file:summary_file.write(f"# {repo_name} Summary不存在\n\n")summary_file.write("README文件不存在。\n")
if __name__ == '__main__':# 配置組織名稱org_name = 'datawhalechina'# 配置 export_direxport_dir = "../database/readme_db"  # 請?zhí)鎿Q為實際readme的目錄路徑summary_dir="../../data_base/knowledge_db/readme_summary"# 請?zhí)鎿Q為實際readme的概括的目錄路徑model="gpt-3.5-turbo"  #deepseek-chat,gpt-3.5-turbo,moonshot-v1-8kmain(org_name,export_dir,summary_dir,model)

• extract_text_from_md() 函數(shù)用來抽取 md 文件中的文本，
• remove_urls() 函數(shù)過濾網頁鏈接以及大模型風控詞。
• generate_llm_summary() 函數(shù)LLM生成每個 readme 的概括。

2.在上述知識庫構建完畢之后，../../data_base/knowledge_db 目錄下就有了目標文件：

上面 函數(shù)本質上起了一個數(shù)據(jù)爬蟲和數(shù)據(jù)清理的過程；

作者這里也貢獻幾個構建個人GIS知識庫的爬蟲代碼：

• 爬取域名下的所有文字數(shù)據(jù)

# 爬取域名下的所有文字數(shù)據(jù)
import requests
import re
import urllib.request
from bs4 import BeautifulSoup
from collections import deque
from html.parser import HTMLParser
from urllib.parse import urlparse
import os
import pandas as pd
import tiktoken
import openai
import numpy as np
from ast import literal_eval
# 正則表達式模式，用于匹配URL
HTTP_URL_PATTERN = r'^http[s]{0,1}://.+$'
# 定義 OpenAI 的 API 密鑰
openai.api_key = 'your api key'
# 定義要爬取的根域名
domain = "leafletjs.com"
full_url = "https://leafletjs.com/"
# 創(chuàng)建一個類來解析 HTML 并獲取超鏈接
class HyperlinkParser(HTMLParser):def __init__(self):super().__init__()# 創(chuàng)建一個列表來存儲超鏈接self.hyperlinks = []# 重寫 HTMLParser 的 handle_starttag 方法以獲取超鏈接def handle_starttag(self, tag, attrs):attrs = dict(attrs)# 如果標簽是錨點標簽且具有 href 屬性，則將 href 屬性添加到超鏈接列表中if tag == "a" and "href" in attrs:self.hyperlinks.append(attrs["href"])
# 函數(shù)：從 URL 獲取超鏈接
def get_hyperlinks(url):# 嘗試打開 URL 并讀取 HTMLtry:# 打開 URL 并讀取 HTMLwith urllib.request.urlopen(url) as response:# 如果響應不是 HTML，則返回空列表if not response.info().get('Content-Type').startswith("text/html"):return []         # 解碼 HTMLhtml = response.read().decode('utf-8')except Exception as e:print(e)return []# 創(chuàng)建 HTML 解析器，然后解析 HTML 以獲取超鏈接parser = HyperlinkParser()parser.feed(html)return parser.hyperlinks
# 函數(shù)：獲取在同一域內的 URL 的超鏈接
def get_domain_hyperlinks(local_domain, url):clean_links = []for link in set(get_hyperlinks(url)):clean_link = None# 如果鏈接是 URL，請檢查是否在同一域內if re.search(HTTP_URL_PATTERN, link):# 解析 URL 并檢查域是否相同url_obj = urlparse(link)if url_obj.netloc == local_domain:clean_link = link# 如果鏈接不是 URL，請檢查是否是相對鏈接else:if link.startswith("/"):link = link[1:]elif (link.startswith("#")or link.startswith("mailto:")or link.startswith("tel:")):continueclean_link = "https://" + local_domain + "/" + linkif clean_link is not None:if clean_link.endswith("/"):clean_link = clean_link[:-1]clean_links.append(clean_link)# 返回在同一域內的超鏈接列表return list(set(clean_links))
# 函數(shù)：爬取網頁
def crawl(url):# 解析 URL 并獲取域名local_domain = urlparse(url).netloc# 創(chuàng)建一個隊列來存儲要爬取的 URLqueue = deque([url])# 創(chuàng)建一個集合來存儲已經看過的 URL（無重復）seen = set([url])# 創(chuàng)建一個目錄來存儲文本文件if not os.path.exists("text/"):os.mkdir("text/")if not os.path.exists("text/"+local_domain+"/"):os.mkdir("text/" + local_domain + "/")# 創(chuàng)建一個目錄來存儲 CSV 文件if not os.path.exists("processed"):os.mkdir("processed")# 當隊列非空時，繼續(xù)爬取while queue:# 從隊列中獲取下一個 URLurl = queue.pop()print(url)  # 用于調試和查看進度# 嘗試從鏈接中提取文本，如果失敗則繼續(xù)處理隊列中的下一項try:# 將來自 URL 的文本保存到 <url>.txt 文件中with open('text/'+local_domain+'/'+url[8:].replace("/", "_") + ".txt", "w", encoding="UTF-8") as f:# 使用 BeautifulSoup 從 URL 獲取文本soup = BeautifulSoup(requests.get(url).text, "html.parser")container = soup.find(class_="container")if container is not None:text = container.get_text()else:text = ""# # 獲取文本但去除標簽# text = soup.get_text()# 如果爬蟲遇到需要 JavaScript 的頁面，它將停止爬取if ("You need to enable JavaScript to run this app." in text):print("由于需要 JavaScript，無法解析頁面 " + url)  # 否則，將文本寫入到文本目錄中的文件中f.write(text)except Exception as e:print("無法解析頁面 " + url)# 獲取 URL 的超鏈接并將它們添加到隊列中for link in get_domain_hyperlinks(local_domain, url):if link not in seen:queue.append(link)seen.add(link)
crawl(full_url)
# 函數(shù)：移除字符串中的換行符
def remove_newlines(serie):serie = serie.str.replace('\n', ' ')serie = serie.str.replace('\\n', ' ')serie = serie.str.replace('  ', ' ')serie = serie.str.replace('  ', ' ')return serie
# 創(chuàng)建一個列表來存儲文本文件
texts=[]
# 獲取文本目錄中的所有文本文件
for file in os.listdir("text/" + domain + "/"):# 打開文件并讀取文本with open("text/" + domain + "/" + file, "r", encoding="UTF-8") as f:text = f.read()# 忽略前 11 行和后 4 行，然后替換 -、_ 和 #update 為空格texts.append((file[11:-4].replace('-',' ').replace('_', ' ').replace('#update',''), text))

• 爬取域名下所有文字數(shù)據(jù)的Selenium版本(爬取某些反爬網頁有奇效，就是慢一點)

import requests
from bs4 import BeautifulSoup
from msedge.selenium_tools import Edge, EdgeOptions
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.common.by import By
from urllib.parse import urlparse
import os
from collections import deque
import re
# Selenium 配置
options = EdgeOptions()
# 禁用地理位置請求
prefs = {"profile.default_content_setting_values.geolocation": 2}
options.add_experimental_option("prefs", prefs)
options.add_argument('--ignore-certificate-errors')
options.use_chromium = True
# options.add_argument("--headless")
options.add_argument("--disable-gpu")
options.add_argument("--disable-extensions")
# 正則表達式模式，用于匹配URL
HTTP_URL_PATTERN = r'^http[s]{0,1}://.+$'
# 定義要爬取的根域名
domain = "python.langchain.com"
base_url = "https://python.langchain.com/docs/"
# 函數(shù)：從 URL 獲取超鏈接
def get_hyperlinks(url, browser):try:browser.get(url)# WebDriverWait(browser, 10).until(EC.presence_of_element_located((By.TAG_NAME, "body")))WebDriverWait(browser, 10).until(EC.presence_of_element_located((By.CLASS_NAME, "menu")))html = browser.page_sourceexcept Exception as e:print(e)return []soup = BeautifulSoup(html, 'html.parser')links = [a.get('href') for a in soup.find_all('a', href=True)]return links
# 函數(shù)：獲取在同一域內的 URL 的超鏈接
def get_domain_hyperlinks(local_domain, base_url, url, browser):clean_links = []for link in set(get_hyperlinks(url, browser)):clean_link = None# 檢查鏈接是否為完整的 URLif re.search(HTTP_URL_PATTERN, link):url_obj = urlparse(link)if url_obj.netloc == local_domain:# 檢查鏈接是否是基礎 URL 或其錨點的變化if link.startswith(base_url) or (url_obj.path == urlparse(base_url).path and url_obj.fragment):clean_link = linkelse:# 處理相對鏈接if link.startswith("/"):complete_link = "https://" + local_domain + linkif complete_link.startswith(base_url):clean_link = complete_linkelif link.startswith("#"):# 處理錨點鏈接clean_link = base_url + linkelif link.startswith("mailto:") or link.startswith("tel:"):continueif clean_link is not None:clean_links.append(clean_link)return list(set(clean_links))
# 函數(shù)：保存文本到文件
def save_text_to_file(url, text, local_domain):# 檢查文本長度，如果太短或為空則不保存if len(text.strip()) < 50:print(f"文本內容太少，不保存: {url}")returnbase_dir = "text"domain_dir = os.path.join(base_dir, local_domain)if not os.path.exists(domain_dir):os.makedirs(domain_dir)filename = f"{domain_dir}/{url.replace('https://', '').replace('/', '_')}.txt"with open(filename, 'w', encoding='utf-8') as file:file.write(text)print(f"已保存: {filename}")
# 函數(shù)：爬取網頁
def crawl(base_url):local_domain = urlparse(base_url).netlocqueue = deque([base_url])seen = set([base_url])# 啟動 Edge 瀏覽器browser = Edge(options=options)try:while queue:url = queue.pop()try:browser.get(url)WebDriverWait(browser, 60).until(EC.presence_of_element_located((By.CLASS_NAME, "theme-doc-markdown")))html = browser.page_source# 使用 BeautifulSoup 解析和處理 HTMLsoup = BeautifulSoup(html, 'html.parser')# 專門提取 <main> 標簽的內容main_content = soup.find('div',class_="theme-doc-markdown markdown")if main_content is not None:text = main_content.get_text()else:continue# 保存文本到文件save_text_to_file(url, text, local_domain)except Exception as e:print("無法解析頁面:", url, "; 錯誤:", e)continuefor link in get_domain_hyperlinks(local_domain, base_url, url, browser):if link not in seen:queue.append(link)seen.add(link)finally:browser.quit()
crawl(base_url)

其中有 mp4 格式，md 格式，以及 pdf 格式，對這些文件的加載方式，該項目將代碼放在了 project/database/create_db.py文件 下，部分代碼如下。其中 pdf 格式文件用 PyMuPDFLoader 加載器，md格式文件用UnstructuredMarkdownLoader加載器：

from langchain.document_loaders import UnstructuredFileLoader
from langchain.document_loaders import UnstructuredMarkdownLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.document_loaders import PyMuPDFLoader
from langchain.vectorstores import Chroma
```python
# 首先實現(xiàn)基本配置
```python
DEFAULT_DB_PATH = "../../data_base/knowledge_db"
DEFAULT_PERSIST_PATH = "../../data_base/vector_db"
... 
...
...
def file_loader(file, loaders):if isinstance(file, tempfile._TemporaryFileWrapper):file = file.nameif not os.path.isfile(file):[file_loader(os.path.join(file, f), loaders) for f in  os.listdir(file)]returnfile_type = file.split('.')[-1]if file_type == 'pdf':loaders.append(PyMuPDFLoader(file))elif file_type == 'md':pattern = r"不存在|風控"match = re.search(pattern, file)if not match:loaders.append(UnstructuredMarkdownLoader(file))elif file_type == 'txt':loaders.append(UnstructuredFileLoader(file))return

(2)文本分割和向量化

文本分割和向量化將上述載入的知識庫文本或進行 token 長度進行分割，該項目利用 Langchain 中的文本分割器根據(jù) chunk_size (塊大小)和 chunk_overlap (塊與塊之間的重疊大小)進行分割。

• chunk_size 指每個塊包含的字符或 Token（如單詞、句子等）的數(shù)量
• chunk_overlap 指兩個塊之間共享的字符數(shù)量，用于保持上下文的連貫性，避免分割丟失上下文信息

tip:可以設置一個最大的 Token 長度，然后根據(jù)這個最大的 Token 長度來切分文檔。這樣切分出來的文檔片段是一個一個均勻長度的文檔片段。而片段與片段之間的一些重疊的內容，能保證檢索的時候能夠檢索到相關的文檔片段。

這部分文本分割代碼也在 project/database/create_db.py 文件，該項目采用了 langchain 中 RecursiveCharacterTextSplitter 文本分割器進行分割。代碼如下：

def create_db(files=DEFAULT_DB_PATH, persist_directory=DEFAULT_PERSIST_PATH, embeddings="openai"):"""該函數(shù)用于加載 PDF 文件，切分文檔，生成文檔的嵌入向量，創(chuàng)建向量數(shù)據(jù)庫。參數(shù):file: 存放文件的路徑。embeddings: 用于生產 Embedding 的模型返回:vectordb: 創(chuàng)建的數(shù)據(jù)庫。"""if files == None:return "can't load empty file"if type(files) != list:files = [files]loaders = [][file_loader(file, loaders) for file in files]docs = []for loader in loaders:if loader is not None:docs.extend(loader.load())# 切分文檔text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=150)split_docs = text_splitter.split_documents(docs)

而在切分好知識庫文本之后，需要對文本進行 向量化，文本向量化代碼文件路徑是project/embedding/call_embedding.py ，文本嵌入方式可選本地 m3e 模型，以及調用 openai 和 zhipuai 的 api 的方式進行文本嵌入。代碼如下：

import os
import syssys.path.append(os.path.dirname(os.path.dirname(__file__)))
sys.path.append(r"../../")
from embedding.zhipuai_embedding import ZhipuAIEmbeddings
from langchain.embeddings.huggingface import HuggingFaceEmbeddings
from langchain.embeddings.openai import OpenAIEmbeddings
from llm.call_llm import parse_llm_api_keydef get_embedding(embedding: str, embedding_key: str = None, env_file: str = None):if embedding == 'm3e':return HuggingFaceEmbeddings(model_name="moka-ai/m3e-base")if embedding_key == None:embedding_key = parse_llm_api_key(embedding)if embedding == "openai":return OpenAIEmbeddings(openai_api_key=embedding_key)elif embedding == "zhipuai":return ZhipuAIEmbeddings(zhipuai_api_key=embedding_key)else:raise ValueError(f"embedding {embedding} not support ")

讀者也可自行配置Emebdding模型；

(3)向量數(shù)據(jù)庫

在對知識庫文本進行分割和向量化后，就需要定義一個向量數(shù)據(jù)庫用來存放文檔片段和對應的向量表示了，在向量數(shù)據(jù)庫中，數(shù)據(jù)被表示為向量形式，每個向量代表一個數(shù)據(jù)項。這些向量可以是數(shù)字、文本、圖像或其他類型的數(shù)據(jù)。

向量數(shù)據(jù)庫使用高效的索引和查詢算法來加速向量數(shù)據(jù)的存儲和檢索過程。
該項目選擇 chromadb 向量數(shù)據(jù)庫（類似的向量數(shù)據(jù)庫還有 faiss 😏等)。定義向量庫對應的代碼也在 /database/create_db.py 文件中，persist_directory 即為本地持久化地址，vectordb.persist() 操作可以持久化向量數(shù)據(jù)庫到本地，后續(xù)可以再次載入本地已有的向量庫。完整的文本分割，獲取向量化，并且定義向量數(shù)據(jù)庫代碼如下：

def create_db(files=DEFAULT_DB_PATH, persist_directory=DEFAULT_PERSIST_PATH, embeddings="openai"):"""該函數(shù)用于加載 PDF 文件，切分文檔，生成文檔的嵌入向量，創(chuàng)建向量數(shù)據(jù)庫。參數(shù):file: 存放文件的路徑。embeddings: 用于生產 Embedding 的模型返回:vectordb: 創(chuàng)建的數(shù)據(jù)庫。"""if files == None:return "can't load empty file"if type(files) != list:files = [files]loaders = [][file_loader(file, loaders) for file in files]docs = []for loader in loaders:if loader is not None:docs.extend(loader.load())# 切分文檔text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=150)split_docs = text_splitter.split_documents(docs)if type(embeddings) == str:embeddings = get_embedding(embedding=embeddings)# 定義持久化路徑persist_directory = '../../data_base/vector_db/chroma'# 加載數(shù)據(jù)庫vectordb = Chroma.from_documents(documents=split_docs,embedding=embeddings,persist_directory=persist_directory  # 允許我們將persist_directory目錄保存到磁盤上) vectordb.persist()return vectordb

③向量檢索和生成

進入了 RAG 的檢索和生成階段，即對問句 Query 向量化后在知識庫文檔向量中匹配出與問句 Query 向量最相似的 top k 個片段，**檢索出知識庫文本文本作為上下文 Context 和問題?起添加到 prompt 中，然后提交給 LLM 生成回答 **。

(1)向量數(shù)據(jù)庫檢索

接下去利用向量數(shù)據(jù)庫來進行高效的檢索。

向量數(shù)據(jù)庫是一種用于有效搜索大規(guī)模高維向量空間中相似度的庫，能夠在大規(guī)模數(shù)據(jù)集中快速找到與給定 query 向量最相似的向量。

代碼如下所示：

question="什么是機器學習"
sim_docs = vectordb.similarity_search(question,k=3)
print(f"檢索到的內容數(shù)：{len(sim_docs)}")
for i, sim_doc in enumerate(sim_docs):print(f"檢索到的第{i}個內容: \n{sim_doc.page_content[:200]}", end="\n--------------\n")

運行結果：

檢索到的內容數(shù)：3
檢索到的第0個內容: 
導，同時也能體會到這三門數(shù)學課在機器學習上碰撞產生的“數(shù)學之美”。
1.1
引言
本節(jié)以概念理解為主，在此對“算法”和“模型”作補充說明?！八惴ā笔侵笍臄?shù)據(jù)中學得“模型”的具
體方法，例如后續(xù)章節(jié)中將會講述的線性回歸、對數(shù)幾率回歸、決策樹等?！八惴ā碑a出的結果稱為“模型”，
通常是具體的函數(shù)或者可抽象地看作為函數(shù)，例如一元線性回歸算法產出的模型即為形如 f(x) = wx + b的一元一次函數(shù)。
--------------檢索到的第1個內容: 
模型：機器學習的一般流程如下：首先收集若干樣本（假設此時有 100 個），然后將其分為訓練樣本
（80 個）和測試樣本（20 個），其中 80 個訓練樣本構成的集合稱為“訓練集”，20 個測試樣本構成的集合
稱為“測試集”，接著選用某個機器學習算法，讓其在訓練集上進行“學習”（或稱為“訓練”），然后產出得到“模型”（或稱為“學習器”），最后用測試集來測試模型的效果。執(zhí)行以上流程時，表示我們已經默
--------------檢索到的第2個內容: 
→_→
歡迎去各大電商平臺選購紙質版南瓜書《機器學習公式詳解》
←_←
第 1 章
緒論
本章作為“西瓜書”的開篇，主要講解什么是機器學習以及機器學習的相關數(shù)學符號，為后續(xù)內容作
鋪墊，并未涉及復雜的算法理論，因此閱讀本章時只需耐心梳理清楚所有概念和數(shù)學符號即可。此外，在
閱讀本章前建議先閱讀西瓜書目錄前頁的《主要符號表》，它能解答在閱讀“西瓜書”過程中產生的大部
分對數(shù)學符號的疑惑。
本章也作為

(2)大模型llm的調用

以該項目 project/qa_chain/model_to_llm.py 代碼為例，**在 project/llm/ 的目錄文件夾下分別定義了 星火spark，智譜glm，文心llm等開源模型api調用的封裝，**并在 project/qa_chain/model_to_llm.py 文件中導入了這些模塊，可以根據(jù)用戶傳入的模型名字進行調用 llm。代碼如下：

def model_to_llm(model:str=None, temperature:float=0.0, appid:str=None, api_key:str=None,Spark_api_secret:str=None,Wenxin_secret_key:str=None):"""星火：model,temperature,appid,api_key,api_secret百度問心：model,temperature,api_key,api_secret智譜：model,temperature,api_keyOpenAI：model,temperature,api_key"""if model in ["gpt-3.5-turbo", "gpt-3.5-turbo-16k-0613", "gpt-3.5-turbo-0613", "gpt-4", "gpt-4-32k"]:if api_key == None:api_key = parse_llm_api_key("openai")llm = ChatOpenAI(model_name = model, temperature = temperature , openai_api_key = api_key)elif model in ["ERNIE-Bot", "ERNIE-Bot-4", "ERNIE-Bot-turbo"]:if api_key == None or Wenxin_secret_key == None:api_key, Wenxin_secret_key = parse_llm_api_key("wenxin")llm = Wenxin_LLM(model=model, temperature = temperature, api_key=api_key, secret_key=Wenxin_secret_key)elif model in ["Spark-1.5", "Spark-2.0"]:if api_key == None or appid == None and Spark_api_secret == None:api_key, appid, Spark_api_secret = parse_llm_api_key("spark")llm = Spark_LLM(model=model, temperature = temperature, appid=appid, api_secret=Spark_api_secret, api_key=api_key)elif model in ["chatglm_pro", "chatglm_std", "chatglm_lite"]:if api_key == None:api_key = parse_llm_api_key("zhipuai")llm = ZhipuAILLM(model=model, zhipuai_api_key=api_key, temperature = temperature)else:raise ValueError(f"model{model} not support!!!")return llm

(3)prompt和構建問答鏈

接下去來到了最后一步，設計完基于知識庫問答的 prompt，就可以結合上述檢索和大模型調用進行答案的生成。構建 prompt 的格式如下，具體可以根據(jù)自己業(yè)務需要進行修改：

from langchain.prompts import PromptTemplate
# template = """基于以下已知信息，簡潔和專業(yè)的來回答用戶的問題。
#             如果無法從中得到答案，請說 "根據(jù)已知信息無法回答該問題" 或 "沒有提供足夠的相關信息"，不允許在答案中添加編造成分。
#             答案請使用中文。
#             總是在回答的最后說“謝謝你的提問！”。
# 已知信息：{context}
# 問題: {question}"""
template = """使用以下上下文來回答最后的問題。如果你不知道答案，就說你不知道，不要試圖編造答
案。最多使用三句話。盡量使答案簡明扼要?？偸窃诨卮鸬淖詈笳f“謝謝你的提問！”。
{context}
問題: {question}
有用的回答:"""QA_CHAIN_PROMPT = PromptTemplate(input_variables=["context","question"],template=template)# 運行 chain
并且構建問答鏈：創(chuàng)建檢索 QA 鏈的方法 RetrievalQA.from_chain_type() 有如下參數(shù)：

參數(shù)介紹

• llm：指定使用的 LLM
• chain type :RetrievalQA.from_chain_type(chain_type=“map_reduce”)，
• 自定義 prompt ：通過在RetrievalQA.from_chain_type()方法中，指定chain_type_kwargs參數(shù)，而該參數(shù)：chain_type_kwargs = {“prompt”: PROMPT}
• 返回源文檔：通過RetrievalQA.from_chain_type()方法中指定：return_source_documents=True參數(shù)；也可以使用RetrievalQAWithSourceChain()方法，返回源文檔的引用（坐標或者叫主鍵、索引）

#自定義 QA 鏈
self.qa_chain = RetrievalQA.from_chain_type(llm=self.llm,retriever=self.retriever,return_source_documents=True,chain_type_kwargs={"prompt":self.QA_CHAIN_PROMPT})

問答鏈效果如下：基于召回結果和 query 結合起來構建的 prompt 效果

question_1 = "什么是南瓜書？"
question_2 = "王陽明是誰？"
result = qa_chain({"query": question_1})
print("大模型+知識庫后回答 question_1 的結果：")
print(result["result"])

大模型+知識庫后回答 question_1 的結果：
南瓜書是對《機器學習》（西瓜書）中難以理解的公式進行解析和補充推導細節(jié)的一本書。謝謝你的提問！

result = qa_chain({"query": question_2})
print("大模型+知識庫后回答 question_2 的結果：")
print(result["result"])

大模型+知識庫后回答 question_2 的結果：
我不知道王陽明是誰，謝謝你的提問！

以上檢索問答鏈代碼都在project/qa_chain/QA_chain_self.py 中，此外該項目還實現(xiàn)了帶記憶的檢索問答鏈，兩種自定義檢索問答鏈內部實現(xiàn)細節(jié)類似，只是調用了不同的 LangChain 鏈。完整帶記憶的檢索問答鏈條代碼 project/qa_chain/Chat_QA_chain_self.py 如下：

from langchain.prompts import PromptTemplate
from langchain.chains import RetrievalQA
from langchain.vectorstores import Chroma
from langchain.chains import ConversationalRetrievalChain
from langchain.memory import ConversationBufferMemory
from langchain.chat_models import ChatOpenAIfrom qa_chain.model_to_llm import model_to_llm
from qa_chain.get_vectordb import get_vectordbclass Chat_QA_chain_self:""""帶歷史記錄的問答鏈  - model：調用的模型名稱- temperature：溫度系數(shù)，控制生成的隨機性- top_k：返回檢索的前k個相似文檔- chat_history：歷史記錄，輸入一個列表，默認是一個空列表- history_len：控制保留的最近 history_len 次對話- file_path：建庫文件所在路徑- persist_path：向量數(shù)據(jù)庫持久化路徑- appid：星火- api_key：星火、百度文心、OpenAI、智譜都需要傳遞的參數(shù)- Spark_api_secret：星火秘鑰- Wenxin_secret_key：文心秘鑰- embeddings：使用的embedding模型- embedding_key：使用的embedding模型的秘鑰（智譜或者OpenAI）  """def __init__(self,model:str, temperature:float=0.0, top_k:int=4, chat_history:list=[], file_path:str=None, persist_path:str=None, appid:str=None, api_key:str=None, Spark_api_secret:str=None,Wenxin_secret_key:str=None, embedding = "openai",embedding_key:str=None):self.model = modelself.temperature = temperatureself.top_k = top_kself.chat_history = chat_history#self.history_len = history_lenself.file_path = file_pathself.persist_path = persist_pathself.appid = appidself.api_key = api_keyself.Spark_api_secret = Spark_api_secretself.Wenxin_secret_key = Wenxin_secret_keyself.embedding = embeddingself.embedding_key = embedding_keyself.vectordb = get_vectordb(self.file_path, self.persist_path, self.embedding,self.embedding_key)def clear_history(self):"清空歷史記錄"return self.chat_history.clear()def change_history_length(self,history_len:int=1):"""保存指定對話輪次的歷史記錄輸入?yún)?shù)：- history_len ：控制保留的最近 history_len 次對話- chat_history：當前的歷史對話記錄輸出：返回最近 history_len 次對話"""n = len(self.chat_history)return self.chat_history[n-history_len:]def answer(self, question:str=None,temperature = None, top_k = 4):""""核心方法，調用問答鏈arguments: - question：用戶提問"""if len(question) == 0:return "", self.chat_historyif len(question) == 0:return ""if temperature == None:temperature = self.temperaturellm = model_to_llm(self.model, temperature, self.appid, self.api_key, self.Spark_api_secret,self.Wenxin_secret_key)#self.memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True)retriever = self.vectordb.as_retriever(search_type="similarity",   search_kwargs={'k': top_k})  #默認similarity，k=4qa = ConversationalRetrievalChain.from_llm(llm = llm,retriever = retriever)#print(self.llm)result = qa({"question": question,"chat_history": self.chat_history})       #result里有question、chat_history、answeranswer =  result['answer']self.chat_history.append((question,answer)) #更新歷史記錄return self.chat_history  #返回本次回答和更新后的歷史記錄

OK，時間有限，分析完畢，各位讀者有興趣可以學習一下，給官方一個Star😀😀😀;

文章參考

• OpenAI官方文檔
• DeepSeek官方文檔
• Mistral官方文檔
• ChatGLM官方文檔

項目地址

• Github地址
• 拓展閱讀
• 專欄文章

---

如果覺得我的文章對您有幫助，三連+關注便是對我創(chuàng)作的最大鼓勵！或者一個star🌟也可以😂.