File size: 4,400 Bytes
5d8fd4f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 | """
Knowledge Engine — retrieves relevant information from a local knowledge base.
How it works:
1. Loads 'knowledge.txt' at startup (one paragraph per blank-line block).
2. For a given query, scores each paragraph using keyword overlap.
3. Returns the highest-scoring paragraph + a boolean indicating confidence.
If confidence is low, the caller (main.py) will escalate to the LLM.
This is intentionally lightweight and fully offline. In the future it can be
swapped for a vector-based retrieval system (FAISS + sentence-transformers)
without changing the rest of the architecture.
"""
import os
import re
from typing import List, Tuple
KNOWLEDGE_FILE = os.path.join(os.path.dirname(__file__), "knowledge.txt")
# A paragraph must score at least this much to be considered a real match.
# Queries below this score are escalated to the LLM fallback.
MIN_RELEVANCE_SCORE = 2
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",
}
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 p.strip()]
def _tokenize(text: str) -> List[str]:
words = re.findall(r"\b[a-z]+\b", text.lower())
return [w for w in words if w not in STOP_WORDS and len(w) > 2]
def _score_paragraph(query_tokens: List[str], paragraph: str) -> int:
para_lower = paragraph.lower()
score = 0
for token in query_tokens:
if re.search(r"\b" + re.escape(token) + r"\b", para_lower):
score += 2
elif token in para_lower:
score += 1
return score
def _strip_knowledge_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",
]
lowered = text.lower().strip()
for prefix in prefixes:
if lowered.startswith(prefix):
return text[len(prefix):].strip()
return text
class KnowledgeEngine:
"""Local keyword-scored knowledge retrieval over knowledge.txt."""
def __init__(self, knowledge_file: str = KNOWLEDGE_FILE):
self.paragraphs: List[str] = _load_paragraphs(knowledge_file)
self._loaded = len(self.paragraphs) > 0
def is_loaded(self) -> bool:
return self._loaded
def query(self, user_input: str) -> Tuple[str, bool]:
"""
Find the most relevant paragraph for the given query.
Returns:
(response, found)
found = True → a high-confidence match was found in the KB
found = False → no confident match; caller should try the LLM
"""
if not self._loaded:
return (
"Knowledge base unavailable. Ensure 'knowledge.txt' exists.",
False,
)
clean_query = _strip_knowledge_prefixes(user_input)
query_tokens = _tokenize(clean_query)
if not query_tokens:
return ("Could you rephrase? I couldn't parse the query.", False)
scored: List[Tuple[int, str]] = [
(_score_paragraph(query_tokens, para), para)
for para in self.paragraphs
]
best_score, best_para = max(scored, key=lambda x: x[0])
if best_score < MIN_RELEVANCE_SCORE:
# Signal to caller: escalate to LLM
return ("", False)
return (best_para, True)
|