"""Task complexity classifier for Qwen3 dual-model routing. Classifies incoming tasks as SIMPLE (route to Qwen3-8B for low-latency) or COMPLEX (route to Qwen3-14B for quality-sensitive work). Classification is fully heuristic — no LLM inference required. """ import re from enum import Enum class TaskComplexity(Enum): """Task complexity tier for model routing.""" SIMPLE = "simple" # Qwen3-8B Q6_K: routine, latency-sensitive COMPLEX = "complex" # Qwen3-14B Q5_K_M: quality-sensitive, multi-step # Keywords strongly associated with complex tasks _COMPLEX_KEYWORDS: frozenset[str] = frozenset( [ "plan", "review", "analyze", "analyse", "triage", "refactor", "design", "architecture", "implement", "compare", "debug", "explain", "prioritize", "prioritise", "strategy", "optimize", "optimise", "evaluate", "assess", "brainstorm", "outline", "summarize", "summarise", "generate code", "write a", "write the", "code review", "pull request", "multi-step", "multi step", "step by step", "backlog prioriti", "issue triage", "root cause", "how does", "why does", "what are the", ] ) # Keywords strongly associated with simple/routine tasks _SIMPLE_KEYWORDS: frozenset[str] = frozenset( [ "status", "list ", "show ", "what is", "how many", "ping", "run ", "execute ", "ls ", "cat ", "ps ", "fetch ", "count ", "tail ", "head ", "grep ", "find file", "read file", "get ", "query ", "check ", "yes", "no", "ok", "done", "thanks", ] ) # Content longer than this is treated as complex regardless of keywords _COMPLEX_CHAR_THRESHOLD = 500 # Short content defaults to simple _SIMPLE_CHAR_THRESHOLD = 150 # More than this many messages suggests an ongoing complex conversation _COMPLEX_CONVERSATION_DEPTH = 6 def classify_task(messages: list[dict]) -> TaskComplexity: """Classify task complexity from a list of messages. Uses heuristic rules — no LLM call required. Errs toward COMPLEX when uncertain so that quality is preserved. Args: messages: List of message dicts with ``role`` and ``content`` keys. Returns: TaskComplexity.SIMPLE or TaskComplexity.COMPLEX """ if not messages: return TaskComplexity.SIMPLE # Concatenate all user-turn content for analysis user_content = ( " ".join( msg.get("content", "") for msg in messages if msg.get("role") in ("user", "human") and isinstance(msg.get("content"), str) ) .lower() .strip() ) if not user_content: return TaskComplexity.SIMPLE # Complexity signals override everything ----------------------------------- # Explicit complex keywords for kw in _COMPLEX_KEYWORDS: if kw in user_content: return TaskComplexity.COMPLEX # Numbered / multi-step instruction list: "1. do this 2. do that" if re.search(r"\b\d+\.\s+\w", user_content): return TaskComplexity.COMPLEX # Code blocks embedded in messages if "```" in user_content: return TaskComplexity.COMPLEX # Long content → complex reasoning likely required if len(user_content) > _COMPLEX_CHAR_THRESHOLD: return TaskComplexity.COMPLEX # Deep conversation → complex ongoing task if len(messages) > _COMPLEX_CONVERSATION_DEPTH: return TaskComplexity.COMPLEX # Simplicity signals ------------------------------------------------------- # Explicit simple keywords for kw in _SIMPLE_KEYWORDS: if kw in user_content: return TaskComplexity.SIMPLE # Short single-sentence messages default to simple if len(user_content) <= _SIMPLE_CHAR_THRESHOLD: return TaskComplexity.SIMPLE # When uncertain, prefer quality (complex model) return TaskComplexity.COMPLEX