如何用個人數據知識庫構建RAG聊天機器人？

01.使用 BeautifulSoup4 抓取網頁數據

所有機器學習（ML）項目的第一步都是收集所需的數據。本項目中，我們使用網頁抓取技術來收集知識庫數據。用 requests 庫獲取網頁并使用 BeautifulSoup4.從網頁中提取信息、解析 HTML 信息并提取段落。

導入 BeautifulSoup4 和 Requests 庫進行網頁抓取

運行 pip install beautifulsoup4 sentence-transformers安裝 BeautifulSoup 和 Sentence Transformers。在數據抓取部分只需要導入requests和 BeautifulSoup。接下來，創建一個 dictionary，其中包含我們要抓取的 URL 格式。在本示例中，我們只從 Towards Data Science 抓取內容，同理也可以從其他網站抓取。

現在，用以下代碼所示的格式從每個存檔頁面獲取數據：

import requests
from bs4 import BeautifulSoup
urls = {
    'Towards Data Science': '< https://towardsdatascience.com/archive/{0}/{1:02d}/{2:02d} >'
    }

此外，我們還需要兩個輔助函數來進行網頁抓取。第一個函數將一年中的天數轉換為月份和日期格式。第二個函數從一篇文章中獲取點贊數。

天數轉換函數相對簡單。寫死每個月的天數，并使用該列表進行轉換。由于本項目僅抓取 2023 年數據，因此我們不需要考慮閏年。如果您愿意，可以根據不同的年份進行修改每個月天數。

點贊計數函數統計 Medium 上文章的點贊數，單位為 “K” （1K=1000）。因此，在函數中需要考慮點贊數中的單位“K”。

def convert_day(day):
    month_list = [31, 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31]
    m = 0
    d = 0
    while day > 0:
        m += 1
        d = day
        day -= month_list[m-1]
    return (m, d)

def get_claps(claps_str):
    if (claps_str is None) or (claps_str == '') or (claps_str.split is None):
        return 0
    split = claps_str.split('K')
    claps = float(split[0])
    return int(claps*1000) if len(split) == 2 else int(claps)

解析 BeautifulSoup4 的網頁抓取響應

現在已經設置好必要的組件，可以進行網頁抓取。為了避免在過程中遇到 429 錯誤（請求過多），我們使用 time 庫，在發送請求之間引入延遲。此外，用 sentence transformers 庫從 Hugging Face 獲取 embedding 模型—— MiniLM 模型。

如前所述，我們只抓取了 2023 年的數據，所以將年份設置為 2023。此外，只需要從第 1 天（1 月 1 日）到第 244 天（8 月 31 日）的數據。根據設定的天數進行循環，每個循環在第一次調用time.sleep()之前會首先設置必要的組件。我們會把天數轉換成月份和日期，并轉成字符串，然后根據 urls 字典組成完整的 URL，最后發送請求獲取 HTML 響應。

獲取 HTML 響應之后，使用 BeautifulSoup 進行解析，并搜索具有特定類名（在代碼中指示）的div元素，該類名表示它是一篇文章。我們從中解析標題、副標題、文章 URL、點贊數、閱讀時長和回應數。隨后，再次使用requests來獲取文章的內容。每次通過請求獲取文章內容后，都會再次調用time.sleep()。此時，我們已經獲取了大部分所需的文章元數據。提取文章的每個段落，并使用我們的 HuggingFace 模型獲得對應的向量。接著，創建一個字典包含該文章段落的所有元信息。

import time
from sentence_transformers import SentenceTransformer

model = SentenceTransformer("sentence-transformers/all-MiniLM-L12-v2")
data_batch = []
year = 2023
for i in range(1, 243):
    month, day = convert_day(i)
    date = '{0}-{1:02d}-{2:02d}'.format(year, month, day)
    for publication, url in urls.items():
        response = requests.get(url.format(year, month, day), allow_redirects=True)
        if not response.url.startswith(url.format(year, month, day)):
            continue
        time.sleep(8)
        soup = BeautifulSoup(response.content, 'html.parser')
        articles = soup.find_all("div","postArticle postArticle--short js-postArticle js-trackPostPresentation js-trackPostScrolls")
        for article in articles:
            title = article.find("h3", class_="graf--title")
            if title is None:
                continue
            title = str(title.contents[0]).replace(u'\xA0', u' ').replace(u'\u200a', u' ')
            subtitle = article.find("h4", class_="graf--subtitle")
            subtitle = str(subtitle.contents[0]).replace(u'\xA0', u' ').replace(u'\u200a', u' ') if subtitle is not None else ''
            article_url = article.find_all("a")[3]['href'].split('?')[0]
            claps = get_claps(article.find_all("button")[1].contents[0])
            reading_time = article.find("span", class_="readingTime")
            reading_time = int(reading_time['title'].split(' ')[0]) if reading_time is not None else 0
            responses = article.find_all("a", class_="button")
            responses = int(responses[6].contents[0].split(' ')[0]) if len(responses) == 7 else (0 if len(responses) == 0 else int(responses[0].contents[0].split(' ')[0]))
            article_res = requests.get(article_url)
            time.sleep(8)
            paragraphs = BeautifulSoup(article_res.content, 'html.parser').find_all("[class*="pw-post-body-paragraph"]")
            for i, paragraph in enumerate(paragraphs):
                embedding = model.encode([paragraph.text])[0].tolist()
                data_batch.append({
                    "_id": f"{article_url}+{i}",
                    "article_url": article_url,
                    "title": title,
                    "subtitle": subtitle,
                    "claps": claps,
                    "responses": responses,
                    "reading_time": reading_time,
                    "publication": publication,
                    "date": date,
                    "paragraph": paragraph.text,
                    "embedding": embedding
                })

最后一步是使用 pickle 處理文件。

filename = "TDS_8_30_2023"
with open(f'{filename}.pkl', 'wb') as f:
    pickle.dump(data_batch, f)

數據呈現

數據可視化十分有用。下面是在 Zilliz Cloud 中數據的樣子。請注意其中的 embedding，這些數據表示了文檔向量，也就是我們根據文章段落生成的向量。

02.將 TDS 數據導入到向量數據庫中

獲取數據后，下一步是將其導入到向量數據庫中。在本項目中，我們使用了一個單獨的 notebook 將數據導入到 Zilliz Cloud，而不是從 Towards Data Science 進行網頁抓取。

要將數據插入 Zilliz Cloud，需按照以下步驟進行操作：

• 連接到 Zilliz Cloud
• 定義 Collection 的參數
• 將數據插入 Zilliz Cloud

設置 Jupyter Notebook

運行 pip install pymilvus python-dotenv 來設置 Jupyter Notebook 并啟動數據導入過程。用 dotenv 庫來管理環境變量。對于pymilvus包，需要導入以下模塊：

• utility 用于檢查集合的狀態
• connections 用于連接到 Milvus 實例
• FieldSchema 用于定義字段的 schema
• CollectionSchema 用于定義 collection schema
• DataType 字段中存儲的數據類型
• Collection 我們訪問 collection 的方式

然后，打開之前 pickle 的數據，獲取環境變量，并連接到 Zilliz Cloud。

import pickle
import os
from dotenv import load_dotenv
from pymilvus import utility, connections, FieldSchema, CollectionSchema, DataType, Collection


filename="TDS_8_30_2023"
with open(f'{filename}.pkl', 'rb') as f:
    data_batch = pickle.load(f)

zilliz_uri = "your_zilliz_uri"
zilliz_token = "your_zilliz_token"
connections.connect(
    uri= zilliz_uri,
    token= zilliz_token
)

設置 Zilliz Cloud 向量數據庫并導入數據

接下來，需要設置 Zilliz Cloud。我們必須創建一個 Collection 來存儲和組織從 TDS 網站抓取的數據。需要兩個常量：dimension（維度）和 collection name（集合名稱），dimension 是指我們的向量具有的維度數。在本項目中，我們使用 384 維的 MiniLM 模型。

Milvus 的全新 Dynamic schema 功能允許我們僅為 Collection 設置 ID 和向量字段，無需考慮其他字段數量和數據類型。注意，需要記住保存的特定字段名稱，因為這對于正確檢索字段至關重要。

DIMENSION=384
COLLECTION_NAME="tds_articles"
fields = [
    FieldSchema(name='id', dtype=DataType.VARCHAR, max_length=200, is_primary=True),
    FieldSchema(name='embedding', dtype=DataType.FLOAT_VECTOR, dim=DIMENSION)
]
schema = CollectionSchema(fields=fields, enable_dynamic_field=True)
collection = Collection(name=COLLECTION_NAME, schema=schema)
index_params = {
    "index_type": "AUTO_INDEX",
    "metric_type": "L2",
    "params": {"nlist": 128},
}
collection.create_index(field_name="embedding", index_params=index_params)

Collection 有兩種插入數據的選項：

• 遍歷數據并逐個插入每個數據
• 批量插入數據

在插入所有數據之后，重要的是刷新集合以進行索引并確保一致性，導入大量數據可能需要一些時間。

for data in data_batch:
    collection.insert([data])
collection.flush()

03.查詢 TDS 文章片段

一切準備就緒后，就可以進行查詢了。

獲取 HuggingFace 模型并設置 Zilliz Cloud 查詢

注意，必須獲取 embedding 模型并設置向量數據庫以查詢 Towards Data Science 知識庫。這一步使用了一個單獨的筆記本。我們將使用dotenv庫來管理環境變量。此外，還需要使用 Sentence Transformers 中的 MiniLM 模型。這一步中，可以重用 Web Scraping 部分提供的代碼。

import os
from dotenv import load_dotenv
from pymilvus import connections, Collection


zilliz_uri = "your_zilliz_uri"
zilliz_token = "your_zilliz_token"


from sentence_transformers import SentenceTransformer


model = SentenceTransformer("sentence-transformers/all-MiniLM-L12-v2")

執行向量搜索查詢

連接到向量數據庫并執行搜索。在本項目中，我們將連接到一個 Zilliz Cloud 實例，并檢索之前創建的集合 tds_articles，用戶要先輸入他們的查詢問題。

接下來，使用 Hugging Face 的 embedding 模型對查詢進行編碼。這個過程將用戶的問題轉換為一個 384 維的向量。然后，使用這個編碼后的查詢向量來搜索向量數據庫。在搜索過程中，需要指定進行 ANN 查詢字段（anns_field）、索引參數、期望的搜索結果數量限制以及我們想要的輸出字段（output fields）。

之前，我們用了 Milvus 的 Dynamic Schema 特性來簡化字段 Schema 定義流程。搜索向量數據庫時，包括所需的動態字段在搜索結果中是必要的。這個特定的場景涉及請求paragraph字段，其中包含文章中每個段落的文本。

connections.connect(uri=zilliz_uri, token=zilliz_token)
collection = Collection(name="tds_articles")
query = input("What would you like to ask Towards Data Science's 2023 publications up to September? ")
embedding = model.encode(query)
closest = collection.search([embedding],
    anns_field='embedding',
    param={"metric_type": "L2",
        "params": {"nprobe": 16}},
    limit=2,
    output_fields=["paragraph"])
print(closest[0][0])
print(closest[0][1])

比如，我在應用中查詢大語言模型相關的信息，返回了以下兩個回答。盡管這些回答提到了“語言模型”并包含一些相關信息，但它們沒有提供關于大型語言模型的詳細解釋。第二個回答在語義上相似，但是不足夠接近我們想要的內容。

04.給向量數據庫知識庫添加內容

到目前為止，我們使用 Zilliz Cloud 作為向量數據庫在 TDS 文章上創建了一個知識庫。雖然能夠輕松地檢索語義上相似的搜索結果，但還沒有達到我們的期望。下一步是通過加入新的框架和技術來增強我們的結果。

05.總結

本教程介紹了如何基于 Towards Data Science 文章構建聊天機器人。我們演示了網頁爬取的過程，創建了知識庫，包括將文本轉換成向量存儲在 Zilliz Cloud 中。然后，我們演示了如何提示用戶進行查詢，將查詢轉化為向量，并查詢向量數據庫。

不過，雖然結果在語義上相似，但并不完全符合我們的期望。在本系列的下一篇中，我們將探討使用 LlamaIndex 來優化查詢。除了這里討論的步驟之外，大家也可以結合 Zilliz Cloud 嘗試替換模型、合并文本或使用其他數據集。