前言

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

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

---

一、個(gè)人知識(shí)庫助手

1.項(xiàng)目介紹

①背景

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

②目標(biāo)意義

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

③主要功能

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

2.項(xiàng)目部署

①環(huán)境要求

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

阿里云199一年的都可以，也可以自己本機(jī)部署；因?yàn)槭茿PI調(diào)用，對電腦性能要求不高；

• 內(nèi)存（RAM）: 至少 4 GB

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

②部署流程

這里作者將整個(gè)流程打包:

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
# 安裝依賴項(xiàng)
pip install -r requirements.txt

運(yùn)行項(xiàng)目:

# 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='嵌入模型編號(hào)' -db_path='知識(shí)庫文件路徑' -persist_path='持久化目錄文件路徑'

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

③核心思想

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

④技術(shù)棧

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

當(dāng)然，這里是因?yàn)槲覀兊膽?yīng)用單一，因此只選擇了向量數(shù)據(jù)庫。如果還具有其他業(yè)務(wù)，如列表信息、歷史記錄、檢索信息等，使用結(jié)構(gòu)化數(shù)據(jù)庫也是必要的。向量數(shù)據(jù)庫和結(jié)構(gòu)化數(shù)據(jù)混合使用更合適。

• 本項(xiàng)目支持本地M3E Embedding模型和 API Embedding結(jié)合的方式進(jìn)行向量化；

3.應(yīng)用詳解

①業(yè)務(wù)流程

1、核心架構(gòu)

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

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

整個(gè) RAG 過程包括以下操作：

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

這里很好理解，之前文章中，LLM基礎(chǔ)學(xué)習(xí)的第三篇詳細(xì)講解了其中的每個(gè)步驟；

②索引 Index

三步走戰(zhàn)略：

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

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

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

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

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

項(xiàng)目官方提供了拉去官網(wǎng)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內(nèi)容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為實(shí)際的目錄路徑# 獲取倉庫列表repos = get_repos(org_name, TOKEN, export_dir)# 打印倉庫名稱if repos:for repo in repos:repo_name = repo['name']# 拉取每個(gè)倉庫的READMEfetch_repo_readme(org_name, repo_name, TOKEN, export_dir)# 清理臨時(shí)文件夾# if os.path.exists('temp'):#     shutil.rmtree('temp')

以上默認(rèn)會(huì)把這些readme文件放在同目錄database下的readme_db文件。其中這些readme文件含有不少無關(guān)信息;

😏再運(yùn)行database/text_summary_readme.py文件可以調(diào)用大模型生成每個(gè)readme文件的摘要并保存到上述知識(shí)庫目錄/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"]# 過濾文本中鏈接防止大語言模型風(fēng)控
def remove_urls(text):# 正則表達(dá)式模式，用于匹配URLurl_pattern = re.compile(r'https?://[^\s]*')# 替換所有匹配的URL為空字符串text = re.sub(url_pattern, '', text)# 正則表達(dá)式模式，用于匹配特定的文本specific_text_pattern = re.compile(r'掃描下方二維碼關(guān)注公眾號(hào)|提取碼|關(guān)注|科學(xué)上網(wǎng)|回復(fù)關(guān)鍵詞|侵權(quán)|版權(quán)|致謝|引用|LICENSE'r'|組隊(duì)打卡|任務(wù)打卡|組隊(duì)學(xué)習(xí)的那些事|學(xué)習(xí)周期|開源內(nèi)容|打卡|組隊(duì)學(xué)習(xí)|鏈接')# 替換所有匹配的特定文本為空字符串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：這個(gè)倉庫名是 {repo_name}. 此倉庫的readme全部內(nèi)容是: {readme_content}\2:請用約200以內(nèi)的中文概括這個(gè)倉庫readme的內(nèi)容,返回的概括格式要求：這個(gè)倉庫名是...,這倉庫內(nèi)容主要是..."openai.api_key = openai_api_key# 具體調(diào)用messages = [{"role": "system", "content": "你是一個(gè)人工智能助手"},{"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風(fēng)控.md")with open(summary_file_path, 'w', encoding='utf-8') as summary_file:summary_file.write(f"# {repo_name} Summary風(fēng)控\n\n")summary_file.write("README內(nèi)容風(fēng)控。\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為實(shí)際readme的目錄路徑summary_dir="../../data_base/knowledge_db/readme_summary"# 請?zhí)鎿Q為實(shí)際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ù)過濾網(wǎng)頁鏈接以及大模型風(fēng)控詞。
• generate_llm_summary() 函數(shù)LLM生成每個(gè) readme 的概括。

2.在上述知識(shí)庫構(gòu)建完畢之后，../../data_base/knowledge_db 目錄下就有了目標(biāo)文件：

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

作者這里也貢獻(xiàn)幾個(gè)構(gòu)建個(gè)人GIS知識(shí)庫的爬蟲代碼：

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

# 爬取域名下的所有文字?jǐn)?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
# 正則表達(dá)式模式，用于匹配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)建一個(gè)類來解析 HTML 并獲取超鏈接
class HyperlinkParser(HTMLParser):def __init__(self):super().__init__()# 創(chuàng)建一個(gè)列表來存儲(chǔ)超鏈接self.hyperlinks = []# 重寫 HTMLParser 的 handle_starttag 方法以獲取超鏈接def handle_starttag(self, tag, attrs):attrs = dict(attrs)# 如果標(biāo)簽是錨點(diǎn)標(biāo)簽且具有 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:# 如果響應(yīng)不是 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ù)：獲取在同一域內(nèi)的 URL 的超鏈接
def get_domain_hyperlinks(local_domain, url):clean_links = []for link in set(get_hyperlinks(url)):clean_link = None# 如果鏈接是 URL，請檢查是否在同一域內(nèi)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)# 返回在同一域內(nèi)的超鏈接列表return list(set(clean_links))
# 函數(shù)：爬取網(wǎng)頁
def crawl(url):# 解析 URL 并獲取域名local_domain = urlparse(url).netloc# 創(chuàng)建一個(gè)隊(duì)列來存儲(chǔ)要爬取的 URLqueue = deque([url])# 創(chuàng)建一個(gè)集合來存儲(chǔ)已經(jīng)看過的 URL（無重復(fù)）seen = set([url])# 創(chuàng)建一個(gè)目錄來存儲(chǔ)文本文件if not os.path.exists("text/"):os.mkdir("text/")if not os.path.exists("text/"+local_domain+"/"):os.mkdir("text/" + local_domain + "/")# 創(chuàng)建一個(gè)目錄來存儲(chǔ) CSV 文件if not os.path.exists("processed"):os.mkdir("processed")# 當(dāng)隊(duì)列非空時(shí)，繼續(xù)爬取while queue:# 從隊(duì)列中獲取下一個(gè) URLurl = queue.pop()print(url)  # 用于調(diào)試和查看進(jìn)度# 嘗試從鏈接中提取文本，如果失敗則繼續(xù)處理隊(duì)列中的下一項(xiàng)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 = ""# # 獲取文本但去除標(biāo)簽# 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 的超鏈接并將它們添加到隊(duì)列中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)建一個(gè)列表來存儲(chǔ)文本文件
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))

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

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")
# 正則表達(dá)式模式，用于匹配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ù)：獲取在同一域內(nèi)的 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:# 檢查鏈接是否是基礎(chǔ) URL 或其錨點(diǎn)的變化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("#"):# 處理錨點(diǎn)鏈接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):# 檢查文本長度，如果太短或?yàn)榭談t不保存if len(text.strip()) < 50:print(f"文本內(nèi)容太少，不保存: {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ù)：爬取網(wǎng)頁
def crawl(base_url):local_domain = urlparse(base_url).netlocqueue = deque([base_url])seen = set([base_url])# 啟動(dòng) 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> 標(biāo)簽的內(nèi)容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, "; 錯(cuò)誤:", 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 格式，對這些文件的加載方式，該項(xiàng)目將代碼放在了 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
# 首先實(shí)現(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"不存在|風(fēng)控"match = re.search(pattern, file)if not match:loaders.append(UnstructuredMarkdownLoader(file))elif file_type == 'txt':loaders.append(UnstructuredFileLoader(file))return

(2)文本分割和向量化

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

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

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

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

def create_db(files=DEFAULT_DB_PATH, persist_directory=DEFAULT_PERSIST_PATH, embeddings="openai"):"""該函數(shù)用于加載 PDF 文件，切分文檔，生成文檔的嵌入向量，創(chuàng)建向量數(shù)據(jù)庫。參數(shù):file: 存放文件的路徑。embeddings: 用于生產(chǎn) 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)

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

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ìn)行分割和向量化后，就需要定義一個(gè)向量數(shù)據(jù)庫用來存放文檔片段和對應(yīng)的向量表示了，在向量數(shù)據(jù)庫中，數(shù)據(jù)被表示為向量形式，每個(gè)向量代表一個(gè)數(shù)據(jù)項(xiàng)。這些向量可以是數(shù)字、文本、圖像或其他類型的數(shù)據(jù)。

向量數(shù)據(jù)庫使用高效的索引和查詢算法來加速向量數(shù)據(jù)的存儲(chǔ)和檢索過程。
該項(xiàng)目選擇 chromadb 向量數(shù)據(jù)庫（類似的向量數(shù)據(jù)庫還有 faiss 😏等)。定義向量庫對應(yīng)的代碼也在 /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: 用于生產(chǎn) 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

③向量檢索和生成

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

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

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

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

代碼如下所示：

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

運(yùn)行結(jié)果：

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

(2)大模型llm的調(diào)用

以該項(xiàng)目 project/qa_chain/model_to_llm.py 代碼為例，**在 project/llm/ 的目錄文件夾下分別定義了 星火spark，智譜glm，文心llm等開源模型api調(diào)用的封裝，**并在 project/qa_chain/model_to_llm.py 文件中導(dǎo)入了這些模塊，可以根據(jù)用戶傳入的模型名字進(jìn)行調(diào)用 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和構(gòu)建問答鏈

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

from langchain.prompts import PromptTemplate
# template = """基于以下已知信息，簡潔和專業(yè)的來回答用戶的問題。
#             如果無法從中得到答案，請說 "根據(jù)已知信息無法回答該問題" 或 "沒有提供足夠的相關(guān)信息"，不允許在答案中添加編造成分。
#             答案請使用中文。
#             總是在回答的最后說“謝謝你的提問！”。
# 已知信息：{context}
# 問題: {question}"""
template = """使用以下上下文來回答最后的問題。如果你不知道答案，就說你不知道，不要試圖編造答
案。最多使用三句話。盡量使答案簡明扼要。總是在回答的最后說“謝謝你的提問！”。
{context}
問題: {question}
有用的回答:"""QA_CHAIN_PROMPT = PromptTemplate(input_variables=["context","question"],template=template)# 運(yùn)行 chain
并且構(gòu)建問答鏈：創(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()方法，返回源文檔的引用（坐標(biāo)或者叫主鍵、索引）

#自定義 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})

問答鏈效果如下：基于召回結(jié)果和 query 結(jié)合起來構(gòu)建的 prompt 效果

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

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

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

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

以上檢索問答鏈代碼都在project/qa_chain/QA_chain_self.py 中，此外該項(xiàng)目還實(shí)現(xiàn)了帶記憶的檢索問答鏈，兩種自定義檢索問答鏈內(nèi)部實(shí)現(xiàn)細(xì)節(jié)類似，只是調(diào)用了不同的 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：調(diào)用的模型名稱- temperature：溫度系數(shù)，控制生成的隨機(jī)性- top_k：返回檢索的前k個(gè)相似文檔- chat_history：歷史記錄，輸入一個(gè)列表，默認(rèn)是一個(gè)空列表- 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：當(dāng)前的歷史對話記錄輸出：返回最近 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):""""核心方法，調(diào)用問答鏈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})  #默認(rèn)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，時(shí)間有限，分析完畢，各位讀者有興趣可以學(xué)習(xí)一下，給官方一個(gè)Star😀😀😀;

文章參考

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

項(xiàng)目地址

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

---

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