"""Local standards knowledge base ingestion. Indexes PDFs/DOCX from configured folders into the vector store under a reserved tenant id. This allows the agentic pipeline to cite RICS standards/exemplar material without requiring uploads through the UI. Files are NOT committed to git; the ingestion simply reads from disk. """ from __future__ import annotations import hashlib import logging from pathlib import Path from langchain_core.documents import Document from app.config import settings from app.ingest.hierarchy import build_hierarchical_documents from app.ingest.pipeline import load_raw_documents, stamp_chunks_for_index from app.retrieval.chunk_role import classify_chunk_role from app.vectorstore.factory import get_vectorstore logger = logging.getLogger(__name__) def _stable_kb_doc_id(path: Path) -> str: # Stable across restarts; changes if file path changes. h = hashlib.sha1(str(path).encode("utf-8"), usedforsecurity=False).hexdigest()[:20] return f"kb-{h}" def iter_knowledge_base_roots() -> list[Path]: """Resolved directory (or single-file parent) roots listed in ``knowledge_base_dirs``.""" raw = (settings.knowledge_base_dirs or "").strip() if not raw: return [] parts = [p.strip() for p in raw.split(",") if p.strip()] root = Path(__file__).resolve().parents[2] # repo root out: list[Path] = [] for part in parts: p = Path(part) if not p.is_absolute(): p = root / p if not p.exists(): continue out.append(p if p.is_dir() else p.parent) # de-dup while preserving order seen: set[str] = set() uniq: list[Path] = [] for p in out: k = str(p.resolve()).lower() if k in seen: continue seen.add(k) uniq.append(p) return uniq def iter_kb_files() -> list[Path]: """List all candidate KB documents from configured dirs.""" raw = (settings.knowledge_base_dirs or "").strip() if not raw: return [] parts = [p.strip() for p in raw.split(",") if p.strip()] root = Path(__file__).resolve().parents[2] # repo root out: list[Path] = [] for part in parts: p = Path(part) if not p.is_absolute(): p = root / p if not p.exists(): continue if p.is_file(): out.append(p) continue for ext in (".pdf", ".docx"): out.extend(sorted(p.rglob(f"*{ext}"))) # de-dup seen: set[str] = set() uniq: list[Path] = [] for p in out: k = str(p.resolve()).lower() if k in seen: continue seen.add(k) uniq.append(p) return uniq def upsert_knowledge_base() -> dict[str, int]: """(Re)index all KB files into the vector store. Returns counts for observability; safe to call at startup. """ if not settings.knowledge_base_enabled: return {"files": 0, "chunks": 0, "skipped": 0} files = iter_kb_files() if not files: logger.info("KB ingest: no knowledge base files found") return {"files": 0, "chunks": 0, "skipped": 0} vs = get_vectorstore() tenant_id = settings.knowledge_base_tenant_id chunks_total = 0 skipped = 0 for fp in files: suf = fp.suffix.lower() if suf not in (".pdf", ".docx"): skipped += 1 continue doc_id = _stable_kb_doc_id(fp) try: # Best-effort upsert: delete prior chunks for this kb doc_id. vs.delete_document(doc_id) raw_docs = load_raw_documents(fp) # Ensure a reasonable source label even without DB Document rows. label = fp.name for d in raw_docs: d.metadata = {**(d.metadata or {}), "source": label} hier = build_hierarchical_documents( raw_docs, doc_id=doc_id, filename=label, chunk_size=settings.chunk_size, chunk_overlap=settings.chunk_overlap, ) stamped = stamp_chunks_for_index(hier, doc_id=doc_id, tenant_id=tenant_id) # Mark provenance as knowledge base for downstream agents and # classify each chunk so retrieval can bias toward boilerplate at # assembly tier. for d in stamped: d.metadata["kb"] = True d.metadata["kb_path"] = str(fp) d.metadata["chunk_role"] = classify_chunk_role(d.page_content or "") vs.add_documents(stamped) chunks_total += len(stamped) except Exception as exc: # noqa: BLE001 logger.warning("KB ingest failed for %s: %s", fp, exc) skipped += 1 if chunks_total: import asyncio from app.retrieval.semantic_cache import invalidate_semantic_cache_for_tenant try: asyncio.get_running_loop() pending = True except RuntimeError: pending = False if pending: asyncio.create_task(invalidate_semantic_cache_for_tenant(tenant_id)) else: asyncio.run(invalidate_semantic_cache_for_tenant(tenant_id)) logger.info("KB ingest complete: files=%d chunks=%d skipped=%d", len(files), chunks_total, skipped) return {"files": len(files), "chunks": chunks_total, "skipped": skipped}