vector-search

name	vector-search
description	Implement semantic vector search for construction data. Build AI-powered search using embeddings and vector databases (Qdrant, ChromaDB) for intelligent querying of specifications, standards, and project documents.
homepage	https://datadrivenconstruction.io
metadata	{"openclaw":{"emoji":"🔢","os":["darwin","linux","win32"],"homepage":"https://datadrivenconstruction.io","requires":{"bins":"[Truncated]","env":"[Truncated]"},"primaryEnv":"OPENAI_API_KEY"}}

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 = {: doc}
             metadata  idx < (metadata):
                payload.update(metadata[idx])

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

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

         (points)

     ():
        
        query_vector = .model.encode(query).tolist()

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

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

         [
            {
                : r.score,
                : r.payload.get(),
                : {k: v  k, v  r.payload.items()  k != }
            }
             r  results
        ]

     ():
        
        
        semantic_results = .search(collection_name, query, limit=limit*)

        
         keyword_filter:
            filtered = [
                r  r  semantic_results
                 keyword_filter.lower()  r[].lower()
            ]
             filtered[:limit]

         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  specs]

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

         (specs)

     ():
        
        collection = .create_collection(collection_name)

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

         [
            {
                : results[][][i],
                : results[][][i],
                : results[][][i]  results[]  {},
                : results[][][i]  results[]  
            }
             i  ((results[][]))
        ]

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"
         .db.search(.collection, query, limit=limit)

     ():
        
        query = 
         .db.search(.collection, query, limit=limit)

Standards and Codes Search

class StandardsSearchEngine:
    """Search engine for building standards and codes"""

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

    def index_standards(self, standards):
        """Index building standards

        Args:
            standards: List of dicts with 'code', 'title', 'section', 'text'
        """
        self.db.create_collection(self.collection, "Building standards and codes")

        documents = [s['text'] for s in standards]
        metadata = [{k: v for k, v in s.items() if k != 'text'} for s in standards]

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

    def find_applicable_standards(self, context, limit=5):
        """Find standards applicable to a given context"""
        return self.db.search(self.collection, context, limit=limit)

    def search_fire_codes(self, query):
        """Search fire safety codes"""
        full_query = f"fire safety code requirement: {query}"
        return .db.search(
            .collection,
            full_query,
            limit=,
            filters={: }
        )

     ():
        
        full_query = 
         .db.search(
            .collection,
            full_query,
            limit=,
            filters={: }
        )

Work Item Search (OpenConstructionEstimate)

class WorkItemSearchEngine:
    """Search engine for construction work items and unit prices"""

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

    def index_work_items(self, items_df):
        """Index work items database

        Args:
            items_df: DataFrame with columns:
                - code: Work item code
                - description: Work description
                - unit: Unit of measure
                - unit_price: Price per unit
                - category: Work category
        """
        self.db.create_collection(self.collection, "Construction work items")

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

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

    def find_similar_work(self, description, limit=10):
        """Find similar work items by description"""
        results = self.db.search(self.collection, description, limit=limit)

        return [
            {
                'description': r['text'],
                'code': r['metadata'].get('code'),
                'unit': r['metadata'].get('unit'),
                : r[].get(),
                : r[]
            }
             r  results
        ]

     ():
        
        matches = .find_similar_work(work_description, limit=)

          matches:
             

        best_match = matches[]
        unit_price = best_match.get(, )

         {
            : best_match[],
            : best_match[],
            : best_match[],
            : unit_price,
            : quantity,
            : unit_price * quantity,
            : best_match[]
        }

RAG for Construction

Retrieval Augmented Generation

from openai import OpenAI

class ConstructionRAG:
    """RAG system for construction queries"""

    def __init__(self, vector_db: ConstructionVectorDB, openai_client=None):
        self.db = vector_db
        self.llm = openai_client or OpenAI()

    def answer_query(self, query, collection, n_context=5):
        """Answer query using RAG"""
        # Retrieve relevant context
        context_docs = self.db.search(collection, query, limit=n_context)

        # Build context string
        context = "\n\n".join([
            f"Document {i+1}:\n{doc['text']}"
            for i, doc in enumerate(context_docs)
        ])

        # Generate answer
        prompt = f"""Based on the following construction documents, answer the query.

Context:
{context}

Query: {query}

Provide a detailed, accurate answer based only on the provided context.
If the context doesn't contain enough information, say so."""

        response = self.llm.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": "You are a construction industry expert."},
                {"role": "user", "content": prompt}
            ]
        )

         {
            : response.choices[].message.content,
            : context_docs,
            : query
        }

     ():
        
        docs = .db.search(collection, topic, limit=)

        context = .join([doc[]  doc  docs])

        prompt = 

        response = .llm.chat.completions.create(
            model=,
            messages=[{: , : prompt}]
        )

         {
            : response.choices[].message.content,
            : (docs)
        }

Document Indexing Pipeline

import os
import pdfplumber
from typing import List, Dict

class DocumentIndexingPipeline:
    """Pipeline for indexing construction documents"""

    def __init__(self, vector_db: ConstructionVectorDB):
        self.db = vector_db
        self.chunk_size = 500
        self.chunk_overlap = 50

    def chunk_text(self, text: str) -> List[str]:
        """Split text into chunks"""
        words = text.split()
        chunks = []

        for i in range(0, len(words), self.chunk_size - self.chunk_overlap):
            chunk = ' '.join(words[i:i + self.chunk_size])
            if len(chunk) > 50:  # Skip very small chunks
                chunks.append(chunk)

        return chunks

    def extract_pdf_text(self, pdf_path: str) -> str:
        """Extract text from PDF"""
        text = ""
        with pdfplumber.open(pdf_path)  pdf:
             page  pdf.pages:
                page_text = page.extract_text()
                 page_text:
                    text += page_text + 
         text

     ():
        
        
         file_path.endswith():
            text = .extract_pdf_text(file_path)
        :
             (file_path, , encoding=)  f:
                text = f.read()

        
        chunks = .chunk_text(text)

        
        base_metadata = metadata  {}
        base_metadata[] = os.path.basename(file_path)

        chunk_metadata = [
            {**base_metadata, : i}
             i  ((chunks))
        ]

        
         .db.index_documents(collection, chunks, chunk_metadata)

     ():
        
         extensions  :
            extensions = [, , ]

        total_indexed = 

         root, _, files  os.walk(directory):
             file  files:
                 (file.endswith(ext)  ext  extensions):
                    file_path = os.path.join(root, file)
                    :
                        count = .index_document(file_path, collection)
                        total_indexed += count
                        ()
                     Exception  e:
                        ()

         total_indexed

Quick Reference

Component	Description	Use Case
Qdrant	High-performance vector DB	Production deployments
ChromaDB	Simple embedded vector DB	Development/testing
SentenceTransformers	Embedding models	Text to vectors
RAG	Retrieval + Generation	Q&A over documents

Embedding Models for Construction

# Recommended models by use case
EMBEDDING_MODELS = {
    'general': 'all-MiniLM-L6-v2',           # Fast, 384 dim
    'multilingual': 'paraphrase-multilingual-MiniLM-L12-v2',  # Multi-language
    'quality': 'all-mpnet-base-v2',          # Better quality, 768 dim
    'construction': 'allenai/scibert_scivocab_uncased'  # Technical texts
}

Resources

• Book: "Data-Driven Construction" by Artem Boiko, Chapter 4.4
• Website: https://datadrivenconstruction.io
• Qdrant: https://qdrant.tech
• ChromaDB: https://www.trychroma.com
• SentenceTransformers: https://www.sbert.net

Next Steps

• See llm-data-automation for LLM integration
• See document-classification-nlp for document categorization
• See rag-construction for RAG applications