61 lines
1.8 KiB
Python
61 lines
1.8 KiB
Python
"""Agent router — selects the right subagent for each step."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
|
|
from ..models.agent import AgentRole
|
|
from ..models.session import TaskStep
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Keyword-based routing hints
|
|
_ROLE_KEYWORDS: dict[AgentRole, list[str]] = {
|
|
AgentRole.COLLECTOR: [
|
|
"gather", "collect", "read", "explore", "search", "find",
|
|
"discover", "analyze", "investigate", "research", "scan",
|
|
"understand", "review existing",
|
|
],
|
|
AgentRole.CODER: [
|
|
"implement", "write", "create", "build", "code", "fix",
|
|
"modify", "refactor", "add", "update", "generate", "develop",
|
|
"edit", "change", "configure", "set up",
|
|
],
|
|
AgentRole.REVIEWER: [
|
|
"review", "validate", "check", "verify", "test", "audit",
|
|
"inspect", "evaluate", "assess", "confirm",
|
|
],
|
|
}
|
|
|
|
|
|
def route_step(step: TaskStep) -> AgentRole:
|
|
"""Determine which agent role should handle this step.
|
|
|
|
Uses the step's declared agent_role if valid, otherwise falls back
|
|
to keyword-based routing.
|
|
"""
|
|
# Respect explicit assignment
|
|
declared = step.agent_role.lower()
|
|
try:
|
|
role = AgentRole(declared)
|
|
return role
|
|
except ValueError:
|
|
pass
|
|
|
|
# Keyword-based fallback
|
|
desc_lower = step.description.lower()
|
|
scores: dict[AgentRole, int] = {role: 0 for role in _ROLE_KEYWORDS}
|
|
|
|
for role, keywords in _ROLE_KEYWORDS.items():
|
|
for kw in keywords:
|
|
if kw in desc_lower:
|
|
scores[role] += 1
|
|
|
|
best = max(scores, key=lambda r: scores[r])
|
|
if scores[best] > 0:
|
|
logger.info("Routed step '%s' to %s (score=%d)", step.description[:60], best, scores[best])
|
|
return best
|
|
|
|
# Default to coder
|
|
return AgentRole.CODER
|