Knowledge Base Preparation

source/scripts/prepare_kb.py

[
  {
    "id": 1,
    "question": "What is a database?",
    "answer": "A database is an organized collection of structured data..."
  },
  {
    "id": 2,
    "question": "What is normalization?",
    "answer": "Normalization is the process of organizing data..."
  }
]

def topic_from_filename(fname: str) -> str:
    fname = fname.lower()
    if "database" in fname or "dbms" in fname:
        return "DBMS"
    if "oops" in fname:
        return "OOPs"
    if "os" in fname:
        return "OS"
    return "uncategorized"

def assign_subtopic(q: str, a: str, topic: str, rules: List[Dict[str, Any]]) -> str:
    text = f"{q} {a}".lower()
    tokens = tokenize(text)

    best_subtopic = None
    best_score = 0.0

    for rule in rules:
        if rule["topic"] != topic:
            continue

        keywords = rule["keywords"]
        matched = sum(1 for kw in keywords if kw in text or kw in tokens)
        coverage = matched / len(keywords)

        if coverage > best_score:
            best_score = coverage
            best_subtopic = rule["subtopic"]

    # confidence threshold
    if best_score >= 0.25:
        return best_subtopic

    # fallback to default subtopic
    fallback = {
        "DBMS": "DBMS Architecture",
        "OOPs": "Classes",
        "OS": "Processes"
    }
    return fallback.get(topic)

def refine_subtopic(q: str, a: str, topic: str, subtopic: str) -> str:
    text = f"{q} {a}".lower()

    if "deadlock" in text:
        if topic == "DBMS" and any(x in text for x in ["transaction", "lock"]):
            return "Deadlocks"
        if topic == "OS" and any(x in text for x in ["process", "resource"]):
            return "Deadlocks"

    if "memory" in text:
        if topic == "OOPs":
            return "Memory Management in OOP"
        if topic == "OS":
            return "Memory Management"

    return subtopic

def difficulty_heuristic(q: str, a: str) -> str:
    text = f"{q} {a}".lower()

    advanced = [
        "mvcc", "2pl", "serializable", "cap theorem", "wal",
        "b+ tree", "page replacement", "banker's algorithm",
        "vtable", "raii"
    ]

    beginner = [
        "what is", "define", "basic", "class", "object",
        "process", "thread", "primary key"
    ]

    if sum(1 for t in advanced if t in text) >= 2:
        return "Advanced"
    if sum(1 for t in beginner if t in text) >= 2:
        return "Beginner"
    return "Intermediate"

def normalize_text(s: str) -> str:
    s = unidecode(s)  # Convert Unicode to ASCII
    s = strip_html(s)  # Remove HTML tags
    s = s.replace("\xa0", " ")  # Replace non-breaking spaces
    s = unicodedata.normalize("NFKC", s)  # Normalize Unicode
    s = re.sub(r"[ \t]+", " ", s)  # Collapse whitespace
    s = re.sub(r"\n\s*\n\s*", "\n\n", s)  # Normalize line breaks
    return s.strip()

def strip_html(text: str) -> str:
    text = re.sub(r"<\s*br\s*/?>", "\n", text, flags=re.I)  # <br> → newline
    text = re.sub(r"<[^>]+>", " ", text)  # Remove all other tags
    return text

[
  {
    "id": "1",
    "question": "What is a database?",
    "answer": "A database is an organized collection...",
    "topic": "DBMS",
    "subtopic": "DBMS Architecture",
    "difficulty": "Beginner",
    "source": "database_qna.json"
  }
]

{"id": "1", "text": "Q: What is a database?\nA: A database is...", "metadata": {"topic": "DBMS", "subtopic": "DBMS Architecture", "difficulty": "Beginner", "source": "database_qna.json"}}
{"id": "2", "text": "Q: What is normalization?\nA: Normalization is...", "metadata": {"topic": "DBMS", "subtopic": "Normalization", "difficulty": "Intermediate", "source": "database_qna.json"}}

pip install unidecode

cd source/scripts
python prepare_kb.py

🔄 Preparing KB with strict topic control...
✅ KB preparation complete
Clean JSON : data/processed/kb_clean.json
Chunks     : data/processed/kb_chunks.jsonl

seen = set()

for obj in raw_items:
    q = normalize_text(obj.get("question", ""))
    a = normalize_text(obj.get("answer", ""))

    if not q or not a:
        continue

    key = (q.lower(), a.lower())
    if key in seen:
        continue
    seen.add(key)

{
  "id": 42,
  "question": "What is <b>normalization</b> in DBMS?",
  "answer": "Normalization is organizing data to reduce redundancy using normal forms like 1NF, 2NF, 3NF."
}

{
  "id": "42",
  "question": "What is normalization in DBMS?",
  "answer": "Normalization is organizing data to reduce redundancy using normal forms like 1NF, 2NF, 3NF.",
  "topic": "DBMS",
  "subtopic": "Normalization",
  "difficulty": "Intermediate",
  "source": "database_qna.json"
}