如何使用FastEmbed和Qdrant構建男士時尚服裝推薦系統

推薦系統無處不在，從 Netflix、Spotify 到 Amazon。但是，如果您想構建一個視覺推薦引擎，一個不僅關注標題或標籤，更關注影像的引擎，該怎麼辦呢？在本文中，您將構建一個男士時尚推薦系統。它將使用影像嵌入和 Qdrant 向量資料庫。您將從原始影像資料到即時視覺推薦。

學習目標

• 影像嵌入如何表示視覺內容
• 如何使用 FastEmbed 生成向量
• 如何使用 Qdrant 儲存和搜尋向量
• 如何構建反饋驅動的推薦引擎
• 如何使用 Streamlit 建立簡單的 UI

用例：T恤和Polo衫的視覺推薦

假設使用者點選了一件時尚的Polo衫。您的時尚推薦系統將不再使用產品標籤，而是推薦外觀相似的T恤和Polo衫。它使用圖片本身來做出決策。

讓我們來探索一下具體方法。

步驟 1：理解影像嵌入

什麼是影像嵌入？

影像嵌入是一個向量，它是一個數字列表。這些數字代表影像中的關鍵特徵。兩張相似的影像在向量空間中具有接近的嵌入。這使得系統能夠測量視覺相似性。

例如，兩件不同的 T 恤可能在畫素上看起來不同。但如果它們的顏色、圖案和紋理相似，它們的嵌入就會接近。這對於時尚推薦系統來說是一項至關重要的能力。

嵌入是如何生成的？

大多數嵌入模型都使用深度學習。CNN（卷積神經網路）提取視覺模式。這些模式成為向量的一部分。

在本例中，我們使用 FastEmbed。此處使用的嵌入模型是：Qdrant/Unicom-ViT-B-32

from fastembed import ImageEmbedding
from typing import List
from dotenv import load_dotenv
import os
load_dotenv()
model = ImageEmbedding(os.getenv("IMAGE_EMBEDDING_MODEL"))
def compute_image_embedding(image_paths: List[str]) -> list[float]:
return list(model.embed(image_paths))

此函式接收一個圖片路徑列表，並返回能夠捕捉這些圖片精髓的向量。

步驟 2：獲取資料集

我們使用了一個包含約 2000 張男士時尚圖片的資料集。您可以在 Kaggle 上找到它。以下是載入資料集的方法：

import shutil, os, kagglehub
from dotenv import load_dotenv
load_dotenv()
kaggle_repo = os.getenv("KAGGLE_REPO")
path = kagglehub.dataset_download(kaggle_repo)
target_folder = os.getenv("DATA_PATH")
def getData():
if not os.path.exists(target_folder):
shutil.copytree(path, target_folder)

此指令碼會檢查目標資料夾是否存在。如果不存在，則會將影像複製到該資料夾​​。

步驟 3：使用Qdrant儲存和搜尋向量

一旦有了嵌入，我們就需要儲存和搜尋它們。這時 Qdrant 就派上用場了。它是一個快速且可擴充套件的向量資料庫。

以下是連線到 Qdrant 向量資料庫的方法：

from qdrant_client import QdrantClient
client = QdrantClient(
url=os.getenv("QDRANT_URL"),
api_key=os.getenv("QDRANT_API_KEY"),
)
This is how to insert the images paired with its embedding to a Qdrant collection:
class VectorStore:
def __init__(self, embed_batch: int = 64, upload_batch: int = 32, parallel_uploads: int = 3):
# ... (initializer code omitted for brevity) ...
def insert_images(self, image_paths: List[str]):
def chunked(iterable, size):
for i in range(0, len(iterable), size):
yield iterable[i:i + size]
for batch in chunked(image_paths, self.embed_batch):
embeddings = compute_image_embedding(batch)  # Batch embed
points = [
models.PointStruct(id=str(uuid.uuid4()), vector=emb, payload={"image_path": img})
for emb, img in zip(embeddings, batch)
]
# Batch upload each sub-batch
self.client.upload_points(
collection_name=self.collection_name,
points=points,
batch_size=self.upload_batch,
parallel=self.parallel_uploads,
max_retries=3,
wait=True
)

這段程式碼獲取影像檔案路徑列表，將其批次轉換為嵌入，並將這些嵌入上傳到 Qdrant 集合。它首先檢查該集合是否存在。然後，它會使用執行緒並行處理影像以加快速度。每幅影像都會獲得一個唯一的 ID，並與其嵌入和路徑一起被封裝成一個“點”。然後，這些點會被分塊上傳到 Qdrant。

搜尋相似影像

def search_similar(query_image_path: str, limit: int = 5):
emb_list = compute_image_embedding([query_image_path])
hits = client.search(
collection_name="fashion_images",
query_vector=emb_list[0],
limit=limit
)
return [{"id": h.id, "image_path": h.payload.get("image_path")} for h in hits]

您輸入一張查詢圖片。系統會使用餘弦相似度指標返回視覺上相似的圖片。

步驟 4：建立帶有反饋的推薦引擎

現在我們更進一步。如果使用者喜歡某些圖片，而不喜歡其他圖片，該怎麼辦？時尚推薦系統能從中學習嗎？

是的。Qdrant 允許我們提供正面和負面的反饋。這樣，它就能返回更好、更個性化的結果。

class RecommendationEngine:
def get_recommendations(self, liked_images:List[str], disliked_images:List[str], limit=10):
recommended = client.recommend(
collection_name="fashion_images",
positive=liked_images,
negative=disliked_images,
limit=limit
)
return [{"id": hit.id, "image_path": hit.payload.get("image_path")} for hit in recommended]

此函式的輸入如下：

• liked_images：使用者點贊過的圖片 ID 列表。
• disliked_images：使用者點踩過的圖片 ID 列表。
• limit（可選）：一個整數，指定要返回的最大推薦數量（預設為 10）。

這將使用之前介紹過的嵌入向量相似度返回推薦的服裝。

這讓你的系統能夠快速適應使用者偏好。

步驟 5：使用Streamlit構建UI

我們使用 Streamlit 構建介面。它簡單、快速，並且是用 Python 編寫的。

使用者可以：

• 瀏覽服裝
• 點贊或點踩商品
• 檢視更新、更優質的推薦

以下是 Streamlit 程式碼：

import streamlit as st
from PIL import Image
import os
from src.recommendation.engine import RecommendationEngine
from src.vector_database.vectorstore import VectorStore
from src.data.get_data import getData
# -------------- Config --------------
st.set_page_config(page_title="🧥 Men's Fashion Recommender", layout="wide")
IMAGES_PER_PAGE = 12
# -------------- Ensure Dataset Exists (once) --------------
@st.cache_resource
def initialize_data():
getData()
return VectorStore(), RecommendationEngine()
vector_store, recommendation_engine = initialize_data()
# -------------- Session State Defaults --------------
session_defaults = {
"liked": {},
"disliked": {},
"current_page": 0,
"recommended_images": vector_store.points,
"vector_store": vector_store,
"recommendation_engine": recommendation_engine,
}
for key, value in session_defaults.items():
if key not in st.session_state:
st.session_state[key] = value
# -------------- Sidebar Info --------------
with st.sidebar:
st.title("🧥 Men's Fashion Recommender")
st.markdown("""
**Discover fashion styles that suit your taste.**  
Like 👍 or dislike 👎 outfits and receive AI-powered recommendations tailored to you.
""")
st.markdown("### 📦 Dataset")
st.markdown("""
- Source: [Kaggle – virat164/fashion-database](https://www.kaggle.com/datasets/virat164/fashion-database)  
- ~2,000 fashion images
""")
st.markdown("### 🧠 How It Works")
st.markdown("""
1. Images are embedded into vector space  
2. You provide preferences via Like/Dislike  
3. Qdrant finds visually similar images  
4. Results are updated in real-time
""")
st.markdown("### ⚙️ Technologies")
st.markdown("""
- **Streamlit** UI  
- **Qdrant** vector DB  
- **Python** backend  
- **PIL** for image handling  
- **Kaggle API** for data
""")
st.markdown("---")
# -------------- Core Logic Functions --------------
def get_recommendations(liked_ids, disliked_ids):
return st.session_state.recommendation_engine.get_recommendations(
liked_images=liked_ids,
disliked_images=disliked_ids,
limit=3 * IMAGES_PER_PAGE
)
def refresh_recommendations():
liked_ids = list(st.session_state.liked.keys())
disliked_ids = list(st.session_state.disliked.keys())
st.session_state.recommended_images = get_recommendations(liked_ids, disliked_ids)
# -------------- Display: Selected Preferences --------------
def display_selected_images():
if not st.session_state.liked and not st.session_state.disliked:
return
st.markdown("### 🧍 Your Picks")
cols = st.columns(6)
images = st.session_state.vector_store.points
for i, (img_id, status) in enumerate(
list(st.session_state.liked.items()) + list(st.session_state.disliked.items())
):
img_path = next((img["image_path"] for img in images if img["id"] == img_id), None)
if img_path and os.path.exists(img_path):
with cols[i % 6]:
st.image(img_path, use_container_width=True, caption=f"{img_id} ({status})")
col1, col2 = st.columns(2)
if col1.button("❌ Remove", key=f"remove_{img_id}"):
if status == "liked":
del st.session_state.liked[img_id]
else:
del st.session_state.disliked[img_id]
refresh_recommendations()
st.rerun()
if col2.button("🔁 Switch", key=f"switch_{img_id}"):
if status == "liked":
del st.session_state.liked[img_id]
st.session_state.disliked[img_id] = "disliked"
else:
del st.session_state.disliked[img_id]
st.session_state.liked[img_id] = "liked"
refresh_recommendations()
st.rerun()
# -------------- Display: Recommended Gallery --------------
def display_gallery():
st.markdown("### 🧠 Smart Suggestions")
page = st.session_state.current_page
start_idx = page * IMAGES_PER_PAGE
end_idx = start_idx + IMAGES_PER_PAGE
current_images = st.session_state.recommended_images[start_idx:end_idx]
cols = st.columns(4)
for idx, img in enumerate(current_images):
with cols[idx % 4]:
if os.path.exists(img["image_path"]):
st.image(img["image_path"], use_container_width=True)
else:
st.warning("Image not found")
col1, col2 = st.columns(2)
if col1.button("👍 Like", key=f"like_{img['id']}"):
st.session_state.liked[img["id"]] = "liked"
refresh_recommendations()
st.rerun()
if col2.button("👎 Dislike", key=f"dislike_{img['id']}"):
st.session_state.disliked[img["id"]] = "disliked"
refresh_recommendations()
st.rerun()
# Pagination
col1, _, col3 = st.columns([1, 2, 1])
with col1:
if st.button("⬅️ Previous") and page > 0:
st.session_state.current_page -= 1
st.rerun()
with col3:
if st.button("➡️ Next") and end_idx < len(st.session_state.recommended_images):
st.session_state.current_page += 1
st.rerun()
# -------------- Main Render Pipeline --------------
st.title("🧥 Men's Fashion Recommender")
display_selected_images()
st.divider()
display_gallery()
This UI closes the loop. It turns a function into a usable product.

小結

您剛剛構建了一個完整的時尚推薦系統。它可以識別圖片、理解視覺特徵並提出智慧建議。

使用 FastEmbed、Qdrant 和 Streamlit，您現在擁有一個強大的推薦系統。它適用於 T 恤、Polo 衫以及任何男士服裝，但也可以適用於任何其他基於圖片的推薦。