Reddit Hybrid Search with FastAPI
Build a hybrid search engine over Reddit posts using dense and sparse vectors.
Introduction
30 Second Summary
Ever searched for something and gotten results that match your exact words but completely miss what you actually meant? Or a search engine that "gets" your intent but misses the specific term you typed? The best search systems solve both problems at once.
In this project, you will build a hybrid search engine that indexes Reddit posts and combines semantic understanding with keyword matching to deliver results that neither approach could achieve alone.
What You'll Build
A FastAPI-powered search API that fetches Reddit posts, indexes them into Qdrant with both dense and sparse vectors, and serves hybrid search results fused together using Reciprocal Rank Fusion.
By the end of this project, you'll have:
- A Reddit data ingestion pipeline that pulls real posts via public JSON endpoints.
- A Qdrant collection with dual dense and sparse vector indexing for semantic and keyword search.
- A FastAPI app with a hybrid search endpoint and a custom Trie-based autocomplete.
- Secret Mission: Add subreddit filtering and relevance score display to your search results.
Are there any prerequisites?
You should be comfortable with basic Python and have a general understanding of how APIs work. FastAPI experience is helpful but not required.
Fetch Reddit Posts
Your environment is locked and loaded. Qdrant is running, your virtual environment is active, and all packages are installed.
Now it's time to feed the search engine some data. You'll write a Python script that pulls real posts from Reddit and saves them locally as structured JSON. The trick is that Reddit exposes a public JSON endpoint for every page. Just append .json to any Reddit URL and you get structured data back. No API key, no OAuth, no signup required.
In this step, get ready to:
- Write a Python function that fetches posts from any subreddit via Reddit's public JSON endpoint.
- Parse each response to extract fields like title, author, score, and selftext.
- Save the structured data to a local JSON file for indexing.
Write the fetch function
Reddit's public JSON endpoint returns paginated data. Each page contains up to 100 posts, and you navigate between pages using an after cursor token. Your function needs to handle pagination, respect rate limits, and extract the fields you care about.
- Create a file called ingest.py in your project directory.
- Add the following code to define the imports and the fetch_subreddit_posts function:
import requests
import json
import time
def fetch_subreddit_posts(subreddit: str, sort: str = "hot", limit: int = 100) -> list[dict]:
"""Fetch posts from a subreddit using Reddit's public JSON API."""
posts = []
after = None
headers = {"User-Agent": "HybridSearchEngine/1.0"}
while len(posts) < limit:
page_size = min(100, limit - len(posts))
url = f"https://www.reddit.com/r/{subreddit}/{sort}.json"
params = {"limit": page_size, "raw_json": 1}
if after:
params["after"] = after
response = requests.get(url, headers=headers, params=params, timeout=30)
if response.status_code == 429:
retry_after = int(response.headers.get("retry-after", "3"))
print(f"Rate limited, sleeping {retry_after}s...")
time.sleep(retry_after)
continue
response.raise_for_status()
data = response.json()
children = data["data"]["children"]
if not children:
break
for child in children:
post = child["data"]
posts.append({
"id": post["id"],
"title": post["title"],
"selftext": post.get("selftext", "")[:500],
"author": post.get("author", "[deleted]"),
"score": post.get("score", 0),
"subreddit": post.get("subreddit", subreddit),
"permalink": post.get("permalink", ""),
"created_utc": post.get("created_utc", 0),
})
after = data["data"].get("after")
if not after:
break
time.sleep(1.5)
return posts[:limit]What does this code do?
The fetch_subreddit_posts function builds a URL like https://www.reddit.com/r/python/hot.json and fetches the page. Reddit returns posts inside a data.children array, where each child has a nested data object containing the actual post fields.
- The User-Agent header identifies your script to Reddit. Without it, requests may be blocked.
- If Reddit returns a 429 status (rate limited), the function sleeps and retries automatically.
- The after cursor handles pagination. Each response includes the token for the next page.
- A 1.5-second delay between pages keeps you under Reddit's ~60 requests/minute limit.
- The selftext field is truncated to 500 characters to keep the dataset manageable.
- Save ingest.py.
- You won't see output yet. The next chunk adds the code that actually calls this function.
Add the main function to save posts to JSON
The fetch function handles a single subreddit. Now you need an orchestrator that fetches from multiple subreddits and writes everything to a file the indexer can read later.
- In ingest.py, add the following code below the fetch_subreddit_posts function:
def main():
subreddits = ["python", "machinelearning"]
all_posts = []
for sub in subreddits:
print(f"Fetching posts from r/{sub}...")
posts = fetch_subreddit_posts(sub, sort="top", limit=50)
all_posts.extend(posts)
print(f" Got {len(posts)} posts")
time.sleep(2)
with open("reddit_posts.json", "w") as f:
json.dump(all_posts, f, indent=2)
print(f"\nSaved {len(all_posts)} posts to reddit_posts.json")
if __name__ == "__main__":
main()What does this code do?
The main() function loops through two subreddits (r/python and r/machinelearning) and fetches the top 50 posts from each. It collects up to 100 posts total.
The sort="top" parameter targets the highest-voted posts, giving you richer content with longer titles and more selftext. The 2-second sleep between subreddits is a courtesy delay to stay within rate limits.
Finally, json.dump writes everything to reddit_posts.json with indentation for readability. This file becomes the input for the indexing step.
- Save ingest.py.
✔️ Awesome, I've got everything!
Great. Double check you've saved your file before moving on.
ⓧ I'd like to double check the full code
import requests
import json
import time
def fetch_subreddit_posts(subreddit: str, sort: str = "hot", limit: int = 100) -> list[dict]:
"""Fetch posts from a subreddit using Reddit's public JSON API."""
posts = []
after = None
headers = {"User-Agent": "HybridSearchEngine/1.0"}
while len(posts) < limit:
page_size = min(100, limit - len(posts))
url = f"https://www.reddit.com/r/{subreddit}/{sort}.json"
params = {"limit": page_size, "raw_json": 1}
if after:
params["after"] = after
response = requests.get(url, headers=headers, params=params, timeout=30)
if response.status_code == 429:
retry_after = int(response.headers.get("retry-after", "3"))
print(f"Rate limited, sleeping {retry_after}s...")
time.sleep(retry_after)
continue
response.raise_for_status()
data = response.json()
children = data["data"]["children"]
if not children:
break
for child in children:
post = child["data"]
posts.append({
"id": post["id"],
"title": post["title"],
"selftext": post.get("selftext", "")[:500],
"author": post.get("author", "[deleted]"),
"score": post.get("score", 0),
"subreddit": post.get("subreddit", subreddit),
"permalink": post.get("permalink", ""),
"created_utc": post.get("created_utc", 0),
})
after = data["data"].get("after")
if not after:
break
time.sleep(1.5)
return posts[:limit]
def main():
subreddits = ["python", "machinelearning"]
all_posts = []
for sub in subreddits:
print(f"Fetching posts from r/{sub}...")
posts = fetch_subreddit_posts(sub, sort="top", limit=50)
all_posts.extend(posts)
print(f" Got {len(posts)} posts")
time.sleep(2)
with open("reddit_posts.json", "w") as f:
json.dump(all_posts, f, indent=2)
print(f"\nSaved {len(all_posts)} posts to reddit_posts.json")
if __name__ == "__main__":
main()Run the ingestion script
Time to pull real data from Reddit. The script will take a few seconds per subreddit because of the built-in rate limiting delays.
- Run the ingestion script by executing this command:
python ingest.py- You should see output like this:
Fetching posts from r/python...
Got 50 posts
Fetching posts from r/machinelearning...
Got 50 posts
Saved 100 posts to reddit_posts.json- Confirm the JSON file was created by running:
ls reddit_posts.json- You should see reddit_posts.json listed in the output.
Why might the post count differ?
Reddit may return fewer than 50 posts for a subreddit if some posts have been deleted or if the subreddit has fewer highly-voted posts available. Anything above 0 per subreddit means the script is working correctly.
Seeing a connection error or 403?
Make sure you are connected to the internet. Reddit occasionally blocks requests without a User-Agent header, but the script already includes one.
If you see a 403 Forbidden error, Reddit may be temporarily blocking your IP. Wait 30 seconds and try again. Some corporate or university networks also block Reddit traffic entirely.
Help me debug my Reddit ingestion script
Your dataset is ready. Next up, you'll index these posts into Qdrant with both dense and sparse vectors to power hybrid search.
Index Posts into Qdrant
You have raw Reddit posts saved in a JSON file. To make them searchable, you need to convert that text into numerical representations (vectors) that capture meaning.
In this step, you will index your posts into Qdrant with two types of vectors: dense embeddings that understand semantic meaning, and sparse BM25 vectors that catch exact keyword matches.
In this step, get ready to:
- Create a Qdrant collection configured for hybrid search with both dense and sparse vectors.
- Generate dense and sparse embeddings for each post using fastembed.
- Upload all posts with their dual vectors and metadata to Qdrant.
Create a Qdrant collection with dual vector configuration
A Qdrant collection is like a database table, but optimized for vector similarity search. Your collection needs two vector spaces: one for dense semantic embeddings and one for sparse keyword vectors.
- Create a new file called indexer.py in the same folder as ingest.py.
- Add the imports, client connection, and collection creation code by pasting this into indexer.py:
import json
from qdrant_client import QdrantClient, models
def index_posts():
client = QdrantClient(url="http://localhost:6333")
collection_name = "reddit_posts"
# Remove the collection if it already exists (for re-runs)
if client.collection_exists(collection_name):
client.delete_collection(collection_name)
# Create collection with named dense and sparse vector spaces
client.create_collection(
collection_name=collection_name,
vectors_config={
"dense": models.VectorParams(
size=384,
distance=models.Distance.COSINE,
)
},
sparse_vectors_config={
"sparse": models.SparseVectorParams(
modifier=models.Modifier.IDF,
)
},
)What does this code do?
The vectors_config dictionary defines a named dense vector space called dense with 384 dimensions and cosine distance. The 384 dimensions match the output of the sentence-transformers/all-MiniLM-L6-v2 embedding model you will use.
The sparse_vectors_config dictionary defines a sparse vector space called sparse with an IDF modifier. This applies Inverse Document Frequency weighting to BM25 term scores, which downweights common words and boosts rare, informative terms.
You will not see a change yet. The next chunk loads and embeds your posts.
Generate dense and sparse embeddings for each post
Each post needs two vector representations. Dense vectors from sentence-transformers/all-MiniLM-L6-v2 capture semantic meaning ("machine learning" is close to "AI training"). Sparse vectors from Qdrant/bm25 capture exact term importance (the word "PyTorch" gets a high weight if it appears in a post).
- Add the document preparation and embedding generation code below the create_collection call, still inside the index_posts() function:
with open("reddit_posts.json", "r") as f:
posts = json.load(f)
# Combine title and body into a single searchable document per post
documents = []
for post in posts:
text = f"{post['title']} {post['selftext']}".strip()
documents.append(text)
# Model identifiers for fastembed (downloaded automatically)
dense_model = "sentence-transformers/all-MiniLM-L6-v2"
sparse_model = "Qdrant/bm25"
print(f"Indexing {len(posts)} posts...")
# Generate both dense and sparse embeddings for all documents
points = []
dense_embeddings = list(
client.embedding.embed(model=dense_model, inputs=documents)
)
sparse_embeddings = list(
client.embedding.embed_sparse(model=sparse_model, inputs=documents)
)What does this code do?
Each post's title and body text are concatenated into a single string. This gives the embedding model full context about what the post is about.
The client.embedding.embed call uses the Qdrant client's built-in Inference API. It delegates to fastembed under the hood, which runs the all-MiniLM-L6-v2 model locally on your CPU via ONNX Runtime. The first run downloads the model weights automatically.
The client.embedding.embed_sparse call generates BM25 sparse vectors, where each dimension corresponds to a vocabulary token and its TF-IDF weight.
You will not see a change yet. The next chunk builds the point structures and uploads them.
Upload posts with dual vectors and metadata to Qdrant
Each post becomes a "point" in Qdrant, carrying both vector representations plus a metadata payload with the original post fields for display in search results.
- Add the point construction, upsert, and main guard below the embedding generation code, still inside index_posts() (the if __name__ block goes outside the function):
for i, post in enumerate(posts):
points.append(
models.PointStruct(
id=i,
vector={
"dense": dense_embeddings[i].tolist(),
},
payload={
"title": post["title"],
"selftext": post["selftext"],
"author": post["author"],
"score": post["score"],
"subreddit": post["subreddit"],
"permalink": post["permalink"],
},
)
)
# Attach sparse vector separately (SparseVector has indices + values)
points[-1].vector["sparse"] = models.SparseVector(
indices=sparse_embeddings[i].indices.tolist(),
values=sparse_embeddings[i].values.tolist(),
)
# Upload all points to the collection in one call
client.upsert(collection_name=collection_name, points=points)
print(f"Indexed {len(points)} posts into Qdrant collection '{collection_name}'")
if __name__ == "__main__":
index_posts()What does this code do?
Each PointStruct bundles two named vectors (dense and sparse) with a payload dictionary. The payload stores the original post metadata so search results can display titles, authors, and links.
The sparse vector is attached separately because SparseVector uses two arrays: indices (which vocabulary tokens are present) and values (how important each token is). This compressed format only stores non-zero entries, making it memory-efficient.
Why hybrid instead of just semantic search?
Dense vectors excel at understanding meaning ("best Python framework" matches posts about Django and Flask). Sparse BM25 vectors excel at exact term matching (searching "asyncio" finds posts with that exact word even if the dense model places them far apart in vector space).
By indexing both, your search engine gets the best of both worlds. In the next step, you will fuse these two signals together with Reciprocal Rank Fusion.
✔️ Awesome, I've got everything!
Great. Double check you have saved indexer.py.
ⓧ I'd like to double check the full code
import json
from qdrant_client import QdrantClient, models
def index_posts():
client = QdrantClient(url="http://localhost:6333")
collection_name = "reddit_posts"
if client.collection_exists(collection_name):
client.delete_collection(collection_name)
client.create_collection(
collection_name=collection_name,
vectors_config={
"dense": models.VectorParams(
size=384,
distance=models.Distance.COSINE,
)
},
sparse_vectors_config={
"sparse": models.SparseVectorParams(
modifier=models.Modifier.IDF,
)
},
)
with open("reddit_posts.json", "r") as f:
posts = json.load(f)
documents = []
for post in posts:
text = f"{post['title']} {post['selftext']}".strip()
documents.append(text)
dense_model = "sentence-transformers/all-MiniLM-L6-v2"
sparse_model = "Qdrant/bm25"
print(f"Indexing {len(posts)} posts...")
points = []
dense_embeddings = list(
client.embedding.embed(model=dense_model, inputs=documents)
)
sparse_embeddings = list(
client.embedding.embed_sparse(model=sparse_model, inputs=documents)
)
for i, post in enumerate(posts):
points.append(
models.PointStruct(
id=i,
vector={
"dense": dense_embeddings[i].tolist(),
},
payload={
"title": post["title"],
"selftext": post["selftext"],
"author": post["author"],
"score": post["score"],
"subreddit": post["subreddit"],
"permalink": post["permalink"],
},
)
)
points[-1].vector["sparse"] = models.SparseVector(
indices=sparse_embeddings[i].indices.tolist(),
values=sparse_embeddings[i].values.tolist(),
)
client.upsert(collection_name=collection_name, points=points)
print(f"Indexed {len(points)} posts into Qdrant collection '{collection_name}'")
if __name__ == "__main__":
index_posts()- Save indexer.py.
- Run the indexer script by running this command:
python indexer.pyWhat should I see?
On the first run, fastembed downloads the model weights (about 80 MB for the dense model). You will see download progress bars followed by:
Indexing 100 posts...
Indexed 100 posts into Qdrant collection 'reddit_posts'The exact number depends on how many posts your ingestion script fetched. Subsequent runs skip the download since the model is cached locally.
Seeing a connection error?
Make sure your Qdrant container is still running. Check with docker ps and look for the qdrant/qdrant image. If it is not listed, restart it with the docker run command from Step 1.
If you see a FileNotFoundError for reddit_posts.json, make sure you are running the command from the same directory where ingest.py saved the file.
Need more help? Help me debug my indexer.py connection or file errors.
Your posts are now indexed with dual vector representations. Next up, you will build a Trie data structure for instant autocomplete suggestions.
Build the Trie Autocomplete
Your posts are indexed and searchable by meaning and keywords. But modern search engines also offer instant suggestions as users type.
In this step, you'll implement a Trie data structure from scratch. A trie stores strings character-by-character in a tree, giving you prefix lookups in O(m) time where m is the prefix length. That means suggestions appear instantly regardless of how many posts you've indexed.
In this step, get ready to:
- Implement the TrieNode and Trie classes with insert and prefix search methods.
- Populate the trie with Reddit post titles and verify prefix matching works.
Create the TrieNode class
Every character in a trie occupies one node. Each node needs to track its children (the next possible characters), whether it marks the end of a complete title, and what that original title was.
- Create a new file called trie.py in your project folder.
- Add the TrieNode class by pasting the following code:
class TrieNode:
def __init__(self):
# Maps each character to the next TrieNode in the path
self.children: dict[str, "TrieNode"] = {}
# True when this node marks the end of a complete title
self.is_end: bool = False
# Stores the original (unmodified case) title at end nodes
self.title: str = ""What does this code do?
Each TrieNode represents a single character position in the tree.
- children is a dictionary that maps each character to its child node. This lets us follow paths character by character.
- is_end flags whether a complete title ends at this node. Without it, we couldn't distinguish between a prefix and a full inserted string.
- title stores the original title text so we can return properly-cased results even though we search in lowercase.
- Save trie.py.
You won't see a change yet. The next chunk builds the Trie class that uses these nodes.
Build the Trie class with insert and prefix search
The Trie class provides two operations. The insert method adds a title character by character. The search_prefix method walks the tree to the end of a prefix, then collects all titles stored below that point.
- In trie.py, add the Trie class below the TrieNode class by pasting this code:
class Trie:
def __init__(self):
self.root = TrieNode()
def insert(self, title: str) -> None:
node = self.root
# Convert to lowercase for case-insensitive matching
key = title.lower()
for char in key:
if char not in node.children:
node.children[char] = TrieNode()
node = node.children[char]
# Mark this node as the end of a complete title
node.is_end = True
node.title = title
def search_prefix(self, prefix: str, max_results: int = 10) -> list[str]:
node = self.root
key = prefix.lower()
# Walk to the node representing the end of the prefix
for char in key:
if char not in node.children:
return []
node = node.children[char]
# Collect all titles below this node
results: list[str] = []
self._collect(node, results, max_results)
return results
def _collect(self, node: TrieNode, results: list[str], max_results: int) -> None:
# Stop collecting once we have enough results
if len(results) >= max_results:
return
if node.is_end:
results.append(node.title)
for child in node.children.values():
if len(results) >= max_results:
return
self._collect(child, results, max_results)What does this code do?
- insert walks (or creates) one node per character of the lowercased title. At the final character, it marks the node as an endpoint and stores the original title.
- search_prefix traverses the tree along the prefix characters. If any character is missing, there are no matches. Otherwise it delegates to _collect.
- _collect recursively visits all child nodes below the prefix, appending titles from endpoint nodes until max_results is reached.
Why build this from scratch instead of using a library?
Libraries like pygtrie exist, but implementing the trie yourself demonstrates CS fundamentals. Interviewers and hiring managers value this because it shows you understand the data structure powering autocomplete, not just how to call a library.
- Save trie.py.
You won't see a visible change yet. Next you'll test that the trie actually returns matching titles from your Reddit data.
Seeing a syntax error when you save?
- Make sure the Trie class is not indented inside the TrieNode class. Both classes should start at the same indentation level (column 0).
- Check that you have a blank line separating the two classes.
Why isn't my trie.py working?
✔️ Awesome, I've got everything!
Great. Double check you've saved your file before moving on.
ⓧ I'd like to double check the full code
class TrieNode:
def __init__(self):
self.children: dict[str, "TrieNode"] = {}
self.is_end: bool = False
self.title: str = ""
class Trie:
def __init__(self):
self.root = TrieNode()
def insert(self, title: str) -> None:
node = self.root
key = title.lower()
for char in key:
if char not in node.children:
node.children[char] = TrieNode()
node = node.children[char]
node.is_end = True
node.title = title
def search_prefix(self, prefix: str, max_results: int = 10) -> list[str]:
node = self.root
key = prefix.lower()
for char in key:
if char not in node.children:
return []
node = node.children[char]
results: list[str] = []
self._collect(node, results, max_results)
return results
def _collect(self, node: TrieNode, results: list[str], max_results: int) -> None:
if len(results) >= max_results:
return
if node.is_end:
results.append(node.title)
for child in node.children.values():
if len(results) >= max_results:
return
self._collect(child, results, max_results)Test the trie with your Reddit post titles
Before wiring the trie into the API, you'll load your Reddit post titles into it and confirm that prefix search returns matching results.
- Test the trie by running this command in your terminal:
python3 -c "
import json
from trie import Trie
t = Trie()
with open('reddit_posts.json') as f:
posts = json.load(f)
for p in posts:
t.insert(p['title'])
print(f'Inserted {len(posts)} titles')
results = t.search_prefix('how')
print(f'Prefix \"how\" returned {len(results)} results:')
for r in results[:5]:
print(f' - {r}')
"What does this command do?
- It imports your Trie class and loads all post titles from reddit_posts.json.
- It inserts every title into the trie, then searches for all titles starting with "how".
- If your data includes posts like "How to learn Python" or "How I built a neural network", those will appear in the results.
- You should see output showing how many titles were inserted, followed by a list of titles starting with "how".
What if I see 0 results?
That's possible if none of your fetched posts have titles starting with "how". Try a different prefix that matches your data. Run this to see the first few titles in your dataset:
python3 -c "import json; posts=json.load(open('reddit_posts.json')); [print(p['title']) for p in posts[:5]]"Pick the first few characters of any title you see and use that as your search prefix.
Seeing ModuleNotFoundError: No module named 'trie'?
- Make sure your terminal is in the same directory as trie.py. Run ls and confirm you see trie.py listed.
- Check that the file is named exactly trie.py (all lowercase, no extra characters).
- Verify your virtual environment is still activated. You should see the environment name in your terminal prompt.
Why can't Python find my trie module?
Your trie is built and tested. Next up, you'll wire it into a FastAPI application alongside the hybrid search endpoint, giving users both instant autocomplete suggestions and full semantic+keyword search results.
Build the Hybrid Search API
You have all the building blocks ready: Reddit posts ingested, vectors indexed in Qdrant, and a Trie ready for autocomplete lookups. Now you need a way for users to actually query all of this.
In this step, you will wire everything together into a FastAPI application that exposes a hybrid search endpoint. The endpoint combines dense semantic search with sparse BM25 keyword search, fused via Reciprocal Rank Fusion for best-of-both-worlds retrieval.
In this step, get ready to:
- Create a FastAPI app that connects to Qdrant and loads the trie on startup.
- Build a hybrid search endpoint using Qdrant's prefetch API with RRF fusion.
- Run the server and test semantic vs keyword queries to see the hybrid advantage.
Create the FastAPI app skeleton
The app needs to connect to Qdrant, know which embedding models to use, and load all post titles into the Trie at startup.
- Create a new file called main.py in your project folder.
- Add the following imports and app configuration:
import json
from fastapi import FastAPI, Query
from qdrant_client import QdrantClient, models
from trie import Trie
# Create the FastAPI application
app = FastAPI(title="Reddit Hybrid Search Engine")
# Connect to the Qdrant instance running in Docker
client = QdrantClient(url="http://localhost:6333")
COLLECTION_NAME = "reddit_posts"
# Model names used by fastembed via the Qdrant Inference API
DENSE_MODEL = "sentence-transformers/all-MiniLM-L6-v2"
SPARSE_MODEL = "Qdrant/bm25"
# Trie instance for autocomplete lookups
autocomplete_trie = Trie()What does this code do?
This sets up the module-level configuration your endpoints will share.
- COLLECTION_NAME points at the Qdrant collection you created in the indexing step.
- DENSE_MODEL and SPARSE_MODEL tell the Qdrant Inference API which fastembed models to use for encoding queries.
- autocomplete_trie is the Trie you built in the previous step, instantiated here so you can populate it at startup.
You won't see a change yet. The next chunk adds the startup event and endpoints.
- Below the module-level code you just added, paste the startup handler and utility endpoints:
@app.on_event("startup")
def load_trie():
# Load all post titles into the trie for autocomplete
with open("reddit_posts.json", "r") as f:
posts = json.load(f)
for post in posts:
autocomplete_trie.insert(post["title"])
@app.get("/autocomplete")
def autocomplete(q: str = Query(..., min_length=1), limit: int = Query(default=10, le=20)):
# Return prefix-matched title suggestions from the trie
suggestions = autocomplete_trie.search_prefix(q, max_results=limit)
return {"query": q, "suggestions": suggestions}
@app.get("/health")
def health():
return {"status": "ok"}What does this code do?
- load_trie() runs once when the server starts. It reads reddit_posts.json and inserts every post title into the Trie so autocomplete works immediately.
- The /autocomplete endpoint takes a query string q and returns up to limit prefix-matched suggestions. This uses the O(prefix-length) lookup from your Trie.
- The /health endpoint is a simple status check that confirms the server is running.
- Save main.py.
Seeing an import error for trie?
- Make sure trie.py is in the same directory as main.py.
- Check that trie.py contains the Trie class (capital T). Python imports are case-sensitive.
Why can't FastAPI find my Trie module?
Build the hybrid search endpoint
This is the core of your search engine. A single query will run through both dense (semantic) and sparse (BM25) retrieval paths, then Qdrant fuses the results using Reciprocal Rank Fusion.
- In main.py, add the following search endpoint above the /health route:
@app.get("/search")
def hybrid_search(q: str = Query(..., min_length=1), limit: int = Query(default=10, le=50)):
# Generate dense and sparse embeddings for the user's query
dense_embedding = list(client.embedding.embed(model=DENSE_MODEL, inputs=[q]))[0]
sparse_embedding = list(client.embedding.embed_sparse(model=SPARSE_MODEL, inputs=[q]))[0]
# Run hybrid search: prefetch from both vector spaces, then fuse with RRF
results = client.query_points(
collection_name=COLLECTION_NAME,
prefetch=[
models.Prefetch(
query=models.SparseVector(
indices=sparse_embedding.indices.tolist(),
values=sparse_embedding.values.tolist(),
),
using="sparse",
limit=20,
),
models.Prefetch(
query=dense_embedding.tolist(),
using="dense",
limit=20,
),
],
query=models.FusionQuery(fusion=models.Fusion.RRF),
limit=limit,
)
# Format the fused results for the API response
search_results = []
for point in results.points:
payload = point.payload
search_results.append({
"title": payload["title"],
"selftext": payload["selftext"][:200],
"author": payload["author"],
"score": payload["score"],
"subreddit": payload["subreddit"],
"permalink": f"https://reddit.com{payload['permalink']}",
"relevance": round(point.score, 4),
})
return {"query": q, "count": len(search_results), "results": search_results}How does hybrid search with RRF work?
The prefetch list tells Qdrant to run two separate retrievals. The first prefetch searches the sparse BM25 vectors (great for exact keyword matches). The second searches the dense semantic vectors (great for meaning-based matches).
FusionQuery with Fusion.RRF merges both result sets using Reciprocal Rank Fusion. RRF ranks each result by 1/(k + rank) from each source, then sums the scores. This means a post that ranks highly in both keyword AND semantic search gets boosted to the top.
Why hybrid instead of just semantic search?
Pure semantic search can miss exact technical terms. If a user searches for "asyncio", dense vectors might return posts about "concurrency" or "async programming" but miss posts that specifically mention the word "asyncio".
Pure keyword search misses meaning. Searching "how to speed up my code" won't find posts titled "Python performance optimization". Hybrid search catches both cases.
- Save main.py.
Seeing a NameError for models?
- Make sure your import at the top of main.py includes models from qdrant_client: from qdrant_client import QdrantClient, models.
- Check that the /search function is placed between the /autocomplete and /health endpoints, not inside another function.
Help me fix my hybrid search endpoint code.
✔️ Awesome, I've got everything!
Great. Double check you have saved main.py with all the endpoints.
ⓧ I'd like to double check the full code
import json
from fastapi import FastAPI, Query
from qdrant_client import QdrantClient, models
from trie import Trie
app = FastAPI(title="Reddit Hybrid Search Engine")
client = QdrantClient(url="http://localhost:6333")
COLLECTION_NAME = "reddit_posts"
DENSE_MODEL = "sentence-transformers/all-MiniLM-L6-v2"
SPARSE_MODEL = "Qdrant/bm25"
autocomplete_trie = Trie()
@app.on_event("startup")
def load_trie():
with open("reddit_posts.json", "r") as f:
posts = json.load(f)
for post in posts:
autocomplete_trie.insert(post["title"])
@app.get("/autocomplete")
def autocomplete(q: str = Query(..., min_length=1), limit: int = Query(default=10, le=20)):
suggestions = autocomplete_trie.search_prefix(q, max_results=limit)
return {"query": q, "suggestions": suggestions}
@app.get("/search")
def hybrid_search(q: str = Query(..., min_length=1), limit: int = Query(default=10, le=50)):
dense_embedding = list(client.embedding.embed(model=DENSE_MODEL, inputs=[q]))[0]
sparse_embedding = list(client.embedding.embed_sparse(model=SPARSE_MODEL, inputs=[q]))[0]
results = client.query_points(
collection_name=COLLECTION_NAME,
prefetch=[
models.Prefetch(
query=models.SparseVector(
indices=sparse_embedding.indices.tolist(),
values=sparse_embedding.values.tolist(),
),
using="sparse",
limit=20,
),
models.Prefetch(
query=dense_embedding.tolist(),
using="dense",
limit=20,
),
],
query=models.FusionQuery(fusion=models.Fusion.RRF),
limit=limit,
)
search_results = []
for point in results.points:
payload = point.payload
search_results.append({
"title": payload["title"],
"selftext": payload["selftext"][:200],
"author": payload["author"],
"score": payload["score"],
"subreddit": payload["subreddit"],
"permalink": f"https://reddit.com{payload['permalink']}",
"relevance": round(point.score, 4),
})
return {"query": q, "count": len(search_results), "results": search_results}
@app.get("/health")
def health():
return {"status": "ok"}Run and test the API
Time to see everything working together. You will start the FastAPI server, then send requests to each endpoint.
- Start the development server by running this command:
uvicorn main:app --reloadWhat does this command do?
- uvicorn is the ASGI server that runs your FastAPI app.
- main:app tells uvicorn to look in main.py for the object named app.
- --reload enables auto-reload so the server restarts whenever you change the code.
- You should see output indicating the server is running at http://127.0.0.1:8000.
- Open a second terminal tab (keep the server running in the first one).
- Test the health endpoint by running:
curl http://localhost:8000/health- You should see {"status":"ok"} in the response.
- Test the autocomplete endpoint by running:
curl "http://localhost:8000/autocomplete?q=python"- You should see a JSON response with a suggestions array containing post titles that start with "python".
- Now test the hybrid search endpoint with a semantic query:
curl "http://localhost:8000/search?q=how+to+speed+up+my+code"- You should see results about performance optimization, even though those posts may not contain the exact phrase "speed up my code". That is the semantic (dense) search doing its job.
- Now test with a keyword-specific query:
curl "http://localhost:8000/search?q=asyncio"- You should see results that specifically mention "asyncio" ranked near the top. That is the BM25 sparse search pulling exact keyword matches forward through the RRF fusion.
Why do both queries return good results?
The semantic query ("how to speed up my code") finds conceptually related posts through dense vectors. The keyword query ("asyncio") finds exact-term matches through sparse BM25 vectors. Because both retrieval paths feed into RRF fusion, each query benefits from whichever signal is stronger for that particular search.
Server not starting?
- Make sure your Qdrant container is still running. Check with docker ps.
- Make sure reddit_posts.json exists in the same directory. The startup event reads this file.
- Check that your virtual environment is activated. You need fastapi and uvicorn installed.
Help me get my FastAPI server running.
Your search engine is live and serving hybrid results. You have proven that combining semantic understanding with keyword precision produces stronger retrieval than either approach alone. Next, you can explore the Secret Mission to add subreddit filtering and formatted relevance scores.
Secret mission
Add Subreddit Filtering and Score Display
Your hybrid search works across all indexed subreddits at once. But what if a user only wants results from r/python? And those raw RRF fusion scores don't mean much to an end user. Time to add production-quality filtering and human-readable relevance percentages.
Clean Up Your Resources
Clean Up Your Resources
Decide whether to keep your resources running, pause them to come back later, or delete them entirely. This project runs entirely locally, so there are no ongoing costs.
Resources you used:
- Qdrant Docker container (running on port 6333).
- Local Python virtual environment with installed packages.
- Project files (ingest.py, indexer.py, trie.py, main.py, reddit_posts.json).
- qdrant_storage directory (Qdrant's data volume).
Keep everything running
No action needed. Choose this if you are still actively building or experimenting with different subreddits and queries.
- Your Qdrant container continues serving search queries on port 6333.
- Your FastAPI app stays available as long as you leave the uvicorn process running.
- Docker Engine uses minimal resources when the container is idle.
Pause - I'll come back to this later
Shut down running processes to free up memory. Your data and files stay intact so you can resume later.
- Stop the uvicorn server by pressing Ctrl+C in the terminal where it is running.
- Stop the Qdrant container by running:
docker stop $(docker ps -q --filter ancestor=qdrant/qdrant)- Your indexed data persists in the qdrant_storage directory. When you want to resume, start the container again with the same docker run command from Step 1.
Delete - I don't want to use this again
Remove all project resources and start fresh. This removes the container, data, and all project files.
- Stop the uvicorn server by pressing Ctrl+C in the terminal where it is running.
- Stop and remove the Qdrant container by running:
docker stop $(docker ps -q --filter ancestor=qdrant/qdrant)
docker rm $(docker ps -aq --filter ancestor=qdrant/qdrant)- Remove the Qdrant storage directory by running:
rm -rf qdrant_storage- Deactivate your virtual environment by running:
deactivate- Remove all project files and the virtual environment directory by running:
rm -f ingest.py indexer.py trie.py main.py reddit_posts.json
rm -rf venv- Optionally, remove the downloaded Qdrant Docker image to free disk space:
docker rmi qdrant/qdrantNice Work!
Nice Work!
Nice work! You just built a production-style hybrid search engine that combines the best of semantic understanding with exact keyword matching over real Reddit data.
You've learned how to:
- Built a data ingestion pipeline that pulls real Reddit posts and indexes them into Qdrant with dual dense (semantic) and sparse (BM25) vector representations.
- Implemented Reciprocal Rank Fusion to combine semantic and keyword search results into a single ranked list that outperforms either approach alone.
- Created a custom Trie data structure for O(prefix-length) autocomplete and served everything through a clean FastAPI REST API.
- Secret Mission: Added subreddit filtering and relevance score display for production-quality search results.
Ready to quiz yourself?