File size: 3,766 Bytes
cb505ff | 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 100 101 102 103 104 | from __future__ import annotations
from typing import Any, Dict, Optional
from src.skills.models import SkillDefinition, SkillResult
from src.skills.registry import SkillRegistry
from src.runtime.agent_runtime import AgentRuntime
from src.runtime.execution_context import ExecutionContext
class ManagerRuntime:
"""Persistent Manager Runtime for orchestration, governance, and routing skill lifecycle."""
ROLE_TO_WORKER_SKILL = {
"patient": "patient-support",
"caregiver": "caregiver-support",
"researcher": "diabetes-research",
"dietary": "diabetes-dietary-coach",
}
INTENT_TO_WORKER_SKILL = {
"diagnosis": "clinical-diagnosis-support",
"treatment": "clinical-treatment-support",
"monitoring": "clinical-monitoring-support",
"general": "clinical-general-support",
}
def __init__(self, registry: SkillRegistry):
self.registry = registry
self.agent_runtime = AgentRuntime(name="ManagerRuntime", registry=registry)
@property
def active_skill(self) -> Optional[SkillDefinition]:
return self.agent_runtime.active_skill
def load_skill(self, skill_id: str, version: Optional[str] = None) -> SkillDefinition:
config = self.agent_runtime.load_skill(skill_id, version=version)
return config.definition
def execute(self, context: ExecutionContext, input_override: Optional[Dict[str, Any]] = None) -> SkillResult:
return self.agent_runtime.execute(context=context, input_override=input_override)
def flush_skill(self) -> None:
self.agent_runtime.flush()
def plan_route(self, context: ExecutionContext) -> Dict[str, Any]:
"""Return the next approved runtime step for the request context."""
role = (context.user_role or "").strip().lower() or None
intent = (context.intent or "").strip().lower() or None
if role is None:
return {
"route_type": "manager_skill",
"manager_skill": "role-routing",
"worker_skill": None,
"role": None,
"intent": None,
}
if role == "clinician" and intent is None:
return {
"route_type": "manager_skill",
"manager_skill": "clinical-intent-routing",
"worker_skill": None,
"role": role,
"intent": None,
}
if role == "clinician":
worker_skill = self.INTENT_TO_WORKER_SKILL.get(intent, self.INTENT_TO_WORKER_SKILL["general"])
else:
worker_skill = self.ROLE_TO_WORKER_SKILL.get(role)
if worker_skill is None:
return {
"route_type": "unroutable",
"manager_skill": None,
"worker_skill": None,
"role": role,
"intent": intent,
}
return {
"route_type": "worker_skill",
"manager_skill": None,
"worker_skill": worker_skill,
"role": role,
"intent": intent or "general",
}
def route_request(self, context: ExecutionContext) -> SkillResult:
"""Execute the required Manager skill or return an approved worker plan."""
plan = self.plan_route(context)
if plan["route_type"] == "manager_skill":
self.load_skill(plan["manager_skill"])
result = self.execute(context)
self.flush_skill()
result.output["route_plan"] = plan
return result
return SkillResult(
skill_id="manager-orchestration",
skill_version="1.0.0",
output={"status": "routing_complete", "route_plan": plan},
)
|