Diffusion, Deep learning 和向量資料庫研究

picture

2024-04-25

Diffusion, Deep learning 和向量資料庫研究

1. Image Generation

參考資料:

常見圖片生成

  • Autoencoder [2006]
  • Variational Autoencoder (VAE) [2014]
  • Flow-based models
  • GAN [2014]
  • PixelRNN [2016]
  • Diffusion [2020]

Defusion Model

Defusion Model 是模仿Markov chain,利用常態分佈(高斯分佈)逐漸對一張圖片增加雜訊,直到看不到原始的圖片(下圖的q),再利用模型一步一步把圖片還原回來(下圖的p) image

可以看下左這張圖逐漸從雜訊中生出一張圖。

Image 1

生成過程

Image 2

我的train壞掉的版本

Diffusion process (擴散過程)

下面是的意思是,我們會逐步對一張圖片加上常態分配的雜訊,這個步驟總共會做T次,其中的一次叫做t。Beta_t是一個會隨著t越大而增加的值,而雜訊的常態分配的平均數(mean)是Beta_t*上一步驟t-1的圖片, 變異數(var)是Beta_t,可以算出雜訊eps, 接著新的圖片會是main + √var*eps

我們模型要訓練的其實是給定一張被雜訊污染的X_t,要求模型幫我們預測出是什麼雜訊污染他,也就是預測eps:

公式

圖片

n_steps = 100
beta = torch.linspace(0.0001, 0.04, n_steps)

def q_xt_xtminus1(xtm1, t):
  # gat
  mean = gather(1. - beta, t) ** 0.5 * xtm1 # √(1−βt)*xtm1
  var = gather(beta, t) # βt I
  eps = torch.randn_like(xtm1) # Noise shaped like xtm1
  return mean + (var ** 0.5) * eps


def gather(consts: torch.Tensor, t: torch.Tensor):
    """
    Gather consts for $t$ and reshape to feature map shape
    用來提取一個tensor中第t個值,並錢reshape
    
    """
    c = consts.gather(-1, t)
    return c.reshape(-1, 1, 1, 1)

Reverse process (逆擴散過程)

上面說到模型需要給定圖片x_t,預測雜訊eps,實際上是要讓模型學會產生eps的常態分配的平均值和標準差。

有了上面的beta_t,可以算出alpha_t = 1 - beta_t, alpha_bar_t = 從 t=0 連乘到t=t,接著按下面步驟算出:

  1. 雜訊的係數 eps_coef = (1-alpha_t )/ √(1-alpha_bar_t)
  2. 平均mean = (1/(√alpha_t))*(x_t - eps_coef * noise)
  3. var: 就是beta_t
  4. 新的雜訊eps: 用torch生成隨機數
  5. 新的圖片 main + √var*eps

公式:

圖片

程式碼:

# Set up some parameters
n_steps = 100
beta = torch.linspace(0.0001, 0.04, n_steps).cuda()
alpha = 1. - beta
alpha_bar = torch.cumprod(alpha, dim=0)

def p_xt(xt, noise, t):
  """
  x_t:被污染的圖片
  noise:模型看著x_t預測出來的雜訊
  t:現在是第幾步驟
  """
  alpha_t = gather(alpha, t)
  alpha_bar_t = gather(alpha_bar, t)
  eps_coef = (1 - alpha_t) / (1 - alpha_bar_t) ** .5
  mean = 1 / (alpha_t ** 0.5) * (xt - eps_coef * noise) # Note minus sign
  var = gather(beta, t)
  eps = torch.randn(xt.shape, device=xt.device)
  return mean + (var ** 0.5) * eps 

Algorithm 演算法

image

Loss Function

學習的時候會使用下面這個loss function來計算,其實就是算出污染x_t圖片的雜訊eps_t與模型預測的雜訊pred_noise_t平方差的平方 (mse)

圖片

Training

重複以下4個步驟直到loss降低 1.訓練時先從我們想訓練的圖片集(q(x_theata)中挑出一張 x_theata 2.設定總訓練步驟數大T(也就是上面設定的n_step),中隨機挑朱一個步驟做 3.先用常態分佈隨機出一個雜訊eps 4.用上面的loss function算出eps與模型預測的雜訊pred_noise的平方差的平方,然後做gradient descent

Algorithm 1 Training
1: repeat
2:   x0 ~ q(x0)
3:   t ~ Uniform({1, ..., T})
4:   ε ~ 𝒩(0, I)
5:   Take gradient descent step on
         ∇θ || ε - εθ(√αtx0 + √1 - αtε, t) ||2
6: until converged
    

Sampling

訓練好之後就可以來產圖了,產圖流程如下:

  1. 先獲得一個雜訊,就做x_T
  2. 執行大T(n_step)次去噪動作,以下為loop
    1. 使用訓練好的model產生預測的雜訊pred_eps
    2. 使用Reverse process (逆擴散過程)的公式,幫圖片降噪,並還要加上一個小雜訊
  3. loop完之後圖片產出
Algorithm 2 Sampling
1: xT ~ 𝒩(0, I)
2: for t = T, ..., 1 do
3:   z ~ 𝒩(0, I) if t > 1, else z = 0
4:   xt-1 = 1/√αt (xt - 1 - αt/√1 - αt εθ(xt, t)) + σtz
5: end for
6: return x0
    

UNet 模型

參考資料

U Net 是用來預測雜訊 pred_eps的模型本體,他可以輸入兩個值:

  1. 用來預測的被污染的圖片x_t
  2. 現在是第幾個步驟t

步驟是

  1. down: 2次的 3*3 convolution 搭配一次2*2 max pool,共執行4次
  2. middle(bottom),兩次的convolution,在這裡要加數步驟t的embedding資訊,模型才會知道現在在第幾步驟
  3. up:2次的 3*3 convolution 搭配一次2*2 up-conv(逆convolution),共執行4次 image

2. Deep Learning

參考資料

名稱的由來

1958年,Frank Rosenblatt 提出 Perceptron 演算法,用於線性分類。1962年,Marvin Minsky 批評其局限性,導致研究熱潮迅速平息。1980年代中期,Multilayer Perceptron(多層感知器)被提出,實際上是多層 Logistic Model 的連接,被稱為 Neural Network(神經網路)。1986年,Rumelhart 等提出 Back-propagation 演算法,可調整神經網路中各單元的權重。然而,隱藏層超過三層時效果不佳,1989年,多數學者認為一層隱藏層足夠。

1999年後,隨著GPU的發展,訓練多層隱藏層變得可能,Multilayer Perceptron 被重新命名為 Deep Learning(深度學習)。2006年,Hinton等人提出使用 Restricted Boltzmann machines (RBM) 來初始化深度學習的權重,吸引了許多研究者的關注。一些人認為使用RBM才算是深度學習,但隨著研究深入,發現不需要RBM也能訓練深度學習模型。因此,Deep Learning 與 Multilayer Perceptron(Neural Network)成為同一種演算法的不同名稱,「深度學習」與「神經網絡」常被交互使用。

深度學習5個組件

1. Activation Function

Activation function 是深度學習中最基本的單位,可以想像它是神經網路中的 一個節點(node),執行最簡單的非線性轉換。 常見的有Sigmoid 和 ReLU等:

Sigmoid 函數:

$$ \sigma(x) = \frac{1}{1 + e^{-x}} $$

ReLU 函數:

$$ \text{ReLU}(x) = \max(0, x) $$

Activation function 可以產生相當於邏輯運算子的效果,舉例來說,如果有四的點,分別為(0,0)、(1,0)、(0,1)、(1,1)。如果要將(0,0)與(1,1)分為一類,(1,0)與(0,1)分為另一類,會發現若單純使用一次方程會不能輕易區分。

但如果我們用以以下兩個Function做轉換

$$ \text{w} = \max(0, x - 0.5 y) $$

$$ \text{z} = \max(0, -0.9x + y) $$ 轉換後的四個點分別為(0,0)、(0.5,0.1)、(1,0)、(0,1),就可以被一次方程分成兩邊。

Image 1

轉換前

Image 2

轉換後

2. 深度學習架構

深度學習架構可以靈活變化以滿足不同需求。這邊使用最基礎的 Fully Connected Feedforward Network 架構,由 Input Layer、Output Layer 和任意數量的 Hidden Layer 組成。

  1. Input Layer:這一層並不是真正的一層,而是表示輸入向量。每個自變數都會輸入到下一層的所有節點中。

  2. Hidden Layer:這層是由任意數量的 Activation Function 組成的,可以有多層。每層的節點數量和使用的 Activation Function 可以不同。例如,第一層可以使用5000個 Sigmoid 函數,而第二層可以使用3000個 ReLU 函數。從 Input Layer 得到的自變數在每個節點會產生一個值,這些值作為新的自變數輸入到下一層。每個 Hidden Layer 的輸出都會輸入到下一層的所有節點中,最終輸出到 Output Layer。

  3. Output Layer:這一層作為分類器,將經過 Hidden Layer 非線性轉換的資料分類。例如,經過多層非線性轉換後,資料點會轉換成容易被分類的型態,找出資料的特徵(Feature),再利用 Output Layer 把資料分類成多個類別。常用的分類函數是 Softmax function,它將 hidden layer 中的資料壓縮到 0 到 1 之間,計算出每個類別的後驗機率,最大值對應的類別即為預測類別。

$$ \text{Softmax}(x_i) = \frac{e^{x_i}}{\sum\limits_{j=1}^N e^{x_j}} $$

  1. Fully Connected Feedforward Network:這是最常見的連接各層的方法。每個節點的輸出都會傳遞到下一層的所有節點,而每個節點也會使用上一層所有節點的輸出。這種結構被稱為 Fully Connected,資料從 Input Layer 傳遞到 Hidden Layer,再傳遞到 Output Layer,呈現順向結構。

這種架構允許任意改變以適應不同的需求,並提供了強大的靈活性和可擴展性,使其成為深度學習領域的基礎。

Image 1

Fully Connected Feedforward Network

3. Loss Function

在深度學習中,評估模型好壞需要使用 Loss Function。Loss Function 通過比較模型預測結果與真實資料來計算一個分數(Loss),分數越低表示預測結果越接近真實資料,模型性能越好。常用的 Loss Function 包括 Cross Entropy,特別是在分類任務中。

Cross Entropy 的優勢在於當模型預測與真實資料相差較大時,微分後的斜率較大,有助於訓練,而平方差公式在相同情況下則較平緩,不利於訓練。Loss Function 必須可微,以便使用 Gradient Descent 找出最符合真實情況的模型。根據不同情況,也可以選擇其他 Loss Function 或自定義 Loss Function。

Cross Entropy

$$ L = -\sum_{i=1}^{N} y_i \log(\hat{y}_i) $$

Image 1

紅色為平方差,黑色為 Cross Entropy

4. Gradient Descent

想像模型中的每個權重 (weight) 代表空間中的一軸,把所有權重帶入 Loss Function 中形成高低起伏的空間,每個點表示一組權重計算出的 Loss。最佳模型擁有最低的 Loss,即整個空間中最底點的位置。計算全部權重組合的 Loss 會耗費大量資源,因此使用 Gradient Descent 來找出答案。

Gradient Descent 從 Loss 組成的波浪中選擇一個點,朝著該點斜率的反方向走一步,重複此步驟直到斜率為0的最底點。要得到此斜率需要把 Loss Function 對每一個權重做偏微分,得到的 值組成一個向量就叫做梯度(Gradient),可用符號∇表示

而為了要求得最低值,需要將原本的權重減去 Gradient,才會朝最低值前進。原先的點差距太遠而非像最低點前進,於是在減 Gradient 時會乘上一個數來限制Gradient 的步伐長度,這個數稱之為 Learning Rate,可用符號 η 表示。以上步驟可表示如下:

$$ w^{(T+1)} = w^{(T)} - \eta \nabla \text{Loss}(w^{(T)}) $$

除了基本的 Gradient Descent,還有許多變形方法,如每次只使用一筆資料的 Stochastic Gradient Descent、處理不均衡自變數的 Adagrad,結合動量概念的 Adam。

5. Backpropagation

參考影片:ML Lecture 7: Backpropagation

Backpropagation是在Deep Leaning Model中計算Gradient的方法,在計算之前我們要先了解微積分的Chain Rule。

Chain Rule:

如果有兩個function如下

$$ \begin{aligned} y &= g(x) \ z &= h(y) \end{aligned} $$

則 $x$ 對 $z$ 的微分如下:

$$ \frac{\mathrm{d}z}{\mathrm{d}x} = \frac{\mathrm{d}z}{\mathrm{d}y} \cdot \frac{\mathrm{d}y}{\mathrm{d}x} $$

又如果有一個值 $s$ 同時影響 $x$ 和 $y$, 而 $x$ 和 $y$ 又影響 $z$ :

$$ \begin{aligned} x &= g(s) \ y &= h(s) \ z &= k(x,y) \end{aligned} $$

則 $s$ 對 $z$ 的微分如下

$$ \frac{\mathrm{d}z}{\mathrm{d}s} = \frac{\partial{z}}{\partial{x}} \cdot \frac{\mathrm{d}x}{\mathrm{d}s} + \frac{\partial{z}}{\partial{y}} \cdot \frac{\mathrm{d}y}{\mathrm{d}s} $$

接著我們先定義深度學習模型會要使用的幾個function:

activation function 用sigmoid function

$$ a = \sigma_{z} $$

填入activation function中的 $z$ 如下

$$ z = x_1w_1 +x_2w_2 +b $$

loss function 用 Cross Entropy

$$ L(\theta) = \sum_{n=1}^{N} C^n(\theta) $$

也可以改寫如下

$$ \frac{\partial L(\theta)}{\partial w} = \sum_{n=1}^{N} \frac{\partial c^n(\theta)}{\partial w} $$

從下圖可以看到完整的function邏輯,input layer傳入參數 $x_1$ , $x_2$ ,並與 $w_1$ , $w_2$ , $b$ 組成 $z$, 將 $z$ 帶入 activation function $\sigma{(z)}$ 產出 $a$, $a$ 再作為下一層layer的參數。

圖片

為了要更新參數 $w$ ,我們需要用 $w$ 對Loss function做偏微分,依照chain rule如下:

$$ \frac{\partial{C}}{\partial{w}} = \frac{\partial{z}}{\partial{w}} \cdot \frac{\partial{C}}{\partial{z}} $$

而我們可以輕易知道 $\frac{\partial{z}}{\partial{w_1}}$ 就是 $x_1$ ,而這個 $x_1$ 從上一層傳過來的,因此也叫做前向傳播。

前向傳播 (Forward Propagation)

$$ \begin{aligned} \frac{\partial{z}}{\partial{w_1}} = x_1 \ \frac{\partial{z}}{\partial{w_2}} = x_2 \end{aligned} $$

而 $\frac{\partial{C}}{\partial{w}}$ 後面的部分是:

$$ \frac{\partial{C}}{\partial{z}} = \frac{\partial{a}}{\partial{z}} \cdot \frac{\partial{C}}{\partial{a}} $$

其中activation function $\sigma(z)$ 的微分已知:

$$ \frac{\partial{a}}{\partial{z}} = \sigma'(z) $$

而 $\frac{\partial{C}}{\partial{a}}$ 又可以繼續如下(因為$a$出來的值又會向下影響到下一層的參數function $z'$ 和 $z''$ )

Chain Rule

$$ \frac{\partial{C}}{\partial{a}} = \frac{\partial{z'}}{\partial{a}} \cdot \frac{\partial{C}}{\partial{z'}} + \frac{\partial{z''}}{\partial{a}} \cdot \frac{\partial{C}}{\partial{z''}} $$

從上面的圖片我們可以知道

$$ \begin{aligned} \frac{\partial{z'}}{\partial{a}} = w_1 \ \frac{\partial{z''}}{\partial{a}} = w_2 \end{aligned} $$

雖然 $\frac{\partial{C}}{\partial{z'}}$ 和 $\frac{\partial{C}}{\partial{z''}}$ 仍然未知,但我們先假設我們知道怎麼算,可以從上面的算是中得出

$$ \frac{\partial{C}}{\partial{z}} = \sigma'(x)\left[w_3 \cdot \frac{\partial{C}}{\partial{z'}} + w_4\cdot \frac{\partial{C}}{\partial{z''}}\right] $$

最後假設 $z'$ 和 $z''$ 下一層就是進入Output layer並產出預測的結果 $y_1$ 和 $y_2$ , 並有相對應的答案(Ground Truth) $\hat{y}_1$ 和 $\hat{y}_2$ ,我們便可以算出 $\frac{\partial{C}}{\partial{z'}}$ 和 $\frac{\partial{C}}{\partial{z''}}$ :

$$ \begin{aligned} \frac{\partial{C}}{\partial{z'}} = \frac{\partial{y_1}}{\partial{z'}} \cdot \frac{\partial{C}}{\partial{y_1}} \ \frac{\partial{C}}{\partial{z''}} = \frac{\partial{y_2}}{\partial{z''}} \cdot \frac{\partial{C}}{\partial{y_2}} \end{aligned} $$

而我們知道了 $\frac{\partial{C}}{\partial{z'}}$ 和 $\frac{\partial{C}}{\partial{z''}}$ 就可以算出 $\frac{\partial{C}}{\partial{z}}$ :

$$ \frac{\partial{C}}{\partial{z}} = \sigma'(x)\left[w_3 \cdot \frac{\partial{y_1}}{\partial{z'}} \cdot \frac{\partial{C}}{\partial{y_1}} + w_4\cdot \frac{\partial{y_2}}{\partial{z''}} \cdot \frac{\partial{C}}{\partial{y_2}}\right] $$

再搭配前向傳播的 $x_1$ 可以組出來:

$$ \frac{\partial{C}}{\partial{w_1}} = x_1 \cdot \frac{\partial{C}}{\partial{z}} \cdot \sigma'(x)\left[w_3 \cdot \frac{\partial{y_1}}{\partial{z'}} \cdot \frac{\partial{C}}{\partial{y_1}} + w_4\cdot \frac{\partial{y_2}}{\partial{z''}} \cdot \frac{\partial{C}}{\partial{y_2}}\right] $$

我們就可以用 $\frac{\partial{C}}{\partial{w_1}}$ 去調整 $w_1$ 了

3. Embedding 與 向量資料庫

參考資料:

Embedding

Embedding 是大型語言模型開發中的一個重要關鍵技術,可以將文字轉成依照向量 (vector)的方式存在,方便輸入到深度學習的模型中。經由特殊Embedding模型embed後的文字,還可以讓類似詞意的文字在向量空間中在一起。Embedding也不一定要限制在詞,也可以針對整個句子的句義抽取出訊息變成 vector (像是Openai 可以提供將句子變成 1,536為度的向量)

One-hot encoding

最簡單的Embedding是 one-hard encoding,就是一個跟字典一樣長的vector,在該詞的位置上標上1,其他都是0。 例如 你好嗎? 就可以變成 , , 的字典,並表示如下:

  1. 你:[1, 0, 0]
  2. 好:[0, 1, 0]
  3. 嗎:[0, 0, 1]

這樣的好處是很好Embedding,壞處是vector會變得太長。

Hugging Face

我們可以用Hugging fac上的模型all-MiniLM-L6-v2來做embed,他會把一整個句子直接抽取成 384為度的向量,如下面所表示

from sentence_transformers import SentenceTransformer

class EmbeddingModel:
    """
    this class is responsible for the embedding model
    """

    def __init__(self, model_name="sentence-transformers/all-MiniLM-L6-v2"):
        self.model: SentenceTransformer = SentenceTransformer(model_name)

    def encode(self, text: list[str]):
        """
        change the texts into embeddings
        """
        return self.model.encode(text)

下面我們將三個句子做Embed

def main():
    embedding_model = EmbeddingModel()
    sentences = ["Hello, World!", "I am Murky!", "I am Happy !"]

    embeddings = embedding_model.encode(sentences)
    print("Dimension: ", len(embeddings[0]))  # 384 個向量
    print("embeddings: ", embeddings)

得到下面結果,三個長度為384的 float list

Dimension:  384

embeddings:  [[-0.03817715  0.03291114 -0.0054594  ... -0.04089032  0.03187141
   0.0181632 ]
 [-0.03684668 -0.03121175  0.11125965 ...  0.00893893 -0.08519208
   0.01065892]
 [-0.03827702 -0.10684672 -0.0039953  ...  0.05886666 -0.01260292
  -0.01910227]]

向量資料庫

向量資料庫可以把Embedding後的vector當作key,並在裡面存放資料。並且可以藉由向量的相似程度來搜尋出相似的資料,主要用在推薦系統,大型語言模型的RAG時使用等。

Qdrant

向量資料庫有很多種,下面介紹Qdrant, Qdrant是開源的Rust語言寫成的向量資料庫,並且可以在本地端用docker架設

docker compose

docker-compose.yml 如下設定

services:
  qdrant:
    image: qdrant/qdrant:v1.6.1
    restart: always
    container_name: qdrant
    ports:
      - "6333:6333"
    volumes:
      - ./qdrant/storage:/qurant/storage
      - ./qdrant/config.yaml:/qurant/config/production.yaml

上面比較重要的是 volumn的設定。

  • /qurant/storage:是設定向量資料庫真實的檔案是要存在你的電腦裡面的哪裡,我就是存在專案資料夾下面的 ./qdrant/storage 資料夾
  • /qurant/config/production.yaml:是設定Qdrand config檔在你的電腦上的真實位置,像是我的就是在專案資料夾下面的 ./qdrant/config.yaml

另外在config.yaml找到 下面這個部份可以設定密碼

service:
  api_key: your_secret_api_key_here

接著在comand line中輸入下面指令就可以啟用了

docker-compose up -d
Qdrand Python SDK

Qdrand有提供Python 的SDK,可使用pip下載

poetry add qdrant-client openai

以下的是完整的程式碼,解說在後面

from app.embedding.model import EmbeddingModel
from qdrant_client import QdrantClient
from qdrant_client.http import models
from qdrant_client.http.models import PointStruct


class QdrantSingleton:
    _instance = None
    _initialized = False

    def __new__(cls, *args, **kwargs):
        if not cls._instance:
            cls._instance = super(QdrantSingleton, cls).__new__(cls)
        return cls._instance

    def __init__(self):
        if not QdrantSingleton._initialized:
            self.qdrant_client = QdrantClient(
                url="http://localhost",
                port=6333,
                # api_key="api_key",
            )
            self.embedding_model = EmbeddingModel()
            QdrantSingleton._initialized = True

    def recreate_collection(self, collection_name: str):
        """
        這個方法會重新建立collection,並設定collection的參數
        https://ithelp.ithome.com.tw/articles/10335513
        """
        # https://python-client.qdrant.tech/qdrant_client.http.models.models#qdrant_client.http.models.models.VectorParams
        # 一個用cosine算距離的向量,長度是384
        vectors_config = models.VectorParams(
            distance=models.Distance.COSINE,
            size=384,
        )

        # m代表每個節點近鄰數量。m值越大,查詢速度越快,但內存和構建時間也會增加。
        # ef_construct這是用於構建圖時的效率參數。較大的ef_construct值會導致更好的查詢品質,但會增加構建時間。代表在構建索引時搜索的節點數量
        hnsw_config = models.HnswConfigDiff(on_disk=True, m=16, ef_construct=100)

        # memmap_threshold是這表示當數據大小超過20000時,將使用內存映射來管理數據,這可以有效地處理大量數據並減少內存使用。
        optimizers_config = models.OptimizersConfigDiff(memmap_threshold=20000)

        self.qdrant_client.recreate_collection(
            collection_name=collection_name,
            vectors_config=vectors_config,
            hnsw_config=hnsw_config,
            optimizers_config=optimizers_config,
        )
        return self.qdrant_client

    def create_collection(self, collection_name: str):
        """
        create collection
        """
        vectors_config = models.VectorParams(
            distance=models.Distance.COSINE,
            size=384,
        )

        hnsw_config = models.HnswConfigDiff(on_disk=True, m=16, ef_construct=100)

        optimizers_config = models.OptimizersConfigDiff(memmap_threshold=20000)
        if not self.qdrant_client.collection_exists(collection_name):
            self.qdrant_client.create_collection(
                collection_name=collection_name,
                vectors_config=vectors_config,
                hnsw_config=hnsw_config,
                optimizers_config=optimizers_config,
            )
        return self.qdrant_client

    def get_embedding(self, text: str) -> list[float]:
        """
        get embedding
        """
        embedding_list = self.embedding_model.encode([text])
        embedding = embedding_list[0]
        embedding_to_float_list = embedding.tolist()
        return embedding_to_float_list

    def upsert_vectors(
        self, vectors: list[list[float]], collection_name: str, data: list
    ):
        """
        upsert vectors
        payload is metadata, can be any data in dict
        """
        for i, vector in enumerate(vectors):
            self.qdrant_client.upsert(
                collection_name=collection_name,
                points=[
                    PointStruct(
                        id=i,
                        vector=vector,
                        payload=data[i],
                    )
                ],
            )
        print("upsert_vectors done")

    def search_for_qdrant(self, text: str, collection_name: str, limit_k: int):
        """
        search for qdrant
        """
        embedding_vector = self.get_embedding(text)
        search_result = self.qdrant_client.search(
            collection_name=collection_name,
            query_vector=embedding_vector,
            limit=limit_k,
            append_payload=True,
        )
        return search_result

這邊看起來叫複雜,但其實這個是python的Singleton的寫法,在呼叫這個class的時候都只會回傳同一個init。最重要的是QdrantClient,這裡要放和Qdrant的連線資訊,另外EmbeddingModel則是上面的Class引入。

class QdrantSingleton:
    _instance = None
    _initialized = False

    def __new__(cls, *args, **kwargs):
        if not cls._instance:
            cls._instance = super(QdrantSingleton, cls).__new__(cls)
        return cls._instance

    def __init__(self):
        if not QdrantSingleton._initialized:
            self.qdrant_client = QdrantClient(
                url="http://localhost",
                port=6333,
                # api_key="api_key",
            )
            self.embedding_model = EmbeddingModel()
            QdrantSingleton._initialized = True

接著我們要在向量資料庫中建立一個collection,collection就像是一般資料庫的table,用來存放我們的資料,由於是練習,我選擇使用recreate_collection,這樣如果呼叫撞名的collection就可以先刪除再重新創一個,如果不想一直刪除也可以選擇上面的create_collection

在這裡我們可以用VectorParams設定向量之間的距離怎麼算,我選擇COSINE,也可以用models.Distance.EUCLID選擇兩點直線距離。

並且VectorParams還可以設定當作key的vector的長度有多長,因為我使用Hugging face的all-MiniLM-L6-v2固定會產生 384 維度的vector,所以寫384

  def recreate_collection(self, collection_name: str):
        """
        這個方法會重新建立collection,並設定collection的參數
        https://ithelp.ithome.com.tw/articles/10335513
        """
        # https://python-client.qdrant.tech/qdrant_client.http.models.models#qdrant_client.http.models.models.VectorParams
        # 一個用cosine算距離的向量,長度是384
        vectors_config = models.VectorParams(
            distance=models.Distance.COSINE,
            size=384,
        )

        # m代表每個節點近鄰數量。m值越大,查詢速度越快,但內存和構建時間也會增加。
        # ef_construct這是用於構建圖時的效率參數。較大的ef_construct值會導致更好的查詢品質,但會增加構建時間。代表在構建索引時搜索的節點數量
        hnsw_config = models.HnswConfigDiff(on_disk=True, m=16, ef_construct=100)

        # memmap_threshold是這表示當數據大小超過20000時,將使用內存映射來管理數據,這可以有效地處理大量數據並減少內存使用。
        optimizers_config = models.OptimizersConfigDiff(memmap_threshold=20000)

        self.qdrant_client.recreate_collection(
            collection_name=collection_name,
            vectors_config=vectors_config,
            hnsw_config=hnsw_config,
            optimizers_config=optimizers_config,
        )
        return self.qdrant_client

呼叫之後進入localhost:6333/dashboard應該可以看到下面的面 image

以下則是利用Hugging face的all-MiniLM-L6-v2將文字產出vector,我設計成一次用一個句子embedding,但要注意回傳的embedding會是 tensor, shape是(1, 384),所以要拿出第0個之後呼叫to_list,轉成float array後才能放入Qdrant

    def get_embedding(self, text: str) -> list[float]:
        """
        get embedding
        """
        embedding_list = self.embedding_model.encode([text])
        embedding = embedding_list[0]
        embedding_to_float_list = embedding.tolist()
        return embedding_to_float_list

接著可以用upsert(其實insert也可以)的方法,把vector當作key, data當作value(在Qdrant裡面叫做payload),data可以是任何型態的資料組成的array,像是python 的dictionary, 會存成json的形式

    def upsert_vectors(
        self, vectors: list[list[float]], collection_name: str, data: list
    ):
        """
        upsert vectors
        payload is metadata, can be any data in dict
        """
        for i, vector in enumerate(vectors):
            self.qdrant_client.upsert(
                collection_name=collection_name,
                points=[
                    PointStruct(
                        id=i,
                        vector=vector,
                        payload=data[i],
                    )
                ],
            )
        print("upsert_vectors done")

最後是查詢,只要提供一段文字,會先進行embed之後,用這個vector去資料庫查詢,並查出最接近的limit_k的資料,然後把裡面的payload拿出來。

    def search_for_qdrant(self, text: str, collection_name: str, limit_k: int):
        """
        search for qdrant
        """
        embedding_vector = self.get_embedding(text)
        search_result = self.qdrant_client.search(
            collection_name=collection_name,
            query_vector=embedding_vector,
            limit=limit_k,
            append_payload=True,
        )
        return search_result
Qdrant實戰演練

以下是我有American Idiots前四句的歌詞,我們把他一個一個embed好,再用embed的vector當作key, 把歌詞的資料當作payload存在向量資料庫

def main():

    # qdrant
    american_idiots = [
        {"id": "1", "lyric": "Don't wanna be an American idiot"},
        {"id": "2", "lyric": "Don't want a nation under the new media"},
        {"id": "3", "lyric": "And can you hear the sound of hysteria?"},
        {"id": "4", "lyric": "The subliminal * America"},
    ]

    qdrant = QdrantSingleton()
    collection_name = "Lyrics"
    qdrant.recreate_collection(collection_name)

    embedding_array = [qdrant.get_embedding(text["lyric"]) for text in american_idiots]

    qdrant.upsert_vectors(embedding_array, collection_name, american_idiots)

localhost:6333/dashboard中可以看到已經存進去了。

image

接著我們用一句話進去搜查,並且設定我們只想要找到最相近的一個值

    query_text = "stupid american"
    search_result = qdrant.search_for_qdrant(query_text, collection_name, limit_k=1)

    print(f"尋找: {query_text}", search_result)

output可以看到最接近的歌詞是"Don't wanna be an American idiot"

尋找: stupid american [ScoredPoint(id=0, version=0, score=0.5378643, payload={'id': '1', 'lyric': "Don't wanna be an American idiot"}, vector=None, shard_key=None)]
murky_avatar

Murky

Software Engineer

To kill the Demon, you need to put a dead witcher into his pocket secretly

Check more from this author

Share to

Back