"""Pharos pack loading + confidence-gated routing. Ported from the Pharos router on MTH (~/lab/projects/pharos/router.py): embedding similarity when sentence-transformers is installed, with a deterministic keyword-overlap fallback so routing still works on a bare deployment. Encoding policy follows the Pharos encoding-comparison results: triples-only is the safe default; richer encodings only above confidence thresholds; walk-only never (it hurt reasoning in the benchmark). """ import json import re from dataclasses import dataclass, field from pathlib import Path DEFAULT_PACKS_DIR = Path(__file__).parent.parent.parent / "packs" @dataclass class Pack: name: str description: str triples: list path: Path def triples_text(self, max_triples: int = 200) -> str: lines = [f"Knowledge domain: {self.description}\n", "Key relationships:"] for t in self.triples[:max_triples]: s = t.get("subject", t.get("s", "")) p = t.get("predicate", t.get("p", "")) o = t.get("object", t.get("o", "")) lines.append(f"- {s} [{p}] {o}") return "\n".join(lines) def descriptor(self) -> str: from collections import Counter counter: Counter = Counter() for t in self.triples[:120]: for k in ("subject", "s", "predicate", "p", "object", "o"): counter.update(re.findall( r"[a-z]{3,}", str(t.get(k, "")).lower().replace("_", " "), )) common = [w for w, _ in counter.most_common(100)] return (f"{self.name.replace('_', ' ')}. {self.description}. " + " ".join(common)) @dataclass class RoutedKnowledge: knowledge_text: str = "" pack_names: list = field(default_factory=list) confidence: float = 0.0 encoding_policy: str = "none" explanation: str = "" class PackLibrary: def __init__(self, packs_dir: Path | str = DEFAULT_PACKS_DIR): self.packs_dir = Path(packs_dir) self.packs: list[Pack] = [] self._load() def _load(self) -> None: if not self.packs_dir.exists(): return # Flat .json packs (Multiverse demo format) ... for f in sorted(self.packs_dir.glob("*.json")): try: data = json.loads(f.read_text()) except json.JSONDecodeError: continue if isinstance(data, dict) and "triples" in data: self.packs.append(Pack( name=data.get("pack_name", f.stem), description=data.get("description", f.stem), triples=data["triples"], path=f, )) # ... and directory packs (triples.json inside, Pharos pack format) for d in sorted(p for p in self.packs_dir.iterdir() if p.is_dir()): tf = d / "triples.json" if not tf.exists(): continue try: data = json.loads(tf.read_text()) except json.JSONDecodeError: continue if isinstance(data, list): triples, desc = data, d.name else: triples = data.get("triples", []) desc = data.get("description", d.name) self.packs.append(Pack(name=d.name, description=desc, triples=triples, path=tf)) def get(self, name: str) -> Pack | None: for p in self.packs: if p.name == name: return p return None class PharosRouter: """Confidence-gated pack selection (see module docstring).""" def __init__(self, library: PackLibrary, match_threshold: float = 0.30, source_threshold: float = 0.55, max_packs: int = 2, embedding_model: str = "all-MiniLM-L6-v2"): self.library = library self.match_threshold = match_threshold self.source_threshold = source_threshold self.max_packs = max_packs self._st_model = None self._embeddings = None self._try_embeddings(embedding_model) def _try_embeddings(self, model_name: str) -> None: try: from sentence_transformers import SentenceTransformer except ImportError: return if not self.library.packs: return self._st_model = SentenceTransformer(model_name) descriptors = [p.descriptor() for p in self.library.packs] self._embeddings = self._st_model.encode( descriptors, convert_to_numpy=True, ) def route(self, query: str) -> RoutedKnowledge: if not self.library.packs: return RoutedKnowledge(explanation="no packs loaded") scored = (self._route_embedding(query) if self._st_model is not None else self._route_keyword(query)) matches = [(p, s) for p, s in scored if s >= self.match_threshold] matches = matches[: self.max_packs] if not matches: return RoutedKnowledge( explanation=f"no pack above threshold {self.match_threshold}", ) top_conf = matches[0][1] policy = "triples_source" if top_conf >= self.source_threshold else "triples" knowledge = "\n\n".join(p.triples_text() for p, _ in matches) return RoutedKnowledge( knowledge_text=knowledge, pack_names=[p.name for p, _ in matches], confidence=round(top_conf, 3), encoding_policy=policy, explanation=(f"top={matches[0][0].name} ({top_conf:.3f}), " f"method={'embedding' if self._st_model else 'keyword'}"), ) def _route_embedding(self, query: str) -> list: import numpy as np q = self._st_model.encode(query, convert_to_numpy=True) sims = self._embeddings @ q / ( np.linalg.norm(self._embeddings, axis=1) * (np.linalg.norm(q) or 1) ) ranked = sorted(zip(self.library.packs, sims.tolist()), key=lambda x: -x[1]) return ranked def _route_keyword(self, query: str) -> list: """Fraction of (content-bearing) query words found in the pack descriptor, with 4-char prefix matching so migration/migrations and share/shared/shares count as hits.""" stop = {"the", "and", "for", "with", "how", "what", "why", "does", "should", "would", "can", "you", "this", "that", "when"} q_words = [w for w in re.findall(r"[a-z]{3,}", query.lower()) if w not in stop] scored = [] for pack in self.library.packs: d_words = set(re.findall(r"[a-z]{3,}", pack.descriptor().lower())) if not q_words or not d_words: scored.append((pack, 0.0)) continue def matched(qw: str) -> bool: if qw in d_words: return True if len(qw) >= 4: prefix = qw[:4] return any(dw.startswith(prefix) or qw.startswith(dw[:4]) for dw in d_words if len(dw) >= 4) return False hits = sum(1 for qw in q_words if matched(qw)) scored.append((pack, hits / len(q_words))) return sorted(scored, key=lambda x: -x[1]) def load_router(packs_dir: Path | str = DEFAULT_PACKS_DIR, **kwargs) -> PharosRouter: return PharosRouter(PackLibrary(packs_dir), **kwargs) if __name__ == "__main__": import sys router = load_router(sys.argv[1] if len(sys.argv) > 1 else DEFAULT_PACKS_DIR) print(f"packs: {[p.name for p in router.library.packs]}") for q in ("How do I add a migration for a new table?", "Why does the webhook skip signature verification?"): r = router.route(q) print(f"\nQ: {q}\n -> {r.pack_names} conf={r.confidence} " f"policy={r.encoding_policy} ({r.explanation})")