Agent Routing Models
Routing tasks to the right agent based on task type, complexity, and agent capability.
Routing Strategies
Rule-Based Routing
class RuleRouter:
def route(self, task: str) -> str:
task_lower = task.lower()
if any(kw in task_lower for kw in ["docker", "container", "image", "volume"]):
return "docker_agent"
elif any(kw in task_lower for kw in ["c++", "rust", "compile", "build"]):
return "build_agent"
elif any(kw in task_lower for kw in ["python", "script", "data"]):
return "python_agent"
elif any(kw in task_lower for kw in ["wsl", "linux", "ubuntu"]):
return "wsl_agent"
else:
return "general_agent"
LLM-Based Router
class LLMRouter:
def __init__(self, router_llm, agents: dict):
self.llm = router_llm
self.agents = agents # agent_name → {description, capabilities}
def route(self, task: str) -> str:
prompt = f"""Available agents:
{self._format_agents()}
User task: {task}
Which agent should handle this? Respond with just the agent name."""
return self.llm.invoke(prompt).strip()
def route_with_confidence(self, task: str) -> tuple[str, float]:
prompt = f"""Available agents:
{self._format_agents()}
User task: {task}
Respond in JSON: {{"agent": "agent_name", "confidence": 0.95, "reason": "brief reason"}}"""
import json
result = json.loads(self.llm.invoke(prompt))
return result["agent"], result["confidence"]
def _format_agents(self) -> str:
lines = []
for name, info in self.agents.items():
lines.append(f"- {name}: {info['description']} (capabilities: {', '.join(info['capabilities'])})")
return "\n".join(lines)
Embedding-Based Router
from sentence_transformers import SentenceTransformer
from sklearn.metrics.pairwise import cosine_similarity
import numpy as np
class EmbeddingRouter:
def __init__(self, agents: dict):
self.encoder = SentenceTransformer("all-MiniLM-L6-v2")
self.agents = agents
# Pre-encode agent descriptions
self.agent_embeddings = {
name: self.encoder.encode(info["description"])
for name, info in agents.items()
}
def route(self, task: str) -> str:
task_embedding = self.encoder.encode(task)
scores = {
name: cosine_similarity([task_embedding], [emb])[0][0]
for name, emb in self.agent_embeddings.items()
}
best = max(scores, key=scores.get)
return best
Hybrid Router
class HybridRouter:
def __init__(self, rule_router: RuleRouter, llm_router: LLMRouter,
embedding_router: EmbeddingRouter):
self.routers = [rule_router, llm_router, embedding_router]
def route(self, task: str) -> str:
votes = {}
for router in self.routers:
result = router.route(task)
votes[result] = votes.get(result, 0) + 1
return max(votes, key=votes.get)
Task Classification Router
class TaskClassifier:
def __init__(self, llm):
self.llm = llm
def classify(self, task: str) -> dict:
prompt = f"""Classify this task:
Task: {task}
Categories:
- type: question|instruction|debugging|analysis|creation
- domain: docker|wsl|windows|cpp|rust|python|general
- complexity: simple|medium|complex
- requires_admin: true|false
- estimated_steps: <number>
Respond in JSON format."""
import json
return json.loads(self.llm.invoke(prompt))
Priority Routing
class PriorityRouter(RuleRouter):
def __init__(self, agents: dict, priority_map: dict = None):
super().__init__()
self.agents = agents
self.priority_map = priority_map or {
"error": 1, # highest priority
"blocking": 2,
"feature": 3,
"question": 4,
"research": 5,
}
def route(self, task: str, urgency: str = "normal") -> tuple[str, int]:
agent = super().route(task)
priority = self.priority_map.get(urgency, 5)
return agent, priority
Pitfalls
- Rule-based: misses edge cases, needs constant updating
- LLM-based: adds latency and cost per routing decision
- Embedding-based: requires pre-encoded descriptions, domain-dependent
- No router handles 100% — always implement fallback agent
- Cold start: new agents need routing rules until sufficient routing data