Update behavior_model.py

behavior_model.py

@@ -1,21 +1,27 @@
+# behavior_model.py
 """
-behavior_model.py
+Improved conversation flow + complexity router.
 
-Lightweight neural conversation flow detector for multi-turn input.
-Uses a zero-shot classifier (transformers) for label scoring and
-embeddings heuristics to consider recent-turn context.
+Exports:
+  - analyze_flow(messages, custom_labels=None, prefer_fast=True) -> dict
+    Adds routing decisions:
+      - flow_label: str
+      - confidence: float
+      - explanation: str
+      - is_complex: bool
+      - complexity_score: float (0.0 - 1.0)
+      - route: "direct" | "planning"  (direct => send to LLM immediately, planning => run planner)
+      - scores: optional dict of label scores (if classifier used)
 
-Output: analyze_flow(messages) -> dict {
-    "flow_label": str,
-    "confidence": float,
-    "explanation": str
-}
-
-Lazy-loads the zero-shot classifier and falls back to heuristics.
+Design goals:
+  - Fast-path for short/simple requests (heuristics only) to reduce latency.
+  - Lazy-load zero-shot classifier only when heuristics are ambiguous.
+  - Thread-safe lazy loading.
 """
 import threading
-from typing import List, Dict
+from typing import List, Dict, Any
 import traceback
+import re
 
 _flow_classifier = None
 _flow_lock = threading.Lock()
@@ -31,6 +37,30 @@ _DEFAULT_LABELS = [
     "closing",
 ]
 
+# fast keyword sets and patterns for complexity heuristics
+_COMPLEX_KEYWORDS = {
+    "task": ["implement", "create", "build", "generate", "write", "develop", "deploy", "setup", "configure", "install", "refactor", "optimize", "benchmark"],
+    "analysis": ["explain", "why", "how", "analyze", "analysis", "compare", "evaluate", "breakdown", "diagnose"],
+    "error": ["error", "exception", "traceback", "stacktrace", "crash", "bug", "not working", "fix", "debug"],
+    "code_signs": ["```", "def ", "function(", "class ", "import ", "console.log", "{", "};", ";", "->", "std::", "#include"],
+    "data": ["dataset", "csv", "json", "table", "rows", "columns", "api", "endpoint"],
+    "math": ["calculate", "compute", "solve", "equation", "integral", "sum", "mean", "variance"]
+}
+
+_QUESTION_WORDS = set(["what","why","how","which","when","where","who","whom","whose","do","does","did","can","could","would","should","is","are","was","were","may","might"])
+
+# weights for heuristic scoring
+_WEIGHTS = {
+    "word_count": 0.15,
+    "sentence_count": 0.05,
+    "has_code": 0.30,
+    "has_numbers_or_urls": 0.05,
+    "task_keywords": 0.20,
+    "analysis_keywords": 0.15,
+    "error_keywords": 0.10,
+    "question_words_density": 0.10
+}
+
 def _load_flow_model():
     global _flow_classifier
     try:
@@ -69,45 +99,93 @@ def _concat_recent_messages(messages: List[Dict], max_chars: int = 1200) -> str:
             break
     return "".join(reversed(parts)).strip()
 
-def analyze_flow(messages: List[Dict], custom_labels: List[str] = None) -> Dict:
-    try:
-        text_blob = _concat_recent_messages(messages, max_chars=1200)
-        labels = custom_labels or _DEFAULT_LABELS
-        _ensure_flow_loaded()
-        if not _flow_classifier or not text_blob:
-            return _heuristic_flow(text_blob)
-
-        result = _flow_classifier(text_blob, candidate_labels=labels, multi_label=False)
-        if not result or 'labels' not in result:
-            return _heuristic_flow(text_blob)
-
-        top_label = result['labels'][0]
-        top_score = float(result['scores'][0] if result.get('scores') else 0.0)
-        explanation_map = {
-            "task_request": "User requests a concrete task — prefer actionable steps.",
-            "clarification": "User asks for clarification — ask concise clarifying questions.",
-            "follow_up": "Continuation — reference prior answer and continue.",
-            "escalation": "User shows dissatisfaction — de-escalate, propose solutions.",
-            "small_talk": "Casual conversation — be friendly and short.",
-            "information_seeking": "Seeking facts — be concise and cite if possible.",
-            "confirmation": "Yes/no or confirmatory — respond succinctly.",
-            "closing": "Conversation ending — provide short wrap-up."
-        }
-        explanation = explanation_map.get(top_label, "Follow user's flow and be concise.")
-        scores = {lbl: float(s) for lbl, s in zip(result.get('labels', []), result.get('scores', []))}
-        return {
-            "flow_label": top_label,
-            "confidence": top_score,
-            "scores": scores,
-            "explanation": explanation
-        }
-    except Exception as e:
-        traceback.print_exc()
-        return _heuristic_flow(_concat_recent_messages(messages))
+def _fast_complexity_score(text: str) -> Dict[str, Any]:
+    """
+    Returns a dict:
+      { score: float (0-1), features: {...}, explanation: str }
+    Higher score -> more complex.
+    """
+    t = (text or "").strip()
+    if not t:
+        return {"score": 0.0, "features": {}, "explanation": "empty text -> trivial"}
+
+    # basic counts
+    words = re.findall(r"\w+", t)
+    word_count = len(words)
+    sentence_count = max(1, len(re.findall(r"[.!?]+", t)) or 1)
+
+    # flags
+    lower = t.lower()
+    has_code = any(sig in lower for sig in _COMPLEX_KEYWORDS["code_signs"])
+    has_numbers = bool(re.search(r"\d+", t))
+    has_url = bool(re.search(r"https?://|www\.|\.[a-z]{2,4}/", lower))
+    has_numbers_or_urls = has_numbers or has_url
+
+    # keyword signals
+    task_kw = sum(1 for k in _COMPLEX_KEYWORDS["task"] if k in lower)
+    analysis_kw = sum(1 for k in _COMPLEX_KEYWORDS["analysis"] if k in lower)
+    error_kw = sum(1 for k in _COMPLEX_KEYWORDS["error"] if k in lower)
+    math_kw = sum(1 for k in _COMPLEX_KEYWORDS["math"] if k in lower)
+    data_kw = sum(1 for k in _COMPLEX_KEYWORDS["data"] if k in lower)
+
+    # question word density
+    qwords = sum(1 for w in re.findall(r"\w+", lower) if w in _QUESTION_WORDS)
+    q_density = qwords / max(1, word_count)
+
+    # compute raw score using weighted features
+    score = 0.0
+    score += min(word_count / 200.0, 1.0) * _WEIGHTS["word_count"]
+    score += min(sentence_count / 6.0, 1.0) * _WEIGHTS["sentence_count"]
+    score += (1.0 if has_code else 0.0) * _WEIGHTS["has_code"]
+    score += (1.0 if has_numbers_or_urls else 0.0) * _WEIGHTS["has_numbers_or_urls"]
+    score += min(task_kw / 3.0, 1.0) * _WEIGHTS["task_keywords"]
+    score += min(analysis_kw / 3.0, 1.0) * _WEIGHTS["analysis_keywords"]
+    score += min(error_kw / 2.0, 1.0) * _WEIGHTS["error_keywords"]
+    score += min(q_density * 2.0, 1.0) * _WEIGHTS["question_words_density"]  # scale
+
+    # minor boosts for data/math keywords
+    if math_kw or data_kw:
+        score = min(score + 0.05, 1.0)
+
+    # normalize (weights sum > 1 so clamp)
+    score = max(0.0, min(score, 1.0))
+
+    features = {
+        "word_count": word_count,
+        "sentence_count": sentence_count,
+        "has_code": has_code,
+        "has_numbers_or_urls": has_numbers_or_urls,
+        "task_kw": task_kw,
+        "analysis_kw": analysis_kw,
+        "error_kw": error_kw,
+        "math_kw": math_kw,
+        "data_kw": data_kw,
+        "q_density": round(q_density, 3)
+    }
+
+    # plain language explanation for fast path
+    expl_parts = []
+    if has_code:
+        expl_parts.append("Detected code-like tokens")
+    if task_kw:
+        expl_parts.append(f"{task_kw} task-related keywords")
+    if analysis_kw:
+        expl_parts.append(f"{analysis_kw} analysis-related keywords")
+    if error_kw:
+        expl_parts.append(f"{error_kw} error/debug keywords")
+    if word_count > 120:
+        expl_parts.append("Long message (>120 words)")
+    if q_density > 0.2:
+        expl_parts.append("High question density")
+
+    explanation = "; ".join(expl_parts) if expl_parts else "No strong complex signals detected"
+
+    return {"score": round(score, 3), "features": features, "explanation": explanation}
 
 def _heuristic_flow(blob: str) -> Dict:
+    # basic fallback from previous implementation, slightly adapted
     b = (blob or "").lower()
-    if any(w in b for w in ["please", "could you", "can you", "i need", "i want"]):
+    if any(w in b for w in ["please", "could you", "can you", "i need", "i want", "please help"]):
         label, conf = "task_request", 0.55
     elif any(w in b for w in ["what do you mean", "clarify", "explain", "how so"]):
         label, conf = "clarification", 0.55
@@ -117,10 +195,134 @@ def _heuristic_flow(blob: str) -> Dict:
         label, conf = "small_talk", 0.6
     elif any(w in b for w in ["error", "not working", "frustrat", "angry", "problem"]):
         label, conf = "escalation", 0.6
-    elif any(w in b for w in ["what is", "who is", "when is", "look up", "search"]):
+    elif any(w in b for w in ["what is", "who is", "when is", "look up", "search", "find"]):
         label, conf = "information_seeking", 0.55
     else:
         label, conf = "follow_up", 0.4
 
-    explanation = f"Fallback heuristic suggests '{label}' (confidence ~{conf}). Mirror user's last message and proceed accordingly."
+    explanation = f"Fallback heuristic suggests '{label}' (confidence ~{conf})."
     return {"flow_label": label, "confidence": conf, "scores": {label: conf}, "explanation": explanation}
+
+def analyze_flow(messages: List[Dict], custom_labels: List[str] = None, prefer_fast: bool = True) -> Dict:
+    """
+    Main entry.
+
+    prefer_fast: if True, prefer heuristic-only decisions when confident to reduce latency.
+    Returns dict with:
+      - flow_label, confidence, explanation
+      - is_complex (bool)
+      - complexity_score (0-1)
+      - route: 'direct' or 'planning'
+      - scores: optional (when classifier used)
+    """
+    try:
+        text_blob = _concat_recent_messages(messages, max_chars=1200)
+        labels = custom_labels or _DEFAULT_LABELS
+
+        # run fast heuristic complexity scoring on the user's latest message (most relevant)
+        last_user_msg = ""
+        if messages:
+            # find last user message content
+            for m in reversed(messages):
+                if m.get("role") == "user" and (m.get("content") or "").strip():
+                    last_user_msg = m.get("content").strip()
+                    break
+
+        fast = _fast_complexity_score(last_user_msg or text_blob)
+        complexity_score = float(fast.get("score", 0.0))
+        features = fast.get("features", {})
+        fast_expl = fast.get("explanation", "")
+
+        # routing heuristics thresholds (tunable)
+        DIRECT_THRESHOLD = 0.25  # <= -> direct (fast)
+        PLANNING_THRESHOLD = 0.60  # >= -> planning (complex)
+        ambig_low = DIRECT_THRESHOLD
+        ambig_high = PLANNING_THRESHOLD
+
+        # quick decision if confident and prefer_fast
+        if prefer_fast and (complexity_score <= ambig_low or complexity_score >= ambig_high):
+            route = "direct" if complexity_score <= ambig_low else "planning"
+            is_complex = complexity_score >= ambig_high
+            # attempt to pick a flow_label via heuristic (fast)
+            fallback = _heuristic_flow(text_blob)
+            label = fallback.get("flow_label", "follow_up")
+            conf = round(0.5 + (0.5 * complexity_score) if is_complex else 0.4, 2)
+            explanation = f"Fast-path decision: route='{route}'. {fast_expl} (score={complexity_score})."
+            return {
+                "flow_label": label,
+                "confidence": conf,
+                "explanation": explanation,
+                "is_complex": bool(is_complex),
+                "complexity_score": round(complexity_score, 3),
+                "route": route,
+                "features": features,
+                "scores": {label: conf}
+            }
+
+        # If ambiguous or prefer classifier, try zero-shot classifier (lazy)
+        _ensure_flow_loaded()
+        if not _flow_classifier or not text_blob:
+            # fallback
+            fallback = _heuristic_flow(text_blob)
+            explanation = f"Classifier unavailable; heuristic fallback. {fast_expl} (score={complexity_score})."
+            # route by heuristic score
+            route = "planning" if complexity_score >= PLANNING_THRESHOLD else "direct"
+            is_complex = complexity_score >= PLANNING_THRESHOLD
+            return {
+                "flow_label": fallback.get("flow_label", "follow_up"),
+                "confidence": fallback.get("confidence", 0.4),
+                "explanation": explanation,
+                "is_complex": bool(is_complex),
+                "complexity_score": round(complexity_score, 3),
+                "route": route,
+                "features": features,
+                "scores": fallback.get("scores")
+            }
+
+        # use classifier to get a more informed flow label
+        try:
+            result = _flow_classifier(text_blob, candidate_labels=labels, multi_label=False)
+            if not result or 'labels' not in result:
+                raise ValueError("classifier returned no labels")
+            top_label = result['labels'][0]
+            top_score = float(result['scores'][0] if result.get('scores') else 0.0)
+            # decide complexity/route combining classifier and heuristic
+            is_complex = complexity_score >= PLANNING_THRESHOLD or top_label in ("task_request", "escalation", "information_seeking")
+            route = "planning" if is_complex or top_score < 0.5 else "direct"
+            explanation = (
+                f"Classifier suggests '{top_label}' (score={round(top_score,2)}). "
+                f"Heuristic complexity score={complexity_score} ({fast_expl}). Routed to '{route}'."
+            )
+            scores = {lbl: float(s) for lbl, s in zip(result.get('labels', []), result.get('scores', []))}
+            return {
+                "flow_label": top_label,
+                "confidence": round(top_score, 3),
+                "explanation": explanation,
+                "is_complex": bool(is_complex),
+                "complexity_score": round(complexity_score, 3),
+                "route": route,
+                "features": features,
+                "scores": scores
+            }
+        except Exception as e:
+            # classifier error -> fallback heuristics
+            traceback.print_exc()
+            fallback = _heuristic_flow(text_blob)
+            route = "planning" if complexity_score >= PLANNING_THRESHOLD else "direct"
+            explanation = f"Classifier error; fallback to heuristic. {fast_expl} (score={complexity_score}). Error: {e}"
+            return {
+                "flow_label": fallback.get("flow_label", "follow_up"),
+                "confidence": fallback.get("confidence", 0.4),
+                "explanation": explanation,
+                "is_complex": complexity_score >= PLANNING_THRESHOLD,
+                "complexity_score": round(complexity_score, 3),
+                "route": route,
+                "features": features,
+                "scores": fallback.get("scores")
+            }
+
+    except Exception as e:
+        traceback.print_exc()
+        return _heuristic_flow(_concat_recent_messages(messages))
+
+# End of behavior_model.py