Spaces:
Sleeping
Sleeping
File size: 5,392 Bytes
b76f199 0b42403 b76f199 5fca0ca b76f199 0b42403 b76f199 0b42403 b76f199 732b14f b76f199 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 | """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}
|