File size: 3,392 Bytes
82f262a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 | """
RoutingMatrix - JSON skill routing with precedence and typed keys.
Deterministic skill index (from matrix), hot-swappable LoRA adapter paths.
Back-compatible with .skill files via load_legacy_skill().
"""
import hashlib
import json
import re
from dataclasses import dataclass
from pathlib import Path
from typing import Dict, List, Optional
@dataclass
class RouteDecision:
skill: Optional[str] = None
token: Optional[str] = None
index: Optional[int] = None
adapter: Optional[str] = None
score: float = 0.0
def is_skill(self) -> bool:
return self.skill is not None
def deterministic_index(name: str, num_slots: int = 64) -> int:
"""Stable skill->index mapping (replaces runtime hash(), which varies per process)."""
return int(hashlib.sha256(name.encode("utf-8")).hexdigest(), 16) % num_slots
class RoutingMatrix:
def __init__(self, matrix_path: Optional[str] = None):
self.skills: List[Dict] = []
self.default: Dict = {}
if matrix_path:
self.load(matrix_path)
def load(self, matrix_path: str):
data = json.loads(Path(matrix_path).read_text(encoding="utf-8"))
self.skills = data.get("skills", [])
self.default = data.get("default", {})
return len(self.skills)
def load_legacy_skill(self, skill_path: str) -> Dict:
"""Adopt a .skill file into the matrix (keeps old runtime working)."""
data = json.loads(Path(skill_path).read_text(encoding="utf-8"))
entry = {
"name": data["name"],
"token": data["token"],
"index": deterministic_index(data["name"]),
"priority": 10,
"patterns": [
{"type": "keyword", "value": p} for p in data.get("trigger_patterns", [])
],
}
self.skills.append(entry)
return entry
def _score(self, text: str, entry: Dict) -> float:
low = text.lower()
score = 0.0
for pat in entry.get("patterns", []):
value = pat.get("value", "")
ptype = pat.get("type", "regex")
if ptype == "regex":
if re.search(value, low):
score += 1.0
elif ptype == "keyword":
if value.lower() in low:
score += 0.8
return score
def route(self, text: str) -> RouteDecision:
"""Highest-scoring skill wins; ties broken by priority."""
best = RouteDecision()
for entry in self.skills:
score = self._score(text, entry)
if score == 0:
continue
entry_prio = entry.get("priority", 0)
best_prio = 0 if best.score == 0 else self._priority_of(best.skill)
if score > best.score or (score == best.score and entry_prio > best_prio):
best = RouteDecision(
skill=entry["name"],
token=entry.get("token"),
index=entry.get("index", deterministic_index(entry["name"])),
adapter=entry.get("adapter"),
score=score,
)
return best
def _priority_of(self, name: Optional[str]) -> int:
if not name:
return 0
for e in self.skills:
if e.get("name") == name:
return e.get("priority", 0)
return 0 |