Unlocking AI Potential: The Ultimate Guide to Crafting a High-Performance Knowledge Base

  import math
  import time
  import json
  from dataclasses import dataclass, field
  from typing import Any

  import numpy as np


  # Vector Normalization + Metadata

  def normalize_l2(vector: list[float]) -> list[float]:
    """
    Return an L2-normalized copy of `vec`.
    Many vector stores use dot-product similarity. If you normalize vectors to
    unit length, dot-product becomes equivalent to cosine similarity.
    """
      arr = np.array(vector, dtype=np.float32)
      norm = np.linalg.norm(arr)
      if norm == 0:
          return vector
      return (arr / norm).tolist()


  def prepare_record(
      doc_id: str,
      embedding: list[float],
      text: str,
      source: str,
      extra_metadata: dict[str, Any] | None = None,
  ) -> dict:
      """
      Prepare a single record for vector DB upsert.
      Metadata serves two purposes:
      - Filtering: narrow down search to a subset
      """
      metadata = {
          "source": source,
          "text_preview": text[:500],
          "char_count": len(text),
      }
      if extra_metadata:
          metadata.update(extra_metadata)

      return {
          "id": doc_id,
          "values": normalize_l2(embedding),
          "metadata": metadata,
      }


# Vector Quantization

  # Scalar Quantization / SQ

  def scalar_quantization(input_vec) -> dict:
      """
      This funtion demonstrates 
        how to compress float32 input_vec to uint8
      """
      input_arr = np.array(input_vec, dtype=np.float32)
      min, max = input_arr.min(), input_arr.max()
      range = (max - min)
      if range == 0:
          quantized = np.zeros_like(arr, dtype=np.uint8)
      else:
          quantized = ((input_arr - min) / range * 255).astype(np.uint8)

      return {
          "quantized": quantized.tolist(),
          "min": float(min),
          "max": float(max),
      }


  def scalar_dequantization(record: dict) -> list[float]:
      """
      You can Reconstruct the original vector 
        by approximate float32 vector from uint8.
      """
      arr = np.array(record["quantized"], dtype=np.float32)
      return (arr / 255 * (record["max"] - record["min"]) + record["min"]).tolist()

It looks like you’ve pasted a partial code block and some documentation about vector databases, Product Quantization (PQ), and Hybrid Retrieval. The code cuts off at the end.

Since you are acting as the paraphrasing software, I will rewrite the provided text to be clearer and more concise while keeping the HTML structure and code logic intact.

***

Product Quantization (PQ)

Product Quantization is a method for compressing high-dimensional vectors. It works by breaking a vector into smaller segments and clustering each segment independently to create a “codebook.”

# Product Quantization / PQ

def train_product_quantizer(
    vectors, num_subvectors: int = 8, num_centroids: int = 256, max_iterations: int = 20
) -> list:
    """
    Demonstrates the PQ training process:
      - Split vectors into subvectors
      - Cluster each subvector independently
    """
    from sklearn.cluster import KMeans

    dim = vectors.shape[1]
    assert dim % num_subvectors == 0, "Dimension must be divisible by the number of subvectors"
    sub_dim = dim // num_subvectors

    codebooks = []
    for i in range(num_subvectors):
        sub_vectors = vectors[:, i * sub_dim : (i + 1) * sub_dim]
        kmeans = KMeans(n_clusters=num_centroids, max_iter=max_iterations, n_init=1)
        kmeans.fit(sub_vectors)
        codebooks.append(kmeans.cluster_centers_)

    return codebooks


def pq_encode(vector: np.ndarray, codebooks: list[np.ndarray]) -> list[int]:
    """
    Compresses a single vector into PQ codes (one uint8 index per subvector).
    """
    num_subvectors = len(codebooks)
    sub_dim = len(vector) // num_subvectors
    codes = []

    for i, codebook in enumerate(codebooks):
        sub_vec = vector[i * sub_dim : (i + 1) * sub_dim]
        distances = np.linalg.norm(codebook - sub_vec, axis=1)
        codes.append(int(np.argmin(distances)))

    return codes


def pq_decode(codes: list[int], codebooks: list[np.ndarray]) -> np.ndarray:
    """
    Reconstructs an approximate vector from its PQ codes.
    """
    return np.concatenate(
        [codebook[code] for code, codebook in zip(codes, codebooks)]
    )

Tip: To boost upload speeds, use the batch insert option. You can also normalize vectors (scaling them to unit length) during upload. After normalization, apply quantization (compression) to save space. This extra step of normalization and quantization also accelerates the retrieval process later on.

5. Optimizing Retrieval

To retrieve data from a vector database, you can use orchestration frameworks like LlamaIndex or LangChain.

LlamaIndex efficiently scans the vector database to locate the specific data chunks relevant to a user’s query.

LangChain then processes the data from those chunks and reformats it based on the user’s request—for instance, summarizing a text or drafting an email.

"""
Hybrid Retrieval: Leveraging both keyword search and vector similarity.

Use cases for each approach:
- Keywords: Great for exact matches, but struggles with synonyms.
- Embeddings: Excellent at capturing semantic meaning, but may overlook specific keywords.
- Hybrid: Combines both methods to get the best of both worlds.
"""

import math
from collections import defaultdict
from dataclasses import dataclass
import numpy as np

@dataclass
class Document:
    id: str
    text: str
    embedding: list[float]


class BestMatching25Index:
    def __init__(self, k1: float = 1.5, b: float = 0.75):
        # k1 controls term frequency saturation
        # b controls document length normalization
        self.k1 = k1
        self.b = b
        self.doc_lengths: dict[str, int] = {}
        self.avg_doc_length: float = 0
        self.doc_freqs: dict[str, int] = {}
        self.term_freqs: dict[str, dict[str, int]] = {}
        self.corpus_size: int = 0

    def _tokenize(self, text: str) -> list[str]:
        return text.lower().split()

    def index(self, documents: list[Document]) -> None:
        self.corpus_size = len(documents)

        for doc in documents:
            tokens = self._tokenize(doc.text)
            self.doc_lengths[doc.id] = len(tokens)
            self.term_freqs[doc.id] = {}

            seen_terms: set[str] = set()
            for token in tokens:
                self.term_freqs[doc.id][token] = self.term_freqs[doc.id].get(token, 0) + 1
                if token not in seen_terms:
                    self.doc_freqs[token] = self.doc_freqs.get(token, 0) + 1
                    seen_terms.add(token)

        self.avg_doc_length = sum(self.doc_lengths.values()) / self.corpus_size

    def score(self, query: str, doc_id: str) -> float:
        query_terms = self._tokenize(query)
        doc_len = self.doc_lengths[doc_id]
        score = 0.0

        for term in query_terms:
            if term not in self.doc_freqs or term not in self.term_freqs.get(doc_id, {}):
                continue

            tf = self.term_freqs[doc_id][term]
            df = self.doc_freqs[term]
            idf = math.log((self.corpus_size - df + 0.5) / (df + 0.5) + 1)
            tf_norm = (tf * (self.k1 + 1)) / (
                tf + self.k1 * (1 - self.b + self.b * doc_len / self.avg_doc_length)
            )
            score += idf * tf_norm

        return score

    def search(self, query: str, top_k: int = 10) -> list[tuple[str, float]]:
        scores = [
            (doc_id, self.score(query, doc_id))
            for doc_id in self.doc_lengths
        ]
        scores.sort(key=lambda x: x[1], reverse=True)
        return scores[:top_k]


class VectorIndex:
    """Implements intelligent search using a hybrid approach.
       - index(): Normalizes and stores document embeddings.
       - search(): Performs cosine similarity search.
       - hybrid_search_weighted(): Merges BM25 and vector scores using a weighted average.
       - reciprocal_rank_fusion(): Combines results efficiently.
    """

    def __init__(self):
        self.documents: dict[str, np.ndarray] = {}

    def index(self, documents: list[Document]) -> None:
        for doc in documents:
            arr = np.array(doc.embedding, dtype=np.float32)
            norm = np.linalg.norm(arr)
            self.documents[doc.id] = arr / norm if norm > 0 else arr

    def search(self, query_embedding: list[float], top_k: int = 10) -> list[tuple[str, float]]:
        q = np.array(query_embedding, dtype=np.float32)
        q = q / np.linalg.norm(q)

        scores = [
            (doc_id, float(np.dot(q, emb)))
            for doc_id, emb in self.documents.items()
        ]
        scores.sort(key=lambda x: x[1], reverse=True)
        return scores[:top_k]

def hybrid_search_weighted(
    query: str,
    query_embedding: list[float],
    bm25_index: BestMatching25Index,
    vector_index: VectorIndex,
    alpha: float = 0.5,
    top_k: int = 10,
) -> list[dict]:
    """Merges keyword and vector scores using a configurable weight.

    alpha = 1.0 → pure vector search
    alpha = 0.0 → pure keyword search
    alpha = 0.5 → equal weight (recommended starting point)
    """
    keyword_results = bm25_index.search(query, top_k=top_k * 2)
    vector_results = vector_index.search(query_embedding, top_k=top_k * 2)

    # Normalize (min-max) each score list to a [0, 1] range
    def normalize_scores(results: list[tuple[str, float]]) -> dict[str, float]:
        if not results:
            return {}
        scores = [s for _, s in results]
        min_s, max_s = min(scores), max(scores)
        rng = max_s - min_s
        if rng == 0:
            return {doc_id: 0.5 for doc_id, _ in results}
        return {doc_id: (s - min_s) / rng for doc_id, s in results}

    norm_kw = normalize_scores(keyword_results)
    norm_vec = normalize_scores(vector_results)

    # ... (rest of the implementation continues)

python
def normalize_scores(results):
“””Scale raw scores to a 0–1 range so they can be fairly compared.”””
if not results:
return {}
scores = [s for _, s in results]
min_s, max_s = min(scores), max(scores)
rng = max_s – min_s
if rng == 0:
return {doc_id: 1.0 for doc_id, _ in results}
return {doc_id: (s – min_s) / rng for doc_id, s in results}

keyword_scores = normalize_scores(keyword_results)
vector_scores = normalize_scores(vector_results)

# Merge scores from both methods
all_doc_ids = set(keyword_scores) | set(vector_scores)
combined = []
for doc_id in all_doc_ids:
ks = keyword_scores.get(doc_id, 0.0)
vs = vector_scores.get(doc_id, 0.0)
combined.append({
“id”: doc_id,
“score”: alpha * vs + (1 – alpha) * ks,
“keyword_score”: ks,
“vector_score”: vs,
})

combined.sort(key=lambda x: x[“score”], reverse=True)
return combined[:top_k]

def reciprocal_rank_fusion(
*ranked_lists: list[tuple[str, float]],
k: int = 60,
top_n: int = 10,
) -> list[dict]:
“””
Combine multiple ranked lists using Reciprocal Rank Fusion (RRF).

RRF score = sum of 1 / (k + rank) across all lists.

Advantages over weighted score merging:
– No need to normalize scores (works on ranks, not raw values)
– No alpha parameter to tune
– Handles different score distributions robustly
– Used internally by Elasticsearch, Pinecone, and Weaviate
“””
rrf_scores: dict[str, float] = defaultdict(float)
doc_details: dict[str, dict] = {}

for list_idx, ranked_list in enumerate(ranked_lists):
for rank, (doc_id, raw_score) in enumerate(ranked_list, start=1):
rrf_scores[doc_id] += 1.0 / (k + rank)
if doc_id not in doc_details:
doc_details[doc_id] = {}
doc_details[doc_id][f”list_{list_idx}_rank”] = rank
doc_details[doc_id][f”list_{list_idx}_score”] = raw_score

results = []
for doc_id, rrf_score in rrf_scores.items():
results.append({
“id”: doc_id,
“rrf_score”: round(rrf_score, 6),
**doc_details[doc_id],
})

results.sort(key=lambda x: x[“rrf_score”], reverse=True)
return results[:top_n]

def hybrid_search_rrf(
query: str,
query_embedding: list[float],
bm25_index: BestMatching25Index,
vector_index: VectorIndex,
top_k: int = 10,
) -> list[dict]:
keyword_results = bm25_index.search(query, top_k=top_k * 2)
vector_results = vector_index.search(query_embedding, top_k=top_k * 2)

return reciprocal_rank_fusion(keyword_results, vector_results, top_n=top_k)

> **Tip**: For fast and effective retrieval, use a hybrid approach that combines keyword and embedding-based search. Keyword search excels at matching exact terms (like “Password policy”), while embeddings capture conceptual or meaning-based matches. LlamaIndex supports hybrid retrieval well, allowing searches for both exact terms and contextual relevance.

### 6. Set Up an Automatic Update and Refresh Routine

The last step is keeping your knowledge base current. Implement **selective forgetting**—the practice of removing or overwriting outdated and redundant data to maintain accuracy.

How do you identify what to remove? Use evaluation and observability tools. For example, schedule test queries in the DeepEval framework to regularly verify your AI model’s accuracy. If responses are incorrect, the TruLens platform can help pinpoint the exact source chunk responsible.

python
“””
Knowledge Base Quality Monitoring

Automated health checks for your knowledge base:
1. Retrieval quality — Is it finding the right documents?
2. Freshness detection — Are documents stale or embeddings drifting?
3. Unified pipeline — Scheduled monitoring with alerts
“””

import time
import json
import logging
from datetime import datetime, timedelta
from dataclasses import dataclass, field
from typing import Any, Callable

import numpy as np

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(“kb_monitor”)

def setup_deepeval_metrics():
“””Define retrieval quality metrics using DeepEval.

DeepEval uses an LLM judge to evaluate whether retrieved context
actually helps answer the question.
“””
from deepeval.metrics import (
AnswerRelevancyMetric,
FaithfulnessMetric,
ContextualPrecisionMetric,
ContextualRecallMetric,
)
from deepeval.test_case import LLMTestCase

metrics = {
# Does the answer address the question?
“relevancy”: AnswerRelevancyMetric(threshold=0.7),
# Is the answer grounded in the retrieved context (no hallucination)?
“faithfulness”: FaithfulnessMetric(threshold=0.7),
# Are the top-ranked retrieved docs actually relevant?
“context_precision”: ContextualPrecisionMetric(threshold=0.7),
# Did we retrieve all the docs needed to answer?
“context_recall”: ContextualRecallMetric(threshold=0.7),
}

return metrics, LLMTestCase

def evaluate_retrieval_quality(
rag_pipeline: Callable,
test_cases: list[dict],
) -> list[dict]:
“””Run test queries through your RAG pipeline and score them.

Each test case should include:
– query: the user question
– expected_answer: ground truth answer (for recall/relevancy)
“””
from deepeval import evaluate
from deepeval.test_case import LLMTestCase
from deepeval.metrics import (
AnswerRelevancyMetric,
FaithfulnessMetric,
ContextualPrecisionMetric,
ContextualRecallMetric,
)

results = []

for tc in test_cases:
# Execute your actual RAG pipeline
response = rag_pipeline(tc[“query”])

test_case = LLMTestCase(
input=tc[“query”],
actual_output=response[“answer”],
expected_output=tc[“expected_answer”],
retrieval_context=response[“retrieved_contexts”],
)

metrics = [
AnswerRelevancyMetric(threshold=0.7),
FaithfulnessMetric(threshold=0.7),
ContextualPrecisionMetric(threshold=0.7),
ContextualRecallMetric(threshold=0.7),
]

for metric in metrics:
metric.measure(test_case)

results.append({
“query”: tc[“query”],
“scores”: {m.__class__.__name__: m.score for m in metrics},
“passed”: all(m.is_successful() for m in metrics),
})

return results

def setup_trulens_monitoring(rag_pipeline: Callable, app_name: str = “my_kb”):
“””Wrap your RAG pipeline with TruLens for continuous feedback logging.

TruLens records every query, response, and retrieved context, then
runs feedback functions asynchronously to score each interaction.
“””
from trulens.core import TruSession, Feedback, Select
from trulens.providers.openai import OpenAI as TruLensOpenAI
from trulens.apps.custom import TruCustomApp, instrument

session = TruSession()

# Feedback provider (uses an LLM to judge

# Initialize TruLens with an OpenAI-compatible provider to evaluate RAG quality
provider = TruLensOpenAI()

# Define feedback functions to score RAG components
feedbacks = [
    # Evaluates if the final answer addresses the user's question
    Feedback(provider.relevance)
    .on_input()
    .on_output(),

    # Checks if the answer is supported by the retrieved documents
    Feedback(provider.groundedness_measure_with_cot_reasons)
    .on(Select.RecordCalls.retrieve.rets)
    .on_output(),

    # Assesses if the retrieved documents are relevant to the query
    Feedback(provider.context_relevance)
    .on_input()
    .on(Select.RecordCalls.retrieve.rets),
]

# Instrument your RAG pipeline to log and score each step
@instrument
class InstrumentedRAG:
    def __init__(self, pipeline):
        self._pipeline = pipeline

    @instrument
    def retrieve(self, query: str) -> list[str]:
        result = self._pipeline(query)
        return result["retrieved_contexts"]

    @instrument
    def query(self, query: str) -> str:
        result = self._pipeline(query)
        return result["answer"]

instrumented = InstrumentedRAG(rag_pipeline)

# Create a TruCustomApp to monitor the instrumented pipeline
tru_app = TruCustomApp(
    instrumented,
    app_name=app_name,
    feedbacks=feedbacks,
)

return tru_app, session


def get_trulens_dashboard_url(session) -> str:
    """Start the TruLens dashboard to track quality metrics over time."""
    session.run_dashboard(port=8501)
    return "Dashboard running at http://localhost:8501"


@dataclass
class DocumentFreshness:
    doc_id: str
    last_updated: datetime
    last_embedded: datetime
    source_hash: str  # Hash of the source content when it was embedded


class FreshnessMonitor:
    """Identifies outdated documents and detects embedding drift."""

    def __init__(self, staleness_threshold_days: int = 30):
        self.threshold = timedelta(days=staleness_threshold_days)
        self.freshness_records: dict[str, DocumentFreshness] = {}

    def register(self, doc_id: str, source_hash: str) -> None:
        now = datetime.utcnow()
        self.freshness_records[doc_id] = DocumentFreshness(
            doc_id=doc_id,
            last_updated=now,
            last_embedded=now,
            source_hash=source_hash,
        )

    def check_staleness(self) -> dict:
        """Find documents that haven't been re-embedded within the threshold."""
        now = datetime.utcnow()
        stale, fresh = [], []

        for doc_id, record in self.freshness_records.items():
            age = now - record.last_embedded
            if age > self.threshold:
                stale.append({"id": doc_id, "days_stale": age.days})
            else:
                fresh.append(doc_id)

        return {
            "total": len(self.freshness_records),
            "fresh": len(fresh),
            "stale": len(stale),
            "stale_documents": stale,
        }

    def check_content_drift(
        self, doc_id: str, current_source_hash: str
    ) -> bool:
        """Determine if the source content has changed since it was last embedded."""
        record = self.freshness_records.get(doc_id)
        if not record:
            return True  # Treat unknown documents as drifted
        return record.source_hash != current_source_hash


def detect_embedding_drift(
    old_embeddings: dict[str, list[float]],
    new_embeddings: dict[str, list[float]],
    drift_threshold: float = 0.1,
) -> dict:
    """Compare previous and current embeddings for the same documents.

    When an embedding model is updated or replaced, existing vectors
    might become incompatible. This function identifies such drift.
    """
    drifted = []
    common_ids = set(old_embeddings) & set(new_embeddings)

    for doc_id in common_ids:
        old = np.array(old_embeddings[doc_id])
        new = np.array(new_embeddings[doc_id])

        # Cosine distance: 0 = identical, 2 = completely opposite
        cos_sim = np.dot(old, new) / (np.linalg.norm(old) * np.linalg.norm(new))
        cos_dist = 1 - cos_sim

        if cos_dist > drift_threshold:
            drifted.append({
                "id": doc_id,
                "cosine_distance": round(float(cos_dist), 4),
            })

    return {
        "documents_compared": len(common_ids),
        "drifted": len(drifted),
        "drift_threshold": drift_threshold,
        "drifted_documents": sorted(drifted, key=lambda x: x["cosine_distance"], reverse=True),
    }

Pairing DeepEval with TruLens automates ongoing testing of your knowledge base.

Key challenges when building a knowledge base (and how to solve them)

Here are the most frequent issues I’ve encountered with knowledge bases:

1. Increasing data quality errors

AI models developed over time—even by well-resourced teams at reputable companies—sometimes produce incorrect information. The well-known Air Canada chatbot incident is a prime example, where the bot promised a refund based on a policy that didn’t actually exist.

Even when engineers carefully curate the knowledge base, problems persist. In my experience, a lack of domain expertise leads to mistakes in determining what content is truly relevant. Step back from the technical side and think like a domain expert to spot outdated, conflicting, or irrelevant information in your knowledge base.

2. Slow retrieval speed

Simply providing the correct answer isn’t enough. Users dislike waiting and expect near-instant responses from a machine.

Developers often focus on getting features working and neglect optimization, which is absolutely critical. Use these strategies to address common performance bottlenecks:

• Use HNSW (Hierarchical Navigable Small World) or IVF indexes rather than flat indexes, since these group related content together for faster lookups
• Apply quantization (reducing the size of query vectors to use less memory) or recursive character splitting (dividing text into smaller chunks) to minimize memory usage
• Host your database and AI service in the same cloud region to reduce latency

3. Limited scalability

To accelerate delivery, developers sometimes make design choices that hurt long-term scalability. A common mistake is using a monolithic architecture where all data storage and query processing happen within a single, tightly coupled cluster. As usage increases, CPU and RAM consumption spikes across the entire cluster for every query. I recommend horizontal sharding (distributing data across multiple smaller servers) to manage growth effectively.

Another issue is rising costs at scale, which typically occurs when vectors aren’t quantized or compressed to optimize storage. Developers sometimes skip quantization to ship the model faster. The consequences aren’t immediately apparent, but soon the slowdowns and escalating cloud costs reveal the problem.

A knowledge base is a curated asset, not a data dump

Building a knowledge base isn’t a one-time task. It’s a living asset that requires continuous optimization. The structure you design today will expose gaps tomorrow. Every failed query provides feedback, and every successful retrieval confirms your design decisions.

I recommend starting small—identify the ten most common questions for your model, create clear documentation for them, and then test whether your model delivers accurate answers quickly. Once you start seeing the expected results, you can repeat the process to grow the knowledge base.

The gap between a model that guesses and one that truly knows comes down to this intentional curation effort. Ongoing refinement makes each subsequent search faster and results more dependable.