如何使用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 衫以及任何男士服装，但也可以适用于任何其他基于图片的推荐。