Build a mini search engine from scratch using an inverted index (hash map), TF-IDF scoring for relevance ranking, and a trie for autocomplete suggestions.
What We Are Building
A search engine does three things: (1) indexes documents so they are searchable, (2) finds relevant documents for a query, and (3) ranks them by relevance. We will build all three using fundamental data structures:
- Inverted Index (Hash Map): Maps each word to the list of documents containing it
- TF-IDF Scoring: Ranks results by how important a word is to a document
- Trie: Provides autocomplete suggestions as the user types
Part 1: The Inverted Index
An inverted index is the backbone of every search engine. Instead of scanning every document for every query (O(n × d) where n is documents and d is average doc length), we precompute a mapping: word → list of documents.
import re
import math
from collections import defaultdict, Counter
class InvertedIndex:
def __init__(self):
self.index = defaultdict(set) # word -> set of doc_ids
self.documents = {} # doc_id -> original text
self.doc_word_counts = {} # doc_id -> Counter of words
self.doc_lengths = {} # doc_id -> total word count
self.num_docs = 0
def tokenize(self, text):
"""Split text into lowercase words, remove punctuation."""
text = text.lower()
words = re.findall(r'[a-z0-9]+', text)
# Remove common stop words
stop_words = {'the', 'a', 'an', 'is', 'are', 'was', 'were', 'in', 'on',
'at', 'to', 'for', 'of', 'and', 'or', 'but', 'it', 'this'}
return [w for w in words if w not in stop_words and len(w) > 1]
def add_document(self, doc_id, text):
"""Index a document."""
self.documents[doc_id] = text
tokens = self.tokenize(text)
self.doc_word_counts[doc_id] = Counter(tokens)
self.doc_lengths[doc_id] = len(tokens)
self.num_docs += 1
# Update inverted index
for word in set(tokens):
self.index[word].add(doc_id)
def search(self, query):
"""Find documents containing ALL query words."""
query_tokens = self.tokenize(query)
if not query_tokens:
return []
# Intersection of document sets for each query word
result_sets = [self.index.get(word, set()) for word in query_tokens]
if not result_sets:
return []
matching_docs = result_sets[0]
for s in result_sets[1:]:
matching_docs = matching_docs.intersection(s)
return list(matching_docs)
Why hash map? Looking up which documents contain a word is O(1) with a hash map. Without an index, we would need to scan every document for every query — unusable at scale.
Part 2: TF-IDF Relevance Scoring
Not all matches are equally relevant. A document mentioning "algorithm" 10 times is more relevant to the query "algorithm" than one mentioning it once. TF-IDF captures this:
- TF (Term Frequency): How often does the word appear in THIS document?
- IDF (Inverse Document Frequency): How rare is this word across ALL documents?
A word that appears in every document (like "the") has low IDF. A word that appears in few documents (like "quicksort") has high IDF.
class TFIDFRanker:
def __init__(self, index):
self.index = index
def tf(self, word, doc_id):
"""Term frequency: word count / total words in document."""
count = self.index.doc_word_counts[doc_id].get(word, 0)
total = self.index.doc_lengths[doc_id]
return count / total if total > 0 else 0
def idf(self, word):
"""Inverse document frequency: log(total_docs / docs_containing_word)."""
doc_freq = len(self.index.index.get(word, set()))
if doc_freq == 0:
return 0
return math.log(self.index.num_docs / doc_freq)
def tf_idf(self, word, doc_id):
"""TF-IDF score for a word in a document."""
return self.tf(word, doc_id) * self.idf(word)
def rank_documents(self, query, doc_ids):
"""Rank documents by relevance to query using TF-IDF."""
query_tokens = self.index.tokenize(query)
scores = []
for doc_id in doc_ids:
score = sum(self.tf_idf(word, doc_id) for word in query_tokens)
scores.append((doc_id, score))
# Sort by score descending
scores.sort(key=lambda x: -x[1])
return scores
Part 3: Trie for Query Autocomplete
As the user types, we suggest completions based on previously indexed words:
class TrieNode:
def __init__(self):
self.children = {}
self.is_word = False
self.doc_frequency = 0 # How many documents contain this word
class SearchTrie:
def __init__(self):
self.root = TrieNode()
def insert(self, word, doc_freq=1):
node = self.root
for char in word:
if char not in node.children:
node.children[char] = TrieNode()
node = node.children[char]
node.is_word = True
node.doc_frequency = doc_freq
def suggest(self, prefix, max_results=5):
"""Get word suggestions for a prefix."""
node = self.root
for char in prefix:
if char not in node.children:
return []
node = node.children[char]
# DFS to find all words under this prefix
suggestions = []
self._collect(node, prefix, suggestions)
# Sort by document frequency (popularity)
suggestions.sort(key=lambda x: -x[1])
return [word for word, freq in suggestions[:max_results]]
def _collect(self, node, current_word, results):
if node.is_word:
results.append((current_word, node.doc_frequency))
for char, child in node.children.items():
self._collect(child, current_word + char, results)
Putting It All Together: The Search Engine
class MiniSearchEngine:
def __init__(self):
self.index = InvertedIndex()
self.ranker = TFIDFRanker(self.index)
self.trie = SearchTrie()
def index_document(self, doc_id, text):
"""Add a document to the search engine."""
self.index.add_document(doc_id, text)
# Update trie with words from this document
for word in set(self.index.tokenize(text)):
doc_freq = len(self.index.index[word])
self.trie.insert(word, doc_freq)
def search(self, query, top_k=5):
"""Search for documents matching the query, ranked by relevance."""
matching_docs = self.index.search(query)
if not matching_docs:
return []
ranked = self.ranker.rank_documents(query, matching_docs)
results = []
for doc_id, score in ranked[:top_k]:
results.append({
'doc_id': doc_id,
'score': round(score, 4),
'snippet': self.index.documents[doc_id][:150] + '...'
})
return results
def autocomplete(self, prefix):
"""Suggest query completions."""
return self.trie.suggest(prefix.lower())
# Demo
engine = MiniSearchEngine()
docs = {
1: "Binary search is an efficient algorithm for finding items in sorted arrays. It works by repeatedly dividing the search interval in half.",
2: "Merge sort is a divide and conquer algorithm that splits arrays into halves, sorts each half, then merges them back together.",
3: "Hash tables provide constant time average lookup by mapping keys to indices using a hash function. Collisions are handled with chaining or open addressing.",
4: "Graph algorithms like BFS and DFS traverse nodes systematically. BFS uses a queue for level-order traversal while DFS uses a stack.",
5: "Dynamic programming solves problems by breaking them into overlapping subproblems and storing results to avoid redundant computation."
}
for doc_id, text in docs.items():
engine.index_document(doc_id, text)
# Search
results = engine.search("algorithm sort")
for r in results:
print(f"Doc {r['doc_id']} (score: {r['score']}): {r['snippet']}")
# Autocomplete
print(engine.autocomplete("sor")) # ['sort', 'sorted', 'sorting']
Complexity Analysis
| Operation | Time Complexity | Data Structure Used |
|---|
| Index a document | O(d) where d = doc length | Hash Map |
| Search (find matches) | O(q) where q = query words | Hash Map (intersection) |
| Rank results | O(r × q) where r = result count | TF-IDF computation |
| Autocomplete | O(p + w) where p = prefix, w = words with prefix | Trie |
Extensions for Real-World Use
- Phrase search: Store word positions in the index to match exact phrases
- Stemming: Reduce "running," "runs," "ran" to common root "run"
- Spell correction: Use edit distance on trie nodes for "did you mean?"
- PageRank: For web search, combine TF-IDF with link-based authority
- Distributed indexing: Shard the inverted index across machines for billions of documents