Spaces:
Sleeping
Sleeping
File size: 2,008 Bytes
a1e4405 | 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 | from __future__ import annotations
from dataclasses import dataclass
from .contracts import CapabilityCapsule, SkillDescriptor, SkillValidationState
from .schemas import NeedTask
@dataclass(frozen=True)
class SkillQuery:
language: str
audience: str
tasks: tuple[NeedTask, ...]
has_evidence: bool
text_length: int
class SkillRegistry:
def __init__(self, skills: list[SkillDescriptor]):
self._skills = {skill.skill_id: skill for skill in skills}
def active(self) -> list[SkillDescriptor]:
return [s for s in self._skills.values() if s.state == SkillValidationState.ACTIVE]
def select(self, query: SkillQuery, *, limit: int = 8) -> list[CapabilityCapsule]:
language = "ja" if query.language.lower().startswith("ja") else "en"
shapes = {task.response_shape for task in query.tasks}
intents = {task.intent for task in query.tasks}
scored: list[tuple[int, SkillDescriptor]] = []
for skill in self.active():
if language not in skill.languages:
continue
score = 50 if skill.baseline else 0
if shapes.intersection(skill.task_shapes): score += 30
if "technical" in skill.capabilities and shapes.intersection({"procedure", "troubleshooting", "comparison"}): score += 20
if "evidence" in skill.capabilities and query.has_evidence: score += 20
if "audience" in skill.capabilities and query.audience != "general": score += 10
if "planning" in skill.capabilities and query.text_length >= 800: score += 25
if intents.intersection({"troubleshooting", "comparison", "procedure"}) and "review" in skill.capabilities: score += 10
if score > 0:
scored.append((score, skill))
scored.sort(key=lambda pair: (-pair[0], pair[1].skill_id))
return [CapabilityCapsule(skill_id=s.skill_id, text=s.capsule, score=score, capabilities=list(s.capabilities)) for score, s in scored[:limit]]
|