Vector Database Deep Dive: Building a Semantic Search Engine with Weaviate
In 2025, semantic search has evolved from a nice-to-have feature to a fundamental requirement for modern applications. While traditional keyword-based search struggles with context and meaning, vector databases like Weaviate enable true understanding through mathematical representations of meaning. This comprehensive guide explores how to build production-ready semantic search systems using Weaviate, covering everything from vector embeddings and similarity algorithms to hybrid search patterns and real-time updates. Whether you're building an intelligent document retrieval system, product recommendation engine, or AI-powered knowledge base, mastering vector databases will transform how you handle unstructured data.
🚀 Why Vector Databases Dominate AI Applications in 2025
Vector databases have become the backbone of modern AI systems by enabling efficient storage and retrieval of high-dimensional embeddings. Unlike traditional databases that match exact values, vector databases find semantically similar content, making them perfect for AI applications that need to understand context and meaning.
- Semantic Understanding: Find content based on meaning rather than exact keyword matches
- Multi-modal Search: Search across text, images, audio, and video using the same interface
- Real-time Performance: Handle millions of vectors with sub-second query times
- Hybrid Capabilities: Combine vector search with traditional filtering and keyword search
- Scalability: Scale horizontally to handle growing datasets and query loads
🔧 Weaviate Architecture: Understanding the Core Components
Weaviate's modular architecture separates storage, computation, and embedding generation, providing flexibility and performance. Understanding these components is crucial for building efficient systems.
- Vector Index: HNSW algorithm for efficient approximate nearest neighbor search
- Modules System: Pluggable components for embeddings, text processing, and more
- GraphQL Interface: Unified query language for both vector and traditional operations
- Replication & Sharding: Built-in high availability and horizontal scaling
- Multi-tenancy: Isolated data partitions for different applications or customers
💻 Complete Weaviate Setup and Configuration
Let's start with a complete Docker-based Weaviate setup with custom modules and optimized configuration for production use.
# docker-compose.yml - Production Weaviate Setup
version: '3.4'
services:
weaviate:
command:
- --host
- 0.0.0.0
- --port
- '8080'
- --scheme
- http
image: cr.weaviate.io/semitechnologies/weaviate:1.24.0
ports:
- 8080:8080
- 50051:50051
restart: on-failure:0
environment:
OPENAI_APIKEY: ${OPENAI_APIKEY}
QUERY_DEFAULTS_LIMIT: 25
AUTHENTICATION_ANONYMOUS_ACCESS_ENABLED: 'true'
PERSISTENCE_DATA_PATH: '/var/lib/weaviate'
DEFAULT_VECTORIZER_MODULE: 'text2vec-openai'
ENABLE_MODULES: 'text2vec-openai,generative-openai,qna-openai'
CLUSTER_HOSTNAME: 'node1'
LOG_LEVEL: 'info'
MAX_IMPORT_BATCH_SIZE: '100'
MAX_IMPORT_CONCURRENT_REQUESTS: '4'
volumes:
- weaviate_data:/var/lib/weaviate
deploy:
resources:
limits:
memory: 8G
reservations:
memory: 4G
# Optional: Add monitoring with Prometheus
prometheus:
image: prom/prometheus:latest
ports:
- "9090:9090"
volumes:
- ./monitoring/prometheus.yml:/etc/prometheus/prometheus.yml
command:
- '--config.file=/etc/prometheus/prometheus.yml'
- '--storage.tsdb.path=/prometheus'
- '--web.console.libraries=/etc/prometheus/console_libraries'
- '--web.console.templates=/etc/prometheus/consoles'
- '--storage.tsdb.retention.time=200h'
- '--web.enable-lifecycle'
grafana:
image: grafana/grafana:latest
ports:
- "3000:3000"
environment:
- GF_SECURITY_ADMIN_PASSWORD=admin
volumes:
- ./monitoring/grafana/dashboards:/var/lib/grafana/dashboards
- ./monitoring/grafana/provisioning:/etc/grafana/provisioning
depends_on:
- prometheus
volumes:
weaviate_data:
🛠️ Python Client Implementation and Schema Design
Proper schema design is crucial for performance and functionality. Here's how to define classes and properties with optimal vectorization settings.
# weaviate_schema.py - Advanced Schema Design
import weaviate
from weaviate.classes.config import Configure, Property, DataType
from weaviate.classes.init import AdditionalConfig, Timeout
import asyncio
class WeaviateSemanticSearch:
def __init__(self, endpoint="http://localhost:8080"):
self.client = weaviate.WeaviateClient(
additional_config=AdditionalConfig(
timeout=Timeout(init=60, query=300, insert=120),
grpc_port_experimental=50051
)
)
self.client.connect_to_local()
async def create_document_schema(self):
"""Create optimized schema for document search"""
document_class = {
"class": "Document",
"description": "A document for semantic search",
"vectorizer": "text2vec-openai",
"moduleConfig": {
"text2vec-openai": {
"model": "text-embedding-3-large",
"modelVersion": "latest",
"type": "text",
"vectorizeClassName": False
},
"generative-openai": {
"model": "gpt-4"
}
},
"properties": [
{
"name": "title",
"dataType": ["text"],
"description": "Document title",
"moduleConfig": {
"text2vec-openai": {
"skip": False,
"vectorizePropertyName": False
}
}
},
{
"name": "content",
"dataType": ["text"],
"description": "Document content",
"moduleConfig": {
"text2vec-openai": {
"skip": False,
"vectorizePropertyName": False
}
}
},
{
"name": "category",
"dataType": ["text"],
"description": "Document category",
"moduleConfig": {
"text2vec-openai": {
"skip": True,
"vectorizePropertyName": False
}
}
},
{
"name": "tags",
"dataType": ["text[]"],
"description": "Document tags",
"moduleConfig": {
"text2vec-openai": {
"skip": True,
"vectorizePropertyName": False
}
}
},
{
"name": "created_at",
"dataType": ["date"],
"description": "Creation timestamp"
},
{
"name": "updated_at",
"dataType": ["date"],
"description": "Last update timestamp"
},
{
"name": "author",
"dataType": ["text"],
"description": "Document author"
},
{
"name": "word_count",
"dataType": ["int"],
"description": "Number of words in document"
},
{
"name": "readability_score",
"dataType": ["number"],
"description": "Document readability score"
}
],
"vectorIndexType": "hnsw",
"vectorIndexConfig": {
"distance": "cosine",
"ef": 128,
"efConstruction": 128,
"maxConnections": 32,
"cleanupIntervalSeconds": 300,
"dynamicEfMin": 100,
"dynamicEfMax": 500,
"dynamicEfFactor": 8
},
"shardingConfig": {
"desiredCount": 3,
"actualCount": 3,
"virtualPerPhysical": 128,
"key": "_id",
"strategy": "hash",
"function": "murmur3"
},
"replicationConfig": {
"factor": 2,
"asyncEnabled": False
}
}
# Create the class
try:
self.client.collections.create_from_dict(document_class)
print("Document schema created successfully")
except Exception as e:
print(f"Schema creation error: {e}")
async def create_multimodal_schema(self):
"""Create schema for multi-modal search (text + images)"""
multimodal_class = {
"class": "MultimodalContent",
"description": "Content with both text and images",
"vectorizer": "text2vec-openai",
"moduleConfig": {
"text2vec-openai": {
"model": "text-embedding-3-large",
"vectorizeClassName": False
}
},
"properties": [
{
"name": "text_content",
"dataType": ["text"],
"description": "Text content for vectorization"
},
{
"name": "image_url",
"dataType": ["text"],
"description": "URL to image file"
},
{
"name": "image_embedding",
"dataType": ["blob"],
"description": "Pre-computed image embeddings"
},
{
"name": "content_type",
"dataType": ["text"],
"description": "Type of content (text, image, mixed)"
}
]
}
self.client.collections.create_from_dict(multimodal_class)
print("Multimodal schema created successfully")
def get_collection(self, class_name):
"""Get collection with proper configuration"""
return self.client.collections.get(
class_name,
consistency_level=weaviate.classes.config.ConsistencyLevel.QUORUM
)
# Initialize and create schemas
async def main():
search_engine = WeaviateSemanticSearch()
await search_engine.create_document_schema()
await search_engine.create_multimodal_schema()
if __name__ == "__main__":
asyncio.run(main())
🚀 Advanced Data Ingestion and Vectorization
Efficient data ingestion is critical for performance. Here's how to implement batch processing, error handling, and custom vectorization.
# data_ingestion.py - Advanced Data Import
import asyncio
import aiohttp
import pandas as pd
from datetime import datetime
import uuid
from typing import List, Dict, Any
import logging
class WeaviateDataManager:
def __init__(self, client):
self.client = client
self.logger = logging.getLogger(__name__)
async def import_documents_batch(self, documents: List[Dict], batch_size: int = 100):
"""Import documents with batch processing and error handling"""
collection = self.client.collections.get("Document")
successful_imports = 0
failed_imports = 0
for i in range(0, len(documents), batch_size):
batch = documents[i:i + batch_size]
batch_objects = []
for doc in batch:
try:
# Validate required fields
if not doc.get('title') or not doc.get('content'):
self.logger.warning(f"Skipping document missing required fields: {doc.get('id', 'unknown')}")
failed_imports += 1
continue
# Create Weaviate object
weaviate_obj = {
"title": doc['title'],
"content": doc['content'],
"category": doc.get('category', 'general'),
"tags": doc.get('tags', []),
"created_at": doc.get('created_at', datetime.now().isoformat()),
"updated_at": doc.get('updated_at', datetime.now().isoformat()),
"author": doc.get('author', 'unknown'),
"word_count": doc.get('word_count', len(doc['content'].split())),
"readability_score": doc.get('readability_score', 0.0)
}
# Add UUID if provided, otherwise generate
if 'id' in doc:
weaviate_obj['id'] = doc['id']
batch_objects.append(weaviate_obj)
except Exception as e:
self.logger.error(f"Error processing document: {e}")
failed_imports += 1
if batch_objects:
try:
# Insert batch with retry logic
result = await self._insert_with_retry(collection, batch_objects)
successful_imports += len(result.successful)
failed_imports += len(result.failed) if hasattr(result, 'failed') else 0
self.logger.info(f"Batch {i//batch_size + 1}: {len(result.successful)} successful, {len(result.failed) if hasattr(result, 'failed') else 0} failed")
except Exception as e:
self.logger.error(f"Batch insert failed: {e}")
failed_imports += len(batch_objects)
# Rate limiting to avoid overwhelming the server
await asyncio.sleep(0.1)
return {
"successful": successful_imports,
"failed": failed_imports,
"total": len(documents)
}
async def _insert_with_retry(self, collection, objects, max_retries=3):
"""Insert with exponential backoff retry logic"""
for attempt in range(max_retries):
try:
return collection.data.insert_many(objects)
except Exception as e:
if attempt == max_retries - 1:
raise e
wait_time = (2 ** attempt) + 1
self.logger.warning(f"Insert failed, retrying in {wait_time}s: {e}")
await asyncio.sleep(wait_time)
async def import_from_csv(self, csv_path: str, **kwargs):
"""Import documents from CSV file"""
df = pd.read_csv(csv_path)
documents = []
for _, row in df.iterrows():
doc = {
'title': row.get('title', ''),
'content': row.get('content', ''),
'category': row.get('category', 'general'),
'tags': row.get('tags', '').split(';') if pd.notna(row.get('tags')) else [],
'author': row.get('author', 'unknown'),
'word_count': row.get('word_count', 0),
'readability_score': row.get('readability_score', 0.0)
}
# Add timestamp if available
if 'created_at' in row and pd.notna(row['created_at']):
doc['created_at'] = row['created_at']
documents.append(doc)
return await self.import_documents_batch(documents, **kwargs)
async def update_document(self, doc_id: str, updates: Dict[str, Any]):
"""Update existing document with partial updates"""
collection = self.client.collections.get("Document")
try:
# Get existing document
existing = collection.query.fetch_object_by_id(doc_id)
if not existing:
raise ValueError(f"Document {doc_id} not found")
# Merge updates
updated_data = {**existing.properties, **updates}
updated_data['updated_at'] = datetime.now().isoformat()
# Update document
collection.data.update(
uuid=doc_id,
properties=updated_data
)
self.logger.info(f"Document {doc_id} updated successfully")
return True
except Exception as e:
self.logger.error(f"Failed to update document {doc_id}: {e}")
return False
async def delete_documents_by_filter(self, filters: Dict[str, Any]):
"""Delete documents matching filter criteria"""
collection = self.client.collections.get("Document")
try:
# Build GraphQL where filter
where_clause = self._build_where_clause(filters)
# Execute batch delete
result = collection.data.delete_many(where=where_clause)
self.logger.info(f"Deleted {result} documents matching filters")
return result
except Exception as e:
self.logger.error(f"Failed to delete documents: {e}")
return 0
def _build_where_clause(self, filters: Dict[str, Any]) -> Dict[str, Any]:
"""Build GraphQL where clause from filters"""
where_conditions = []
for field, value in filters.items():
if isinstance(value, (list, tuple)):
where_conditions.append({
"path": [field],
"operator": "ContainsAny",
"valueText": value
})
elif isinstance(value, str):
where_conditions.append({
"path": [field],
"operator": "Equal",
"valueText": value
})
elif isinstance(value, (int, float)):
where_conditions.append({
"path": [field],
"operator": "Equal",
"valueNumber": value
})
elif isinstance(value, dict):
# Handle range queries
if 'min' in value and 'max' in value:
where_conditions.append({
"path": [field],
"operator": "And",
"operands": [
{"path": [field], "operator": "GreaterThanEqual", "valueNumber": value['min']},
{"path": [field], "operator": "LessThanEqual", "valueNumber": value['max']}
]
})
if len(where_conditions) == 1:
return where_conditions[0]
else:
return {
"operator": "And",
"operands": where_conditions
}
# Example usage
async def example_import():
from weaviate_schema import WeaviateSemanticSearch
search_engine = WeaviateSemanticSearch()
data_manager = WeaviateDataManager(search_engine.client)
# Sample documents
sample_docs = [
{
"title": "Introduction to Machine Learning",
"content": "Machine learning is a subset of artificial intelligence that enables computers to learn without being explicitly programmed.",
"category": "AI",
"tags": ["machine-learning", "ai", "tutorial"],
"author": "AI Researcher",
"word_count": 150,
"readability_score": 8.5
},
{
"title": "Deep Learning Fundamentals",
"content": "Deep learning uses neural networks with multiple layers to learn complex patterns in large datasets.",
"category": "AI",
"tags": ["deep-learning", "neural-networks", "advanced"],
"author": "ML Engineer",
"word_count": 200,
"readability_score": 7.8
}
]
result = await data_manager.import_documents_batch(sample_docs)
print(f"Import result: {result}")
if __name__ == "__main__":
asyncio.run(example_import())
🔍 Advanced Query Patterns and Semantic Search
Weaviate's GraphQL interface enables powerful query patterns. Here are advanced search techniques for production systems.
# semantic_search.py - Advanced Query Patterns
import weaviate
from weaviate.classes.query import Filter, HybridFusion
from typing import List, Dict, Any, Optional
import numpy as np
class AdvancedSemanticSearch:
def __init__(self, client):
self.client = client
self.collection = client.collections.get("Document")
async def semantic_search(self, query: str, limit: int = 10,
filters: Optional[Dict] = None,
certainty: float = 0.7):
"""Basic semantic search with filters"""
response = self.collection.query.near_text(
query=query,
limit=limit,
filters=self._build_filters(filters) if filters else None,
certainty=certainty,
return_metadata=weaviate.classes.query.MetadataQuery(certainty=True, distance=True)
)
return self._format_results(response.objects)
async def hybrid_search(self, query: str, alpha: float = 0.5,
limit: int = 10, filters: Optional[Dict] = None):
"""Hybrid search combining vector and keyword search"""
response = self.collection.query.hybrid(
query=query,
alpha=alpha, # 0 = keyword, 1 = vector
limit=limit,
filters=self._build_filters(filters) if filters else None,
fusion_type=HybridFusion.RELATIVE_SCORE,
return_metadata=weaviate.classes.query.MetadataQuery(
score=True,
explain_score=True,
certainty=True
)
)
return self._format_results(response.objects)
async def multimodal_search(self, text_query: str, image_embedding: List[float],
alpha: float = 0.7, limit: int = 10):
"""Multi-modal search combining text and image vectors"""
# For multi-modal, you'd need a custom implementation
# This is a simplified version
response = self.collection.query.near_text(
query=text_query,
limit=limit,
return_metadata=weaviate.classes.query.MetadataQuery(certainty=True)
)
return self._format_results(response.objects)
async def generative_search(self, query: str, limit: int = 5,
generate_prompt: str = None):
"""Search with generative AI augmentation"""
if not generate_prompt:
generate_prompt = """
Summarize the key points from these documents in relation to the query: {query}
Documents:
{documents}
"""
response = self.collection.generate.near_text(
query=query,
limit=limit,
grouped_task=generate_prompt,
return_metadata=weaviate.classes.query.MetadataQuery(certainty=True)
)
return {
"results": self._format_results(response.objects),
"generated_summary": response.generated
}
async def faceted_search(self, query: str, facets: List[str],
limit: int = 10):
"""Search with faceted filtering and aggregation"""
response = self.collection.aggregate.over_all(
filters=Filter.by_property("category").equal("AI"),
return_metrics=[weaviate.classes.query.Metrics("word_count").count().maximum().minimum().mean()]
)
search_results = await self.semantic_search(query, limit)
return {
"search_results": search_results,
"facets": {
"word_count_stats": response.attributes[0] if response.attributes else {}
}
}
async def conversational_search(self, conversation_history: List[Dict],
current_query: str, limit: int = 5):
"""Context-aware search using conversation history"""
# Build context from conversation history
context = " ".join([f"Q: {msg['query']} A: {msg.get('response', '')}"
for msg in conversation_history[-3:]]) # Last 3 exchanges
enhanced_query = f"Context: {context}. Current question: {current_query}"
return await self.semantic_search(enhanced_query, limit)
async def similarity_graph(self, doc_id: str, depth: int = 2,
limit_per_depth: int = 3):
"""Find similar documents and build a similarity graph"""
similar_docs = {}
# Get initial document
initial_doc = self.collection.query.fetch_object_by_id(doc_id)
if not initial_doc:
return {"error": "Document not found"}
similar_docs[doc_id] = {
"document": initial_doc.properties,
"similar": []
}
# Find similar documents recursively
await self._find_similar_recursive(doc_id, similar_docs, depth, limit_per_depth)
return similar_docs
async def _find_similar_recursive(self, source_id: str, graph: Dict,
depth: int, limit: int, current_depth: int = 0):
"""Recursively find similar documents"""
if current_depth >= depth:
return
# Find similar documents
source_doc = self.collection.query.fetch_object_by_id(source_id)
if not source_doc:
return
similar_response = self.collection.query.near_object(
near_object=source_id,
limit=limit,
return_metadata=weaviate.classes.query.MetadataQuery(certainty=True)
)
for obj in similar_response.objects:
if obj.uuid not in graph:
graph[obj.uuid] = {
"document": obj.properties,
"similar": []
}
# Add to similarity list
if obj.uuid != source_id:
graph[source_id]["similar"].append({
"id": obj.uuid,
"certainty": obj.metadata.certainty,
"title": obj.properties.get('title', '')
})
# Recursive call for next depth
await self._find_similar_recursive(
obj.uuid, graph, depth, limit, current_depth + 1
)
def _build_filters(self, filters: Dict) -> Filter:
"""Build Weaviate filter from dictionary"""
filter_conditions = []
for field, value in filters.items():
if isinstance(value, (list, tuple)):
filter_conditions.append(
Filter.by_property(field).contains_any(value)
)
elif isinstance(value, str):
filter_conditions.append(
Filter.by_property(field).equal(value)
)
elif isinstance(value, dict):
if 'min' in value and 'max' in value:
filter_conditions.append(
Filter.by_property(field).greater_or_equal(value['min']). \
less_or_equal(value['max'])
)
if len(filter_conditions) == 1:
return filter_conditions[0]
else:
# Combine multiple filters with AND
combined_filter = filter_conditions[0]
for condition in filter_conditions[1:]:
combined_filter = combined_filter & condition
return combined_filter
def _format_results(self, objects) -> List[Dict]:
"""Format search results for API response"""
formatted = []
for obj in objects:
formatted.append({
"id": obj.uuid,
"title": obj.properties.get('title', ''),
"content": obj.properties.get('content', ''),
"category": obj.properties.get('category', ''),
"tags": obj.properties.get('tags', []),
"author": obj.properties.get('author', ''),
"certainty": getattr(obj.metadata, 'certainty', None),
"distance": getattr(obj.metadata, 'distance', None),
"score": getattr(obj.metadata, 'score', None),
"explanation": getattr(obj.metadata, 'explain_score', None)
})
return formatted
# Example usage
async def search_examples():
from weaviate_schema import WeaviateSemanticSearch
search_engine = WeaviateSemanticSearch()
semantic_search = AdvancedSemanticSearch(search_engine.client)
# Basic semantic search
results = await semantic_search.semantic_search(
"machine learning algorithms",
limit=5,
filters={"category": "AI"}
)
print("Semantic search results:", results)
# Hybrid search
hybrid_results = await semantic_search.hybrid_search(
"neural networks deep learning",
alpha=0.7,
limit=5
)
print("Hybrid search results:", hybrid_results)
# Generative search
generative_results = await semantic_search.generative_search(
"explain machine learning concepts",
limit=3
)
print("Generative search results:", generative_results)
if __name__ == "__main__":
asyncio.run(search_examples())
📊 Performance Optimization and Monitoring
Production vector databases require careful performance tuning and monitoring. Here are optimization strategies for 2025:
- Index Tuning: Optimize HNSW parameters (ef, efConstruction, maxConnections) for your data Sharding Strategy: Implement custom sharding based on access patterns
⚡ Key Takeaways
- Schema Design Matters: Proper class and property configuration significantly impacts performance and functionality
- Hybrid Search Excellence: Combine vector and keyword search for the best of both worlds
- Production Readiness: Implement proper error handling, monitoring, and backup strategies
- Multi-modal Capabilities: Weaviate can handle text, images, and custom embeddings simultaneously
- Scalability First: Design for horizontal scaling from the beginning with proper sharding
- Generative Integration: Leverage Weaviate's built-in generative AI modules for enhanced search
- Monitoring Essential: Implement comprehensive monitoring for performance and reliability
❓ Frequently Asked Questions
- How does Weaviate compare to other vector databases like Pinecone or Chroma?
- Weaviate stands out with its GraphQL interface, multi-modal capabilities, and built-in generative AI modules. While Pinecone excels in pure vector search performance and Chroma offers simplicity, Weaviate provides a complete ecosystem with hybrid search, filtering, and AI integration out of the box. It's particularly strong for complex applications needing both vector and traditional database features.
- What's the optimal batch size for importing data into Weaviate?
- For optimal performance, use batch sizes between 50-200 objects. Smaller batches increase network overhead, while larger batches can cause memory issues and timeouts. The sweet spot depends on your object size and network latency. Monitor your import performance and adjust accordingly. For large-scale imports, consider parallelizing across multiple workers with appropriate rate limiting.
- Can I use custom embedding models with Weaviate?
- Yes, Weaviate supports custom embedding models through its modules system. You can implement custom vectorizers or bring your own pre-computed embeddings. For custom models, you'll need to implement a vectorizer module or use the 'none' vectorizer and provide embeddings directly. This flexibility allows integration with specialized models for different domains or languages.
- How do I handle data updates and real-time synchronization?
- Weaviate supports real-time updates through its GraphQL API. For synchronization with external systems, implement change data capture (CDC) patterns or use Weaviate's webhook system. For large-scale updates, use batch operations with proper error handling. Remember that vector updates require re-embedding, so consider the computational cost of frequent updates.
- What's the best way to scale Weaviate for high-traffic applications?
- Implement horizontal scaling with proper sharding configuration, use read replicas for query load distribution, and implement caching at multiple levels. For write-heavy applications, consider sharding by time or category. Monitor performance metrics and use connection pooling. For ultimate scalability, consider Weaviate Cloud with managed scaling or Kubernetes deployment with auto-scaling.
- How do I ensure data consistency and backup in production?
- Use Weaviate's replication features for high availability, implement regular snapshot backups, and use consistent read/write consistency levels. For critical data, enable synchronous replication and regular backup exports. Monitor disk usage and implement alerting for capacity planning. Test your backup and recovery procedures regularly.
💬 Have you implemented semantic search with Weaviate or other vector databases? Share your experiences, challenges, or performance tips in the comments below! If you found this guide helpful, please share it with your team or on social media to help others master vector databases.
About LK-TECH Academy — Practical tutorials & explainers on software engineering, AI, and infrastructure. Follow for concise, hands-on guides.
Posts
