KYC Compliance Search with Hybrid RAG

Build a hybrid search engine over KYC documents with AI risk summaries.

Introduction

30 Second Summary

When a bank screens a new customer, compliance officers need to search through hundreds of pages of regulations. Sometimes they need the exact regulatory code. Other times they need a conceptual match for an unusual scenario.

In this project, you will build a hybrid search assistant that combines semantic understanding with keyword precision to search KYC compliance documents. You will also generate AI-powered risk summaries and push those insights to a Power BI dashboard.

What You'll Build

A FastAPI-powered compliance search API that uses hybrid retrieval (dense embeddings + sparse keyword matching) with Reciprocal Rank Fusion to deliver high-precision results, generate risk summaries via an LLM, and surface insights on a Power BI dashboard.

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

  • A Qdrant collection with hybrid indexing (dense + sparse vectors) over sample KYC compliance documents.
  • A search API with a tunable alpha parameter for balancing semantic vs keyword retrieval, plus a comparison endpoint that demonstrates hybrid search advantages.
  • An AI-generated risk summary pipeline that synthesizes search results into actionable compliance assessments and pushes them to a Power BI Push Dataset.
  • Secret Mission: Add a compliance alert system that automatically flags high-risk queries and pushes real-time alerts to Power BI when risk scores exceed a threshold.

Are there any prerequisites?

You should be comfortable writing Python scripts and understand the basics of REST APIs. You will also need Docker Desktop installed on your machine to run Qdrant locally.

For the Power BI integration in Step 5, you need a Power BI Pro or Premium license and an Azure AD app registration. The core search system (Steps 1-4) works without Power BI.

Ingest Compliance Documents with Hybrid Indexing

Your environment is ready and your sample compliance documents are in place. Now you need to get those documents into a format that your search engine can actually query.

In this step, you will write an ingestion script that creates a Qdrant collection with two types of vectors: dense vectors for semantic understanding and sparse vectors for exact keyword matching. Each document gets embedded with both models, giving your search engine two complementary signals to work with.

In this step, get ready to:
  • Create a Qdrant collection configured for hybrid search with dense and sparse vector fields.
  • Ingest 8 KYC compliance documents with dual embeddings and metadata.
  • Verify the ingestion via the Qdrant dashboard.
Create the collection with hybrid vector configuration

A standard vector database collection stores one type of vector per document. For hybrid search, you need two: a dense vector that captures meaning and a sparse vector that captures exact keywords.

  • Create a new file called ingest.py in your project folder.
  • Add the imports, model constants, and the collection creation function by pasting this code:
import json
from qdrant_client import QdrantClient, models

DENSE_MODEL = "sentence-transformers/all-MiniLM-L6-v2"
SPARSE_MODEL = "prithivida/Splade_PP_en_v1"
COLLECTION_NAME = "kyc_compliance"


def create_collection(client: QdrantClient):
    """Create a Qdrant collection with hybrid search support."""
    if client.collection_exists(COLLECTION_NAME):
        client.delete_collection(COLLECTION_NAME)

    client.create_collection(
        collection_name=COLLECTION_NAME,
        vectors_config={
            "dense": models.VectorParams(
                size=client.get_embedding_size(DENSE_MODEL),
                distance=models.Distance.COSINE,
            )
        },
        sparse_vectors_config={
            "sparse": models.SparseVectorParams()
        },
    )
    print(f"Collection '{COLLECTION_NAME}' created.")

What does this code do?

The two model constants define which embedding models FastEmbed will use. sentence-transformers/all-MiniLM-L6-v2 produces 384-dimensional dense vectors that capture semantic similarity. prithivida/Splade_PP_en_v1 produces sparse vectors that give higher weight to specific keywords (like SPLADE-style learned term importance).

The create_collection function first deletes any existing collection with the same name (so you can safely re-run). Then it creates a new collection with two vector configurations: a named "dense" field using cosine distance, and a named "sparse" field for sparse vectors. The get_embedding_size() call automatically determines the correct vector dimension (384) from the model.

Why two types of vectors?

Dense vectors understand meaning. Searching for "money laundering risk" will find documents about "illicit fund transfers" even if those exact words never appear. Sparse vectors understand exact terms. Searching for "SAR" or "$5,000" will surface documents containing those precise strings.

Compliance queries need both. A query like "PEP screening for sanctioned jurisdictions" combines conceptual intent with regulatory terminology that must match exactly.

Add the document ingestion function

Now you need a function that takes each compliance document, generates both dense and sparse embeddings, and stores them as points in Qdrant with their metadata.

  • Add the following function below create_collection in ingest.py:
def ingest_documents(client: QdrantClient, documents: list):
    """Ingest documents with both dense and sparse embeddings."""
    points = []

    for i, doc in enumerate(documents):
        text = f"{doc['title']}. {doc['content']}"

        dense_embedding = client.get_embedding(text, model=DENSE_MODEL)
        sparse_embedding = client.get_sparse_embedding(text, model=SPARSE_MODEL)

        points.append(
            models.PointStruct(
                id=i,
                vector={
                    "dense": dense_embedding,
                    "sparse": sparse_embedding,
                },
                payload={
                    "doc_id": doc["id"],
                    "title": doc["title"],
                    "section": doc["section"],
                    "risk_category": doc["risk_category"],
                    "effective_date": doc["effective_date"],
                    "content": doc["content"],
                },
            )
        )

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

What does this code do?

For each document, the function concatenates the title and content into a single text string. This gives the embedding model richer context than content alone.

It then generates two embeddings via FastEmbed: get_embedding() for the dense vector and get_sparse_embedding() for the sparse vector. Both get stored in a PointStruct alongside the document metadata (section, risk category, effective date). This metadata will let you filter and display results later.

Run the ingestion script and verify in the dashboard

The final piece ties everything together: connect to Qdrant, load the JSON documents, create the collection, and ingest.

  • Add the main function at the bottom of ingest.py:
def main():
    client = QdrantClient("http://localhost:6333")

    with open("sample_documents.json", "r") as f:
        documents = json.load(f)

    create_collection(client)
    ingest_documents(client, documents)

    info = client.get_collection(COLLECTION_NAME)
    print(f"Collection info: {info.points_count} points indexed.")


if __name__ == "__main__":
    main()

What does this code do?

The main() function connects to your local Qdrant instance, loads the 8 KYC documents from sample_documents.json, creates the hybrid collection, and ingests all documents. The final get_collection() call confirms the point count so you know everything landed successfully.

  • Save ingest.py.

✔️ Awesome, I've got everything!

Great. Make sure you have saved ingest.py before continuing.

ⓧ I'd like to double check the full code

import json
from qdrant_client import QdrantClient, models

DENSE_MODEL = "sentence-transformers/all-MiniLM-L6-v2"
SPARSE_MODEL = "prithivida/Splade_PP_en_v1"
COLLECTION_NAME = "kyc_compliance"


def create_collection(client: QdrantClient):
    """Create a Qdrant collection with hybrid search support."""
    if client.collection_exists(COLLECTION_NAME):
        client.delete_collection(COLLECTION_NAME)

    client.create_collection(
        collection_name=COLLECTION_NAME,
        vectors_config={
            "dense": models.VectorParams(
                size=client.get_embedding_size(DENSE_MODEL),
                distance=models.Distance.COSINE,
            )
        },
        sparse_vectors_config={
            "sparse": models.SparseVectorParams()
        },
    )
    print(f"Collection '{COLLECTION_NAME}' created.")


def ingest_documents(client: QdrantClient, documents: list):
    """Ingest documents with both dense and sparse embeddings."""
    points = []

    for i, doc in enumerate(documents):
        text = f"{doc['title']}. {doc['content']}"

        dense_embedding = client.get_embedding(text, model=DENSE_MODEL)
        sparse_embedding = client.get_sparse_embedding(text, model=SPARSE_MODEL)

        points.append(
            models.PointStruct(
                id=i,
                vector={
                    "dense": dense_embedding,
                    "sparse": sparse_embedding,
                },
                payload={
                    "doc_id": doc["id"],
                    "title": doc["title"],
                    "section": doc["section"],
                    "risk_category": doc["risk_category"],
                    "effective_date": doc["effective_date"],
                    "content": doc["content"],
                },
            )
        )

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


def main():
    client = QdrantClient("http://localhost:6333")

    with open("sample_documents.json", "r") as f:
        documents = json.load(f)

    create_collection(client)
    ingest_documents(client, documents)

    info = client.get_collection(COLLECTION_NAME)
    print(f"Collection info: {info.points_count} points indexed.")


if __name__ == "__main__":
    main()
  • Run the ingestion script in your terminal:
python ingest.py

First run downloads embedding models

The first time you run this script, FastEmbed will download both embedding models from Hugging Face. This is a one-time download that may take a minute or two depending on your internet connection. Subsequent runs will use the cached models and complete in seconds.

  • You should see output like this:
Collection 'kyc_compliance' created.
Ingested 8 documents into 'kyc_compliance'.
Collection info: 8 points indexed.
  • Open your browser and navigate to http://localhost:6333/dashboard.
  • Click on the kyc_compliance collection.
  • Confirm that 8 points are indexed and that both vector fields (dense and sparse) appear in the collection configuration.

Not seeing 8 points?

  • Make sure your Qdrant Docker container is still running. Check with docker ps in your terminal.
  • Verify that sample_documents.json is in the same folder as ingest.py. A FileNotFoundError means the file path is wrong.
  • If you see a connection error, confirm Qdrant is accessible at http://localhost:6333 by visiting it in your browser.

Help me troubleshoot my Qdrant ingestion script. I ran python ingest.py but didn't get the expected 8 points indexed output.

Your compliance documents are now indexed with dual vectors, ready for intelligent retrieval. Next up, you will build the FastAPI search endpoints that query this collection using Reciprocal Rank Fusion to combine both signals into a single ranked result list.

Build the Hybrid Search API

Your compliance documents are indexed in Qdrant with both dense and sparse vectors. Now you need to make them searchable through an API.

The challenge with compliance search is that a simple keyword search misses conceptual matches ("money laundering risk" should find EDD guidelines), while pure semantic search can miss exact regulatory terms like "SAR" or "$5,000 threshold." In this step, you'll build a FastAPI search API that combines both approaches using Reciprocal Rank Fusion.

In this step, get ready to:
  • Create a search module with hybrid, dense-only, and sparse-only search functions.
  • Build a FastAPI application with a /search endpoint.
  • Add a /compare endpoint that demonstrates hybrid retrieval advantages.
Create the FastAPI application with a search endpoint

FastAPI lets you expose your search logic as a REST API with built-in request validation and interactive documentation. You'll create the API shell first, then build the search module it depends on.

  • Create a new file called app.py in your project folder and paste the following code:
import os
from datetime import datetime, timezone

from dotenv import load_dotenv
from fastapi import FastAPI, Query
from qdrant_client import QdrantClient

from search import hybrid_search, dense_only_search, sparse_only_search

load_dotenv()

app = FastAPI(
    title="KYC Compliance Search Assistant",
    description="Hybrid retrieval over KYC guidelines with AI-generated risk summaries",
)

qdrant = QdrantClient("http://localhost:6333")


@app.get("/search")
def search(
    q: str = Query(..., description="Search query"),
    limit: int = Query(5, description="Number of results"),
    alpha: float = Query(0.6, description="Balance: 0=keyword, 1=semantic"),
):
    """Perform hybrid search over compliance documents."""
    results = hybrid_search(qdrant, q, alpha=alpha, limit=limit)
    return {"query": q, "alpha": alpha, "results": results}

What does this code do?

This sets up a FastAPI application with a single /search endpoint. The endpoint accepts three query parameters:

  • q is the search query text (required).
  • limit controls how many results to return (default 5).
  • alpha balances keyword vs semantic search (default 0.6, meaning slightly more weight toward semantic understanding).

The extra imports (os, datetime, load_dotenv) will be used in a later step when you add AI-generated summaries.

  • Save app.py.

You cannot run this file yet because it imports from search.py, which does not exist. You'll create that next.

Implement the hybrid search function

The core of hybrid retrieval is combining two different search signals. Dense vectors capture meaning ("money laundering" matches "illicit fund transfers"), while sparse vectors capture exact terms ("SAR" matches documents containing that exact acronym). Reciprocal Rank Fusion merges both ranked lists without needing to normalize their scores.

  • Create a new file called search.py and paste the following code for the hybrid search function:
from qdrant_client import QdrantClient, models

DENSE_MODEL = "sentence-transformers/all-MiniLM-L6-v2"
SPARSE_MODEL = "prithivida/Splade_PP_en_v1"
COLLECTION_NAME = "kyc_compliance"


def hybrid_search(client: QdrantClient, query: str, alpha: float = 0.6, limit: int = 5):
    """
    Perform hybrid search combining dense and sparse retrieval with RRF.
    alpha is not used directly here since RRF handles fusion,
    but we use prefetch limits to bias toward one signal.
    """
    results = client.query_points(
        collection_name=COLLECTION_NAME,
        query=models.FusionQuery(fusion=models.Fusion.RRF),
        prefetch=[
            models.Prefetch(
                query=models.Document(text=query, model=DENSE_MODEL),
                using="dense",
                limit=20,
            ),
            models.Prefetch(
                query=models.Document(text=query, model=SPARSE_MODEL),
                using="sparse",
                limit=20,
            ),
        ],
        limit=limit,
    ).points

    return [
        {
            "score": point.score,
            "doc_id": point.payload["doc_id"],
            "title": point.payload["title"],
            "section": point.payload["section"],
            "risk_category": point.payload["risk_category"],
            "content": point.payload["content"],
        }
        for point in results
    ]

What does this code do?

The hybrid_search function uses Qdrant's Query API with RRF fusion. Here is how it works:

  • models.FusionQuery(fusion=models.Fusion.RRF) tells Qdrant to merge results from multiple prefetch queries using Reciprocal Rank Fusion. RRF scores each document based on its rank position rather than raw scores, which avoids the problem of dense and sparse scores being on completely different scales.
  • Each Prefetch runs a separate retrieval pass. The dense prefetch finds semantically similar documents, while the sparse prefetch finds keyword-matched documents.
  • models.Document(text=query, model=DENSE_MODEL) tells Qdrant to embed the query server-side using FastEmbed. No separate embedding step is needed in your code.

Now add the two comparison functions below the hybrid_search function in the same search.py file:

def dense_only_search(client: QdrantClient, query: str, limit: int = 5):
    """Pure semantic search using only dense vectors."""
    results = client.query_points(
        collection_name=COLLECTION_NAME,
        query=models.Document(text=query, model=DENSE_MODEL),
        using="dense",
        limit=limit,
    ).points

    return [
        {
            "score": point.score,
            "doc_id": point.payload["doc_id"],
            "title": point.payload["title"],
            "section": point.payload["section"],
            "risk_category": point.payload["risk_category"],
            "content": point.payload["content"],
        }
        for point in results
    ]


def sparse_only_search(client: QdrantClient, query: str, limit: int = 5):
    """Pure keyword search using only sparse vectors."""
    results = client.query_points(
        collection_name=COLLECTION_NAME,
        query=models.Document(text=query, model=SPARSE_MODEL),
        using="sparse",
        limit=limit,
    ).points

    return [
        {
            "score": point.score,
            "doc_id": point.payload["doc_id"],
            "title": point.payload["title"],
            "section": point.payload["section"],
            "risk_category": point.payload["risk_category"],
            "content": point.payload["content"],
        }
        for point in results
    ]

Why do you need separate search functions?

These functions isolate each retrieval mode so you can compare them side-by-side. dense_only_search queries the dense vector field directly (pure semantic), while sparse_only_search queries the sparse field (pure keyword matching via SPLADE).

Having all three modes lets you prove that hybrid search outperforms either individual approach on compliance queries. This is the /compare endpoint you'll add next.

  • Save search.py.
  • Start the FastAPI development server by running this command:
fastapi dev app.py
  • You should see output indicating the server started at http://127.0.0.1:8000.
  • Open a new terminal tab (keep the server running) and test the /search endpoint by running:
curl "http://127.0.0.1:8000/search?q=PEP+screening+requirements"

You should see a JSON response containing ranked compliance documents, with the PEP screening document appearing near the top of the results.

Server not starting?

  • If you see a ModuleNotFoundError, make sure your virtual environment is activated and you installed all dependencies in Step 1.
  • If you see an import error for search, confirm that search.py is in the same folder as app.py.
  • If port 8000 is already in use, stop whatever process is using it or add --port 8001 to the fastapi dev command.

help me debug my FastAPI server startup error

Add a compare endpoint

To demonstrate why hybrid search matters for compliance text, you'll add an endpoint that runs the same query through all three retrieval modes and returns results side-by-side.

  • In app.py, add the following /compare endpoint below the /search endpoint:
@app.get("/compare")
def compare_search_modes(
    q: str = Query(..., description="Search query"),
    limit: int = Query(5, description="Number of results"),
):
    """Compare pure dense, pure sparse, and hybrid search results."""
    return {
        "query": q,
        "dense_only": dense_only_search(qdrant, q, limit=limit),
        "sparse_only": sparse_only_search(qdrant, q, limit=limit),
        "hybrid_rrf": hybrid_search(qdrant, q, limit=limit),
    }

What does this endpoint show?

The /compare endpoint returns three sets of results for the same query. For a query like "high-value transaction from sanctioned jurisdiction," you'll see dense search finding semantically related documents about EDD and sanctions, sparse search finding documents with exact keyword matches, and hybrid RRF combining both for the best overall ranking.

  • Save app.py. The dev server auto-reloads with your changes.
  • Test the /compare endpoint by running:
curl "http://127.0.0.1:8000/compare?q=suspicious+transaction+over+5000"

You should see three arrays of results. Notice how the hybrid results combine the strengths of both approaches, surfacing the SAR filing document (exact keyword match on "$5,000") alongside the transaction monitoring document (semantic match on suspicious transactions).

Not seeing different results across modes?

  • With only 8 documents, the rankings may look similar across modes. Try a more specific query like CTR filing threshold cash where keyword matching gives an edge.
  • If you get a connection error, make sure your Qdrant container is still running on port 6333.

help me understand why my hybrid search results look the same as dense-only results

Why hybrid over pure semantic?

Compliance documents are unique because they combine precise regulatory terms ("$10,000 CTR threshold," "30 calendar days") with broad concepts ("risk-based approach," "adequate controls"). Pure semantic search misses exact numerical thresholds. Pure keyword search misses conceptual relationships.

Hybrid search with RRF gives you both. A document that matches on exact terms AND meaning ranks highest.

✔️ Awesome, I've got everything!

Great. Double check that both search.py and app.py are saved.

ⓧ I'd like to double check the full code

Here is the complete search.py:

from qdrant_client import QdrantClient, models

DENSE_MODEL = "sentence-transformers/all-MiniLM-L6-v2"
SPARSE_MODEL = "prithivida/Splade_PP_en_v1"
COLLECTION_NAME = "kyc_compliance"


def hybrid_search(client: QdrantClient, query: str, alpha: float = 0.6, limit: int = 5):
    """
    Perform hybrid search combining dense and sparse retrieval with RRF.
    alpha is not used directly here since RRF handles fusion,
    but we use prefetch limits to bias toward one signal.
    """
    results = client.query_points(
        collection_name=COLLECTION_NAME,
        query=models.FusionQuery(fusion=models.Fusion.RRF),
        prefetch=[
            models.Prefetch(
                query=models.Document(text=query, model=DENSE_MODEL),
                using="dense",
                limit=20,
            ),
            models.Prefetch(
                query=models.Document(text=query, model=SPARSE_MODEL),
                using="sparse",
                limit=20,
            ),
        ],
        limit=limit,
    ).points

    return [
        {
            "score": point.score,
            "doc_id": point.payload["doc_id"],
            "title": point.payload["title"],
            "section": point.payload["section"],
            "risk_category": point.payload["risk_category"],
            "content": point.payload["content"],
        }
        for point in results
    ]


def dense_only_search(client: QdrantClient, query: str, limit: int = 5):
    """Pure semantic search using only dense vectors."""
    results = client.query_points(
        collection_name=COLLECTION_NAME,
        query=models.Document(text=query, model=DENSE_MODEL),
        using="dense",
        limit=limit,
    ).points

    return [
        {
            "score": point.score,
            "doc_id": point.payload["doc_id"],
            "title": point.payload["title"],
            "section": point.payload["section"],
            "risk_category": point.payload["risk_category"],
            "content": point.payload["content"],
        }
        for point in results
    ]


def sparse_only_search(client: QdrantClient, query: str, limit: int = 5):
    """Pure keyword search using only sparse vectors."""
    results = client.query_points(
        collection_name=COLLECTION_NAME,
        query=models.Document(text=query, model=SPARSE_MODEL),
        using="sparse",
        limit=limit,
    ).points

    return [
        {
            "score": point.score,
            "doc_id": point.payload["doc_id"],
            "title": point.payload["title"],
            "section": point.payload["section"],
            "risk_category": point.payload["risk_category"],
            "content": point.payload["content"],
        }
        for point in results
    ]

Here is the complete app.py:

import os
from datetime import datetime, timezone

from dotenv import load_dotenv
from fastapi import FastAPI, Query
from qdrant_client import QdrantClient

from search import hybrid_search, dense_only_search, sparse_only_search

load_dotenv()

app = FastAPI(
    title="KYC Compliance Search Assistant",
    description="Hybrid retrieval over KYC guidelines with AI-generated risk summaries",
)

qdrant = QdrantClient("http://localhost:6333")


@app.get("/search")
def search(
    q: str = Query(..., description="Search query"),
    limit: int = Query(5, description="Number of results"),
    alpha: float = Query(0.6, description="Balance: 0=keyword, 1=semantic"),
):
    """Perform hybrid search over compliance documents."""
    results = hybrid_search(qdrant, q, alpha=alpha, limit=limit)
    return {"query": q, "alpha": alpha, "results": results}


@app.get("/compare")
def compare_search_modes(
    q: str = Query(..., description="Search query"),
    limit: int = Query(5, description="Number of results"),
):
    """Compare pure dense, pure sparse, and hybrid search results."""
    return {
        "query": q,
        "dense_only": dense_only_search(qdrant, q, limit=limit),
        "sparse_only": sparse_only_search(qdrant, q, limit=limit),
        "hybrid_rrf": hybrid_search(qdrant, q, limit=limit),
    }

Your search API is live and returning ranked compliance results. Next up, you'll add AI-generated risk summaries that synthesize retrieved passages into actionable compliance guidance.

Add AI-Generated Risk Summaries

Your hybrid search returns relevant compliance passages, but compliance officers need more than raw document matches. They need a synthesized risk assessment that tells them the severity level, which regulations apply, and what actions to take.

In this step, you'll integrate the OpenAI API to generate AI-powered risk summaries from your search results. The LLM takes the top retrieved passages and produces a structured assessment with a severity rating.

In this step, get ready to:
  • Create a summarizer module that generates risk assessments using OpenAI.
  • Add a /search-with-summary endpoint to the FastAPI application.
  • Test the full pipeline with a compliance scenario query.
Create the risk summary generator

The summarizer takes the top search results and sends them to an LLM with a system prompt tailored for compliance analysis. The LLM returns a structured risk assessment that a compliance officer can act on immediately.

  • Create a new file called summarizer.py in your project folder.
  • Add the imports and OpenAI client initialization by pasting this code:
import os
from openai import OpenAI
from dotenv import load_dotenv

# Load environment variables and initialize the OpenAI client
load_dotenv()

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

What does this code do?

  • load_dotenv() reads your .env file so the OpenAI API key is available as an environment variable.
  • OpenAI(api_key=os.getenv("OPENAI_API_KEY")) creates a client instance authenticated with your key.
  • Below the client initialization, add the generate_risk_summary function:
def generate_risk_summary(query: str, search_results: list) -> dict:
    """Generate an AI-powered risk summary from search results."""
    # Build context from the top 3 results with their metadata
    context = "\n\n".join(
        f"[{r['section']} - {r['risk_category']} Risk] {r['title']}\n{r['content']}"
        for r in search_results[:3]
    )

    # Send the query and retrieved context to the LLM
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": (
                    "You are a KYC compliance analyst. Based on the retrieved "
                    "compliance guidelines, provide a concise risk assessment. "
                    "Include: 1) Risk Severity (Low/Medium/High/Critical), "
                    "2) Applicable regulations and requirements, "
                    "3) Recommended actions. Keep the summary under 150 words."
                ),
            },
            {
                "role": "user",
                "content": f"Query: {query}\n\nRelevant Guidelines:\n{context}",
            },
        ],
        temperature=0.3,
    )

    summary_text = response.choices[0].message.content

    # Extract severity level from the generated text
    severity = "Medium"
    for level in ["Critical", "High", "Medium", "Low"]:
        if level.lower() in summary_text.lower():
            severity = level
            break

    return {
        "query": query,
        "severity": severity,
        "summary": summary_text,
        "sources": [r["doc_id"] for r in search_results[:3]],
        "confidence": search_results[0]["score"] if search_results else 0.0,
    }

What does this code do?

  • The function joins the top 3 search results into a formatted context string that includes each document's section, risk category, title, and full content.
  • It sends this context to gpt-4o-mini with a system prompt that instructs the model to act as a KYC compliance analyst.
  • temperature=0.3 keeps responses consistent and factual rather than creative.
  • After receiving the summary text, it scans for severity keywords in priority order (Critical first, then High, Medium, Low) to extract a structured severity rating.
  • The function returns a dictionary with the query, severity, full summary text, source document IDs, and a confidence score from the top search result.
  • Save summarizer.py.
  • Confirm the file exists by running:
ls summarizer.py

You should see summarizer.py listed in the output.

File not showing up?

  • Make sure you saved the file in the same project folder where search.py and app.py live.
  • Check that the filename is exactly summarizer.py (not summariser.py or summary.py).

Help me troubleshoot why my summarizer.py file isn't appearing in my project folder.

Why use RAG instead of sending all documents directly?

Sending your entire compliance document library to the LLM on every query would be expensive and hit token limits quickly. By using RAG (Retrieval-Augmented Generation), you first retrieve only the most relevant passages with hybrid search, then send just those to the LLM.

This keeps costs low, stays within context windows, and produces more focused answers because the model only sees relevant information.

✔️ Awesome, I've got everything!

Great. Double check you've saved summarizer.py and let's move on to wiring it into the API.

ⓧ I'd like to double check the full code

import os
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))


def generate_risk_summary(query: str, search_results: list) -> dict:
    """Generate an AI-powered risk summary from search results."""
    context = "\n\n".join(
        f"[{r['section']} - {r['risk_category']} Risk] {r['title']}\n{r['content']}"
        for r in search_results[:3]
    )

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": (
                    "You are a KYC compliance analyst. Based on the retrieved "
                    "compliance guidelines, provide a concise risk assessment. "
                    "Include: 1) Risk Severity (Low/Medium/High/Critical), "
                    "2) Applicable regulations and requirements, "
                    "3) Recommended actions. Keep the summary under 150 words."
                ),
            },
            {
                "role": "user",
                "content": f"Query: {query}\n\nRelevant Guidelines:\n{context}",
            },
        ],
        temperature=0.3,
    )

    summary_text = response.choices[0].message.content

    severity = "Medium"
    for level in ["Critical", "High", "Medium", "Low"]:
        if level.lower() in summary_text.lower():
            severity = level
            break

    return {
        "query": query,
        "severity": severity,
        "summary": summary_text,
        "sources": [r["doc_id"] for r in search_results[:3]],
        "confidence": search_results[0]["score"] if search_results else 0.0,
    }
Add the /search-with-summary endpoint

Now you need to wire the summarizer into your FastAPI application so it's accessible as an API endpoint. You also need a placeholder for the Power BI integration that you'll build in the next step.

  • Create a new file called powerbi.py in your project folder with this placeholder code:
def push_risk_assessment(assessment: dict):
    """Placeholder - full implementation in the next step."""
    pass

Why create a placeholder?

The app.py file will import push_risk_assessment from powerbi.py. Without this file, Python raises an ImportError when the server starts. You'll replace this stub with the real Power BI integration in the next step.

  • Save powerbi.py.
  • Now open app.py and find these import lines near the top:
from search import hybrid_search, dense_only_search, sparse_only_search
  • Add these two new import lines directly below it:
from summarizer import generate_risk_summary
from powerbi import push_risk_assessment
  • Next, find the /search endpoint function (the one with @app.get("/search")). Add the following new endpoint directly below the closing return statement of that function, before the /compare endpoint:
@app.get("/search-with-summary")
def search_with_summary(
    q: str = Query(..., description="Search query"),
    limit: int = Query(5, description="Number of results"),
):
    """Perform hybrid search and generate an AI risk summary."""
    results = hybrid_search(qdrant, q, limit=limit)
    summary = generate_risk_summary(q, results)
    summary["timestamp"] = datetime.now(timezone.utc).isoformat()

    # Push to Power BI if configured (placeholder for now)
    if os.getenv("POWERBI_DATASET_ID"):
        push_risk_assessment(summary)

    return {"query": q, "results": results, "risk_summary": summary}

What does this endpoint do?

  • It calls hybrid_search to retrieve the most relevant compliance passages.
  • It passes those results to generate_risk_summary which sends them to the LLM for analysis.
  • It adds a UTC timestamp to the summary for audit trail purposes.
  • It conditionally pushes results to Power BI if the dataset ID environment variable is configured.
  • Save app.py.

Seeing an ImportError?

  • Make sure powerbi.py exists in the same folder as app.py. The import will fail without it.
  • Check that summarizer.py is also in the same folder. Both files must be importable.
  • Verify there are no typos in the import lines. The function names must match exactly.

Help me fix an ImportError when starting my FastAPI server after adding the summarizer and powerbi imports.

✔️ Awesome, I've got everything!

Great. Double check you've saved app.py and let's test the new endpoint.

ⓧ I'd like to double check the full code

import os
from datetime import datetime, timezone

from dotenv import load_dotenv
from fastapi import FastAPI, Query
from qdrant_client import QdrantClient

from search import hybrid_search, dense_only_search, sparse_only_search
from summarizer import generate_risk_summary
from powerbi import push_risk_assessment

load_dotenv()

app = FastAPI(
    title="KYC Compliance Search Assistant",
    description="Hybrid retrieval over KYC guidelines with AI-generated risk summaries",
)

qdrant = QdrantClient("http://localhost:6333")


@app.get("/search")
def search(
    q: str = Query(..., description="Search query"),
    limit: int = Query(5, description="Number of results"),
    alpha: float = Query(0.6, description="Balance: 0=keyword, 1=semantic"),
):
    """Perform hybrid search over compliance documents."""
    results = hybrid_search(qdrant, q, alpha=alpha, limit=limit)
    return {"query": q, "alpha": alpha, "results": results}


@app.get("/search-with-summary")
def search_with_summary(
    q: str = Query(..., description="Search query"),
    limit: int = Query(5, description="Number of results"),
):
    """Perform hybrid search and generate an AI risk summary."""
    results = hybrid_search(qdrant, q, limit=limit)
    summary = generate_risk_summary(q, results)
    summary["timestamp"] = datetime.now(timezone.utc).isoformat()

    if os.getenv("POWERBI_DATASET_ID"):
        push_risk_assessment(summary)

    return {"query": q, "results": results, "risk_summary": summary}


@app.get("/compare")
def compare_search_modes(
    q: str = Query(..., description="Search query"),
    limit: int = Query(5, description="Number of results"),
):
    """Compare pure dense, pure sparse, and hybrid search results."""
    return {
        "query": q,
        "dense_only": dense_only_search(qdrant, q, limit=limit),
        "sparse_only": sparse_only_search(qdrant, q, limit=limit),
        "hybrid_rrf": hybrid_search(qdrant, q, limit=limit),
    }
Test the complete pipeline

Time to verify that the search, summarization, and structured output all work together. You'll query for a realistic compliance scenario and check that the AI-generated summary makes sense.

Make sure your OpenAI API key is set

Before testing, confirm that your .env file has a real OpenAI API key in the OPENAI_API_KEY field (not the placeholder text). If you haven't added your key yet, update .env now.

  • Stop your running FastAPI server by pressing Ctrl+C in the terminal where it's running.
  • Restart the server by running:
fastapi dev app.py
  • Open a new terminal tab and test the /search-with-summary endpoint by running:
curl "http://localhost:8000/search-with-summary?q=high-value+transaction+from+sanctioned+jurisdiction"
  • You should see a JSON response containing both the results array (retrieved documents) and a risk_summary object with severity, summary, sources, and timestamp fields.

What should I see?

The risk_summary.severity should be "Critical" or "High" for this query because it involves sanctions and high-value transactions. The summary field contains the LLM-generated assessment with applicable regulations and recommended actions.

The sources array shows which document IDs the summary was built from (e.g., sanctions-001, edd-001). The confidence score reflects how well the top result matched your query.

Getting a 500 error or timeout?

  • Check your terminal running the FastAPI server for error details. An AuthenticationError means your OpenAI API key is invalid or missing from .env.
  • If you see a ModuleNotFoundError for openai, make sure your virtual environment is activated and openai is installed.
  • If the Qdrant container is not running, start it again with docker start followed by the container ID.

Help me debug why my /search-with-summary endpoint is returning a 500 error.

Your search assistant now combines retrieval precision with AI-powered analysis. Next up, you'll connect this pipeline to Power BI so risk assessments surface on a live dashboard.

Push Risk Summaries to Power BI

Your search assistant now generates intelligent risk summaries from compliance documents. The next challenge is making those summaries visible to decision-makers who live in dashboards, not terminals.

In this step, you will connect your Python application to Power BI using a Push Dataset. This lets your code send risk assessment rows directly to a Power BI table in real time, where analysts can build reports and monitor compliance activity.

In this step, get ready to:
  • Register an Azure AD application and configure it to access the Power BI REST API.
  • Write a Python module that authenticates via MSAL and pushes risk assessment data to a Push Dataset.
  • Create the Push Dataset, run compliance queries, and verify results appear in Power BI.
Register an Azure AD application

Your Python code needs permission to write data into Power BI on your behalf. You will register a service principal in Microsoft Entra ID (formerly Azure AD) and grant it the Dataset.ReadWrite.All permission.

  • Open the Azure Portal at https://portal.azure.com and sign in.
  • Search for App registrations in the top search bar and select it.
  • Click New registration.
  • Enter KYC Compliance Push as the application name.
  • Under Supported account types, select Accounts in this organizational directory only.
  • Leave Redirect URI blank and click Register.
  • On the Overview page, copy the Application (client) ID and Directory (tenant) ID. You will need both shortly.

Next, create a client secret so your Python code can authenticate without a browser.

  • In the left menu under Manage, click Certificates & secrets.
  • Click New client secret.
  • Enter a description like KYC push secret and set an expiration (6 months is fine for a project).
  • Click Add.
  • Copy the Value column immediately. This is your client secret and it will not be shown again.

Now grant the Power BI API permission.

  • In the left menu, click API permissions.
  • Click Add a permission.
  • Select Power BI Service from the Microsoft APIs tab.
  • Select Delegated permissions.
  • Check Dataset.ReadWrite.All and click Add permissions.
  • Click Grant admin consent for your tenant, then confirm with Yes.

Why a service principal?

A service principal lets your Python script authenticate to Power BI without requiring a human to sign in each time. It uses a client ID and secret instead of interactive browser login.

The Dataset.ReadWrite.All permission specifically allows your app to create and write to Push Datasets.

You also need to enable service principal access in your Power BI workspace.

  • Open the Power BI service at https://app.powerbi.com.
  • Navigate to your workspace (or create a new one called KYC Compliance).
  • Click the three-dot menu on the workspace, then select Workspace access.
  • Add your service principal (search by the app name KYC Compliance Push) as a Member or Admin.
  • Click Add.

How to find your workspace ID

Your workspace ID is the GUID in the Power BI URL after /groups/. For example, in https://app.powerbi.com/groups/abc123-def456/..., the workspace ID is abc123-def456.

Finally, update your .env file with the real credentials you just collected.

  • Open .env in your editor.
  • Replace your-tenant-id-here with the Directory (tenant) ID from the app's Overview page.
  • Replace your-client-id-here with the Application (client) ID.
  • Replace your-client-secret-here with the secret value you copied.
  • Replace your-workspace-id-here with the workspace GUID from the Power BI URL.
  • Leave POWERBI_DATASET_ID as your-dataset-id-here for now. You will fill it in after creating the dataset.
  • Save .env.

Can't find Power BI Service in the permissions list?

Make sure your Azure AD tenant has a Power BI Pro or Premium license assigned. The Power BI Service API only appears in tenants with an active Power BI subscription.

help me troubleshoot Azure AD permissions for Power BI

Write the Power BI integration module

With credentials in place, you will create powerbi.py. This module handles MSAL authentication, pushing rows to the Push Dataset, and a one-time dataset creation function.

  • Install the MSAL library by running:
pip install msal

What is MSAL?

MSAL (Microsoft Authentication Library) handles the OAuth 2.0 client credentials flow. It requests an access token from Microsoft Entra ID using your client ID and secret, then your code attaches that token to Power BI API requests.

  • Create a new file called powerbi.py in your project folder.
  • Add the imports, configuration constants, and authentication function by pasting the following code:
import os
import json

import msal
import requests
from dotenv import load_dotenv

load_dotenv()

# Load Power BI credentials from environment variables
TENANT_ID = os.getenv("AZURE_TENANT_ID", "")
CLIENT_ID = os.getenv("AZURE_CLIENT_ID", "")
CLIENT_SECRET = os.getenv("AZURE_CLIENT_SECRET", "")
WORKSPACE_ID = os.getenv("POWERBI_WORKSPACE_ID", "")
DATASET_ID = os.getenv("POWERBI_DATASET_ID", "")
TABLE_NAME = os.getenv("POWERBI_TABLE_NAME", "RiskAssessments")


def get_access_token() -> str:
    """Authenticate to Power BI using MSAL service principal."""
    app = msal.ConfidentialClientApplication(
        client_id=CLIENT_ID,
        authority=f"https://login.microsoftonline.com/{TENANT_ID}",
        client_credential=CLIENT_SECRET,
    )
    scope = ["https://analysis.windows.net/powerbi/api/.default"]
    result = app.acquire_token_for_client(scopes=scope)
    if "access_token" in result:
        return result["access_token"]
    raise Exception(f"Failed to get token: {result.get('error_description')}")

What does this code do?

The constants at the top pull your Azure credentials from .env so nothing sensitive is hardcoded.

The get_access_token() function creates an MSAL confidential client application, requests a token using the client credentials flow, and returns the bearer token. The scope https://analysis.windows.net/powerbi/api/.default tells Azure you want access to all Power BI API permissions granted to the app.

  • Below the get_access_token function, add the push_risk_assessment function that sends data to Power BI:

def push_risk_assessment(assessment: dict):
    """Push a risk assessment to Power BI Push Dataset."""
    if not all([TENANT_ID, CLIENT_ID, CLIENT_SECRET, WORKSPACE_ID, DATASET_ID]):
        print("Power BI credentials not configured. Skipping push.")
        return

    access_token = get_access_token()
    url = (
        f"https://api.powerbi.com/v1.0/myorg/groups/{WORKSPACE_ID}"
        f"/datasets/{DATASET_ID}/tables/{TABLE_NAME}/rows"
    )
    row = {
        "Query": assessment["query"][:200],
        "Severity": assessment["severity"],
        "Summary": assessment["summary"][:500],
        "Timestamp": assessment["timestamp"],
        "Confidence": round(assessment["confidence"], 4),
        "Sources": ", ".join(assessment["sources"]),
    }
    response = requests.post(
        url=url,
        data=json.dumps([row]),
        headers={"Content-Type": "application/json", "Authorization": f"Bearer {access_token}"},
    )
    if response.status_code == 200:
        print(f"Pushed risk assessment to Power BI: {assessment['severity']}")
    else:
        print(f"Power BI push failed: {response.status_code} - {response.text}")

What does this code do?

The function first checks that all credentials are present. If any are missing, it gracefully skips the push (this is why app.py could run even before Power BI was configured).

It formats the assessment into a row matching the Push Dataset schema, then POSTs the JSON array to the Power BI rows endpoint. The [:200] and [:500] slices prevent exceeding column length limits.

  • Finally, add the create_push_dataset function at the bottom of the file. This is a one-time setup function you will call manually:

def create_push_dataset(access_token: str) -> str:
    """Create the Power BI Push Dataset (run once during setup)."""
    url = (
        f"https://api.powerbi.com/v1.0/myorg/groups/{WORKSPACE_ID}"
        f"/datasets?defaultRetentionPolicy=basicFIFO"
    )
    body = {
        "name": "KYC Risk Assessments",
        "defaultMode": "Push",
        "tables": [
            {
                "name": TABLE_NAME,
                "columns": [
                    {"name": "Query", "dataType": "string"},
                    {"name": "Severity", "dataType": "string"},
                    {"name": "Summary", "dataType": "string"},
                    {"name": "Timestamp", "dataType": "DateTime"},
                    {"name": "Confidence", "dataType": "Double"},
                    {"name": "Sources", "dataType": "string"},
                ],
            }
        ],
    }
    response = requests.post(
        url=url,
        data=json.dumps(body),
        headers={"Content-Type": "application/json", "Authorization": f"Bearer {access_token}"},
    )
    if response.status_code == 201:
        dataset_id = response.json()["id"]
        print(f"Created Push Dataset: {dataset_id}")
        return dataset_id
    else:
        raise Exception(f"Failed to create dataset: {response.status_code} - {response.text}")

What does this code do?

This function sends a POST request to create a new Push Dataset in your Power BI workspace. The defaultRetentionPolicy=basicFIFO parameter tells Power BI to automatically remove old rows when storage fills up.

The schema defines six columns matching the row structure in push_risk_assessment. On success, it returns the new dataset ID which you will add to your .env file.

  • Save powerbi.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved powerbi.py with all three functions.

ⓧ I'd like to double check the full code

import os
import json

import msal
import requests
from dotenv import load_dotenv

load_dotenv()

TENANT_ID = os.getenv("AZURE_TENANT_ID", "")
CLIENT_ID = os.getenv("AZURE_CLIENT_ID", "")
CLIENT_SECRET = os.getenv("AZURE_CLIENT_SECRET", "")
WORKSPACE_ID = os.getenv("POWERBI_WORKSPACE_ID", "")
DATASET_ID = os.getenv("POWERBI_DATASET_ID", "")
TABLE_NAME = os.getenv("POWERBI_TABLE_NAME", "RiskAssessments")


def get_access_token() -> str:
    """Authenticate to Power BI using MSAL service principal."""
    app = msal.ConfidentialClientApplication(
        client_id=CLIENT_ID,
        authority=f"https://login.microsoftonline.com/{TENANT_ID}",
        client_credential=CLIENT_SECRET,
    )
    scope = ["https://analysis.windows.net/powerbi/api/.default"]
    result = app.acquire_token_for_client(scopes=scope)
    if "access_token" in result:
        return result["access_token"]
    raise Exception(f"Failed to get token: {result.get('error_description')}")


def push_risk_assessment(assessment: dict):
    """Push a risk assessment to Power BI Push Dataset."""
    if not all([TENANT_ID, CLIENT_ID, CLIENT_SECRET, WORKSPACE_ID, DATASET_ID]):
        print("Power BI credentials not configured. Skipping push.")
        return

    access_token = get_access_token()
    url = (
        f"https://api.powerbi.com/v1.0/myorg/groups/{WORKSPACE_ID}"
        f"/datasets/{DATASET_ID}/tables/{TABLE_NAME}/rows"
    )
    row = {
        "Query": assessment["query"][:200],
        "Severity": assessment["severity"],
        "Summary": assessment["summary"][:500],
        "Timestamp": assessment["timestamp"],
        "Confidence": round(assessment["confidence"], 4),
        "Sources": ", ".join(assessment["sources"]),
    }
    response = requests.post(
        url=url,
        data=json.dumps([row]),
        headers={"Content-Type": "application/json", "Authorization": f"Bearer {access_token}"},
    )
    if response.status_code == 200:
        print(f"Pushed risk assessment to Power BI: {assessment['severity']}")
    else:
        print(f"Power BI push failed: {response.status_code} - {response.text}")


def create_push_dataset(access_token: str) -> str:
    """Create the Power BI Push Dataset (run once during setup)."""
    url = (
        f"https://api.powerbi.com/v1.0/myorg/groups/{WORKSPACE_ID}"
        f"/datasets?defaultRetentionPolicy=basicFIFO"
    )
    body = {
        "name": "KYC Risk Assessments",
        "defaultMode": "Push",
        "tables": [
            {
                "name": TABLE_NAME,
                "columns": [
                    {"name": "Query", "dataType": "string"},
                    {"name": "Severity", "dataType": "string"},
                    {"name": "Summary", "dataType": "string"},
                    {"name": "Timestamp", "dataType": "DateTime"},
                    {"name": "Confidence", "dataType": "Double"},
                    {"name": "Sources", "dataType": "string"},
                ],
            }
        ],
    }
    response = requests.post(
        url=url,
        data=json.dumps(body),
        headers={"Content-Type": "application/json", "Authorization": f"Bearer {access_token}"},
    )
    if response.status_code == 201:
        dataset_id = response.json()["id"]
        print(f"Created Push Dataset: {dataset_id}")
        return dataset_id
    else:
        raise Exception(f"Failed to create dataset: {response.status_code} - {response.text}")

Getting a ModuleNotFoundError for msal?

Make sure your virtual environment is activated before running pip install msal. If you see the error at import time, run the install again in your active environment.

help me fix a ModuleNotFoundError for msal in my Python project

Create the dataset and verify the integration

You will now run the dataset creation function once, grab the dataset ID, restart your FastAPI server, and send real queries through the pipeline.

  • Stop your running FastAPI server (press Ctrl+C in the terminal where it is running).
  • Create the Push Dataset by running this command:
python -c "from powerbi import get_access_token, create_push_dataset; token = get_access_token(); create_push_dataset(token)"

What should I see?

You should see output like Created Push Dataset: xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx. The long string is your new dataset ID.

  • Copy the dataset ID from the terminal output.
  • Open .env and replace your-dataset-id-here with the dataset ID you just copied.
  • Save .env.
  • Restart the FastAPI server by running:
fastapi dev app.py
  • Run a compliance query by opening this URL in your browser:
http://localhost:8000/search-with-summary?q=high-value+transaction+from+sanctioned+jurisdiction

Check your terminal. You should see a line like Pushed risk assessment to Power BI: High confirming the push succeeded.

  • Run two or three more queries to populate the dataset with variety:
http://localhost:8000/search-with-summary?q=PEP+screening+requirements+for+government+officials
http://localhost:8000/search-with-summary?q=SAR+filing+thresholds+for+transactions+over+25000

What does each query do?

Each time you hit the /search-with-summary endpoint, the pipeline performs hybrid search, generates an AI risk summary, and pushes a row to your Power BI Push Dataset. Each row includes the query, severity, summary text, timestamp, confidence score, and source document IDs.

Now verify the data arrived in Power BI and create a quick dashboard tile.

  • Return to the Power BI service at https://app.powerbi.com.
  • Open your workspace and find the KYC Risk Assessments dataset.
  • Click Create a report (or + New then Report).
  • Add a Table visual and drag the Timestamp, Query, Severity, and Summary columns into it.
  • To add severity color coding, select the table, open Format settings, and apply conditional formatting on the Severity column (red for Critical, orange for High, yellow for Medium, green for Low).
  • Pin the visual to a new dashboard.

You should see 3-4 rows showing your recent compliance queries with their risk levels and AI-generated summaries.

Power BI push returning 401 or 403?

A 401 means the access token is invalid or expired. Double check that AZURE_TENANT_ID, AZURE_CLIENT_ID, and AZURE_CLIENT_SECRET are correct in your .env file.

A 403 means the service principal does not have workspace access. Make sure you added it as a Member or Admin in the workspace access settings. Also verify that the Power BI admin has enabled Allow service principals to use Power BI APIs in the tenant settings under Developer settings.

help me troubleshoot Power BI Push Dataset authentication errors

Your compliance search assistant is now feeding risk assessments into an enterprise dashboard in real time. In the Secret Mission, you will add a threshold-based alert system that pushes urgent notifications through a streaming dataset for near-instant visibility.

Secret mission

Add a Compliance Alert System

Your search assistant returns risk summaries on demand. But what if a query is so dangerous it should trigger an immediate alert? Build an automated alert pipeline that monitors for high-risk patterns and pushes real-time alerts to Power BI the instant risk scores cross a threshold.

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.

The Qdrant container runs locally at no cost. Power BI Push Datasets require a Pro or Premium license ($10-20/user/month) and persist in your workspace until deleted. OpenAI API costs were negligible for this project.

Cost warning

If you have a Power BI Pro license, your Push Dataset and Streaming Dataset remain active in your workspace. They do not incur additional charges beyond your existing license, but leaving unused datasets can clutter your workspace.

If you are on a Power BI Pro trial, be aware the trial may expire. Delete the datasets before the trial ends to avoid losing access to cleanup actions.

Resources you used:

  • Qdrant Docker container running on ports 6333 and 6334.
  • Docker named volume qdrant-storage storing indexed vector data.
  • Project folder on your Desktop with Python files, virtual environment, and .env credentials.
  • Power BI Push Dataset (KYC Risk Assessments) in your Power BI workspace.
  • Power BI Streaming Dataset (if you completed the Secret Mission) in your Power BI workspace.
  • Azure AD app registration (service principal) used for Power BI authentication.

Keep everything running

No action needed. Choose this if you are still actively building or experimenting with the search assistant.

  • Your Qdrant Docker container continues running locally at no cost.
  • Your Power BI datasets persist in your workspace and remain available for dashboards.
  • Your Azure AD app registration stays active for future API calls.
  • Your project folder and virtual environment remain on your Desktop.

Pause - I'll come back to this later

Shut down running processes to free up system resources. Your data and files remain intact for when you return.

  • Stop the Qdrant Docker container by running:
  • Find your container ID first with:
docker ps
  • Then stop it:
docker stop <container_id>
  • Replace <container_id> with the ID shown in the docker ps output.
  • Your data persists in the qdrant-storage named volume. When you are ready to resume, run docker start <container_id> and everything picks up where you left off.
  • Power BI datasets remain in your workspace but stop receiving new data until you restart the FastAPI server.

Delete - I don't want to use this again

Remove all project resources and start fresh. This cannot be undone.

Remove the Qdrant Docker container and its data volume.

  • Find your container ID:
docker ps -a
  • Stop and remove the container:
docker stop <container_id>
docker rm <container_id>
  • Remove the named volume:
docker volume rm qdrant-storage

Delete your project folder.

  • On macOS/Linux:
cd ~/Desktop
rm -rf <your-project-folder>
  • On Windows (PowerShell):
cd ~\Desktop
Remove-Item -Recurse -Force <your-project-folder>
  • Replace <your-project-folder> with the name of the folder you created for this project.

Delete your Power BI datasets.

  • Open the Power BI service and navigate to the workspace containing your datasets.
  • Select More options (...) next to the KYC Risk Assessments dataset, then choose Delete.
  • Select Delete to confirm.
  • Repeat for the Streaming Dataset if you completed the Secret Mission.
  • Delete any dashboard tiles that referenced these datasets.

Delete your Azure AD app registration.

  • Sign in to the Microsoft Entra admin center.
  • Browse to Entra ID > App registrations.
  • Select the application you registered for this project.
  • From the Overview page, select Delete.
  • Read the deletion consequences, check the confirmation box, and select Delete to confirm.

Revoke your OpenAI API key (optional).

  • If you created a dedicated API key for this project, revoke it from your OpenAI account dashboard under API keys.
  • If you use this key for other projects, keep it and simply remove the .env file (already handled by deleting the project folder).

Nice Work!

Nice Work!

You just built a production-grade compliance search assistant that financial institutions actually need.

What you learned:

  • Built a hybrid search engine combining dense semantic embeddings with sparse keyword matching over compliance documents, fused with Reciprocal Rank Fusion to balance precision and recall for regulatory queries.
  • Created a FastAPI-powered search API with a /compare endpoint that demonstrates the value of hybrid retrieval over pure semantic or pure keyword search.
  • Integrated OpenAI-powered risk summaries with Power BI Push Datasets to surface AI-generated compliance insights alongside operational data.
  • Built a real-time compliance alert system that monitors for high-risk query patterns and pushes alerts to a Power BI Streaming Dataset when risk scores exceed a threshold.

Ready to quiz yourself?