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