Spaces:
Running
Running
| """The knowledge base, indexed in memory. | |
| Building the index parses ~15 MB of JSON and tokenizes ~12,000 documents — a few | |
| seconds, dominated by the 5.8 MB course-details file. That's too slow to do inside | |
| the first request and pointless to do before the app can serve a page, so it runs | |
| on a background thread started at app startup. Everything else (the landing page, | |
| sign-in, the profile API) is available immediately; only chat waits. | |
| """ | |
| from __future__ import annotations | |
| import logging | |
| import os | |
| import threading | |
| import time | |
| from pathlib import Path | |
| from .doc import DOMAINS, Doc, SOURCE_LABELS | |
| from .index import Index, snippet | |
| __all__ = ["DOMAINS", "Doc", "SOURCE_LABELS", "Index", "snippet", | |
| "start_warmup", "get_index", "ready", "describe"] | |
| log = logging.getLogger("foresight.kb") | |
| KB_DIR = Path(os.environ.get("FORESIGHT_KB_DIR") or | |
| (Path(__file__).resolve().parent.parent.parent / "knowledge-base")) | |
| _index: Index | None = None | |
| _error: str | None = None | |
| _elapsed: float = 0.0 | |
| _lock = threading.Lock() | |
| _done = threading.Event() | |
| def _build() -> None: | |
| global _index, _error, _elapsed | |
| started = time.monotonic() | |
| try: | |
| from . import normalize | |
| docs = normalize.load_all(KB_DIR) | |
| idx = Index(docs) | |
| with _lock: | |
| _index = idx | |
| _elapsed = time.monotonic() - started | |
| log.info("kb: indexed %d documents in %.1fs — %s", | |
| len(docs), _elapsed, idx.stats()["by_source"]) | |
| except Exception as err: # never take the app down with us | |
| _error = str(err) | |
| log.exception("kb: index build failed") | |
| finally: | |
| _done.set() | |
| def start_warmup() -> None: | |
| """Kick off the index build. Safe to call more than once.""" | |
| if _done.is_set() or getattr(start_warmup, "_started", False): | |
| return | |
| start_warmup._started = True # type: ignore[attr-defined] | |
| threading.Thread(target=_build, name="kb-index", daemon=True).start() | |
| def get_index(timeout: float = 30.0) -> Index: | |
| """The index, waiting for the warm-up if it's still running.""" | |
| if _index is None: | |
| start_warmup() | |
| _done.wait(timeout) | |
| if _index is None: | |
| raise RuntimeError(_error or "knowledge base index is not ready yet") | |
| return _index | |
| def ready() -> bool: | |
| return _index is not None | |
| def describe() -> dict: | |
| out: dict = {"ready": ready(), "dir": str(KB_DIR)} | |
| if _index is not None: | |
| out.update(_index.stats()) | |
| out["build_seconds"] = round(_elapsed, 1) | |
| if _error: | |
| out["error"] = _error | |
| return out | |