Spaces:

anris05
/

chatbot

Running on Zero

File size: 8,041 Bytes

"""
Intent classification helpers used by the agent router.

This module is intentionally rule-based (regex + fuzzy topic-name matching)
rather than LLM-based, since the LLM provider abstraction (llm/) hasn't
been wired in yet. It's isolated behind plain functions so it can be
swapped for an LLM-based classifier later without changing router.py's
interface — router.py only calls extract_topics(), is_comparison_query(),
and is_followup_query().

No keyword-based *retrieval* happens here — this only decides ROUTING.
Actual knowledge retrieval is always embedding-based (see rag/retriever.py).
"""

import difflib
import json
import re

import config

# ---------------------------------------------------------------------------
# Topic vocabulary (loaded once from topic_metadata.json)
# ---------------------------------------------------------------------------

_FUZZY_MATCH_CUTOFF = 0.85


def _load_known_topics() -> list:
    """Returns the list of canonical topic names, e.g. ['Binary Search', ...]."""
    with open(config.TOPIC_METADATA_PATH, "r", encoding="utf-8") as f:
        metadata = json.load(f)
    return sorted({entry["topic"] for entry in metadata.values() if entry.get("topic")})


_KNOWN_TOPICS = None


def get_known_topics() -> list:
    global _KNOWN_TOPICS
    if _KNOWN_TOPICS is None:
        _KNOWN_TOPICS = _load_known_topics()
    return _KNOWN_TOPICS


def refresh_known_topics() -> None:
    """Call after topic_metadata.json changes (e.g. after a re-index) to
    force reloading the topic vocabulary on next use."""
    global _KNOWN_TOPICS
    _KNOWN_TOPICS = None


# ---------------------------------------------------------------------------
# Comparison detection
# ---------------------------------------------------------------------------

_COMPARISON_PATTERNS = [
    re.compile(r"\bdifference between\b", re.IGNORECASE),
    re.compile(r"\bvs\.?\b", re.IGNORECASE),
    re.compile(r"\bversus\b", re.IGNORECASE),
    re.compile(r"\bcompare(d|s)?\b", re.IGNORECASE),
    re.compile(r"\bwhich (one )?is (better|faster|more efficient)\b", re.IGNORECASE),
    re.compile(r"\bpros and cons of\b", re.IGNORECASE),
]


def is_comparison_query(query: str) -> bool:
    return any(p.search(query) for p in _COMPARISON_PATTERNS)


# ---------------------------------------------------------------------------
# Follow-up detection
# ---------------------------------------------------------------------------

_FOLLOWUP_PATTERNS = [
    re.compile(r"^\s*(explain|say|show)?\s*(that|this|it)\b.*again\b", re.IGNORECASE),
    re.compile(r"^\s*what (do you mean|about (that|it|this))\b", re.IGNORECASE),
    re.compile(r"^\s*(can you|could you)\s+(clarify|elaborate|expand)\b", re.IGNORECASE),
    re.compile(r"^\s*(and|so)\b", re.IGNORECASE),
    re.compile(r"^\s*(why|how) (is that|does that work)\b", re.IGNORECASE),
    re.compile(r"\b(more|another) example\b", re.IGNORECASE),
    re.compile(r"^\s*(simpler|simplify|dumb it down|eli5)\b", re.IGNORECASE),
]

# Bare pronoun references with no topic name mentioned, e.g. "explain it",
# "why is that", "tell me more"
_PRONOUN_ONLY_PATTERN = re.compile(
    r"\b(it|that|this|those|these)\b", re.IGNORECASE
)

_WORKED_EXAMPLE_PATTERN = re.compile(
    r"\d+(?:\s*,\s*\d+){1,}.*\b(this|that|same|additional info|additional information)\b.*\b(search|sort|algorithm|method|approach|traversal)\b",
    re.IGNORECASE,
)

_CODE_REQUEST_PATTERN = re.compile(r"\b(code|implementation|logic)\b", re.IGNORECASE)

def is_followup_query(query: str, has_conversation_history: bool) -> bool:
    """
    A query is treated as a follow-up when:
      1. There IS prior conversation to follow up on, AND
      2. Either it matches a known follow-up phrasing pattern, OR
         it's short, contains a bare pronoun reference, and mentions no
         known topic by name (i.e. it can't stand on its own).
    """
    if not has_conversation_history:
        return False

    if any(p.search(query) for p in _FOLLOWUP_PATTERNS):
        return True

    if _WORKED_EXAMPLE_PATTERN.search(query):
        return True

    word_count = len(query.strip().split())
    has_pronoun = bool(_PRONOUN_ONLY_PATTERN.search(query)) or bool(_CODE_REQUEST_PATTERN.search(query))
    mentions_topic = bool(extract_topics(query))

    # if has_pronoun and not mentions_topic and word_count <= 8:
    #     return True
    if not mentions_topic and word_count <= 8 and has_conversation_history:
        return True

    return False

_GENERIC_COMPARISON_PATTERN = re.compile(
    r"\b(compare|comparison|vs\.?|versus|difference|different from|how does .* (differ|compare))\b.*"
    r"\b(other|similar|related|different|alternative)\b",
    re.IGNORECASE,
)


def is_generic_comparison_followup(query: str) -> bool:
    """
    Matches comparison phrasing that references *other* algorithms without
    naming a second one explicitly, e.g. "compare with other graph
    algorithms", "how does this differ from similar approaches".
    """
    return bool(_GENERIC_COMPARISON_PATTERN.search(query))

def topics_in_recent_history(recent_messages: list) -> list:
    """
    Scans recent conversation turns (both user and assistant messages) for
    known topic mentions. Order preserved, most-recently-mentioned last.
    """
    topics = []
    for message in recent_messages:
        for topic in extract_topics(message.get("content", "")):
            if topic not in topics:
                topics.append(topic)
            else:
                topics.remove(topic)
                topics.append(topic)  # bump to most-recent position
    return topics


# ---------------------------------------------------------------------------
# Topic extraction (fuzzy match against known topic names)
# ---------------------------------------------------------------------------

def extract_topics(query: str) -> list:
    """
    Returns the list of known topic names mentioned in the query, matched
    via substring + fuzzy matching (handles typos/casing/partial names like
    "binary search" or "BFS traversal").

    This is NOT used for retrieval (retrieval stays purely embedding-based)
    — it's only used for ROUTING decisions: deciding single-topic vs.
    comparison, and picking which topics a comparison should retrieve.
    """
    query_lower = query.lower()
    found = []

    for topic in get_known_topics():
        topic_lower = topic.lower()
        # Strip a trailing parenthetical, e.g. "Kruskal's Algorithm (MST)"
        # -> "kruskal's algorithm", since people ask using the plain name,
        # not the parenthetical qualifier.
        topic_core = re.sub(r"\s*\([^)]*\)\s*$", "", topic_lower).strip()

        # Direct substring match (full name or core name without qualifier)
        if topic_lower in query_lower or (topic_core and topic_core in query_lower):
            found.append(topic)
            continue

        # Acronym match, e.g. "Depth-First Search" -> "DFS", "Breadth-First
        # Search" -> "BFS". Common in how people actually ask DSA questions.
        acronym = "".join(w[0] for w in re.split(r"[\s\-]+", topic) if w).lower()
        if len(acronym) >= 3 and re.search(rf"\b{re.escape(acronym)}\b", query_lower):
            found.append(topic)
            continue

        # Fuzzy match against each sliding n-gram window of the query,
        # sized to the topic's word count, to catch typos/near-matches.
        topic_words = topic_lower.split()
        query_words = query_lower.split()
        window = len(topic_words)
        for i in range(len(query_words) - window + 1):
            candidate = " ".join(query_words[i : i + window])
            ratio = difflib.SequenceMatcher(None, candidate, topic_lower).ratio()
            if ratio >= _FUZZY_MATCH_CUTOFF:
                found.append(topic)
                break

    # De-duplicate while preserving order
    seen = set()
    deduped = []
    for t in found:
        if t not in seen:
            seen.add(t)
            deduped.append(t)
    return deduped