bb9311704a
- cascade.py: add TYPE_CHECKING guard for TaskComplexity to fix F821 - _base.py: add noqa: F401 to re-exported agno tool imports (FileTools, PythonTools, ShellTools, Toolkit) — these are used by other modules - test_three_strike.py: rename unused loop var `i` to `_` (B007) - Run tox -e format to auto-fix I001 import-sort and F401 errors across 23 files (isort ordering and unused pytest import) Fixes #1247
170 lines
4.2 KiB
Python
170 lines
4.2 KiB
Python
"""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
|