Vector Search for Construction

Overview

Based on DDC methodology (Chapter 4.4), this skill implements semantic vector search for construction data. Move beyond keyword matching - find documents and data by meaning, not just words.

Book Reference: "Современные технологии работы с данными" / "Modern Data Technologies"

"Векторные базы данных позволяют находить семантически похожие документы, даже если они используют разную терминологию." — DDC Book, Chapter 4.4

Quick Start

from sentence_transformers import SentenceTransformer
from qdrant_client import QdrantClient
from qdrant_client.models import VectorParams, Distance, PointStruct

# Initialize embedding model
model = SentenceTransformer('all-MiniLM-L6-v2')

# Create Qdrant client (in-memory for demo)
client = QdrantClient(":memory:")

# Create collection
client.create_collection(
    collection_name="construction_docs",
    vectors_config=VectorParams(size=384, distance=Distance.COSINE)
)

# Sample construction documents
documents = [
    "Concrete mix design for C30 grade with water-cement ratio 0.45",
    "Steel reinforcement specifications for structural columns",
    "Waterproofing membrane installation for basement walls",
    "Fire-rated door specifications for escape routes"
]

# Index documents
for idx, doc in enumerate(documents):
    embedding = model.encode(doc).tolist()
    client.upsert(
        collection_name="construction_docs",
        points=[PointStruct(id=idx, vector=embedding, payload={"text": doc})]
    )

# Search
query = "basement moisture protection"
query_vector = model.encode(query).tolist()
results = client.search(
    collection_name="construction_docs",
    query_vector=query_vector,
    limit=3
)

for result in results:
    print(f"Score: {result.score:.3f} - {result.payload['text']}")

Vector Database Setup

Qdrant Setup

from qdrant_client import QdrantClient
from qdrant_client.models import (
    VectorParams, Distance, PointStruct,
    Filter, FieldCondition, MatchValue
)
import uuid

class ConstructionVectorDB:
    """Vector database for construction documents and data"""

    def __init__(self, host="localhost", port=6333, in_memory=False):
        if in_memory:
            self.client = QdrantClient(":memory:")
        else:
            self.client = QdrantClient(host=host, port=port)

        self.model = SentenceTransformer('all-MiniLM-L6-v2')
        self.collections = {}

    def create_collection(self, name, description=None):
        """Create a new collection"""
        self.client.create_collection(
            collection_name=name,
            vectors_config=VectorParams(
                size=384,  # Dimension for all-MiniLM-L6-v2
                distance=Distance.COSINE
            )
        )
        self.collections[name] = description

    def index_documents(self, collection_name, documents, metadata=None):
        """Index documents with embeddings"""
        points = []

        for idx, doc in enumerate(documents):
            embedding = self.model.encode(doc).tolist()

            payload = {"text": doc}
            if metadata and idx < len(metadata):
                payload.update(metadata[idx])

            points.append(PointStruct(
                id=str(uuid.uuid4()),
                vector=embedding,
                payload=payload
            ))

        self.client.upsert(
            collection_name=collection_name,
            points=points
        )

        return len(points)

    def search(self, collection_name, query, limit=5, filters=None):
        """Semantic search"""
        query_vector = self.model.encode(query).tolist()

        search_filter = None
        if filters:
            conditions = [
                FieldCondition(key=k, match=MatchValue(value=v))
                for k, v in filters.items()
            ]
            search_filter = Filter(must=conditions)

        results = self.client.search(
            collection_name=collection_name,
            query_vector=query_vector,
            limit=limit,
            query_filter=search_filter
        )

        return [
            {
                'score': r.score,
                'text': r.payload.get('text'),
                'metadata': {k: v for k, v in r.payload.items() if k != 'text'}
            }
            for r in results
        ]

    def hybrid_search(self, collection_name, query, keyword_filter=None, limit=5):
        """Combine semantic search with keyword filtering"""
        # First semantic search
        semantic_results = self.search(collection_name, query, limit=limit*2)

        # Then keyword filter if provided
        if keyword_filter:
            filtered = [
                r for r in semantic_results
                if keyword_filter.lower() in r['text'].lower()
            ]
            return filtered[:limit]

        return semantic_results[:limit]

ChromaDB Alternative

import chromadb
from chromadb.utils import embedding_functions

class ChromaConstructionDB:
    """ChromaDB-based vector search for construction"""

    def __init__(self, persist_directory=None):
        if persist_directory:
            self.client = chromadb.PersistentClient(path=persist_directory)
        else:
            self.client = chromadb.Client()

        self.embedding_fn = embedding_functions.SentenceTransformerEmbeddingFunction(
            model_name="all-MiniLM-L6-v2"
        )

    def create_collection(self, name):
        """Create or get collection"""
        return self.client.get_or_create_collection(
            name=name,
            embedding_function=self.embedding_fn
        )

    def index_specifications(self, collection_name, specs):
        """Index construction specifications"""
        collection = self.create_collection(collection_name)

        ids = [f"spec_{i}" for i in range(len(specs))]
        documents = [s['text'] for s in specs]
        metadatas = [{k: v for k, v in s.items() if k != 'text'} for s in specs]

        collection.add(
            ids=ids,
            documents=documents,
            metadatas=metadatas
        )

        return len(specs)

    def search(self, collection_name, query, n_results=5, where=None):
        """Search specifications"""
        collection = self.create_collection(collection_name)

        results = collection.query(
            query_texts=[query],
            n_results=n_results,
            where=where
        )

        return [
            {
                'id': results['ids'][0][i],
                'text': results['documents'][0][i],
                'metadata': results['metadatas'][0][i] if results['metadatas'] else {},
                'distance': results['distances'][0][i] if results['distances'] else None
            }
            for i in range(len(results['ids'][0]))
        ]

Construction-Specific Applications

Specification Search

class SpecificationSearchEngine:
    """Search engine for construction specifications"""

    def __init__(self, db: ConstructionVectorDB):
        self.db = db
        self.collection = "specifications"

    def index_specifications(self, specs_df):
        """Index specifications from DataFrame"""
        self.db.create_collection(self.collection, "Construction specifications")

        documents = specs_df['description'].tolist()
        metadata = specs_df.drop('description', axis=1).to_dict('records')

        return self.db.index_documents(self.collection, documents, metadata)

    def find_similar_specs(self, query, category=None, limit=5):
        """Find similar specifications"""
        filters = {'category': category} if category else None
        return self.db.search(self.collection, query, limit=limit, filters=filters)

    def find_related_materials(self, material_name, limit=10):
        """Find specifications related to a material"""
        query = f"specifications for {material_name} materials"
        return self.db.search(self.col