This article was originally published on the Actian blog.
In this tutorial, you'll build a search system that finds manufacturing defect photos by text description or by example image, filtered by part category and defect status, running entirely on local hardware. You'll need Python with uv, Docker, and about 605MB of disk space for the CLIP model weights.
OpenAI's CLIP turns images and text into comparable 512-dimensional vectors, and Actian VectorAI DB stores those vectors with metadata and searches them. After the one-time model download, nothing leaves the machine.
What the finished system does
A text search for "surface scratch on metal component", filtered to metal nuts, returned these top five results.
| Score | Category | Defect | Image |
|---|---|---|---|
| 0.2814 | metal_nut | scratch | 003.png |
| 0.2691 | metal_nut | scratch | 011.png |
| 0.2607 | metal_nut | color | 007.png |
| 0.2588 | metal_nut | scratch | 000.png |
| 0.2470 | metal_nut | bent | 014.png |
Encoding the query took 18.4 ms and the search took 2.1 ms, for 20.5 ms in total.
A photo search with line_capture_0412.png, limited to defective parts, returned five hazelnut defects.
| Score | Category | Defect | Image |
|---|---|---|---|
| 0.8912 | hazelnut | crack | 006.png |
| 0.8744 | hazelnut | crack | 002.png |
| 0.8593 | hazelnut | 010.png | |
| 0.8471 | hazelnut | hole | 004.png |
| 0.8226 | hazelnut | crack | 013.png |
Encoding took 41.7 ms and the search took 1.9 ms, for 43.6 ms in total. Both runs were on a CPU-only machine. CLIP encoding dominates the time, and the database search stays around 2 ms.
Text scores cluster around 0.26 while image scores cluster around 0.89, so only compare scores within one query mode.
Why use a database over FAISS
FAISS can save and reload an index with write_index, and it can restrict a search to an allowed list of IDs, as described in the wiki. It still leaves three jobs to you in a plant setting.
- Metadata management falls to you. FAISS stores vectors and integer IDs only. You'd keep part categories and defect types in a sidecar store such as SQLite, build category-to-ID maps, and reconcile both after a failed ingestion run.
- Filtered graph search costs recall. HNSW indexes lose recall under selective filters at default settings. The benchmark section below measures that tradeoff.
- Serving takes extra work. FAISS lives inside one Python process. Several inspection stations would each need a copy or a service layer you build yourself.
VectorAI DB stores a metadata payload with each vector, applies filters during the search, and runs as a service. FAISS remains a good fit for quick notebook comparisons. For a broader comparison, read Best Python Vector Database Libraries.
How it works
- The model is
openai/clip-vit-base-patch32, with a ViT-B/32 image encoder and a 12-layer text transformer. Both project into 512 dimensions. - Each stored record carries four payload fields, which are
part_category,defect_type,image_path, andis_defect. - One index answers both text and image queries.
OpenAI's model card says CLIP isn't intended for general deployment and recommends in-domain testing, so measure retrieval quality on your own images.
1. Set up the project
uv init defect-search && cd defect-search uv add torch transformers pillow actian-vectorai-client matplotlib Download the CLIP weights once. The checkpoint is about 605MB and caches in ~/.cache/huggingface.
uv run huggingface-cli download openai/clip-vit-base-patch32 For a fully air-gapped machine, copy ~/.cache/huggingface across and set HF_HUB_OFFLINE=1.
2. Start VectorAI DB
services: vectorai: image: actian/vectorai:latest ports: - "6573:6573" # REST API - "6574:6574" # gRPC API, used by the Python SDK - "6575:6575" # local web UI volumes: - ./vectorai_data:/var/lib/actian-vectorai environment: - ACTIAN_VECTORAI_ACCEPT_EULA=YES restart: unless-stopped Port 6574 is the gRPC API the Python SDK uses. The volume mount is what keeps the index on disk across restarts. See the VectorAI DB Docker setup guide for more options.
docker-compose up -d 
Confirm the database is reachable with smoke_test.py.
# smoke_test.py from importlib import import_module VectorAIClient = import_module("actian_vectorai").VectorAIClient with VectorAIClient("localhost:6574") as client: print(client.health_check()) uv run smoke_test.py Server details mean the database is ready. A connection error means the container isn't up yet.
3. Get a corpus
This tutorial uses MVTec AD, which has 5,354 high-resolution images across 15 categories (5 textures and 10 objects) and more than 70 defect types. Unpack it to ./mvtec_ad.
MVTec AD is licensed under CC BY-NC-SA 4.0 for non-commercial research only. For commercial use, point the loader at your own inspection archive. Any
<part_category>/<split>/<defect_type>/<image>layout works, andscan_corpus()is the only function you'd change.
4. Write the shared encoder
from __future__ import annotations from pathlib import Path from typing import Sequence import torch from PIL import Image from transformers import CLIPModel, CLIPProcessor MODEL_ID = "openai/clip-vit-base-patch32" EMBED_DIM = 512 def _features(out) -> torch.Tensor: """ transformers 4.x returns a Tensor from get_image_features and get_text_features. transformers 5.x returns a BaseModelOutputWithPooling whose pooler_output holds the projected vector. Handle both so this does not break on upgrade. """ return out if isinstance(out, torch.Tensor) else out.pooler_output class ClipEncoder: def __init__(self, model_id: str = MODEL_ID, device: str | None = None) -> None: self.device = device or ("cuda" if torch.cuda.is_available() else "cpu") self.model = CLIPModel.from_pretrained(model_id).to(self.device).eval() self.processor = CLIPProcessor.from_pretrained(model_id) @torch.inference_mode() def encode_images(self, paths: Sequence[str | Path]) -> torch.Tensor: images = [Image.open(p).convert("RGB") for p in paths] inputs = self.processor(images=images, return_tensors="pt").to(self.device) vecs = _features(self.model.get_image_features(**inputs)) return torch.nn.functional.normalize(vecs, dim=-1).cpu() @torch.inference_mode() def encode_text(self, texts: Sequence[str]) -> torch.Tensor: inputs = self.processor( text=list(texts), return_tensors="pt", padding=True, truncation=True ).to(self.device) vecs = _features(self.model.get_text_features(**inputs)) return torch.nn.functional.normalize(vecs, dim=-1).cpu() ClipEncoder encodes images and text into the same 512-dimensional space and uses CUDA when it's available. It normalizes every vector, so cosine similarity becomes a dot product.
The _features() helper handles a breaking change. In transformers 4.x, get_image_features returns a tensor. In 5.x, it returns a BaseModelOutputWithPooling, and the usual one-liner fails with AttributeError: 'BaseModelOutputWithPooling' object has no attribute 'shape'.
5. Index the defect corpus
# ingest.py from __future__ import annotations import argparse import time from pathlib import Path from actian_vectorai import ( VectorAIClient, PointStruct, VectorParams, Distance, CollectionExistsError, ) from clip_encoder import ClipEncoder, EMBED_DIM IMAGE_SUFFIXES = {".png", ".jpg", ".jpeg", ".bmp"} def scan_corpus(root: Path) -> list[dict]: """Read metadata straight out of the directory layout.""" records: list[dict] = [] for path in sorted(root.rglob("*")): if path.suffix.lower() not in IMAGE_SUFFIXES: continue if "ground_truth" in path.parts: # segmentation masks, not photos continue try: rel = path.relative_to(root) part_category, _split, defect_type = rel.parts[0], rel.parts[1], rel.parts[2] except (ValueError, IndexError): continue records.append( { "image_path": str(path), "part_category": part_category, "defect_type": defect_type, "is_defect": defect_type != "good", } ) return records def main() -> None: ap = argparse.ArgumentParser() ap.add_argument("--data", default="./mvtec_ad") ap.add_argument("--collection", default="defect_search") ap.add_argument("--url", default="localhost:6574") ap.add_argument("--batch-size", type=int, default=32) args = ap.parse_args() data_root = Path(args.data) if not data_root.is_dir(): raise SystemExit( f"Dataset directory not found: {data_root}\n" "Download and extract MVTec AD, then pass its directory with --data." ) records = scan_corpus(data_root) if not records: raise SystemExit( f"No dataset images found under {data_root}\n" "Expected paths like <data>/<category>/test/<defect>/*.png." ) print(f"Found {len(records)} images") encoder = ClipEncoder() print(f"CLIP running on {encoder.device}") with VectorAIClient(args.url) as client: try: client.collections.create( args.collection, vectors_config=VectorParams(size=EMBED_DIM, distance=Distance.Cosine), ) print(f"Created collection '{args.collection}'") except CollectionExistsError: print(f"Collection '{args.collection}' already exists, appending") started = time.perf_counter() done = 0 for start in range(0, len(records), args.batch_size): batch = records[start : start + args.batch_size] vectors = encoder.encode_images([r["image_path"] for r in batch]) client.points.upsert( args.collection, points=[ PointStruct(id=start + offset, vector=vec.tolist(), payload=record) for offset, (record, vec) in enumerate(zip(batch, vectors)) ], ) done += len(batch) elapsed = time.perf_counter() - started rate = done / elapsed if elapsed else 0.0 remaining = (len(records) - done) / rate if rate else 0.0 print( f"\r{done}/{len(records)} images " f"| {rate:.1f} img/s | ~{remaining/60:.1f} min left", end="", flush=True, ) print(f"\nIndexed {done} images in {(time.perf_counter()-started)/60:.1f} min") print(f"Collection now holds {client.points.count(args.collection)} points") if __name__ == "__main__": main() scan_corpus() reads each image's metadata from its folder path, skips the ground_truth segmentation masks, and marks is_defect as true for anything outside the good folder. The script creates a 512-dimensional cosine collection and only catches CollectionExistsError, so other failures such as a dimension mismatch still stop the run.
uv run ingest.py --data ./mvtec_ad The sample run found 5,354 images, encoded them on CPU at 2.1 images per second, and finished in 42.5 minutes with 5,354 points in the collection. Expect 45 to 90 minutes on CPU and a few minutes on a modern GPU.
To check persistence, restart the container and run a query.
docker compose restart uv run query.py --text "cracked surface" Results come back immediately with no re-encoding, because the vectors were written to the mounted volume.
6. Run queries
# query.py from __future__ import annotations import argparse import time from actian_vectorai import VectorAIClient, FilterBuilder, Field from clip_encoder import ClipEncoder COLLECTION = "defect_search" # must match the name ingest.py created def build_filter(part_category: str | None = None, defect_only: bool = False): """Metadata filter, applied inside the search rather than after it.""" builder = FilterBuilder() if part_category: builder = builder.must(Field("part_category").eq(part_category)) if defect_only: builder = builder.must(Field("is_defect").eq(True)) return None if builder.is_empty() else builder.build() def search_by_text(client, encoder, text, part_category=None, defect_only=False, limit=5): vector = encoder.encode_text([text])[0].tolist() return client.points.search( COLLECTION, vector, limit=limit, filter=build_filter(part_category, defect_only), with_payload=True, ) def search_by_image(client, encoder, image_path, part_category=None, defect_only=False, limit=5): vector = encoder.encode_images([image_path])[0].tolist() return client.points.search( COLLECTION, vector, limit=limit, filter=build_filter(part_category, defect_only), with_payload=True, ) def main() -> None: ap = argparse.ArgumentParser() ap.add_argument("--text") ap.add_argument("--image") ap.add_argument("--part-category") ap.add_argument("--defect-only", action="store_true") ap.add_argument("--limit", type=int, default=5) args = ap.parse_args() if not args.text and not args.image: raise SystemExit("Pass --text or --image") encoder = ClipEncoder() with VectorAIClient("localhost:6574") as client: t0 = time.perf_counter() if args.text: results = search_by_text(client, encoder, args.text, args.part_category, args.defect_only, args.limit) else: results Source：https://dev.to/actiandev/build-an-offline-visual-defect-search-system-with-clip-and-vectorai-db-4pllRelated posts
Build an AI summarizer with background jobs: retries, signed webhooks and Postgres over one MCP connection
Building a Defensible Inventory for CVE-2026-21932
Central Florida Builds for Defense: What a Local Subcontractor Needs in Place Now
Convert OKLCH to HEX in the Browser: A Colour Picker That Never Uploads
How I Automated My Side Project to Earn While I Sleep Using Claude
