如何用 CLIP 和 Actian VectorAI DB 建立離線視覺缺陷搜尋系統

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.

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.

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, and is_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, and scan_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 = search_by_image(client, encoder, args.image,
                                      args.part_category, args.defect_only, args.limit)
        total = (time.perf_counter() - t0) * 1000

    print(f"  score  {'category':<12} {'defect':<16} path")
    for r in results:
        p = r.payload
        print(f"{r.score:7.4f}  {p['part_category']:<12} {p['defect_type']:<16} {p['image_path']}")
    print(f"\ntotal {total:.1f} ms")


if __name__ == "__main__":
    main()

build_filter() chains must() conditions for part category and defect status, and the database applies them inside the search. COLLECTION has to match the name ingest.py created, or searches fail with a collection-not-found error.

uv run query.py --text "surface scratch on metal component" --part-category metal_nut
uv run query.py --image ./line_capture_0412.png --defect-only

The article also includes an optional FastAPI wrapper, app.py, that loads the encoder at startup and exposes GET /search with q, an optional part_category, and limit. It returns the score, image path, and defect type for each hit.

7. Benchmark it

# benchmark.py
def measure_recall_at_k(client, encoder, collection, test_records, k=5) -> float:
    hits = 0
    for record in test_records:
        vec = encoder.encode_images([record["image_path"]])[0].tolist()
        results = client.points.search(
            collection, vector=vec, limit=k + 1, with_payload=True
        )
        neighbours = [
            r for r in results if r.payload["image_path"] != record["image_path"]
        ][:k]
        if any(r.payload["defect_type"] == record["defect_type"] for r in neighbours):
            hits += 1
    return hits / len(test_records) if test_records else 0.0

Recall@5 counts a hit when at least one of the five nearest neighbors shares the query's defect type. The query image is excluded, because otherwise every query retrieves itself at a similarity of 1.0 and recall reads 100%.

uv run benchmark.py --out benchmark.csv
uv run make_chart.py --csv benchmark.csv --hardware "Ryzen 7 5800X, 32GB RAM, CPU-only"

The benchmark ran 100 queries per corpus size on a Ryzen 7 5800X with 32GB of RAM, CPU only.

FAISS is faster at this scale because it runs in-process with no network hop, while VectorAI DB's gRPC calls add a fixed overhead. Recall stayed at or above 0.974. In exchange, you get metadata filtering, durable writes, and multi-client access.

The article also measured filtered HNSW search directly, using 20,000 vectors and a filter that admits 1% of the corpus.

Getting recall above 0.95 took 64 times the default efSearch and raised query time from 0.12 ms to 2.62 ms. That test used random Gaussian vectors, which is close to a worst case for graph traversal. CLIP embeddings tend to cluster, so expect better recall on your own data.

Where the bottleneck sits

A 512-dimensional float32 vector takes 2KB, so all of MVTec's vectors fit in about 11MB against roughly 5GB of images. Encoding is the constraint. At about 2 images per second on CPU, a 50,000-image archive takes about a weekend. Vectors from different models aren't comparable, so switching models means re-encoding everything.

Wrapping up

You now have an offline defect search system that answers text and photo queries from one index, filters by part and defect status inside the search, and keeps its data on local hardware.

What's next

• Pull the image with docker pull actian/vectorai:latest and try VectorAI DB Community Edition, which is self-hosted and free up to 5,000 vectors.
• Read the VectorAI DB documentation.
• Watch an edge retrieval pipeline run without an internet connection.
• Ask questions in the Actian Discord.