Update services/kb_creation.py

services/kb_creation.py

@@ -93,7 +93,6 @@ ERROR_TERMS = ["error", "issue", "fail", "failure", "not working", "cannot", "ca
 
 STEP_VERBS = ["navigate", "select", "scan", "verify", "confirm", "print", "move", "complete", "click", "open", "choose", "enter", "update", "save", "delete", "create", "attach", "assign"]
 
-# Module vocabulary → canonical tags
 MODULE_VOCAB = {
     "picking": ["pick", "picking", "pick release", "wave", "allocation"],
     "receiving": ["receive", "receiving", "inbound", "asn", "appointment"],
@@ -116,17 +115,14 @@ def _infer_intent_tag(section_title: str) -> str:
     return "neutral"
 
 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"
     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")
+        if "role" in t: tags.append("role_access")
+        if "security" in t: tags.append("security")
     if intent == "neutral" and any(term in t for term in ERROR_TERMS):
         intent = "errors"
         tags.append("errors")
@@ -141,7 +137,6 @@ def _derive_module_tags(text: str, filename: str, section_title: str) -> List[st
     for mod, syns in MODULE_VOCAB.items():
         if any(s in tokens for s in syns):
             found.append(mod)
-    # default: if none found, try broad guess
     if not found:
         if "inventory" in tokens or "adjust" in tokens or "uom" in tokens or "cycle" in tokens:
             found = ["inventory"]
@@ -175,7 +170,6 @@ def ingest_documents(folder_path: str) -> None:
                     final_intent = "errors"
                 elif base_intent == "neutral" and derived_intent in ("steps", "prereqs"):
                     final_intent = derived_intent
-
                 module_tags = _derive_module_tags(chunk, file, section_title)
 
                 embedding = model.encode(chunk).tolist()
@@ -186,13 +180,13 @@ def ingest_documents(folder_path: str) -> None:
                     "chunk_index": c_idx,
                     "title": doc_title,
                     "collection": "SOP",
-                    "intent_tag": final_intent,
-                    "topic_tags": topic_tags,
-                    "module_tags": module_tags,
+                    "intent_tag": final_intent,                                      # str
+                    "topic_tags": ", ".join(topic_tags) if topic_tags else "",       # str (NOT list)
+                    "module_tags": ", ".join(module_tags) if module_tags else "",    # str (NOT list)
                 }
                 try:
                     collection.add(ids=[doc_id], embeddings=[embedding], documents=[chunk], metadatas=[meta])
-                except Exception:
+                except Exception as e1:
                     try:
                         collection.delete(ids=[doc_id])
                         collection.add(ids=[doc_id], embeddings=[embedding], documents=[chunk], metadatas=[meta])
@@ -374,7 +368,6 @@ def _extract_modules_from_query(query: str) -> List[str]:
     for mod, syns in MODULE_VOCAB.items():
         if any(s in q for s in syns):
             found.append(mod)
-    # broad guess if user says 'inventory'
     if not found and ("inventory" in q or "adjust" in q):
         found = ["inventory"]
     return list(sorted(set(found)))
@@ -388,15 +381,17 @@ def _intent_weight(meta: dict, user_intent: str) -> float:
     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)):
+    topics = (meta or {}).get("topic_tags", "") or ""
+    topic_list = [t.strip() for t in topics.split(",") if t.strip()]
+    if user_intent == "errors" and (any(k in st for k in ["escalation", "permissions", "access", "known issues"]) or ("permissions" in topic_list)):
         return 0.7
     return -0.2
 
 def _module_weight(meta: Dict[str, Any], user_modules: List[str]) -> float:
     if not user_modules:
         return 0.0
-    doc_modules = (meta or {}).get("module_tags", []) or []
+    raw = (meta or {}).get("module_tags", "") or ""
+    doc_modules = [m.strip() for m in raw.split(",") if m.strip()] if isinstance(raw, str) else (raw or [])
     overlap = len(set(user_modules) & set(doc_modules))
     if overlap == 0:
         return -0.4   # demote different modules to avoid wrong SOP
@@ -406,8 +401,8 @@ 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"))
-    topic_tokens = _tokenize_meta_value(' '.join((meta.get("topic_tags") or [])))
-    module_tokens = _tokenize_meta_value(' '.join((meta.get("module_tags") or [])))
+    topic_tokens = _tokenize_meta_value((meta.get("topic_tags") or ""))
+    module_tokens = _tokenize_meta_value((meta.get("module_tags") or ""))
     meta_tokens = set(fn_tokens + title_tokens + section_tokens + topic_tokens + module_tokens)
     if not meta_tokens or not q_terms:
         return 0.0
@@ -570,19 +565,19 @@ def get_best_steps_section_text(filename: str) -> str:
     return "\n\n".join(texts).strip()
 
 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 [])
+        topics = (m.get("topic_tags") or "")
+        topic_list = [t.strip() for t in topics.split(",") if t.strip()]
         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)
+            or ("permissions" in topic_list)
         ):
             t = (d.get("text") or "").strip()
             if t: