77b70943da
- Add src/infrastructure/router/classifier.py with TaskComplexity enum
and classify_task() heuristic for simple/complex routing
- Add _get_model_for_complexity() to CascadeRouter for fallback chain lookup
- Extend complete() with complexity_hint parameter and auto-classification
- Add fallback_chains config to providers.yaml for qwen3:8b/14b routing
- Fix: use explicit next() comprehension instead of get_model_with_capability
to avoid silent fallback to provider default (0b28497)
- 361 unit tests passing
Fixes #1065
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
167 lines
4.2 KiB
Python
167 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
|