""" rag_service.py — LibBee v3.1 Fixes applied: 1. _find_corpus_index replaced with O(1) reverse-lookup dict (_content_to_idx). Old implementation was O(n * fetch_k) per query — a major bottleneck. 2. fetch_k corrected: max(top_k * 3, 15) — old expression max(top_k*3, top_k) was always top_k*3 (max was a no-op). Now enforces a floor of 15. 3. _content_to_idx rebuilt whenever corpus is loaded (init, cache load, rebuild). 4. audit_knowledge_base removed from get_stats() — now only called explicitly via a dedicated audit() method to avoid disk reads on every /rag-status poll. 5. _write / cache saves use os.replace for atomic writes. """ import hashlib import json import logging import os import re import time from pathlib import Path from typing import List, Optional import numpy as np from src.config import get_settings logger = logging.getLogger(__name__) CHUNK_MAX_CHARS = 1600 CHUNK_OVERLAP_CHARS = 200 class RAGService: def __init__(self): self.vectorstore = None self.bm25 = None self.bm25_corpus: List[str] = [] self.bm25_meta: List[dict] = [] self._embeddings = None self._ready = False self._kb_hash = "" # O(1) reverse lookup: chunk_text -> corpus index self._content_to_idx: dict[str, int] = {} def is_ready(self) -> bool: return self._ready def _knowledge_dir(self) -> Path: return get_settings().kb_dir def _cache_dir(self) -> Path: return get_settings().rag_cache_dir def _hash_knowledge_base(self, txt_files: List[Path]) -> str: digest = hashlib.sha256() for path in txt_files: stat = path.stat() digest.update(path.name.encode("utf-8")) digest.update(str(stat.st_mtime_ns).encode("utf-8")) digest.update(str(stat.st_size).encode("utf-8")) return digest.hexdigest() def _cache_paths(self) -> tuple[Path, Path, Path]: cache_dir = self._cache_dir() return cache_dir / "faiss_index", cache_dir / "bm25_cache.json", cache_dir / "kb_state.json" def _build_reverse_index(self) -> None: """Build O(1) content -> index lookup. Call after any corpus change.""" self._content_to_idx = { chunk.strip(): i for i, chunk in enumerate(self.bm25_corpus) } async def initialize(self, openai_api_key: str) -> None: t0 = time.time() logger.info("RAGService: starting initialization") settings = get_settings() knowledge_dir = settings.kb_dir txt_files = sorted(knowledge_dir.glob("*.txt")) if knowledge_dir.exists() else [] self._kb_hash = self._hash_knowledge_base(txt_files) if txt_files else "" try: from langchain_openai import OpenAIEmbeddings from langchain_community.vectorstores import FAISS from rank_bm25 import BM25Okapi self._embeddings = OpenAIEmbeddings( model="text-embedding-3-small", openai_api_key=openai_api_key ) if self._try_load_cached_indexes(FAISS, BM25Okapi): self._build_reverse_index() self._ready = True logger.info( "RAGService: loaded cached indexes — %d chunks", len(self.bm25_corpus) ) return chunks, metadatas = self._load_and_chunk_all() if not chunks: logger.error("RAGService: no chunks loaded — check knowledge_dir: %s", knowledge_dir) return self.vectorstore = await FAISS.afrom_texts( chunks, self._embeddings, metadatas=metadatas ) self.bm25_corpus = chunks self.bm25_meta = metadatas tokenized = [c.lower().split() for c in chunks] self.bm25 = BM25Okapi(tokenized) self._build_reverse_index() self._save_cached_indexes() self._ready = True logger.info( "RAGService: ready — %d chunks indexed in %.1fs", len(chunks), time.time() - t0 ) except Exception as exc: logger.error("RAGService initialization failed: %s", exc, exc_info=True) def _try_load_cached_indexes(self, FAISS, BM25Okapi) -> bool: faiss_dir, bm25_path, state_path = self._cache_paths() if not (faiss_dir.exists() and bm25_path.exists() and state_path.exists()): return False try: state = json.loads(state_path.read_text(encoding="utf-8")) if state.get("kb_hash") != self._kb_hash: logger.info("RAGService: KB changed (hash mismatch) — rebuilding indexes") return False self.vectorstore = FAISS.load_local( str(faiss_dir), self._embeddings, allow_dangerous_deserialization=True ) bm25_cache = json.loads(bm25_path.read_text(encoding="utf-8")) self.bm25_corpus = bm25_cache["corpus"] self.bm25_meta = bm25_cache["meta"] tokenized = [c.lower().split() for c in self.bm25_corpus] self.bm25 = BM25Okapi(tokenized) return True except Exception as exc: logger.warning("Failed loading cached RAG index: %s — will rebuild", exc) return False def _save_cached_indexes(self) -> None: if not self.vectorstore: return faiss_dir, bm25_path, state_path = self._cache_paths() faiss_dir.mkdir(parents=True, exist_ok=True) self.vectorstore.save_local(str(faiss_dir)) # Atomic write for bm25 cache bm25_tmp = bm25_path.with_suffix(".tmp") bm25_tmp.write_text( json.dumps( {"corpus": self.bm25_corpus, "meta": self.bm25_meta}, ensure_ascii=False ), encoding="utf-8", ) os.replace(bm25_tmp, bm25_path) # Atomic write for state state_tmp = state_path.with_suffix(".tmp") state_tmp.write_text( json.dumps({"kb_hash": self._kb_hash, "chunk_count": len(self.bm25_corpus)}), encoding="utf-8", ) os.replace(state_tmp, state_path) def _load_and_chunk_all(self) -> tuple[List[str], List[dict]]: all_chunks: List[str] = [] all_meta: List[dict] = [] knowledge_dir = self._knowledge_dir() if not knowledge_dir.exists(): logger.error("Knowledge directory not found: %s", knowledge_dir) return [], [] txt_files = sorted(knowledge_dir.glob("*.txt")) for fpath in txt_files: try: text = fpath.read_text(encoding="utf-8").strip() if not text: continue source_url = "" title = fpath.stem.replace("_", " ").replace("-", " ") for line in text.splitlines()[:8]: line = line.strip() if line.startswith("SOURCE:"): source_url = line.replace("SOURCE:", "").strip() elif line.startswith("TITLE:"): title = line.replace("TITLE:", "").strip() for chunk in self._chunk_text(text): all_chunks.append(chunk) all_meta.append( {"source": source_url, "title": title, "filename": fpath.name} ) except Exception as exc: logger.warning("Failed to load %s: %s", fpath.name, exc) logger.info("RAGService: loaded %d chunks from %d files", len(all_chunks), len(txt_files)) return all_chunks, all_meta def _chunk_text(self, text: str) -> List[str]: chunks: List[str] = [] sections = re.split(r"\n(?=(?:TOPIC:|[A-Z][A-Z\s,/&()\-]+:)\s)", text) for section in sections: section = section.strip() if not section or len(section) < 30: continue if len(section) <= CHUNK_MAX_CHARS: chunks.append(section) else: start = 0 while start < len(section): end = start + CHUNK_MAX_CHARS chunk = section[start:end].strip() if chunk: chunks.append(chunk) if end >= len(section): break start += CHUNK_MAX_CHARS - CHUNK_OVERLAP_CHARS return chunks async def hybrid_search(self, query: str, top_k: int = 5, alpha: float = 0.6) -> List[dict]: if not self._ready: logger.warning("hybrid_search called before ready — returning empty results") return [] try: # fetch_k: retrieve 3x candidates, minimum floor of 15 fetch_k = max(top_k * 3, 15) dense_results = await self.vectorstore.asimilarity_search_with_score( query, k=fetch_k ) tokenized_query = query.lower().split() bm25_scores = self.bm25.get_scores(tokenized_query) top_bm25_idx = np.argsort(bm25_scores)[::-1][:fetch_k].tolist() # Weighted RRF fusion # alpha controls dense list weight; (1-alpha) controls BM25 list weight K = 60 rrf: dict[int, float] = {} for rank, (doc, _score) in enumerate(dense_results): idx = self._find_corpus_index(doc.page_content) if idx >= 0: rrf[idx] = rrf.get(idx, 0.0) + alpha * (1.0 / (K + rank + 1)) for rank, idx in enumerate(top_bm25_idx): rrf[idx] = rrf.get(idx, 0.0) + (1 - alpha) * (1.0 / (K + rank + 1)) sorted_idx = sorted(rrf, key=lambda i: rrf[i], reverse=True)[:top_k] results = [] for idx in sorted_idx: if idx < len(self.bm25_corpus): meta = self.bm25_meta[idx] results.append( { "content": self.bm25_corpus[idx], "source": meta.get("source", ""), "title": meta.get("title", ""), "filename": meta.get("filename", ""), "score": round(rrf[idx], 6), } ) return results except Exception as exc: logger.error("hybrid_search error: %s", exc, exc_info=True) return [] def _find_corpus_index(self, content: str) -> int: """O(1) reverse lookup using pre-built dict. Falls back to -1 if not found.""" return self._content_to_idx.get(content.strip(), -1) def audit_knowledge_base(self) -> dict: """Scan KB files for duplicate titles. Call explicitly — not on every stats poll.""" files = sorted(self._knowledge_dir().glob("*.txt")) titles: dict[str, str] = {} duplicates = [] for path in files: title = path.stem try: text = path.read_text(encoding="utf-8", errors="ignore")[:500] first_title = next( ( line.replace("TITLE:", "").strip() for line in text.splitlines() if line.startswith("TITLE:") ), title, ) except Exception: first_title = title if first_title in titles: duplicates.append( {"title": first_title, "files": [titles[first_title], path.name]} ) else: titles[first_title] = path.name return {"file_count": len(files), "duplicate_titles": duplicates[:20]} def get_stats(self) -> dict: """Lightweight stats — no disk reads beyond what's already in memory.""" return { "ready": self._ready, "chunk_count": len(self.bm25_corpus), "knowledge_dir": str(self._knowledge_dir()), "vectorstore_loaded": self.vectorstore is not None, "bm25_loaded": self.bm25 is not None, "kb_hash": self._kb_hash, }