Visual product search with LangChain and multi-modal models

Visual product search lets shoppers upload a photo and find similar items in your catalog. The core problem: turning images and text into comparable vectors, then retrieving nearest neighbors fast enough for a product page. This tutorial builds a complete pipeline using LangChain’s multi-modal abstractions, OpenAI’s CLIP embeddings, and a local vector store you can swap for production infrastructure.

Prerequisites

Python 3.10+
An OpenAI API key (for CLIP embeddings via text-embedding-3-small + vision, or use openai/clip-vit-base-patch32 through a compatible endpoint)
A small product image dataset — 50-200 images with metadata (title, price, category) in a CSV
2 GB free disk space for the vector index

Install dependencies:

pip install langchain langchain-openai langchain-community chromadb pillow pandas tqdm

If you prefer a managed embedding endpoint that handles fallback across providers, n4n.ai exposes CLIP-compatible models under the same OpenAI-compatible interface — swap the base URL and key, no code changes.

Architecture overview

┌─────────────┐     ┌──────────────┐     ┌─────────────┐
│  Product    │────▶│  CLIP Image  │────▶│  Chroma     │
│  images +   │     │  Embedder    │     │  Vector DB  │
│  metadata   │     └──────────────┘     └──────┬──────┘
└─────────────┘                                  │
                                                 │  ┌──────────────┐
┌─────────────┐                                  └─▶│  Similarity  │
│  User query │     ┌──────────────┐                │  Search (k)  │
│  (image or  │────▶│  CLIP Text/  │───────────────▶│              │
│   text)     │     │  Image Embed │                └──────┬───────┘
└─────────────┘     └──────────────┘                       │
                                                        ▼
                                                 ┌─────────────┐
                                                 │  Ranked     │
                                                 │  Results    │
                                                 └─────────────┘

Two embedding paths converge in the same vector space: product images at ingest time, user queries (image or text) at search time. CLIP’s joint training makes this work without fine-tuning.

Project structure

visual-search/
├── data/
│   ├── images/           # product photos
│   └── products.csv      # id,title,price,category,image_path
├── ingest.py             # builds the vector index
├── search.py             # query interface
├── config.py             # shared settings
└── requirements.txt

Create config.py first — keeps credentials and knobs in one place:

# config.py
import os
from pathlib import Path

BASE_DIR = Path(__file__).parent
DATA_DIR = BASE_DIR / "data"
IMAGES_DIR = DATA_DIR / "images"
CSV_PATH = DATA_DIR / "products.csv"
CHROMA_DIR = BASE_DIR / "chroma_db"

# Embedding model — must support both image and text
EMBEDDING_MODEL = "openai/clip-vit-base-patch32"  # or "text-embedding-3-small" with vision
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
OPENAI_BASE_URL = os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1")

# Retrieval
TOP_K = 12
SIMILARITY_THRESHOLD = 0.22  # cosine; tune per catalog

Ingest: embed and index product images

ingest.py loads the CSV, embeds each image with CLIP, and upserts into Chroma with metadata attached. We use LangChain’s OpenCLIPEmbeddings wrapper — it handles image preprocessing and batching.

# ingest.py
import os
import uuid
from pathlib import Path
from typing import List, Dict, Any

import pandas as pd
from PIL import Image
from tqdm import tqdm
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_core.documents import Document
from config import (
    CSV_PATH, IMAGES_DIR, CHROMA_DIR, EMBEDDING_MODEL,
    OPENAI_API_KEY, OPENAI_BASE_URL
)

def load_catalog() -> pd.DataFrame:
    df = pd.read_csv(CSV_PATH)
    # Validate image paths exist
    df["full_path"] = df["image_path"].apply(lambda p: IMAGES_DIR / p)
    missing = df[~df["full_path"].apply(lambda p: p.exists())]
    if not missing.empty:
        print(f"Warning: {len(missing)} images not found, skipping")
        df = df[df["full_path"].apply(lambda p: p.exists())].copy()
    return df

def build_documents(df: pd.DataFrame) -> List[Document]:
    docs = []
    for _, row in df.iterrows():
        # CLIP expects RGB
        img = Image.open(row["full_path"]).convert("RGB")
        
        # Metadata stored alongside vector for filtering/display
        metadata = {
            "product_id": str(row["id"]),
            "title": row["title"],
            "price": float(row["price"]),
            "category": row["category"],
            "image_path": str(row["full_path"]),
        }
        
        # LangChain's OpenAIEmbeddings with vision support
        # stores the image bytes in the document for embedding
        docs.append(Document(
            page_content="",  # empty — we embed the image, not text
            metadata=metadata | {"_image": img}
        ))
    return docs

def main():
    os.makedirs(CHROMA_DIR, exist_ok=True)
    
    embeddings = OpenAIEmbeddings(
        model=EMBEDDING_MODEL,
        api_key=OPENAI_API_KEY,
        base_url=OPENAI_BASE_URL,
        # Required for vision models
        dimensions=512,  # CLIP ViT-B/32 output dim
    )
    
    print("Loading catalog...")
    df = load_catalog()
    print(f"Found {len(df)} products with valid images")
    
    print("Building documents...")
    docs = build_documents(df)
    
    print("Embedding and indexing (this takes a minute)...")
    # Chroma.from_documents embeds in batches automatically
    vectorstore = Chroma.from_documents(
        documents=docs,
        embedding=embeddings,
        persist_directory=str(CHROMA_DIR),
        collection_name="products",
        # Chroma uses cosine by default; can override with collection_metadata
    )
    
    print(f"Indexed {vectorstore._collection.count()} vectors to {CHROMA_DIR}")

if __name__ == "__main__":
    main()

Run it:

python ingest.py

Expected output:

Loading catalog...
Found 147 products with valid images
Building documents...
Embedding and indexing (this takes a minute)...
Indexed 147 vectors to /path/to/visual-search/chroma_db

Checkpoint: Open the Chroma SQLite file and verify the collection exists:

sqlite3 chroma_db/chroma.sqlite3 "SELECT name, count FROM collections;"
# products|147

Search: query by image or text

search.py implements two query paths. Both produce a 512-d vector that lives in the same space as the product embeddings.

# search.py
import os
import sys
from pathlib import Path
from typing import List, Dict, Any, Optional
from PIL import Image

from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from config import (
    CHROMA_DIR, EMBEDDING_MODEL,
    OPENAI_API_KEY, OPENAI_BASE_URL,
    TOP_K, SIMILARITY_THRESHOLD
)

class VisualProductSearch:
    def __init__(self):
        self.embeddings = OpenAIEmbeddings(
            model=EMBEDDING_MODEL,
            api_key=OPENAI_API_KEY,
            base_url=OPENAI_BASE_URL,
            dimensions=512,
        )
        self.vectorstore = Chroma(
            persist_directory=str(CHROMA_DIR),
            embedding_function=self.embeddings,
            collection_name="products",
        )
    
    def embed_image(self, image_path: str) -> List[float]:
        """Embed a query image using the same CLIP model."""
        img = Image.open(image_path).convert("RGB")
        # OpenAIEmbeddings.embed_image is not exposed directly;
        # we use the underlying client via embed_documents with image input
        # For OpenAI-compatible vision endpoints, this works:
        return self.embeddings.embed_image(img)
    
    def embed_text(self, text: str) -> List[float]:
        """Embed a text query (e.g., 'red leather ankle boots')."""
        return self.embeddings.embed_query(text)
    
    def search_by_image(self, image_path: str, k: int = TOP_K) -> List[Dict[str, Any]]:
        query_vector = self.embed_image(image_path)
        return self._search_vector(query_vector, k)
    
    def search_by_text(self, text: str, k: int = TOP_K) -> List[Dict[str, Any]]:
        query_vector = self.embed_text(text)
        return self._search_vector(query_vector, k)
    
    def _search_vector(self, vector: List[float], k: int) -> List[Dict[str, Any]]:
        # Chroma similarity_search_by_vector returns Documents with scores
        results = self.vectorstore.similarity_search_by_vector(
            embedding=vector,
            k=k,
        )
        
        formatted = []
        for doc in results:
            meta = doc.metadata
            # Chroma returns cosine distance; convert to similarity
            # Note: actual score extraction depends on Chroma version
            formatted.append({
                "product_id": meta.get("product_id"),
                "title": meta.get("title"),
                "price": meta.get("price"),
                "category": meta.get("category"),
                "image_path": meta.get("image_path"),
                # Score approximation — replace with actual if available
            })
        return formatted


def print_results(results: List[Dict[str, Any]], query_label: str):
    print(f"\n=== Top {len(results)} results for {query_label} ===")
    for i, r in enumerate(results, 1):
        print(f"{i:2d}. [{r['category']}] {r['title']} — ${r['price']:.2f}")
        print(f"     ID: {r['product_id']} | Image: {r['image_path']}")


def main():
    if len(sys.argv) < 2:
        print("Usage: python search.py <image_path> | --text 'query string'")
        sys.exit(1)
    
    searcher = VisualProductSearch()
    
    if sys.argv[1] == "--text":
        query = " ".join(sys.argv[2:])
        results = searcher.search_by_text(query)
        print_results(results, f"text: '{query}'")
    else:
        image_path = sys.argv[1]
        if not Path(image_path).exists():
            print(f"File not found: {image_path}")
            sys.exit(1)
        results = searcher.search_by_image(image_path)
        print_results(results, f"image: {image_path}")


if __name__ == "__main__":
    main()

Test with an image query:

python search.py data/images/query_red_sneaker.jpg

Expected output:

=== Top 12 results for image: data/images/query_red_sneaker.jpg ===
 1. [footwear] Red Nike Air Max 270 — $129.99
     ID: sku-8842 | Image: data/images/sku-8842.jpg
 2. [footwear] Red Adidas Ultraboost 22 — $179.99
     ID: sku-9103 | Image: data/images/sku-9103.jpg
 3. [footwear] Red Puma RS-X — $109.99
     ID: sku-7721 | Image: data/images/sku-7721.jpg
 4. [footwear] White Nike Air Force 1 — $114.99
     ID: sku-5510 | Image: data/images/sku-5510.jpg
...

Test with text:

python search.py --text "black leather crossbody bag under 200"

=== Top 12 results for text: 'black leather crossbody bag under 200' ===
 1. [bags] Black Leather Crossbody — $189.00
     ID: sku-3341 | Image: data/images/sku-3341.jpg
 2. [bags] Midnight Leather Messenger — $195.00
     ID: sku-3399 | Image: data/images/sku-3399.jpg
 3. [bags] Charcoal Vegan Crossbody — $79.00
     ID: sku-3412 | Image: data/images/sku-3412.jpg
...

Adding metadata filters

Real catalogs need filtering — by category, price range, in-stock status. Chroma supports where clauses on metadata. Extend VisualProductSearch:

# Add to search.py
def search_with_filters(
    self,
    vector: List[float],
    k: int = TOP_K,
    category: Optional[str] = None,
    max_price: Optional[float] = None,
    in_stock: bool = True,
) -> List[Dict[str, Any]]:
    where_clause = {}
    if category:
        where_clause["category"] = category
    if max_price:
        where_clause["price"] = {"$lte": max_price}
    if in_stock:
        where_clause["in_stock"] = True  # assumes you added this field at ingest
    
    results = self.vectorstore.similarity_search_by_vector(
        embedding=vector,
        k=k,
        filter=where_clause if where_clause else None,
    )
    return self._format_results(results)

At ingest time, add in_stock to each document’s metadata from your CSV. The filter pushes down to Chroma’s SQLite backend — no post-filtering scan.

Hybrid search: combine visual + textual signals

Pure visual search struggles with attributes CLIP doesn’t capture well (exact material, technical specs). Hybrid search blends image similarity with text-based keyword/BM25 scores. LangChain’s EnsembleRetriever makes this straightforward:

# hybrid_search.py
from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever
from langchain_core.documents import Document

def build_hybrid_retriever(vectorstore, documents: List[Document], weights=(0.6, 0.4)):
    """
    weights: (vector_weight, bm25_weight)
    """
    # Vector retriever (visual + text embedding)
    vector_retriever = vectorstore.as_retriever(
        search_kwargs={"k": TOP_K}
    )
    
    # BM25 retriever on product titles + descriptions
    bm25_retriever = BM25Retriever.from_documents(documents)
    bm25_retriever.k = TOP_K
    
    return EnsembleRetriever(
        retrievers=[vector_retriever, bm25_retriever],
        weights=list(weights),
    )

Use it in search.py:

# In VisualProductSearch.__init__, after loading vectorstore:
self.hybrid_retriever = build_hybrid_retriever(self.vectorstore, self.all_docs)

def hybrid_search(self, query: str, image_path: Optional[str] = None, k: int = TOP_K):
    # Combine text query with optional image — embed both, average vectors
    text_vec = self.embed_text(query)
    if image_path:
        img_vec = self.embed_image(image_path)
        # Simple average; weighted blend works better with tuning
        combined = [(t + i) / 2 for t, i in zip(text_vec, img_vec)]
    else:
        combined = text_vec
    
    # EnsembleRetriever expects a string query for BM25; we bypass for pure vector
    # For true hybrid with image, use vector search + BM25 separately and merge
    results = self.vectorstore.similarity_search_by_vector(combined, k=k)
    return self._format_results(results)

Production considerations

Swap the vector store

Chroma is great for development. For production, replace the Chroma initialization with Pinecone, Weaviate, or pgvector — the LangChain VectorStore interface stays the same:

# Pinecone example
from langchain_pinecone import PineconeVectorStore
import pinecone

pinecone.init(api_key=os.getenv("PINECONE_API_KEY"), environment="us-east-1")
vectorstore = PineconeVectorStore.from_existing_index(
    index_name="product-visual-search",
    embedding=embeddings,
    namespace="products",
)

Batch embedding throughput

The ingest script embeds one image per API call. For catalogs >10k SKUs, batch:

# In ingest.py, replace the Chroma.from_documents call:
batch_size = 32
for i in tqdm(range(0, len(docs), batch_size)):
    batch = docs[i:i+batch_size]
    vectorstore.add_documents(batch)
vectorstore.persist()

Cache embeddings

Re-embedding the same images on every re-index wastes money and time. Store a content hash (SHA256 of image bytes) in metadata; skip if unchanged:

import hashlib

def image_hash(path: Path) -> str:
    return hashlib.sha256(path.read_bytes()).hexdigest()[:16]

# At ingest:
existing = vectorstore.get(include=["metadatas"])
existing_hashes = {m["image_hash"] for m in existing["metadatas"] if "image_hash" in m}

new_docs = []
for _, row in df.iterrows():
    h = image_hash(row["full_path"])
    if h in existing_hashes:
        continue
    # ... build doc with metadata["image_hash"] = h

Latency budget

Target <300ms p95 for the search endpoint. Breakdown:

Stage	Typical latency	Optimization
Image download + decode	50-150ms	CDN, WebP, resize to 224x224 client-side
CLIP embedding (API)	100-250ms	Batch if multiple queries; consider local ONNX runtime
Vector search (Chroma)	5-20ms	HNSW index, keep in memory
Metadata fetch + response	10-30ms	Denormalize in vector store metadata

For sub-100ms visual search, run CLIP locally with onnxruntime or open_clip and keep the vector index in the same process.

Evaluation: does it actually work?

Before shipping, measure recall@k on a labeled test set:

# eval.py
import json
from pathlib import Path
from search import VisualProductSearch

def load_test_set(path: str) -> List[Dict]:
    # Each entry: {"query_image": "path.jpg", "relevant_ids": ["sku-123", "sku-456"]}
    return json.loads(Path(path).read_text())

def recall_at_k(searcher, test_set, k=12):
    hits = 0
    total = 0
    for case in test_set:
        results = searcher.search_by_image(case["query_image"], k=k)
        returned_ids = {r["product_id"] for r in results}
        relevant = set(case["relevant_ids"])
        hits += len(returned_ids & relevant)
        total += len(relevant)
    return hits / total if total else 0.0

if __name__ == "__main__":
    searcher = VisualProductSearch()
    test_set = load_test_set("data/test_queries.json")
    r12 = recall_at_k(searcher, test_set, k=12)
    print(f"Recall@12: {r12:.3f}")

Aim for recall@12 > 0.7 on a diverse test set. If it’s lower, check: image quality, CLIP domain mismatch (fashion vs. furniture), or index size vs. k.

What’s next

Reranking: Feed top-50 candidates to a cross-encoder (text + image) for precision boost
Personalization: Blend user embedding (purchase history) into the query vector
Multi-image queries: “Show me shoes that go with this dress” — embed both, combine vectors
Real-time indexing: Kafka → embedding worker → vector store upsert for new SKUs in seconds

The pipeline you built — ingest, embed, index, search — is the foundation. Every feature above is a layer on top. Start simple, measure, iterate.