| """ |
| ๐ฒ๐ฆ Moroccan Cultural Knowledge Retriever |
| Lightweight, ultra-fast BM25 / token-matching retriever over the |
| verified Moroccan Cultural Gold Fact-Pack to guarantee 100% factual accuracy. |
| """ |
|
|
| import os |
| import json |
| import re |
| import logging |
| from typing import Optional, Dict, Any, List |
|
|
| logger = logging.getLogger(__name__) |
|
|
| class MoroccanKnowledgeRetriever: |
| """ |
| Indexes verified Moroccan facts (history, dynasties, gastronomy, proverbs, civic laws) |
| and retrieves exact ground-truth context for inference queries. |
| """ |
|
|
| def __init__(self, knowledge_file: Optional[str] = None): |
| if knowledge_file is None: |
| local_facts = os.path.join(os.path.dirname(os.path.abspath(__file__)), "moroccan_facts.jsonl") |
| if os.path.exists(local_facts): |
| knowledge_file = local_facts |
| else: |
| project_root = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")) |
| knowledge_file = os.path.join(project_root, "data/darija_llm/moroccan_cultural_gold.jsonl") |
|
|
| self.knowledge_file = knowledge_file |
| self.documents: List[Dict[str, Any]] = [] |
| self._load_knowledge() |
|
|
| def _normalize_word(self, w: str) -> str: |
| |
| for prefix in ["ูุงู", "ูุงู", "ุจุงู", "ูุงู", "ูู", "ุงู"]: |
| if w.startswith(prefix) and len(w) > len(prefix) + 2: |
| w = w[len(prefix):] |
| break |
| for single in ["ู", "ู", "ุจ", "ู", "ุฏ"]: |
| if w.startswith(single) and len(w) > len(single) + 2: |
| w = w[len(single):] |
| break |
| return w |
|
|
| def _tokenize(self, text: str) -> List[str]: |
| |
| clean = re.sub(r"[^\w\s]", " ", text) |
| clean = clean.replace("ุฃ", "ุง").replace("ุฅ", "ุง").replace("ุข", "ุง").replace("ุฉ", "ู").replace("ู", "ู") |
| raw_words = [w for w in clean.lower().split() if len(w) >= 2] |
| normalized = [self._normalize_word(w) for w in raw_words] |
| return list(set(raw_words + normalized)) |
|
|
| def _load_knowledge(self): |
| if not os.path.exists(self.knowledge_file): |
| logger.warning(f"Knowledge file not found at: {self.knowledge_file}") |
| return |
|
|
| with open(self.knowledge_file, "r", encoding="utf-8") as f: |
| for line in f: |
| if line.strip(): |
| item = json.loads(line.strip()) |
| inst = item.get("instruction", "") |
| resp = item.get("response", "") |
| tokens = self._tokenize(inst + " " + resp) |
| self.documents.append({ |
| "instruction": inst, |
| "response": resp, |
| "category": item.get("category", "general"), |
| "tokens": set(tokens) |
| }) |
| logger.info(f"Loaded {len(self.documents)} verified Moroccan facts into retriever.") |
|
|
| def _get_ngrams(self, text: str, n: int = 3) -> set: |
| clean = re.sub(r"[^\w]", "", text.replace(" ", "")) |
| return {clean[i:i+n] for i in range(len(clean) - n + 1)} if len(clean) >= n else {clean} |
|
|
| def retrieve(self, query: str, threshold: float = 0.25) -> Optional[Dict[str, Any]]: |
| """ |
| Retrieves the best matching verified fact using token overlap and character n-grams. |
| """ |
| if not self.documents: |
| return None |
|
|
| q_tokens = set(self._tokenize(query)) |
| q_ngrams = self._get_ngrams(query) |
| if not q_tokens and not q_ngrams: |
| return None |
|
|
| best_doc = None |
| best_score = 0.0 |
|
|
| for doc in self.documents: |
| |
| doc_tokens = doc["tokens"] |
| token_intersection = len(q_tokens.intersection(doc_tokens)) |
| token_score = token_intersection / max(1, len(q_tokens)) |
|
|
| |
| doc_ngrams = self._get_ngrams(doc["instruction"] + " " + doc["response"][:200]) |
| ngram_intersection = len(q_ngrams.intersection(doc_ngrams)) |
| ngram_score = ngram_intersection / max(1, len(q_ngrams)) |
|
|
| combined_score = (token_score * 0.4) + (ngram_score * 0.6) |
|
|
| if combined_score > best_score: |
| best_score = combined_score |
| best_doc = doc |
|
|
| if best_doc and best_score >= threshold: |
| return { |
| "matched_instruction": best_doc["instruction"], |
| "verified_fact": best_doc["response"], |
| "category": best_doc["category"], |
| "confidence": round(best_score, 3) |
| } |
|
|
| return None |
|
|
