microdao-daarion/crews/agromatrix_crew/depth_classifier.py

"""
Depth Classifier для Степана.

classify_depth(text, has_doc_context, last_topic, user_profile) → "light" | "deep"

Без залежності від crewai — чистий Python.
Fail-closed: помилка → "deep".
"""

from __future__ import annotations

import logging
import re
from typing import Literal

from crews.agromatrix_crew.telemetry import tlog

logger = logging.getLogger(__name__)

# ─── Patterns ────────────────────────────────────────────────────────────────

_DEEP_ACTION_RE = re.compile(
    r'\b(зроби|зробити|перевір|перевірити|порахуй|порахувати|підготуй|підготувати'
    r'|онови|оновити|створи|створити|запиши|записати|зафіксуй|зафіксувати'
    r'|внеси|внести|проаналізуй|проаналізувати|порівняй|порівняти'
    r'|розрахуй|розрахувати|сплануй|спланувати|покажи|показати'
    r'|заплануй|запланувати|закрий|закрити|відкрий|відкрити)\b',
    re.IGNORECASE | re.UNICODE,
)

_DEEP_URGENT_RE = re.compile(
    r'\b(аварія|терміново|критично|тривога|невідкладно|alert|alarm|critical)\b',
    re.IGNORECASE | re.UNICODE,
)

_DEEP_DATA_RE = re.compile(
    r'\b(\d[\d.,]*)\s*(га|кг|л|т|мм|°c|°f|%|гектар|літр|тонн)',
    re.IGNORECASE | re.UNICODE,
)

_LIGHT_GREET_RE = re.compile(
    r'^(привіт|добрий\s+\w+|доброго\s+\w+|hello|hi|hey|ок|окей|добре|зрозумів|зрозуміла'
    r'|дякую|дякуй|спасибі|чудово|супер|ясно|зрозуміло|вітаю|вітання)[\W]*$',
    re.IGNORECASE | re.UNICODE,
)

_DEEP_INTENTS = frozenset({
    'plan_week', 'plan_day', 'plan_vs_fact', 'show_critical_tomorrow', 'close_plan'
})

# ─── Intent detection (inline, no crewai dependency) ─────────────────────────

def _detect_intent(text: str) -> str:
    t = text.lower()
    if 'сплануй' in t and 'тиж' in t:
        return 'plan_week'
    if 'сплануй' in t:
        return 'plan_day'
    if 'критично' in t or 'на завтра' in t:
        return 'show_critical_tomorrow'
    if 'план/факт' in t or 'план факт' in t:
        return 'plan_vs_fact'
    if 'закрий план' in t:
        return 'close_plan'
    return 'general'


# ─── Public API ───────────────────────────────────────────────────────────────

def classify_depth(
    text: str,
    has_doc_context: bool = False,
    last_topic: str | None = None,
    user_profile: dict | None = None,
    session: dict | None = None,
) -> Literal["light", "deep"]:
    """
    Визначає глибину обробки запиту.

      light — Степан відповідає сам, без запуску під-агентів
      deep  — повний orchestration flow з делегуванням

    v3: session — SessionContext; якщо last_depth=="light" і короткий follow-up
        без action verbs → stability_guard повертає "light" без подальших перевірок.

    Правило fail-closed: при будь-якій помилці повертає "deep".
    """
    try:
        t = text.strip()

        # ── Intent Stability Guard (v3) ────────────────────────────────────────
        # Якщо попередня взаємодія була light і поточне повідомлення ≤6 слів
        # без action verbs / urgent → утримуємо в light без зайвих перевірок.
        if (
            session
            and session.get("last_depth") == "light"
            and not _DEEP_ACTION_RE.search(t)
            and not _DEEP_URGENT_RE.search(t)
        ):
            word_count_guard = len(t.split())
            if word_count_guard <= 6:
                tlog(logger, "stability_guard_triggered", chat_id="n/a",
                     words=word_count_guard, last_depth="light")
                return "light"

        # Explicit greetings / social acks → always light
        if _LIGHT_GREET_RE.match(t):
            tlog(logger, "depth", depth="light", reason="greeting")
            return "light"

        word_count = len(t.split())

        # Follow-up heuristic: ≤6 words + last_topic + no action verbs + no urgent → light
        # Handles: "а на завтра?", "а по полю 12?", "а якщо дощ?" etc.
        if (
            word_count <= 6
            and last_topic is not None
            and not _DEEP_ACTION_RE.search(t)
            and not _DEEP_URGENT_RE.search(t)
        ):
            tlog(logger, "depth", depth="light", reason="short_followup_last_topic",
                 words=word_count, last_topic=last_topic)
            return "light"

        # Very short follow-ups without last_topic → light (≤4 words, no verbs)
        if word_count <= 4 and not _DEEP_ACTION_RE.search(t) and not _DEEP_URGENT_RE.search(t):
            tlog(logger, "depth", depth="light", reason="short_followup", words=word_count)
            return "light"

        # Active doc context → deep
        if has_doc_context:
            tlog(logger, "depth", depth="deep", reason="has_doc_context")
            return "deep"

        # Urgency keywords → always deep
        if _DEEP_URGENT_RE.search(t):
            tlog(logger, "depth", depth="deep", reason="urgent_keyword")
            return "deep"

        # Explicit action verbs → deep
        if _DEEP_ACTION_RE.search(t):
            tlog(logger, "depth", depth="deep", reason="action_verb")
            return "deep"

        # Numeric measurements → deep
        if _DEEP_DATA_RE.search(t):
            tlog(logger, "depth", depth="deep", reason="numeric_data")
            return "deep"

        # Intent-based deep trigger
        detected = _detect_intent(t)
        if detected in _DEEP_INTENTS:
            tlog(logger, "depth", depth="deep", reason="intent", intent=detected)
            return "deep"

        tlog(logger, "depth", depth="light", reason="no_deep_signal")
        return "light"

    except Exception as exc:
        logger.warning("classify_depth error, defaulting to deep: %s", exc)
        return "deep"