打造我們的 RAG 系統的第一步就是要先處理好我們知識的來源：「資料庫」！
今天的內容是要建立一個能用 語意搜尋 的資料庫，也就是 向量資料庫（Vector Database）。
而真正的文本在存進去之前，我們要先做前處理。因為一篇文章太長了，AI 一次吃不完。所以我們得先把它切成一小塊一小塊的片段，我們稱為 chunk。
接著再用 Embedding Model 將 chunk 向量化，最後把這些向量存進資料庫，讓之後的檢索能快速找到最相關的內容。
這次實作的例子，我會用我自己在這次鐵人賽寫的前面 25 篇文章當資料 🤭 大家可以找一份自己的文件來一起實作哦！
那麼話不多說，今天我們要來蓋圖書館啦～～
本系列實作的架構圖：
ithomeNLP_RAG/
├── requirements.txt
├── data/
├── db/ # qdrant vector db
├── indexer.py # store vector
├── retriever.py # vector search
├── reranker.txt # rerank
└── frontend.py # generation + UI
本系列實作會使用到的套件（requirements.txt）：
# 數據處理
numpy==1.26.4
pandas==2.2.3
# 向量化
torch==2.6.0
transformers==4.51.3
sentence-transformers==4.1.0
FlagEmbedding>=0.3.0
huggingface-hub==0.31.1
# LangChain
langchain-community==0.3.23
langchain-core==0.3.59
langchain-huggingface==0.2.0
# 向量資料庫
qdrant-client==1.14.3
# LLM
google-generativeai==0.8.5
# Web UI
streamlit==1.45.0
我們今天會做的是 indexer.py
Chunking 是要把文字切小塊一點，這樣在之後檢索時，才能更精準地找到內容。不過，切太小會讓上下文斷掉，切太大又會讓模型難以理解整體的意思。
所以通常可以設定「重疊區塊（overlap）」，讓每個 chunk 都與前一塊有一部分重疊，避免語意被切開意思就斷掉了。
這裡只是單純的用文長來 chunking。想要更精細一點的話，也可以用段落、句子、標點符號等等的，自然語言上的標記來做切分～
import os
import uuid
from typing import List, Dict
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct
from langchain_community.embeddings import HuggingFaceEmbeddings
CHUNK_SIZE = 500 # 每個 chunk 最多 500 字
CHUNK_OVERLAP = 100 # 與前一塊重疊 100 字
def chunk_text(text, size=CHUNK_SIZE, overlap=CHUNK_OVERLAP):
chunks = []
start = 0
while start < len(text):
end = start + size
chunk = text[start:end]
chunks.append(chunk)
start += size - overlap
return chunks
chunked_docs = []
for doc in docs:
chunks = chunk_text(doc["content"])
for i, c in enumerate(chunks):
chunked_docs.append({
"id": f"{doc['id']}_chunk{i}",
"source": doc['id'],
"chunk_index": i,
"content": c
})
print(f"總共切成 {len(chunked_docs)} 個 chunks")
在完成 chunking 後，下一步就是把文字轉成模型能理解的數字向量（Embedding），並存進向量資料庫。
我們這裡用的工具有兩個：
建立完成後，你就擁有一個可以「語意搜尋」的知識庫啦！
DB_DIR = "./db"
EMBEDDING_MODEL = "BAAI/bge-m3"
COLLECTION_NAME = "ithome_nlp"
class EmbeddingProcessor:
def __init__(self):
# Qdrant 本地 DB
self.client = QdrantClient(path=DB_DIR)
# HuggingFace Embedding
self.embedding = HuggingFaceEmbeddings(model_name=EMBEDDING_MODEL)
# Collection
self.collection_name = COLLECTION_NAME
try:
chunk_info = self.client.get_collection(self.collection_name)
print(f"{self.collection_name} 集合連接成功，包含 {chunk_info.points_count} 條記錄")
except Exception:
self.client.create_collection(
collection_name=self.collection_name,
vectors_config=VectorParams(
size=1024,
distance=Distance.COSINE
)
)
print(f"已創建向量集合: {self.collection_name} ")
def _store_dense_vectors(self, chunks: List[Dict], batch_size=BATCH_SIZE):
"""向量化並儲存到 Qdrant"""
try:
total_batches = (len(chunks) + batch_size - 1) // batch_size
for i in range(0, len(chunks), batch_size):
batch_num = i // batch_size + 1
batch_chunks = chunks[i:i+batch_size]
content_list = [c["content"] for c in batch_chunks]
# 向量化
batch_vectors = self.embedding.embed_documents(content_list)
print("成功向量化 batch")
points = []
# 建立資料點
for j, (chunk_dict, vector) in enumerate(zip(batch_chunks, batch_vectors)):
point_id = self.generate_point_id(chunk_dict["id"], j)
payload = {
"source": chunk_dict["source"],
"chunk_index": chunk_dict["chunk_index"],
"content": chunk_dict["content"],
"vector_type": "dense"
}
point = PointStruct(id=point_id, vector=vector, payload=payload)
points.append(point)
# 將資料點上傳進 db
if points:
self.client.upsert(collection_name=self.collection_name, points=points)
print(f"✅ 已上傳 batch {batch_num}/{total_batches}")
return True
except Exception as e:
print(f"向量化和儲存時發生錯誤: {e}")
return False
def generate_point_id(self, chunk_id: str, index: int) -> str:
"""生成唯一的 point ID"""
point_id_string = f"{chunk_id}_{index}"
return str(uuid.uuid5(uuid.NAMESPACE_DNS, point_id_string))
今天我們完成了 RAG 系統的第一步 🎉 🎉 這一步也是整個 RAG 流程的基礎。有了這個向量資料庫就像是讓我們的 AI 助手有一個可以隨時查找資料的「圖書館」。等到使用者提問時，就可以快速找到最相關的書去整理生成答案！
明天我們就要進入 Retrieval 向量檢索 的環節囉～
iThome鐵人賽
看影片追技術