Build a Hybrid RAG Security Intel System

Combine dense and sparse search to query real CVE vulnerability data.

Introduction

30 Second Summary

When you search for something, you might type an exact product code or ask a vague question like "best budget laptop for students." These are fundamentally different types of queries. Most search systems are only good at one of them.

In this project, you will build a Security Vulnerability Intelligence System that combines two search methods (dense semantic search and sparse keyword search) to handle both exact identifiers and natural language questions over real vulnerability data.

What You'll Build

A hybrid RAG pipeline that ingests real CVE data from NIST's National Vulnerability Database, indexes it with both dense and sparse vectors in Qdrant, and uses Reciprocal Rank Fusion to deliver superior retrieval quality across mixed query types.

By the end of this project, you'll have:

  • A Python pipeline that fetches real CVE vulnerability data from NIST's National Vulnerability Database and indexes it with both dense vectors and sparse BM25 vectors in Qdrant.
  • Three retrieval strategies (dense-only, sparse-only, and hybrid with RRF fusion) with measurable quality comparisons showing where each method fails and hybrid wins.
  • A Groq-powered answer generation pipeline that produces grounded vulnerability intelligence reports from hybrid-retrieved context.
  • Secret Mission: Add severity-based metadata filtering to scope hybrid search results by vulnerability criticality.

Are there any prerequisites?

This project assumes you have prior experience building a basic RAG pipeline and are comfortable writing Python. You also need an existing Groq account with an API key (free tier, no credit card required).

Set Up Your Environment

Before you can build a hybrid retrieval pipeline, you need the right libraries in place. Your system will combine dense semantic embeddings with sparse keyword matching, so you need a vector database client that supports both search modes natively.

In this step, you'll install the Python packages, configure your Groq API credentials, and set up the project's file structure.

In this step, get ready to:
  • Install qdrant-client with FastEmbed and the Groq SDK.
  • Configure your Groq API key and project settings.
  • Create the project file structure and verify the Groq connection.
Install the required Python packages

Your project depends on two packages. qdrant-client is the Python client for Qdrant, an open-source vector database. The [fastembed] extra bundles FastEmbed for local CPU-based embeddings, covering both dense and sparse models.

  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir cve-intel
cd cve-intel
  • Confirm the folder exists by running:
ls

You should see the cve-intel folder listed in the output (or you'll already be inside it).

  • Install both packages by running this command:
pip install "qdrant-client[fastembed]==1.18.0" groq==1.4.0

What does this command do?

qdrant-client[fastembed]==1.18.0 installs the Qdrant Python client at version 1.18.0 with the FastEmbed extension. FastEmbed runs embedding models locally on your CPU, so you don't need a GPU or external embedding API.

groq==1.4.0 installs the Groq SDK at version 1.4.0, which you'll use to call Groq's LLM for answer generation.

This may take a minute as it downloads dependencies including ONNX runtime for running the embedding models.

  • Verify the installation succeeded by running:
python -c "from qdrant_client import QdrantClient; print('OK')"

You should see OK printed in your terminal.

Seeing an error instead of OK?

  • If you see a ModuleNotFoundError, the install may not have completed. Re-run the pip install command and check for error messages in the output.
  • If you're using a virtual environment, make sure it's activated before installing.
  • On macOS with zsh, you may need single quotes around the package name: pip install 'qdrant-client[fastembed]==1.18.0'.

help me troubleshoot my qdrant-client installation error

Set up your Groq API key

Your project needs a Groq API key to generate answers from retrieved context. You'll store the key in an environment variable and create a configuration file that loads it.

  • Create a .env file inside your cve-intel folder. This file documents which environment variable your project expects:
GROQ_API_KEY=your-api-key-here

What does this file do?

The .env file is a reference showing which environment variables the project needs. Replace your-api-key-here with your actual Groq API key from console.groq.com.

  • Set the environment variable in your terminal session so Python can read it.

macOS

  • Export your API key by running this command (replace your-api-key-here with your actual key):
export GROQ_API_KEY=your-api-key-here

Windows

  • Set your API key by running this command (replace your-api-key-here with your actual key):
set GROQ_API_KEY=your-api-key-here
  • Create a config.py file inside your cve-intel folder with the following code:
import os

GROQ_API_KEY = os.environ.get("GROQ_API_KEY", "your-api-key-here")
GROQ_MODEL = "llama-3.3-70b-versatile"

COLLECTION_NAME = "cve-intel"
DENSE_MODEL = "sentence-transformers/all-MiniLM-L6-v2"
SPARSE_MODEL = "Qdrant/bm25"

NVD_API_URL = "https://services.nvd.nist.gov/rest/json/cves/2.0"

What does this code do?

  • GROQ_API_KEY reads your API key from the environment variable you just exported. The fallback string is just a placeholder for when the variable isn't set.
  • GROQ_MODEL specifies which LLM to use for answer generation.
  • DENSE_MODEL and SPARSE_MODEL define the two embedding models for hybrid search: a semantic model and a BM25 keyword model.
  • NVD_API_URL points to the NIST National Vulnerability Database API where you'll fetch real CVE data.
  • Save config.py.

✔️ Awesome, I've got everything!

Great. Double check you've saved your file.

ⓧ I'd like to double check the full code

import os

GROQ_API_KEY = os.environ.get("GROQ_API_KEY", "your-api-key-here")
GROQ_MODEL = "llama-3.3-70b-versatile"

COLLECTION_NAME = "cve-intel"
DENSE_MODEL = "sentence-transformers/all-MiniLM-L6-v2"
SPARSE_MODEL = "Qdrant/bm25"

NVD_API_URL = "https://services.nvd.nist.gov/rest/json/cves/2.0"
Create your project file structure and verify Groq

Your project uses four Python files, each with a single responsibility. You already created config.py. Now create the remaining three skeleton files.

  • Create ingest.py inside your cve-intel folder with this placeholder comment:
# skeleton - to be filled in next step
  • Create search.py with this placeholder comment:
# skeleton - to be filled in later steps
  • Create main.py with this placeholder comment:
# skeleton - to be filled in later steps
  • Confirm all files are present by running:
ls

You should see config.py, ingest.py, main.py, search.py, and .env listed in the output.

  • Verify your Groq connection by running this test command:
python -c "
from groq import Groq
from config import GROQ_API_KEY, GROQ_MODEL
client = Groq(api_key=GROQ_API_KEY)
response = client.chat.completions.create(
    model=GROQ_MODEL,
    messages=[{'role': 'user', 'content': 'Say hello in one word.'}],
    max_tokens=10
)
print(response.choices[0].message.content)
"

You should see a short response from the LLM, like Hello or Hi. This confirms your Groq API key works and the SDK is properly installed.

Seeing an authentication error?

  • Make sure you exported your GROQ_API_KEY environment variable in the same terminal session where you're running the command.
  • Verify your key is valid by signing in to console.groq.com and checking the API Keys section.
  • If you opened a new terminal tab, you'll need to re-export the variable.

help me troubleshoot my Groq API authentication error

Why is the first run slower?

The first time you use FastEmbed in a later step, it will download the embedding models (about 90 MB each) to a local cache. Subsequent runs load from cache and are much faster.

Your environment is ready. Next up, you'll fetch real CVE vulnerability data from the National Vulnerability Database and process it into documents ready for indexing.

Fetch and Process CVE Data

Your environment is set up and ready to pull in real vulnerability data. The National Vulnerability Database (NVD) is the U.S. government's repository of security vulnerability information, and it exposes a free JSON API that anyone can query.

In this step, you will build the data ingestion layer that fetches CVE records from the NVD and transforms them into documents your hybrid search engine can index. Each document combines a human-readable text field with structured metadata for filtering later.

In this step, get ready to:
  • Write a function that fetches CVE records from the NVD API.
  • Parse raw API responses into structured documents with metadata.
  • Validate the data pipeline by printing a sample document.
Write the CVE fetch function

The NVD API at https://services.nvd.nist.gov/rest/json/cves/2.0 returns vulnerability records as JSON. It requires no API key for low-volume use (up to 5 requests per 30 seconds).

  • Open ingest.py in your editor and replace the skeleton comment with the following imports and fetch function:
import requests
import time
from config import NVD_API_URL


def fetch_cves(results_per_page=50):
    """Fetch recent CVEs from the NVD API."""
    params = {
        "resultsPerPage": results_per_page,
        "pubStartDate": "2024-01-01T00:00:00.000",
        "pubEndDate": "2024-06-30T00:00:00.000",
    }
    response = requests.get(NVD_API_URL, params=params)
    response.raise_for_status()
    return response.json()

What does this code do?

The fetch_cves function sends a GET request to the NVD API with three query parameters.

  • resultsPerPage limits how many CVE records come back in a single response (50 by default).
  • pubStartDate and pubEndDate narrow the results to vulnerabilities published in the first half of 2024.
  • raise_for_status() throws an exception if the API returns an error code, so failures are caught immediately.

Why does requests work without a separate install?

The requests library was installed automatically as a transitive dependency of qdrant-client[fastembed] in the previous step. You do not need to install it separately.

  • Save ingest.py.

Seeing an import error for requests?

If Python cannot find requests, install it directly by running pip install requests in your terminal. Also confirm you are using the same Python environment where you installed qdrant-client.

help me debug a missing requests import in my Python project

Parse CVE records into documents

The NVD API returns a deeply nested JSON structure. Each vulnerability record contains the CVE ID, a description in multiple languages, CVSS severity metrics, and a publication timestamp.

Your parser needs to flatten this nested data into a clean document format with a searchable text field and a structured metadata dictionary.

  • Add the following parse_cve_records function below fetch_cves in ingest.py:
def parse_cve_records(raw_data):
    """Parse NVD API response into documents with metadata."""
    documents = []
    vulnerabilities = raw_data.get("vulnerabilities", [])

    for item in vulnerabilities:
        cve = item.get("cve", {})
        cve_id = cve.get("id", "UNKNOWN")

        descriptions = cve.get("descriptions", [])
        description = ""
        for desc in descriptions:
            if desc.get("lang") == "en":
                description = desc.get("value", "")
                break

        severity = "UNKNOWN"
        cvss_score = 0.0
        metrics = cve.get("metrics", {})
        if "cvssMetricV31" in metrics:
            cvss_data = metrics["cvssMetricV31"][0]["cvssData"]
            severity = cvss_data.get("baseSeverity", "UNKNOWN")
            cvss_score = cvss_data.get("baseScore", 0.0)
        elif "cvssMetricV2" in metrics:
            cvss_data = metrics["cvssMetricV2"][0]["cvssData"]
            cvss_score = cvss_data.get("baseScore", 0.0)
            if cvss_score >= 9.0:
                severity = "CRITICAL"
            elif cvss_score >= 7.0:
                severity = "HIGH"
            elif cvss_score >= 4.0:
                severity = "MEDIUM"
            else:
                severity = "LOW"

        published = cve.get("published", "")

        text = f"{cve_id}: {description} (Severity: {severity}, CVSS: {cvss_score})"

        documents.append({
            "text": text,
            "metadata": {
                "cve_id": cve_id,
                "severity": severity,
                "cvss_score": cvss_score,
                "published": published,
                "description": description,
            },
        })

    return documents

What does this code do?

The parser walks through each item in the API's vulnerabilities array and extracts the fields your search engine needs.

  • It pulls the English-language description by filtering on lang == "en".
  • It checks for cvssMetricV31 first (the newer scoring standard), then falls back to cvssMetricV2 with manual severity mapping.
  • The text field concatenates the CVE ID, description, and severity into a single searchable string.
  • The metadata dictionary stores individual fields so you can filter on them later (for example, filtering by severity in the Secret Mission).

Why handle two CVSS versions?

Not all CVEs have CVSS v3.1 scores. Older vulnerabilities or those awaiting reanalysis may only carry a CVSS v2.0 score. The v2.0 schema does not include a baseSeverity text label, so the code maps the numeric score to a severity category manually.

  • Save ingest.py.
Validate your data pipeline

Before moving on, you need to confirm that your fetch and parse functions work end to end. A quick test script at the bottom of ingest.py will call the NVD API and print a sample document.

  • Add the following test block at the bottom of ingest.py:
if __name__ == "__main__":
    print("Fetching CVEs from NVD API...")
    raw = fetch_cves(50)
    docs = parse_cve_records(raw)
    print(f"Parsed {len(docs)} CVE documents.")
    if docs:
        print(f"\nSample document:\n{docs[0]['text'][:200]}")
        print(f"Metadata: {docs[0]['metadata']}")

What does this code do?

The if __name__ == "__main__" guard means this code only runs when you execute ingest.py directly. It will not run when other files import functions from ingest.py.

It fetches 50 CVEs, parses them, then prints the count and a truncated sample so you can visually confirm the data looks correct.

  • Save ingest.py.
  • Run the data pipeline by executing this command in your terminal:
python ingest.py
  • You should see output showing the number of parsed documents and a sample CVE entry with its metadata fields.

What should I see?

Your terminal output should look similar to this:

Fetching CVEs from NVD API...
Parsed 50 CVE documents.

Sample document:
CVE-2024-XXXXX: [description text]... (Severity: HIGH, CVSS: 7.5)
Metadata: {'cve_id': 'CVE-2024-XXXXX', 'severity': 'HIGH', 'cvss_score': 7.5, 'published': '2024-01-...', 'description': '...'}

The exact CVE IDs and scores will vary depending on what the NVD returns, but you should see 50 documents with populated metadata.

Getting a connection error or timeout?

The NVD API is a public government service that can occasionally be slow or temporarily unavailable. Wait 30 seconds and try again.

If you see a 403 error, you may have exceeded the rate limit of 5 requests per 30 seconds. Wait at least 30 seconds before retrying.

help me troubleshoot NVD API connection issues in my Python script

✔️ Awesome, I've got everything!

Great. Double check you have saved ingest.py before moving on.

ⓧ I'd like to double check the full code

Here is the complete ingest.py file at this point:

import requests
import time
from config import NVD_API_URL


def fetch_cves(results_per_page=50):
    """Fetch recent CVEs from the NVD API."""
    params = {
        "resultsPerPage": results_per_page,
        "pubStartDate": "2024-01-01T00:00:00.000",
        "pubEndDate": "2024-06-30T00:00:00.000",
    }
    response = requests.get(NVD_API_URL, params=params)
    response.raise_for_status()
    return response.json()


def parse_cve_records(raw_data):
    """Parse NVD API response into documents with metadata."""
    documents = []
    vulnerabilities = raw_data.get("vulnerabilities", [])

    for item in vulnerabilities:
        cve = item.get("cve", {})
        cve_id = cve.get("id", "UNKNOWN")

        descriptions = cve.get("descriptions", [])
        description = ""
        for desc in descriptions:
            if desc.get("lang") == "en":
                description = desc.get("value", "")
                break

        severity = "UNKNOWN"
        cvss_score = 0.0
        metrics = cve.get("metrics", {})
        if "cvssMetricV31" in metrics:
            cvss_data = metrics["cvssMetricV31"][0]["cvssData"]
            severity = cvss_data.get("baseSeverity", "UNKNOWN")
            cvss_score = cvss_data.get("baseScore", 0.0)
        elif "cvssMetricV2" in metrics:
            cvss_data = metrics["cvssMetricV2"][0]["cvssData"]
            cvss_score = cvss_data.get("baseScore", 0.0)
            if cvss_score >= 9.0:
                severity = "CRITICAL"
            elif cvss_score >= 7.0:
                severity = "HIGH"
            elif cvss_score >= 4.0:
                severity = "MEDIUM"
            else:
                severity = "LOW"

        published = cve.get("published", "")

        text = f"{cve_id}: {description} (Severity: {severity}, CVSS: {cvss_score})"

        documents.append({
            "text": text,
            "metadata": {
                "cve_id": cve_id,
                "severity": severity,
                "cvss_score": cvss_score,
                "published": published,
                "description": description,
            },
        })

    return documents


if __name__ == "__main__":
    print("Fetching CVEs from NVD API...")
    raw = fetch_cves(50)
    docs = parse_cve_records(raw)
    print(f"Parsed {len(docs)} CVE documents.")
    if docs:
        print(f"\nSample document:\n{docs[0]['text'][:200]}")
        print(f"Metadata: {docs[0]['metadata']}")

Your data pipeline is pulling real vulnerability intelligence from the NVD. Next up, you will load these documents into a hybrid vector store that indexes them with both dense semantic vectors and sparse BM25 vectors.

Build the Hybrid Vector Store

In the last step, you built a data pipeline that fetches and parses real CVE records from the NVD API. Now you have structured documents ready to be searched.

The core challenge of searching vulnerability data is that queries come in two flavors: exact identifiers like "CVE-2024-3094" (keyword match) and semantic questions like "supply chain attacks" (meaning match). A single vector type cannot handle both well. In this step, you will build a Qdrant collection that stores BOTH dense semantic vectors and sparse BM25 keyword vectors for every document.

In this step, get ready to:
  • Create a Qdrant collection configured for both dense and sparse vectors.
  • Ingest CVE documents with dual vector representations and metadata payloads.
  • Verify ingestion by running a test query that returns results.
Create the hybrid collection

A hybrid collection needs two vector configurations: one for dense embeddings (semantic meaning) and one for sparse vectors (keyword matching). Qdrant lets you define both on the same collection so every document gets indexed both ways simultaneously.

  • Open search.py in your editor. Replace the skeleton comment with the following imports and function:
from qdrant_client import QdrantClient, models
from config import COLLECTION_NAME, DENSE_MODEL, SPARSE_MODEL


def create_collection(client):
    """Create a Qdrant collection with both dense and sparse vectors."""
    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 of 384 dimensions using cosine similarity. This matches the output shape of the sentence-transformers/all-MiniLM-L6-v2 model.

The sparse_vectors_config dictionary defines a named sparse vector space using BM25 with an IDF modifier. IDF (Inverse Document Frequency) weights rare terms higher, which is exactly what makes keyword search effective for unique identifiers like CVE IDs.

  • Save search.py.

Seeing an import error?

Make sure you installed qdrant-client with the fastembed extra in Step 1. The models module comes from qdrant_client and provides all the configuration classes.

Also confirm that config.py exists in the same directory as search.py. Python needs both files in the same folder to resolve the import.

Help me debug my import error in search.py

Ingest documents with dual vectors

Each CVE document needs to be stored as a single point in Qdrant with two vector representations. The FastEmbed integration lets you pass raw text and a model name, and Qdrant handles the embedding computation locally on your CPU.

  • Add the following function below create_collection in search.py:
def ingest_documents(client, documents):
    """Ingest documents with both dense and sparse vectors."""
    points = []
    for i, doc in enumerate(documents):
        point = models.PointStruct(
            id=i,
            vector={
                "dense": models.Document(
                    text=doc["text"],
                    model=DENSE_MODEL,
                ),
                "sparse": models.Document(
                    text=doc["text"],
                    model=SPARSE_MODEL,
                ),
            },
            payload=doc["metadata"],
        )
        points.append(point)

    client.upsert(collection_name=COLLECTION_NAME, points=points)
    print(f"Ingested {len(points)} documents into '{COLLECTION_NAME}'.")

What does this code do?

Each PointStruct bundles an ID, a vector dictionary, and a payload. The vector dictionary contains two entries: "dense" and "sparse". Each entry wraps the document text with a model name so FastEmbed knows which model to use for embedding.

The payload stores all the CVE metadata (ID, severity, CVSS score, publication date, description). This metadata travels with search results, so you can display rich information without a second lookup.

Why two models on the same text?

The dense model (sentence-transformers/all-MiniLM-L6-v2) captures semantic meaning. It understands that "remote code execution" and "arbitrary command injection" are related concepts.

The sparse model (Qdrant/bm25) captures exact keyword overlap. It excels at matching precise identifiers like "CVE-2024-3094" that dense models often mangle into unrelated embeddings.

  • Save search.py.

Getting a model download error?

The first time you run the ingestion, FastEmbed downloads the embedding models. This requires an internet connection. If you see a timeout or connection error, check your network and retry.

Help me fix my FastEmbed model download issue

Verify ingestion with a test query

Your functions are defined but you have not run them yet. You need to confirm that documents are ingested correctly and that queries return results with payloads before moving forward.

  • Run this verification script in your terminal:
python -c "
from qdrant_client import QdrantClient, models
from config import COLLECTION_NAME, DENSE_MODEL
from ingest import fetch_cves, parse_cve_records
from search import create_collection, ingest_documents

client = QdrantClient(':memory:')
raw = fetch_cves(50)
docs = parse_cve_records(raw)
create_collection(client)
ingest_documents(client, docs)

results = client.query_points(
    collection_name=COLLECTION_NAME,
    query=models.Document(text='remote code execution', model=DENSE_MODEL),
    using='dense',
    limit=3,
    with_payload=True,
)
for r in results.points:
    print(f'{r.payload[\"cve_id\"]} - {r.payload[\"severity\"]}')
"

What does this verification do?

This script creates a temporary in-memory Qdrant instance, fetches CVE data, ingests it using your functions, then runs a dense-only query for "remote code execution". If everything works, you will see three CVE IDs printed with their severity levels.

The in-memory instance disappears when the script ends. This is expected. You will wire up a persistent workflow in the next step.

  • You should see output showing the ingestion count (e.g. Ingested 50 documents into 'cve-intel'.) followed by three CVE IDs with severity ratings.

Not seeing any results?

If you see Ingested 0 documents, the NVD API might have returned an empty response. Check your internet connection and try again. The API is rate-limited to 5 requests per 30 seconds without a key.

If you see a KeyError or AttributeError, double-check that the vector names "dense" and "sparse" in your ingest_documents function match the names in create_collection exactly.

Help me debug why my Qdrant ingestion is not returning results

✔️ Awesome, I've got everything!

Great. Double check you have saved search.py with both functions before moving on.

ⓧ I'd like to double check the full code

Here is the complete search.py file at the end of this step:

from qdrant_client import QdrantClient, models
from config import COLLECTION_NAME, DENSE_MODEL, SPARSE_MODEL


def create_collection(client):
    """Create a Qdrant collection with both dense and sparse vectors."""
    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,
            ),
        },
    )


def ingest_documents(client, documents):
    """Ingest documents with both dense and sparse vectors."""
    points = []
    for i, doc in enumerate(documents):
        point = models.PointStruct(
            id=i,
            vector={
                "dense": models.Document(
                    text=doc["text"],
                    model=DENSE_MODEL,
                ),
                "sparse": models.Document(
                    text=doc["text"],
                    model=SPARSE_MODEL,
                ),
            },
            payload=doc["metadata"],
        )
        points.append(point)

    client.upsert(collection_name=COLLECTION_NAME, points=points)
    print(f"Ingested {len(points)} documents into '{COLLECTION_NAME}'.")

Your hybrid vector store is ready. Every CVE is now indexed for both semantic similarity and exact keyword matching. Next up, you will build three retrieval strategies to query this collection and see exactly where each method shines and fails.

Implement Three Retrieval Strategies

Your Qdrant collection is loaded with 50 CVE documents, each indexed with both dense and sparse vectors. Now comes the core question: how do you actually search across those two vector spaces?

In this step, you'll build three retrieval functions that each answer queries differently. Dense search uses semantic understanding. Sparse search uses exact keyword matching. Hybrid search combines both using Reciprocal Rank Fusion to get the best of both worlds.

In this step, get ready to:
  • Build a dense-only search function using semantic vector similarity.
  • Build a sparse-only search function using BM25 keyword matching.
  • Build a hybrid search function that fuses both strategies with RRF.
Build the dense search function

Dense search converts your query into a 384-dimensional embedding and finds documents whose embeddings are closest in vector space. This captures semantic meaning but struggles with exact identifiers like CVE IDs.

  • In search.py, add the following function below your ingest_documents function:
def dense_search(client, query, limit=5):
    """Search using only dense (semantic) vectors."""
    response = client.query_points(
        collection_name=COLLECTION_NAME,
        query=models.Document(text=query, model=DENSE_MODEL),
        using="dense",
        limit=limit,
        with_payload=True,
    )
    return response.points

What does this code do?

The query_points() method takes your query text, embeds it using the same sentence-transformers/all-MiniLM-L6-v2 model that indexed the documents, and finds the 5 closest vectors by cosine distance.

The using="dense" parameter tells Qdrant to search only the dense vector space. The with_payload=True flag ensures results include the CVE metadata you stored earlier.

  • Save search.py.
Build the sparse search function

Sparse search uses BM25 to find documents by exact keyword overlap. It excels at matching specific identifiers and technical terms but misses semantically related concepts that use different words.

  • In search.py, add this function below dense_search:
def sparse_search(client, query, limit=5):
    """Search using only sparse (BM25) vectors."""
    response = client.query_points(
        collection_name=COLLECTION_NAME,
        query=models.Document(text=query, model=SPARSE_MODEL),
        using="sparse",
        limit=limit,
        with_payload=True,
    )
    return response.points

What does this code do?

This function mirrors dense_search but targets the sparse vector space instead. The Qdrant/bm25 model converts your query into a sparse representation that matches documents based on term frequency and inverse document frequency (IDF).

If you search for "CVE-2024-3094", BM25 will find that exact token. Dense search would treat it as noise because the embedding model has no semantic meaning for arbitrary ID strings.

  • Save search.py.

Seeing a NameError?

Make sure both functions are inside search.py and that your imports at the top still include from config import COLLECTION_NAME, DENSE_MODEL, SPARSE_MODEL. If you accidentally deleted the import line, add it back.

Help me debug a NameError in my search functions.

Build the hybrid search function with RRF fusion

Hybrid search runs both dense and sparse queries in parallel, then merges the results using Reciprocal Rank Fusion. RRF assigns each document a fused score based on its rank in each result list, so documents that rank well in both searches bubble to the top.

Why RRF instead of simple score averaging?

Dense and sparse scores live on completely different scales. Cosine similarity returns values between 0 and 1, while BM25 scores can be any positive number. You cannot meaningfully average them.

RRF sidesteps this by ignoring raw scores entirely. It only looks at rank positions: a document ranked #1 in both lists gets a higher fused score than a document ranked #1 in one and #10 in the other. Qdrant performs this fusion server-side so you don't need to normalize anything yourself.

  • In search.py, add this function below sparse_search:
def hybrid_search(client, query, limit=5, severity_filter=None):
    """Search using hybrid dense + sparse with RRF fusion."""
    query_filter = None
    if severity_filter:
        query_filter = models.Filter(
            must=[
                models.FieldCondition(
                    key="severity",
                    match=models.MatchValue(value=severity_filter),
                )
            ]
        )

    prefetch = [
        models.Prefetch(
            query=models.Document(text=query, model=SPARSE_MODEL),
            using="sparse",
            limit=20,
            filter=query_filter,
        ),
        models.Prefetch(
            query=models.Document(text=query, model=DENSE_MODEL),
            using="dense",
            limit=20,
            filter=query_filter,
        ),
    ]

    response = client.query_points(
        collection_name=COLLECTION_NAME,
        prefetch=prefetch,
        query=models.FusionQuery(fusion=models.Fusion.RRF),
        limit=limit,
        with_payload=True,
    )
    return response.points

What does this code do?

The prefetch list tells Qdrant to run two independent searches. The first fetches 20 candidates from the sparse (BM25) index. The second fetches 20 candidates from the dense (semantic) index.

The models.FusionQuery(fusion=models.Fusion.RRF) line applies Reciprocal Rank Fusion to merge both candidate lists into one ranked result. The final limit=5 returns only the top 5 fused results.

The severity_filter parameter defaults to None and does nothing for now. You'll use it later in the Secret Mission to scope results by vulnerability severity.

  • Save search.py.
  • Verify all three functions work by running this quick test in your terminal:
python -c "from search import dense_search, sparse_search, hybrid_search; print('All three retrieval functions imported successfully.')"

You should see the message All three retrieval functions imported successfully. printed with no errors.

Getting an ImportError?

Check that all three functions are defined at the module level in search.py (not nested inside another function). Also confirm you have no indentation errors. Python is strict about consistent spacing.

Help me fix my import error in search.py.

✔️ Awesome, I've got everything!

Great. Double check you've saved search.py with all three retrieval functions.

ⓧ I'd like to double check the full code

from qdrant_client import QdrantClient, models
from config import COLLECTION_NAME, DENSE_MODEL, SPARSE_MODEL


def create_collection(client):
    """Create a Qdrant collection with both dense and sparse vectors."""
    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,
            ),
        },
    )


def ingest_documents(client, documents):
    """Ingest documents with both dense and sparse vectors."""
    points = []
    for i, doc in enumerate(documents):
        point = models.PointStruct(
            id=i,
            vector={
                "dense": models.Document(
                    text=doc["text"],
                    model=DENSE_MODEL,
                ),
                "sparse": models.Document(
                    text=doc["text"],
                    model=SPARSE_MODEL,
                ),
            },
            payload=doc["metadata"],
        )
        points.append(point)

    client.upsert(collection_name=COLLECTION_NAME, points=points)
    print(f"Ingested {len(points)} documents into '{COLLECTION_NAME}'.")


def dense_search(client, query, limit=5):
    """Search using only dense (semantic) vectors."""
    response = client.query_points(
        collection_name=COLLECTION_NAME,
        query=models.Document(text=query, model=DENSE_MODEL),
        using="dense",
        limit=limit,
        with_payload=True,
    )
    return response.points


def sparse_search(client, query, limit=5):
    """Search using only sparse (BM25) vectors."""
    response = client.query_points(
        collection_name=COLLECTION_NAME,
        query=models.Document(text=query, model=SPARSE_MODEL),
        using="sparse",
        limit=limit,
        with_payload=True,
    )
    return response.points


def hybrid_search(client, query, limit=5, severity_filter=None):
    """Search using hybrid dense + sparse with RRF fusion."""
    query_filter = None
    if severity_filter:
        query_filter = models.Filter(
            must=[
                models.FieldCondition(
                    key="severity",
                    match=models.MatchValue(value=severity_filter),
                )
            ]
        )

    prefetch = [
        models.Prefetch(
            query=models.Document(text=query, model=SPARSE_MODEL),
            using="sparse",
            limit=20,
            filter=query_filter,
        ),
        models.Prefetch(
            query=models.Document(text=query, model=DENSE_MODEL),
            using="dense",
            limit=20,
            filter=query_filter,
        ),
    ]

    response = client.query_points(
        collection_name=COLLECTION_NAME,
        prefetch=prefetch,
        query=models.FusionQuery(fusion=models.Fusion.RRF),
        limit=limit,
        with_payload=True,
    )
    return response.points

Your three retrieval strategies are ready. Next up, you'll pit them against each other on real queries and see exactly where dense fails, where sparse fails, and where hybrid wins.

Compare Retrieval Quality and Generate Answers

Your three retrieval strategies are built and ready to prove their worth. Now it is time to put them head-to-head on real queries and see where each one fails.

In this step, you will wire everything together in main.py. You will run a comparison showing that dense search misses exact CVE IDs, sparse search misses semantic intent, and hybrid search captures both. Then you will connect Groq to generate grounded vulnerability reports from the hybrid results.

In this step, get ready to:
  • Build a comparison function that runs all three retrieval strategies side by side.
  • Run the pipeline and observe where dense-only and sparse-only search fail.
  • Generate a Groq-powered vulnerability report from hybrid-retrieved context.
Create the comparison function

The comparison function runs all three retrieval strategies on the same query and prints the results side by side. This makes it visually obvious where each method succeeds or fails.

  • Open main.py in your editor and replace the skeleton comment with the following imports:
from qdrant_client import QdrantClient
from groq import Groq
from config import GROQ_API_KEY, GROQ_MODEL, COLLECTION_NAME
from ingest import fetch_cves, parse_cve_records
from search import create_collection, ingest_documents, dense_search, sparse_search, hybrid_search

What do these imports do?

You are pulling in everything needed for the full pipeline. QdrantClient creates the in-memory vector store. Groq connects to the LLM API. The remaining imports bring in your data ingestion and search functions from the modules you built in earlier steps.

  • Below the imports in main.py, add the compare_strategies function:
def compare_strategies(client, query):
    """Compare all three retrieval strategies on a query."""
    print(f"\n{'='*60}")
    print(f"Query: {query}")
    print(f"{'='*60}")

    # Run all three search strategies
    dense_results = dense_search(client, query)
    sparse_results = sparse_search(client, query)
    hybrid_results = hybrid_search(client, query)

    # Print top 3 from each strategy
    print(f"\nDense (semantic) results:")
    for r in dense_results[:3]:
        print(f"  [{r.score:.3f}] {r.payload['cve_id']} - {r.payload['severity']}")

    print(f"\nSparse (BM25) results:")
    for r in sparse_results[:3]:
        print(f"  [{r.score:.3f}] {r.payload['cve_id']} - {r.payload['severity']}")

    print(f"\nHybrid (RRF) results:")
    for r in hybrid_results[:3]:
        print(f"  [{r.score:.3f}] {r.payload['cve_id']} - {r.payload['severity']}")

    return hybrid_results

What does this function do?

  • It takes a single query and runs it through dense_search, sparse_search, and hybrid_search in sequence.
  • For each strategy, it prints the top 3 results with their relevance scores, CVE IDs, and severity levels.
  • It returns the hybrid results so you can feed them into answer generation later.
  • Now add the initial main() function below compare_strategies. This version initializes the pipeline and runs the comparison:
def main():
    print("Initializing Qdrant (in-memory)...")
    client = QdrantClient(":memory:")

    print("Fetching CVE data from NVD...")
    raw_data = fetch_cves(50)
    documents = parse_cve_records(raw_data)
    print(f"Loaded {len(documents)} CVE records.")

    print("Creating hybrid collection and ingesting documents...")
    create_collection(client)
    ingest_documents(client, documents)

    # Test with three query types: exact ID, semantic, and technical
    test_queries = [
        documents[0]["metadata"]["cve_id"],
        "remote code execution in web servers",
        "memory corruption buffer overflow vulnerability",
    ]

    for query in test_queries:
        compare_strategies(client, query)


if __name__ == "__main__":
    main()

What does this pipeline do?

  • It creates an in-memory Qdrant client, fetches 50 CVEs from the NVD API, and ingests them with both dense and sparse vectors.
  • The three test queries exercise different retrieval strengths. The first is an exact CVE ID (pulled dynamically from your data). The second is a semantic description. The third combines technical keywords.
  • Each query runs through all three strategies so you can compare results side by side.
  • Save main.py.

Seeing an import error?

  • Make sure all four files (config.py, ingest.py, search.py, main.py) are in the same folder.
  • Double check that requests is installed. It is a dependency of qdrant-client so it should already be available.

Why can't Python find my modules?

Run the comparison and observe results

With the comparison pipeline ready, you can now see the retrieval quality differences in action.

  • Run the pipeline by executing this command:
python main.py

The first run will take a moment as FastEmbed downloads the embedding models. After that, you will see three comparison blocks printed to your terminal.

What should I see?

For the exact CVE ID query, sparse (BM25) search will rank the correct CVE at the top because BM25 excels at matching exact tokens. Dense search will likely rank it lower or return semantically similar but incorrect CVEs.

For the semantic queries, dense search will surface relevant vulnerabilities based on meaning. Sparse search may miss them because the exact words do not appear in the document text.

Hybrid search consistently ranks the correct results at or near the top for all three query types. This is Reciprocal Rank Fusion in action. It merges the ranked lists from both strategies so neither failure mode dominates.

Seeing a connection error from NVD?

  • The NVD API allows 5 requests per 30 seconds without a key. If you ran the script multiple times quickly, wait 30 seconds and try again.
  • Check that your machine has internet access and can reach services.nvd.nist.gov.

Why is my NVD API request failing?

Wire up Groq for answer generation

The comparison proves hybrid search finds the right documents. Now you will feed those documents into Groq's LLM to generate a grounded vulnerability report that cites specific CVE IDs.

  • In main.py, add the generate_answer function above your compare_strategies function (below the imports):
def generate_answer(groq_client, query, context_docs):
    """Generate a grounded answer using Groq."""
    # Format retrieved documents as context for the LLM
    context = "\n\n".join(
        f"- {doc.payload['cve_id']}: {doc.payload['description']} "
        f"(Severity: {doc.payload['severity']}, CVSS: {doc.payload['cvss_score']})"
        for doc in context_docs
    )

    # Build the prompt with system instructions and user query
    messages = [
        {
            "role": "system",
            "content": (
                "You are a security vulnerability analyst. Answer questions "
                "based ONLY on the provided CVE context. If the context does not "
                "contain relevant information, say so. Cite CVE IDs in your response."
            ),
        },
        {
            "role": "user",
            "content": f"Context:\n{context}\n\nQuestion: {query}",
        },
    ]

    # Call Groq with low temperature for factual answers
    response = groq_client.chat.completions.create(
        model=GROQ_MODEL,
        messages=messages,
        temperature=0.1,
        max_tokens=500,
    )
    return response.choices[0].message.content

What does this function do?

  • It formats the hybrid search results into a structured context string with CVE IDs, descriptions, and severity scores.
  • The system prompt constrains the LLM to answer only from the provided context and cite CVE IDs. This is the "grounded" part of RAG. The model cannot hallucinate vulnerabilities that are not in the retrieved documents.
  • A temperature of 0.1 keeps responses factual and deterministic.
  • Now update your main() function to add the Groq answer generation. Find the for query in test_queries: loop and add the following code below it (still inside main(), before the if __name__ block):
    print("\n\nGenerating vulnerability report with Groq...")
    groq_client = Groq(api_key=GROQ_API_KEY)
    query = "What are the most critical vulnerabilities?"
    results = hybrid_search(client, query, severity_filter="CRITICAL")
    if not results:
        results = hybrid_search(client, query)

    answer = generate_answer(groq_client, query, results)
    print(f"\nQuery: {query}")
    print(f"\nAnswer:\n{answer}")

What does this addition do?

  • It initializes the Groq client with your API key from config.py.
  • It runs a hybrid search with severity_filter="CRITICAL" to scope results to only the most severe vulnerabilities. If no CRITICAL CVEs exist in your batch, it falls back to an unfiltered hybrid search.
  • The retrieved documents are passed to generate_answer which sends them to the llama-3.3-70b-versatile model via Groq for a grounded response.
  • Save main.py and run the full pipeline again:
python main.py

You should see the three comparison blocks from before, followed by a generated vulnerability report that cites specific CVE IDs from your data.

Getting an authentication error from Groq?

  • Make sure your GROQ_API_KEY environment variable is set before running the script. You can export it with export GROQ_API_KEY=your-key or ensure your .env file is loaded.
  • If you see a 401 error, your API key may be invalid or expired. Generate a new one at console.groq.com.
  • If you see a 429 rate limit error, wait a few seconds and try again. Groq's free tier has rate limits but retries automatically.

Why is my Groq API call failing?

✔️ Awesome, I've got everything!

Great. Double check you have saved main.py and that the generated vulnerability report printed successfully.

ⓧ I'd like to double check the full code

Here is the complete main.py file:

from qdrant_client import QdrantClient
from groq import Groq
from config import GROQ_API_KEY, GROQ_MODEL, COLLECTION_NAME
from ingest import fetch_cves, parse_cve_records
from search import create_collection, ingest_documents, dense_search, sparse_search, hybrid_search


def generate_answer(groq_client, query, context_docs):
    """Generate a grounded answer using Groq."""
    context = "\n\n".join(
        f"- {doc.payload['cve_id']}: {doc.payload['description']} "
        f"(Severity: {doc.payload['severity']}, CVSS: {doc.payload['cvss_score']})"
        for doc in context_docs
    )

    messages = [
        {
            "role": "system",
            "content": (
                "You are a security vulnerability analyst. Answer questions "
                "based ONLY on the provided CVE context. If the context does not "
                "contain relevant information, say so. Cite CVE IDs in your response."
            ),
        },
        {
            "role": "user",
            "content": f"Context:\n{context}\n\nQuestion: {query}",
        },
    ]

    response = groq_client.chat.completions.create(
        model=GROQ_MODEL,
        messages=messages,
        temperature=0.1,
        max_tokens=500,
    )
    return response.choices[0].message.content


def compare_strategies(client, query):
    """Compare all three retrieval strategies on a query."""
    print(f"\n{'='*60}")
    print(f"Query: {query}")
    print(f"{'='*60}")

    dense_results = dense_search(client, query)
    sparse_results = sparse_search(client, query)
    hybrid_results = hybrid_search(client, query)

    print(f"\nDense (semantic) results:")
    for r in dense_results[:3]:
        print(f"  [{r.score:.3f}] {r.payload['cve_id']} - {r.payload['severity']}")

    print(f"\nSparse (BM25) results:")
    for r in sparse_results[:3]:
        print(f"  [{r.score:.3f}] {r.payload['cve_id']} - {r.payload['severity']}")

    print(f"\nHybrid (RRF) results:")
    for r in hybrid_results[:3]:
        print(f"  [{r.score:.3f}] {r.payload['cve_id']} - {r.payload['severity']}")

    return hybrid_results


def main():
    print("Initializing Qdrant (in-memory)...")
    client = QdrantClient(":memory:")

    print("Fetching CVE data from NVD...")
    raw_data = fetch_cves(50)
    documents = parse_cve_records(raw_data)
    print(f"Loaded {len(documents)} CVE records.")

    print("Creating hybrid collection and ingesting documents...")
    create_collection(client)
    ingest_documents(client, documents)

    test_queries = [
        documents[0]["metadata"]["cve_id"],
        "remote code execution in web servers",
        "memory corruption buffer overflow vulnerability",
    ]

    for query in test_queries:
        compare_strategies(client, query)

    print("\n\nGenerating vulnerability report with Groq...")
    groq_client = Groq(api_key=GROQ_API_KEY)
    query = "What are the most critical vulnerabilities?"
    results = hybrid_search(client, query, severity_filter="CRITICAL")
    if not results:
        results = hybrid_search(client, query)

    answer = generate_answer(groq_client, query, results)
    print(f"\nQuery: {query}")
    print(f"\nAnswer:\n{answer}")


if __name__ == "__main__":
    main()

Secret mission

Add Severity-Based Metadata Filtering

In production, security analysts don't want to sift through low-severity noise when triaging critical incidents. Can you add metadata filtering to your hybrid search so analysts can scope queries to only CRITICAL or HIGH severity vulnerabilities?

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally with an in-memory vector store, so there are no ongoing costs or cloud resources to worry about.

Decide whether to keep your project files and packages installed, or remove them entirely.

Resources you used:

  • Local project files: config.py, ingest.py, search.py, main.py, test_filter.py, and .env.
  • Python packages: qdrant-client[fastembed] and groq.

Keep everything running

No action needed. Choose this if you plan to keep experimenting with hybrid search queries or extending the project.

  • Your project files stay on your Desktop, ready to run again anytime.
  • The Qdrant vector store only exists in memory while the script runs. It does not persist between sessions, so there is nothing to shut down.
  • Your Groq API key remains in .env. Groq's free tier has no charges.

Pause - I'll come back to this later

There are no running processes to pause. Qdrant runs in-memory only while your script executes, so nothing persists between sessions.

  • Your project files and installed packages remain on your machine, ready for next time.
  • No ongoing costs accrue while you are away.

Delete - I don't want to use this again

Remove all project files and uninstall the Python packages you added for this project.

  • Delete your project files by navigating to the folder on your Desktop where you created main.py, search.py, and the other files, then running:
rm config.py ingest.py search.py main.py test_filter.py .env
  • Uninstall the Python packages by running:
pip uninstall qdrant-client groq -y
  • Your Groq account is free and can be left as-is. No action needed there.

Nice Work!

Nice Work!

Nice work! You just built a production-pattern hybrid RAG pipeline that combines dense semantic search with sparse BM25 keyword search over real security vulnerability data.

You've learned how to:

  • Build a hybrid search pipeline combining dense vector embeddings with sparse BM25 keyword search, fused server-side via Reciprocal Rank Fusion in Qdrant.
  • Ingest real CVE vulnerability data from NIST's National Vulnerability Database and measure retrieval quality differences between dense-only, sparse-only, and hybrid strategies.
  • Generate grounded vulnerability intelligence reports using Groq's free-tier LLM with context from hybrid-retrieved documents.
  • Secret Mission: Added severity-based metadata filtering to scope hybrid search results by vulnerability criticality.

Ready to quiz yourself?