Update services/kb_creation.py

services/kb_creation.py

@@ -1,3 +1,5 @@
+
+# services/kb_creation.py
 import os
 import re
 import pickle
@@ -6,15 +8,15 @@ from docx import Document
 from sentence_transformers import SentenceTransformer
 import chromadb
 
-# --------------------------- ChromaDB setup ---------------------------
+# ----------------------- ChromaDB setup -----------------------
 CHROMA_PATH = os.path.join(os.getcwd(), "chroma_db")
 client = chromadb.PersistentClient(path=CHROMA_PATH)
 collection = client.get_or_create_collection(name="knowledge_base")
 
-# --------------------------- Embedding model ---------------------------
+# ----------------------- Embedding model -----------------------
 model = SentenceTransformer("sentence-transformers/all-MiniLM-L6-v2")
 
-# --------------------------- BM25 (lightweight) ---------------------------
+# ----------------------- BM25 (lightweight) -----------------------
 BM25_INDEX_FILE = os.path.join(CHROMA_PATH, "bm25_index.pkl")
 bm25_docs: List[Dict[str, Any]] = []
 bm25_inverted: Dict[str, List[int]] = {}
@@ -24,7 +26,7 @@ bm25_ready: bool = False
 BM25_K1 = 1.5
 BM25_B = 0.75
 
-# --------------------------- Utilities ---------------------------
+# ----------------------- Utilities -----------------------
 def _tokenize(text: str) -> List[str]:
     if not text:
         return []
@@ -34,18 +36,13 @@ def _tokenize(text: str) -> List[str]:
 def _normalize_query(q: str) -> str:
     q = (q or "").strip().lower()
     q = re.sub(r"[^\w\s]", " ", q)
-    q = re.sub(
-        r"\b(facing|get|getting|got|seeing|receiving|encountered|having|observing|issue|problem)\b",
-        " ",
-        q,
-    )
     q = re.sub(r"\s+", " ", q).strip()
     return q
 
 def _tokenize_meta_value(val: Optional[str]) -> List[str]:
     return _tokenize(val or "")
 
-# --------------------------- DOCX parsing & chunking ---------------------------
+# ----------------------- DOCX parsing & chunking -----------------------
 def _split_by_sections(doc: Document) -> List[Tuple[str, List[str]]]:
     sections: List[Tuple[str, List[str]]] = []
     current_title = None
@@ -78,49 +75,88 @@ def _chunk_text_with_context(doc_title: str, section_title: str, paragraphs: Lis
     for i in range(0, len(words), max_words):
         chunk_body = ' '.join(words[i:i + max_words]).strip()
         if chunk_body:
-            chunks.append(chunk_body)  # no doc/section headers inside text
+            chunks.append(chunk_body)
     if not chunks:
         chunks = [body]
     return chunks
 
-# --------------------------- Intent tagging (auto) ---------------------------
+# ----------------------- Intent tagging -----------------------
+SECTION_STEPS_HINTS = ["process steps", "procedure", "how to", "workflow", "instructions", "steps"]
+SECTION_ERRORS_HINTS = ["common errors", "resolution", "troubleshooting", "known issues", "escalation", "escalation path", "permissions", "access"]
+
+PERMISSION_TERMS = [
+    "permission", "permissions", "access", "access right", "authorization", "authorisation",
+    "role", "role access", "role mapping", "security", "security profile", "privilege", "insufficient",
+    "not allowed", "not authorized", "denied", "restrict"
+]
+
+ERROR_TERMS = ["error", "issue", "fail", "failure", "not working", "cannot", "can't"]
+STEP_VERBS = ["navigate", "select", "scan", "verify", "confirm", "print", "move", "complete", "click", "open", "choose", "enter", "update", "save", "delete", "create", "attach", "assign"]
+
 def _infer_intent_tag(section_title: str) -> str:
     st = (section_title or "").lower()
-    if any(k in st for k in ["process steps", "procedure", "how to", "workflow", "instructions"]):
+    if any(k in st for k in SECTION_STEPS_HINTS):
         return "steps"
-    if any(k in st for k in ["common errors", "resolution", "troubleshooting"]):
+    if any(k in st for k in SECTION_ERRORS_HINTS):
         return "errors"
-    if any(k in st for k in ["pre-requisites", "prerequisites"]):
+    if "pre" in st and "requisite" in st:
         return "prereqs"
     if any(k in st for k in ["purpose", "overview", "introduction"]):
         return "purpose"
     return "neutral"
 
-# --------------------------- Ingestion ---------------------------
+def _derive_semantic_intent_from_text(text: str) -> Tuple[str, List[str]]:
+    """Return ('errors'|'steps'|'neutral', topic_tags) by scanning the text."""
+    t = (text or "").lower()
+    tags: List[str] = []
+    intent = "neutral"
+    # permissions/access first (override)
+    if any(term in t for term in PERMISSION_TERMS):
+        intent = "errors"
+        tags.append("permissions")
+        if "role" in t:
+            tags.append("role_access")
+        if "security" in t:
+            tags.append("security")
+    # generic errors
+    if intent == "neutral" and any(term in t for term in ERROR_TERMS):
+        intent = "errors"
+        tags.append("errors")
+    # steps indicators
+    if intent == "neutral" and any(v in t for v in STEP_VERBS):
+        intent = "steps"
+        tags.append("procedure")
+    return intent, list(set(tags))
+
+# ----------------------- Ingestion -----------------------
 def ingest_documents(folder_path: str) -> None:
-    print(f"📂 Checking folder: {folder_path}")
+    print(f"[KB] Checking folder: {folder_path}")
     files = [f for f in os.listdir(folder_path) if f.lower().endswith('.docx')]
-    print(f"Found {len(files)} Word files: {files}")
+    print(f"[KB] Found {len(files)} Word files: {files}")
     if not files:
-        print("⚠️ No .docx files found. Please check the folder path.")
+        print("[KB] WARNING: No .docx files found. Please check the folder path.")
         return
-
     global bm25_docs, bm25_inverted, bm25_df, bm25_avgdl, bm25_ready
     bm25_docs, bm25_inverted, bm25_df = [], {}, {}
     bm25_avgdl, bm25_ready = 0.0, False
-
     for file in files:
         file_path = os.path.join(folder_path, file)
         doc_title = os.path.splitext(file)[0]
         doc = Document(file_path)
         sections = _split_by_sections(doc)
         total_chunks = 0
-
         for s_idx, (section_title, paras) in enumerate(sections):
             chunks = _chunk_text_with_context(doc_title, section_title, paras, max_words=900)
             total_chunks += len(chunks)
-            intent_tag = _infer_intent_tag(section_title)
+            base_intent = _infer_intent_tag(section_title)
             for c_idx, chunk in enumerate(chunks):
+                derived_intent, topic_tags = _derive_semantic_intent_from_text(chunk)
+                # choose strongest intent: errors overrides steps
+                final_intent = base_intent
+                if derived_intent == "errors":
+                    final_intent = "errors"
+                elif base_intent == "neutral" and derived_intent in ("steps", "prereqs"):
+                    final_intent = derived_intent
                 embedding = model.encode(chunk).tolist()
                 doc_id = f"{file}:{s_idx}:{c_idx}"
                 meta = {
@@ -129,7 +165,8 @@ def ingest_documents(folder_path: str) -> None:
                     "chunk_index": c_idx,
                     "title": doc_title,
                     "collection": "SOP",
-                    "intent_tag": intent_tag,  # NEW
+                    "intent_tag": final_intent,
+                    "topic_tags": topic_tags,
                 }
                 try:
                     collection.add(ids=[doc_id], embeddings=[embedding], documents=[chunk], metadatas=[meta])
@@ -138,29 +175,31 @@ def ingest_documents(folder_path: str) -> None:
                         collection.delete(ids=[doc_id])
                         collection.add(ids=[doc_id], embeddings=[embedding], documents=[chunk], metadatas=[meta])
                     except Exception as e2:
-                        print(f"❌ Upsert failed for {doc_id}: {e2}")
-
+                        print(f"[KB] ERROR: Upsert failed for {doc_id}: {e2}")
                 tokens = _tokenize(chunk)
                 tf: Dict[str, int] = {}
-                for t in tokens:
-                    tf[t] = tf.get(t, 0) + 1
+                for tkn in tokens:
+                    tf[tkn] = tf.get(tkn, 0) + 1
                 idx = len(bm25_docs)
-                bm25_docs.append({"id": doc_id, "text": chunk, "tokens": tokens, "tf": tf, "length": len(tokens), "meta": meta})
-
+                bm25_docs.append({
+                    "id": doc_id,
+                    "text": chunk,
+                    "tokens": tokens,
+                    "tf": tf,
+                    "length": len(tokens),
+                    "meta": meta,
+                })
                 seen = set()
                 for term in tf.keys():
                     bm25_inverted.setdefault(term, []).append(idx)
                     if term not in seen:
                         bm25_df[term] = bm25_df.get(term, 0) + 1
                         seen.add(term)
-
-        print(f"📄 Ingested {file} → {total_chunks} chunks")
-
+        print(f"[KB] Ingested {file} → {total_chunks} chunks")
     N = len(bm25_docs)
     if N > 0:
         bm25_avgdl = sum(d["length"] for d in bm25_docs) / float(N)
         bm25_ready = True
-
     payload = {
         "bm25_docs": bm25_docs,
         "bm25_inverted": bm25_inverted,
@@ -172,9 +211,10 @@ def ingest_documents(folder_path: str) -> None:
     os.makedirs(CHROMA_PATH, exist_ok=True)
     with open(BM25_INDEX_FILE, "wb") as f:
         pickle.dump(payload, f)
-    print(f"✅ BM25 index saved: {BM25_INDEX_FILE}")
-    print(f"✅ Documents ingested. Total entries in Chroma: {collection.count()}")
+    print(f"[KB] BM25 index saved: {BM25_INDEX_FILE}")
+    print(f"[KB] Documents ingested. Total entries in Chroma: {collection.count()}")
 
+# ----------------------- BM25 load -----------------------
 def _load_bm25_index() -> None:
     global bm25_docs, bm25_inverted, bm25_df, bm25_avgdl, bm25_ready
     if not os.path.exists(BM25_INDEX_FILE):
@@ -188,13 +228,13 @@ def _load_bm25_index() -> None:
         bm25_avgdl = payload.get("bm25_avgdl", 0.0)
         bm25_ready = len(bm25_docs) > 0
         if bm25_ready:
-            print(f"✅ BM25 index loaded: {BM25_INDEX_FILE} (docs={len(bm25_docs)})")
+            print(f"[KB] BM25 index loaded: {BM25_INDEX_FILE} (docs={len(bm25_docs)})")
     except Exception as e:
-        print(f"⚠️ Could not load BM25 index: {e}")
+        print(f"[KB] WARNING: Could not load BM25 index: {e}")
 
 _load_bm25_index()
 
-# --------------------------- BM25 search ---------------------------
+# ----------------------- BM25 search -----------------------
 def _bm25_score_for_doc(query_terms: List[str], doc_idx: int) -> float:
     if not bm25_ready or doc_idx < 0 or doc_idx >= len(bm25_docs):
         return 0.0
@@ -240,7 +280,7 @@ def bm25_search(query: str, top_k: int = 50) -> List[Tuple[int, float]]:
     scored.sort(key=lambda x: x[1], reverse=True)
     return scored[:top_k]
 
-# --------------------------- Semantic-only ---------------------------
+# ----------------------- Semantic-only -----------------------
 def search_knowledge_base(query: str, top_k: int = 10) -> dict:
     query_embedding = model.encode(query).tolist()
     res = collection.query(
@@ -248,16 +288,14 @@ def search_knowledge_base(query: str, top_k: int = 10) -> dict:
         n_results=top_k,
         include=['documents', 'metadatas', 'distances']
     )
-    docs_ll = res.get("documents", [[]]) or [[]]
-    metas_ll = res.get("metadatas", [[]]) or [[]]
-    dists_ll = res.get("distances", [[]]) or [[]]
-    ids_ll = res.get("ids", [[]]) or [[]]
-
+    docs_ll = res.get("documents", [[ ]]) or [[ ]]
+    metas_ll = res.get("metadatas", [[ ]]) or [[ ]]
+    dists_ll = res.get("distances", [[ ]]) or [[ ]]
+    ids_ll = res.get("ids", [[ ]]) or [[ ]]
     documents = docs_ll[0] if docs_ll else []
     metadatas = metas_ll[0] if metas_ll else []
     distances = dists_ll[0] if dists_ll else []
     ids = ids_ll[0] if ids_ll else []
-
     if not ids and documents:
         synthesized = []
         for i, m in enumerate(metadatas):
@@ -266,9 +304,7 @@ def search_knowledge_base(query: str, top_k: int = 10) -> dict:
             idx = (m or {}).get("chunk_index", i)
             synthesized.append(f"{fn}:{sec}:{idx}")
         ids = synthesized
-
-    print(f"🔎 KB search → {len(documents)} docs (top_k={top_k}); "
-          f"first distance: {distances[0] if distances else 'n/a'}; ids={len(ids)}")
+    print(f"[KB] search → {len(documents)} docs (top_k={top_k}); first distance: {distances[0] if distances else 'n/a'}; ids={len(ids)}")
     return {
         "documents": documents,
         "metadatas": metadatas,
@@ -276,21 +312,27 @@ def search_knowledge_base(query: str, top_k: int = 10) -> dict:
         "ids": ids,
     }
 
-# --------------------------- Hybrid (BM25 + Embeddings + Intent + Action) ---------------------------
+# ----------------------- Hybrid search helpers -----------------------
 ACTION_SYNONYMS = {
     "create": ["create", "creation", "add", "new", "generate"],
     "update": ["update", "modify", "change", "edit"],
     "delete": ["delete", "remove"],
     "navigate": ["navigate", "go to", "open"],
-    # NOTE: 'perform' REMOVED to avoid wrong boosts like Appointment "performed..."
 }
 
+ERROR_INTENT_TERMS = [
+    "error", "issue", "fail", "not working", "resolution", "fix",
+    "permission", "permissions", "access", "no access", "authorization", "authorisation",
+    "role", "role mapping", "not authorized", "permission denied", "insufficient privileges",
+    "escalation", "escalation path", "access right"
+]
+
 def _detect_user_intent(query: str) -> str:
     q = (query or "").lower()
-    if any(k in q for k in ["steps", "procedure", "how to", "navigate", "perform", "do", "process"]):
-        return "steps"
-    if any(k in q for k in ["error", "issue", "fail", "not working", "resolution", "fix"]):
+    if any(k in q for k in ERROR_INTENT_TERMS):
         return "errors"
+    if any(k in q for k in ["steps", "procedure", "how to", "navigate", "process", "do", "perform"]):
+        return "steps"
     if any(k in q for k in ["pre-requisite", "prerequisites", "requirement", "requirements"]):
         return "prereqs"
     if any(k in q for k in ["purpose", "overview", "introduction"]):
@@ -313,13 +355,18 @@ def _intent_weight(meta: dict, user_intent: str) -> float:
         return 1.0
     if tag in ["purpose", "prereqs"] and user_intent in ["steps", "errors"]:
         return -0.6
+    st = (meta or {}).get("section", "").lower()
+    topics = (meta or {}).get("topic_tags", []) or []
+    if user_intent == "errors" and (any(k in st for k in ["escalation", "permissions", "access", "known issues"]) or ("permissions" in topics)):
+        return 0.7
     return -0.2
 
 def _meta_overlap(meta: Dict[str, Any], q_terms: List[str]) -> float:
     fn_tokens = _tokenize_meta_value(meta.get("filename"))
     title_tokens = _tokenize_meta_value(meta.get("title"))
     section_tokens = _tokenize_meta_value(meta.get("section"))
-    meta_tokens = set(fn_tokens + title_tokens + section_tokens)
+    topic_tokens = _tokenize_meta_value(' '.join((meta.get("topic_tags") or [])))
+    meta_tokens = set(fn_tokens + title_tokens + section_tokens + topic_tokens)
     if not meta_tokens or not q_terms:
         return 0.0
     qset = set(q_terms)
@@ -368,7 +415,6 @@ def hybrid_search_knowledge_base(query: str, top_k: int = 10, alpha: float = 0.6
     bm25_hits = bm25_search(norm_query, top_k=max(50, top_k * 5))
     bm25_max = max([s for _, s in bm25_hits], default=1.0)
     bm25_norm_pairs = [(idx, (score / bm25_max) if bm25_max > 0 else 0.0) for idx, score in bm25_hits]
-
     bm25_id_to_norm, bm25_id_to_text, bm25_id_to_meta = {}, {}, {}
     for idx, nscore in bm25_norm_pairs:
         d = bm25_docs[idx]
@@ -378,8 +424,8 @@ def hybrid_search_knowledge_base(query: str, top_k: int = 10, alpha: float = 0.6
 
     union_ids = set(sem_ids) | set(bm25_id_to_norm.keys())
 
-    gamma = 0.25  # meta overlap
-    delta = 0.35  # intent boost
+    gamma = 0.30  # meta overlap
+    delta = 0.45  # intent boost (stronger for errors)
     epsilon = 0.30  # action weight
 
     combined_records_ext: List[Tuple[str, float, float, str, Dict[str, Any], float, float, float]] = []
@@ -392,20 +438,15 @@ def hybrid_search_knowledge_base(query: str, top_k: int = 10, alpha: float = 0.6
             sem_meta = sem_metas[pos] if pos < len(sem_metas) else {}
         else:
             sem_sim, sem_dist, sem_text, sem_meta = 0.0, None, "", {}
-
         bm25_sim = bm25_id_to_norm.get(cid, 0.0)
         bm25_text = bm25_id_to_text.get(cid, "")
         bm25_meta = bm25_id_to_meta.get(cid, {})
-
         text = sem_text if sem_text else bm25_text
         meta = sem_meta if sem_meta else bm25_meta
-
         m_overlap = _meta_overlap(meta, q_terms)
         intent_boost = _intent_weight(meta, user_intent)
         act_wt = _action_weight(text, actions)
-
         final_score = alpha * sem_sim + beta * bm25_sim + gamma * m_overlap + delta * intent_boost + epsilon * act_wt
-
         combined_records_ext.append(
             (cid, final_score, (sem_dist if sem_dist is not None else 999.0), text, meta, m_overlap, intent_boost, act_wt)
         )
@@ -423,7 +464,9 @@ def hybrid_search_knowledge_base(query: str, top_k: int = 10, alpha: float = 0.6
         total_intent = sum(max(0.0, r[6]) for r in recs)
         total_action = sum(max(0.0, r[7]) for r in recs)
         total_penalty = sum(min(0.0, r[6]) for r in recs) + sum(min(0.0, r[7]) for r in recs)
-        return total_score + 0.4 * total_overlap + 0.6 * total_intent + 0.5 * total_action + 0.3 * total_penalty
+        esc_weight = 0.3 if any("escalation" in (r[4] or {}).get("section", "").lower() for r in recs) else 0.0
+        perm_weight = 0.3 if any("permissions" in ((r[4] or {}).get("topic_tags") or []) for r in recs) else 0.0
+        return total_score + 0.4 * total_overlap + 0.7 * total_intent + 0.5 * total_action + 0.3 * total_penalty + esc_weight + perm_weight
 
     best_doc, best_doc_prior = None, -1.0
     for fn, recs in doc_groups.items():
@@ -441,7 +484,6 @@ def hybrid_search_knowledge_base(query: str, top_k: int = 10, alpha: float = 0.6
 
     reordered = best_recs + other_recs
     top = reordered[:top_k]
-
     documents = [t[3] for t in top]
     metadatas = [t[4] for t in top]
     distances = [t[2] for t in top]
@@ -460,9 +502,8 @@ def hybrid_search_knowledge_base(query: str, top_k: int = 10, alpha: float = 0.6
         "actions": actions,
     }
 
-# --------------------------- Section fetch helpers (for full output) ---------------------------
+# ----------------------- Section fetch helpers -----------------------
 def get_section_text(filename: str, section: str) -> str:
-    """Concatenate all chunk texts for a given filename+section."""
     texts: List[str] = []
     for d in bm25_docs:
         m = d.get("meta", {})
@@ -473,7 +514,6 @@ def get_section_text(filename: str, section: str) -> str:
     return "\n\n".join(texts).strip()
 
 def get_best_steps_section_text(filename: str) -> str:
-    """Return combined text of all 'steps' sections in the given SOP (filename)."""
     texts: List[str] = []
     for d in bm25_docs:
         m = d.get("meta", {})
@@ -483,7 +523,27 @@ def get_best_steps_section_text(filename: str) -> str:
                 texts.append(t)
     return "\n\n".join(texts).strip()
 
-# --- Admin helpers (optional; unchanged) ---
+def get_best_errors_section_text(filename: str) -> str:
+    """Return combined text of all error/permission/escalation chunks for the given SOP."""
+    texts: List[str] = []
+    for d in bm25_docs:
+        m = d.get("meta", {})
+        sec = (m.get("section") or "").lower()
+        topics = (m.get("topic_tags") or [])
+        if m.get("filename") == filename and (
+            m.get("intent_tag") == "errors"
+            or "error" in sec
+            or "escalation" in sec
+            or "permission" in sec
+            or "access" in sec
+            or ("permissions" in topics)
+        ):
+            t = (d.get("text") or "").strip()
+            if t:
+                texts.append(t)
+    return "\n\n".join(texts).strip()
+
+# ----------------------- Admin helpers -----------------------
 def get_kb_runtime_info() -> Dict[str, Any]:
     return {
         "chroma_path": CHROMA_PATH,
@@ -513,5 +573,4 @@ def reset_kb(folder_path: str) -> Dict[str, Any]:
         result["info"] = get_kb_runtime_info()
         return result
     except Exception as e:
-        return {"status": "ERROR", "error": f"{e}", "info": get_kb_runtime_info()}
-
+        return {"status": "ERROR", "error": f"{e}", "info": get_kb_runtime_info()}