Update main.py

main.py

@@ -1,13 +1,5 @@
-
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
-"""
-main.py
-
-- For 'steps' intent: skip LLM rewriter; render full numbered steps directly.
-- Exact = high overlap OR high combined score. If exact → status=OK; else PARTIAL.
-- No 'partial' word override for steps (prevents false PARTIAL).
-"""
 
 import os, json, re, requests, builtins
 from typing import Optional, Any, Dict, List, Tuple
@@ -32,8 +24,10 @@ VERIFY_SSL = os.getenv("SERVICENOW_SSL_VERIFY", "true").lower() in ("1", "true",
 GEMINI_SSL_VERIFY = os.getenv("GEMINI_SSL_VERIFY", "true").lower() in ("1", "true", "yes")
 
 def safe_str(e: Any) -> str:
-    try: return builtins.str(e)
-    except Exception: return "<error stringify failed>"
+    try:
+        return builtins.str(e)
+    except Exception:
+        return "<error stringify failed>"
 
 load_dotenv()
 os.environ["POSTHOG_DISABLED"] = "true"
@@ -77,7 +71,7 @@ class TicketStatusInput(BaseModel):
     sys_id: Optional[str] = None
     number: Optional[str] = None
 
-# --- filters & extractors (same behavior as your file, but without LLM) ---
+# --- filters & extractors (no LLM) ---
 STRICT_OVERLAP = 3
 MAX_SENTENCES_CONCISE = 6
 
@@ -119,29 +113,41 @@ def _filter_context_for_query(context: str, query: str) -> Tuple[str, Dict[str,
     return "\n".join(kept).strip(), {'mode': 'concise', 'matched_count': 0, 'all_sentences': len(sentences)}
 
 STEP_LINE_REGEX = re.compile(r"^\s*(?:\d+[\.)]\s+|[•\-]\s+)", re.IGNORECASE)
-NAV_LINE_REGEX = re.compile(r"(navigate\s+to|>\s*)", re.IGNORECASE)
-PROCEDURE_VERBS = ["log in","select","scan","verify","confirm","print","move","complete","click","open","navigate","choose","enter","update","save","delete","create","attach","assign"]
-VERB_START_REGEX = re.compile(r"^\s*(?:" + "|".join([re.escape(v) for v in PROCEDURE_VERBS]) + r")\b", re.IGNORECASE)
-NON_PROC_ANY_REGEX = re.compile("|".join([re.escape(v) for v in ["to ensure","as per","purpose","pre-requisites","prerequisites","overview","introduction","organized manner","structured","help users","objective"]]), re.IGNORECASE)
-ACTION_SYNS_FLAT = {"create":["create","creation","add","new","generate"],"update":["update","modify","change","edit"],"delete":["delete","remove"],"navigate":["navigate","go to","open"]}
+NAV_LINE_REGEX  = re.compile(r"(navigate\s+to|>\s*)", re.IGNORECASE)
+PROCEDURE_VERBS = [
+    "log in","select","scan","verify","confirm","print",
+    "move","complete","click","open","navigate","choose",
+    "enter","update","save","delete","create","attach","assign"
+]
+VERB_START_REGEX   = re.compile(r"^\s*(?:" + "|".join([re.escape(v) for v in PROCEDURE_VERBS]) + r")\b", re.IGNORECASE)
+NON_PROC_ANY_REGEX = re.compile("|".join([re.escape(v) for v in [
+    "to ensure","as per","purpose","pre-requisites","prerequisites","overview","introduction",
+    "organized manner","structured","help users","objective"
+]]), re.IGNORECASE)
+ACTION_SYNS_FLAT   = {"create":["create","creation","add","new","generate"],"update":["update","modify","change","edit"],"delete":["delete","remove"],"navigate":["navigate","go to","open"]}
 
 def _action_in_line(ln: str, target_actions: List[str]) -> bool:
     s = (ln or "").lower()
     for act in target_actions:
         for syn in ACTION_SYNS_FLAT.get(act, [act]):
-            if syn in s: return True
+            if syn in s:
+                return True
     return False
 
 def _is_procedural_line(ln: str) -> bool:
     s = (ln or "").strip()
-    if not s: return False
-    if NON_PROC_ANY_REGEX.search(s): return False
+    if not s:
+        return False
+    if NON_PROC_ANY_REGEX.search(s):
+        return False
     if STEP_LINE_REGEX.match(s):
         if s.lstrip().startswith(("•","-")):
             return bool(VERB_START_REGEX.search(s) or NAV_LINE_REGEX.search(s))
         return True
-    if VERB_START_REGEX.match(s): return True
-    if NAV_LINE_REGEX.search(s): return True
+    if VERB_START_REGEX.match(s):
+        return True
+    if NAV_LINE_REGEX.search(s):
+        return True
     return False
 
 def _extract_steps_only(text: str, max_lines: Optional[int] = None, target_actions: Optional[List[str]] = None) -> str:
@@ -150,25 +156,29 @@ def _extract_steps_only(text: str, max_lines: Optional[int] = None, target_actio
     for ln in lines:
         if _is_procedural_line(ln):
             normalized = re.sub(r"^\s*(?:\d+[\.)]\s+|[•\-]\s+)", "", ln).strip().lower()
-            if normalized in seen: continue
-            if target_actions and not _action_in_line(ln, target_actions): continue
-            kept.append(ln); seen.add(normalized)
-            if max_lines is not None and len(kept) >= max_lines: break
+            if normalized in seen:
+                continue
+            if target_actions and not _action_in_line(ln, target_actions):
+                continue
+            kept.append(ln)
+            seen.add(normalized)
+            if max_lines is not None and len(kept) >= max_lines:
+                break
+    return "\n".join(kept).strip() if kept else (text or "").strip()
+
+# ---- Errors extractor ----
+def _extract_errors_only(text: str, max_lines: int = 12) -> str:
+    lines = [ln.strip() for ln in (text or "").splitlines() if ln.strip()]
+    kept: List[str] = []
+    for ln in lines:
+        if STEP_LINE_REGEX.match(ln) or ln.lower().startswith(("error", "resolution", "fix", "verify", "check")):
+            kept.append(ln)
+            if len(kept) >= max_lines:
+                break
     return "\n".join(kept).strip() if kept else (text or "").strip()
 
-def _format_steps_markdown(lines: List[str]) -> str:
-    items, seen_norm = [], set()
-    for i, ln in enumerate(lines, start=1):
-        s = (ln or "").strip()
-        if not s: continue
-        s = re.sub(r"^\s*(?:\d+[\.)]\s+|[•\-]\s+)", "", s).strip()
-        norm = s.lower()
-        if norm in seen_norm: continue
-        seen_norm.add(norm)
-        items.append(f"{i}. {s}")
-    return "\n".join(items).strip()
-    
-ESCALATION_REGEX = re.compile(r"^\\s*Escalation Path|^\\s*\\→\\s*|\\s+→\\s+", re.IGNORECASE)
+# ---- Escalation extractor (optional) ----
+ESCALATION_REGEX = re.compile(r"^\s*Escalation Path|\s+→\s+", re.IGNORECASE)
 
 def _extract_escalation_only(text: str, max_lines: int = 3) -> str:
     lines = [ln.strip() for ln in (text or "").splitlines() if ln.strip()]
@@ -180,48 +190,32 @@ def _extract_escalation_only(text: str, max_lines: int = 3) -> str:
                 break
     return "\n".join(kept).strip()
 
-# --- conversation intents (unchanged) ---
-def _is_incident_intent(msg_norm: str) -> bool:
-    intents = ["create ticket","create a ticket","raise ticket","raise a ticket","open ticket","open a ticket","create incident","create an incident","raise incident","raise an incident","open incident","open an incident","log ticket","log an incident","generate ticket","create snow ticket","raise snow ticket","raise service now ticket","create service now ticket","raise sr","open sr"]
-    return any(p in msg_norm for p in intents)
-
-def _is_generic_issue(msg_norm: str) -> bool:
-    generic = ["issue","have an issue","having an issue","got an issue","problem","have a problem","help","need help","support","need support","please help","need assistance","assist me","facing issue","facing a problem","got a problem"]
-    return any(p == msg_norm or p in msg_norm for p in generic) or len(msg_norm.split()) <= 2
-
-def _merge_number_only_lines(lines: list[str]) -> list[str]:
-    """
-    Merge lines that are just '1', '2', '3', ... with the next non-empty,
-    non-number-only line so the final output becomes '1. <text>'.
-    Also skips consecutive number-only lines safely.
-    """
-    merged: list[str] = []
-    i = 0
-    n = len(lines)
-
+# ---- merge number-only lines helper ----
+def _merge_number_only_lines(lines: List[str]) -> List[str]:
+    merged: List[str] = []
+    i, n = 0, len(lines)
     def is_number_only(s: str) -> bool:
         return bool(re.fullmatch(r"\s*\d+\s*", s or ""))
-
     while i < n:
         curr = (lines[i] or "").strip()
-        # Case A: this line is only a number (e.g., "3")
         if is_number_only(curr):
-            # Find the next line that actually has text content (not just another number)
             j = i + 1
             while j < n and is_number_only(lines[j]):
                 j += 1
             if j < n:
                 merged.append(f"{curr.strip()}. {lines[j].strip()}")
-                i = j + 1   # skip the content line we just merged
+                i = j + 1
             else:
-                # Dangling number at end → skip it
                 i += 1
         else:
             merged.append(curr)
             i += 1
-
     return merged
 
+# ---- strip any existing leading numbering/bullets ----
+def _strip_leading_mark(s: str) -> str:
+    return re.sub(r"^\s*(?:\d+[\.)]\s+|[•\-]\s+)", "", (s or "")).strip()
+
 @app.get("/")
 async def health_check():
     return {"status": "ok"}
@@ -231,12 +225,14 @@ async def chat_with_ai(input_data: ChatInput):
     try:
         msg_norm = (input_data.user_message or "").lower().strip()
 
-        # incident & generic handlers (unchanged)
+        # incident & generic handlers
         if _is_incident_intent(msg_norm):
             return {
-                "bot_response": ("Okay, let’s create a ServiceNow incident.\n\n"
-                                 "Please provide:\n• Short Description (one line)\n"
-                                 "• Detailed Description (steps, error text, IDs, site, environment)"),
+                "bot_response": (
+                    "Okay, let’s create a ServiceNow incident.\n\n"
+                    "Please provide:\n• Short Description (one line)\n"
+                    "• Detailed Description (steps, error text, IDs, site, environment)"
+                ),
                 "status": (input_data.prev_status or "PARTIAL"),
                 "context_found": False, "ask_resolved": False, "suggest_incident": False,
                 "show_incident_form": True, "followup": None, "top_hits": [], "sources": [],
@@ -244,12 +240,14 @@ async def chat_with_ai(input_data: ChatInput):
             }
         if _is_generic_issue(msg_norm):
             return {
-                "bot_response": ("Sure, I can help. Please describe your issue:\n"
-                                 "• Module/area (e.g., Picking, Receiving, Trailer Close)\n"
-                                 "• Exact error message text/code (copy-paste)\n"
-                                 "• IDs involved (Order#, Load ID, Shipment#)\n"
-                                 "• Warehouse/site & environment (prod/test)\n"
-                                 "• When it started and how many users are impacted"),
+                "bot_response": (
+                    "Sure, I can help. Please describe your issue:\n"
+                    "• Module/area (e.g., Picking, Receiving, Trailer Close)\n"
+                    "• Exact error message text/code (copy-paste)\n"
+                    "• IDs involved (Order#, Load ID, Shipment#)\n"
+                    "• Warehouse/site & environment (prod/test)\n"
+                    "• When it started and how many users are impacted"
+                ),
                 "status": "NO_KB_MATCH",
                 "context_found": False, "ask_resolved": False, "suggest_incident": False,
                 "followup": "Please reply with the above details.", "top_hits": [], "sources": [],
@@ -258,60 +256,68 @@ async def chat_with_ai(input_data: ChatInput):
 
         # Hybrid KB search
         kb_results = hybrid_search_knowledge_base(input_data.user_message, top_k=10, alpha=0.6, beta=0.4)
-        # Build a small context window from top chunks
+
+        # Build a small context window
         def extract_kb_context(kb_results: Optional[Dict[str, Any]], top_chunks: int = 2) -> Dict[str, Any]:
             if not kb_results or not isinstance(kb_results, dict):
                 return {"context": "", "sources": [], "top_hits": [], "context_found": False, "best_score": None, "best_combined": None}
             documents = kb_results.get("documents") or []
             metadatas = kb_results.get("metadatas") or []
             distances = kb_results.get("distances") or []
-            combined = kb_results.get("combined_scores") or []
-            items = []
+            combined  = kb_results.get("combined_scores") or []
+            items: List[Dict[str, Any]] = []
             for i, doc in enumerate(documents):
                 text = doc.strip() if isinstance(doc, str) else ""
-                if not text: continue
-                meta = metadatas[i] if i < len(metadatas) and isinstance(metadatas[i], dict) else {}
+                if not text:
+                    continue
+                meta  = metadatas[i] if i < len(metadatas) and isinstance(metadatas[i], dict) else {}
                 score = distances[i] if i < len(distances) else None
-                comb = combined[i] if i < len(combined) else None
+                comb  = combined[i]  if i < len(combined)  else None
                 m = dict(meta)
                 if score is not None: m["distance"] = score
-                if comb is not None: m["combined"] = comb
+                if comb  is not None: m["combined"] = comb
                 items.append({"text": text, "meta": m})
             selected = items[:max(1, top_chunks)]
-            context = "\n\n---\n\n".join([s["text"] for s in selected]) if selected else ""
-            sources = [s["meta"] for s in selected]
+            context  = "\n\n---\n\n".join([s["text"] for s in selected]) if selected else ""
+            sources  = [s["meta"] for s in selected]
             best_distance = None
             if distances:
-                try: best_distance = min([d for d in distances if d is not None])
-                except Exception: best_distance = None
+                try:
+                    best_distance = min([d for d in distances if d is not None])
+                except Exception:
+                    best_distance = None
             best_combined = None
             if combined:
-                try: best_combined = max([c for c in combined if c is not None])
-                except Exception: best_combined = None
-            return {"context": context, "sources": sources, "top_hits": [], "context_found": bool(selected),
-                    "best_score": best_distance, "best_combined": best_combined}
+                try:
+                    best_combined = max([c for c in combined if c is not None])
+                except Exception:
+                    best_combined = None
+            return {
+                "context": context, "sources": sources, "top_hits": [], "context_found": bool(selected),
+                "best_score": best_distance, "best_combined": best_combined
+            }
 
-        kb_ctx = extract_kb_context(kb_results, top_chunks=2)
-        context_raw = kb_ctx.get("context", "") or ""
+        kb_ctx        = extract_kb_context(kb_results, top_chunks=2)
+        context_raw   = kb_ctx.get("context", "") or ""
         filtered_text, filt_info = _filter_context_for_query(context_raw, input_data.user_message)
 
         detected_intent = kb_results.get("user_intent", "neutral")
-        actions = kb_results.get("actions", [])
-        best_doc = kb_results.get("best_doc")
-        best_distance = kb_ctx.get("best_score")
-        best_combined = kb_ctx.get("best_combined")
-        top_meta = (kb_results.get("metadatas") or [{}])[0] if (kb_results.get("metadatas") or []) else {}
+        actions         = kb_results.get("actions", [])
+        best_doc        = kb_results.get("best_doc")
+        best_distance   = kb_ctx.get("best_score")
+        best_combined   = kb_ctx.get("best_combined")
+        top_meta        = (kb_results.get("metadatas") or [{}])[0] if (kb_results.get("metadatas") or []) else {}
 
-        short_query = len((input_data.user_message or "").split()) <= 4
-        gate_combined_ok = 0.58 if short_query else 0.55
-        gate_combined_no_kb = 0.22 if short_query else 0.28
-        gate_distance_no_kb = 2.0
+        short_query          = len((input_data.user_message or "").split()) <= 4
+        gate_combined_ok     = 0.58 if short_query else 0.55
+        gate_combined_no_kb  = 0.22 if short_query else 0.28
+        gate_distance_no_kb  = 2.0
 
         exact_by_filter = (filt_info.get('mode') == 'exact' and filt_info.get('matched_count', 0) > 0)
-        high_conf = (best_combined is not None and best_combined >= gate_combined_ok)
-        exact = bool(exact_by_filter or high_conf)
+        high_conf       = (best_combined is not None and best_combined >= gate_combined_ok)
+        exact           = bool(exact_by_filter or high_conf)
 
-        # --- STEPS intent: full SOP steps first (no LLM rewrite) ---
+        # --- STEPS intent: full SOP steps (numbered) ---
         if detected_intent == "steps" and best_doc:
             full_steps = get_best_steps_section_text(best_doc)
             if not full_steps:
@@ -323,26 +329,17 @@ async def chat_with_ai(input_data: ChatInput):
             else:
                 context = _extract_steps_only(filtered_text, max_lines=MAX_SENTENCES_CONCISE, target_actions=actions)
 
-            # Render numbered list
-            
             raw_lines = [ln.strip() for ln in context.splitlines() if ln.strip()]
-
-# If everything is a single paragraph, defensively split on ". "
             if len(raw_lines) == 1:
-               parts = [p.strip() for p in re.split(r"\.\s+(?=[A-Z0-9])", raw_lines[0]) if p.strip()]
-               raw_lines = parts if len(parts) > 1 else raw_lines
-
-# ✅ NEW: merge number-only lines with their following text lines
+                parts = [p.strip() for p in re.split(r"\.\s+(?=[A-Z0-9])", raw_lines[0]) if p.strip()]
+                raw_lines = parts if len(parts) > 1 else raw_lines
             raw_lines = _merge_number_only_lines(raw_lines)
+            cleaned_lines = [_strip_leading_mark(ln) for ln in raw_lines]
 
-# Render as a clean Markdown numbered list
-            #bot_text = _format_steps_markdown(raw_lines)
-            bot_text = "\n".join(
-    [f"{i+1}. {re.sub(r'^\\s*(?:\\d+[\\.)]\\s+|[•\\-]\\s+)', '', ln).strip()}" for i, ln in enumerate(raw_lines)])
+            # Final numbering without regex inside f-string
+            bot_text = "\n".join([f"{i+1}. {ln}" for i, ln in enumerate(cleaned_lines)])
 
-            # Status ONLY from exact/high confidence gates
             status = "OK" if exact else "PARTIAL"
-
             return {
                 "bot_response": bot_text,
                 "status": status,
@@ -360,12 +357,11 @@ async def chat_with_ai(input_data: ChatInput):
                 },
             }
 
-        # --- Non-steps: keep concise behavior ---
+        # --- Non-steps intents ---
         context = filtered_text
         if detected_intent == "errors":
             errs = _extract_errors_only(context, max_lines=MAX_SENTENCES_CONCISE)
             esc  = _extract_escalation_only(context, max_lines=3)
-    # Join errors + escalation neatly (no extra LLM, no steps)
             context = (errs + ("\n" + esc if esc else "")).strip()
         elif "navigate" in msg_norm or "menu" in msg_norm or "screen" in msg_norm:
             context = _extract_steps_only(context, max_lines=MAX_SENTENCES_CONCISE)
@@ -375,14 +371,15 @@ async def chat_with_ai(input_data: ChatInput):
             (best_combined is None or best_combined < gate_combined_no_kb) and (best_distance is None or best_distance >= gate_distance_no_kb)
         ):
             second_try = (input_data.prev_status or "").upper() == "NO_KB_MATCH"
-            clarify = ("I couldn’t find matching content in the KB yet. To help me narrow it down, please share:\n\n"
-                       "• Module/area (e.g., Picking, Receiving, Trailer Close)\n"
-                       "• Exact error message text/code (copy-paste)\n"
-                       "• IDs involved (Order#, Load ID, Shipment#)\n"
-                       "• Warehouse/site & environment (prod/test)\n"
-                       "• When it started and how many users are impacted\n\n"
-                       "Reply with these details and I’ll search again.") if not second_try else \
-                      "I still don’t find a relevant KB match for this scenario even after clarification."
+            clarify = (
+                "I couldn’t find matching content in the KB yet. To help me narrow it down, please share:\n\n"
+                "• Module/area (e.g., Picking, Receiving, Trailer Close)\n"
+                "• Exact error message text/code (copy-paste)\n"
+                "• IDs involved (Order#, Load ID, Shipment#)\n"
+                "• Warehouse/site & environment (prod/test)\n"
+                "• When it started and how many users are impacted\n\n"
+                "Reply with these details and I’ll search again."
+            ) if not second_try else "I still don’t find a relevant KB match for this scenario even after clarification."
             return {
                 "bot_response": clarify,
                 "status": "NO_KB_MATCH",
@@ -395,7 +392,7 @@ async def chat_with_ai(input_data: ChatInput):
 
         # Default response for non-steps
         bot_text = context.strip()
-        status = "OK" if exact else "PARTIAL"
+        status   = "OK" if exact else "PARTIAL"
         return {
             "bot_response": bot_text,
             "status": status,