anveshai/knowledge_engine.py

"""
Knowledge Engine v2 — BM25-ranked multi-passage retrieval.

Improvements over v1:
  · BM25 ranking (Okapi BM25) instead of raw keyword overlap
  · Multi-passage synthesis — retrieves top-K paragraphs and merges them
  · Phrase-level matching with bonus scoring
  · TF-IDF-style IDF weighting for rare/common term discrimination
  · Dynamic MIN_RELEVANCE threshold based on query complexity
  · Zero external dependencies (pure stdlib + math)
"""

from __future__ import annotations

import math
import os
import re
from collections import Counter
from typing import List, Tuple


KNOWLEDGE_FILE = os.path.join(os.path.dirname(__file__), "knowledge.txt")

# BM25 hyper-parameters
K1: float = 1.5   # term saturation
B:  float = 0.75  # length normalisation

# How many top passages to retrieve & synthesise
TOP_K: int = 3

# Minimum BM25 score for a passage to be considered a real match
MIN_BM25_SCORE: float = 1.0

STOP_WORDS = {
    "a", "an", "the", "is", "are", "was", "were", "be", "been", "being",
    "have", "has", "had", "do", "does", "did", "will", "would", "shall",
    "should", "may", "might", "must", "can", "could", "to", "of", "in",
    "on", "at", "by", "for", "with", "about", "against", "between", "into",
    "through", "during", "before", "after", "above", "below", "from",
    "up", "down", "out", "off", "over", "under", "again", "and", "but",
    "or", "nor", "so", "yet", "both", "either", "neither", "not", "no",
    "what", "which", "who", "whom", "this", "that", "these", "those",
    "i", "me", "my", "myself", "we", "our", "you", "your", "he", "she",
    "it", "they", "them", "their", "tell", "explain", "describe", "give",
    "me", "some", "information", "about", "please", "could", "how", "when",
    "where", "why", "let",
}

# ─────────────────────────────────────────────────────────────────────────────
# Text processing helpers
# ─────────────────────────────────────────────────────────────────────────────

def _load_paragraphs(filepath: str) -> List[str]:
    if not os.path.exists(filepath):
        return []
    with open(filepath, "r", encoding="utf-8") as f:
        content = f.read()
    raw = re.split(r"\n\s*\n", content.strip())
    return [p.strip() for p in raw if len(p.strip()) > 10]


def _tokenize(text: str) -> List[str]:
    words = re.findall(r"\b[a-z]{2,}\b", text.lower())
    return [w for w in words if w not in STOP_WORDS]


def _extract_phrases(text: str, n: int = 2) -> List[str]:
    tokens = _tokenize(text)
    return [" ".join(tokens[i:i+n]) for i in range(len(tokens)-n+1)]


def _strip_question_prefixes(text: str) -> str:
    prefixes = [
        "what is", "what are", "who is", "who are", "explain", "define",
        "tell me about", "describe", "how does", "why is", "when was",
        "where is", "history of", "meaning of", "knowledge:", "knowledge :",
        "learn about", "facts about", "information about", "can you explain",
        "could you explain", "please explain", "give me information about",
    ]
    lowered = text.lower().strip()
    for prefix in prefixes:
        if lowered.startswith(prefix):
            remainder = text[len(prefix):].strip(" ?.,")
            if remainder:
                return remainder
    return text


# ─────────────────────────────────────────────────────────────────────────────
# BM25 index
# ─────────────────────────────────────────────────────────────────────────────

class BM25Index:
    """Okapi BM25 retrieval index over a list of paragraphs."""

    def __init__(self, paragraphs: List[str]):
        self.paragraphs   = paragraphs
        self.n_docs       = len(paragraphs)
        self._token_lists = [_tokenize(p) for p in paragraphs]
        self._tf_lists    = [Counter(tl) for tl in self._token_lists]
        self._doc_lengths = [len(tl) for tl in self._token_lists]
        self._avgdl       = (
            sum(self._doc_lengths) / self.n_docs if self.n_docs else 1.0
        )
        self._idf         = self._compute_idf()

    def _compute_idf(self) -> dict[str, float]:
        idf: dict[str, float] = {}
        for token_list in self._token_lists:
            for tok in set(token_list):
                idf[tok] = idf.get(tok, 0) + 1
        result: dict[str, float] = {}
        for tok, df in idf.items():
            result[tok] = math.log(
                (self.n_docs - df + 0.5) / (df + 0.5) + 1
            )
        return result

    def score(self, query_tokens: List[str], doc_idx: int) -> float:
        tf_map = self._tf_lists[doc_idx]
        dl     = self._doc_lengths[doc_idx]
        score  = 0.0
        for tok in query_tokens:
            if tok not in tf_map:
                continue
            idf_val = self._idf.get(tok, 0.0)
            tf_val  = tf_map[tok]
            numerator   = tf_val * (K1 + 1)
            denominator = tf_val + K1 * (1 - B + B * dl / self._avgdl)
            score += idf_val * (numerator / denominator)
        return score

    def phrase_bonus(self, phrase: str, doc_idx: int) -> float:
        """Extra score if a phrase appears verbatim in the paragraph."""
        if phrase in self.paragraphs[doc_idx].lower():
            return 2.0
        return 0.0

    def retrieve(
        self, query: str, top_k: int = TOP_K
    ) -> List[Tuple[float, str]]:
        clean     = _strip_question_prefixes(query)
        q_tokens  = _tokenize(clean)
        q_phrases = _extract_phrases(clean, 2)

        if not q_tokens:
            return []

        scores: List[Tuple[float, int]] = []
        for idx in range(self.n_docs):
            s = self.score(q_tokens, idx)
            for phrase in q_phrases:
                s += self.phrase_bonus(phrase, idx)
            scores.append((s, idx))

        scores.sort(reverse=True)
        return [
            (s, self.paragraphs[idx])
            for s, idx in scores[:top_k]
            if s >= MIN_BM25_SCORE
        ]


# ─────────────────────────────────────────────────────────────────────────────
# Multi-passage synthesiser
# ─────────────────────────────────────────────────────────────────────────────

def _synthesise(passages: List[Tuple[float, str]], query: str) -> str:
    """
    Merge the top-K retrieved passages into a coherent response.
    Deduplicates sentences that appear across passages.
    """
    if not passages:
        return ""

    if len(passages) == 1:
        return passages[0][1]

    seen_sentences: set[str] = set()
    merged_sentences: List[str] = []

    for _score, para in passages:
        sentences = re.split(r"(?<=[.!?])\s+", para.strip())
        for sent in sentences:
            normalised = sent.strip().lower()
            if normalised not in seen_sentences and len(sent.strip()) > 10:
                seen_sentences.add(normalised)
                merged_sentences.append(sent.strip())

    return " ".join(merged_sentences)


# ─────────────────────────────────────────────────────────────────────────────
# Public interface
# ─────────────────────────────────────────────────────────────────────────────

class KnowledgeEngine:
    """
    BM25-powered local knowledge retrieval.

    Usage:
        ke = KnowledgeEngine()
        response, found = ke.query("What is quantum computing?")
    """

    def __init__(self, knowledge_file: str = KNOWLEDGE_FILE):
        self._paragraphs = _load_paragraphs(knowledge_file)
        self._loaded     = len(self._paragraphs) > 0
        self._index      = BM25Index(self._paragraphs) if self._loaded else None

    def is_loaded(self) -> bool:
        return self._loaded

    def query(self, user_input: str, top_k: int = TOP_K) -> Tuple[str, bool]:
        """
        Find the most relevant passage(s) for the given query using BM25.

        Returns:
            (response_text, found)
            found=True  → high-confidence match(es) retrieved
            found=False → no confident match; caller should escalate to LLM
        """
        if not self._loaded or self._index is None:
            return ("Knowledge base unavailable. Ensure 'knowledge.txt' exists.", False)

        hits = self._index.retrieve(user_input, top_k=top_k)

        if not hits:
            return ("", False)

        response = _synthesise(hits, user_input)
        return (response, True)

    def get_context(self, user_input: str, top_k: int = 2) -> str:
        """
        Return up to top_k relevant passages as a context block (for LLM prompts).
        Returns empty string if nothing relevant is found.
        """
        if not self._loaded or self._index is None:
            return ""
        hits = self._index.retrieve(user_input, top_k=top_k)
        if not hits:
            return ""
        return "\n\n".join(para for _, para in hits)