File size: 1,490 Bytes
a3bcd92 5f83674 a3bcd92 5f83674 a3bcd92 5f83674 a3bcd92 5f83674 a3bcd92 5f83674 |
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 |
"""Minimal orchestration controller used in legacy tests."""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any, Dict
@dataclass
class _RegistryState:
agents: Dict[str, Any] = field(default_factory=dict)
adaptors: Dict[str, Any] = field(default_factory=dict)
evaluators: Dict[str, Any] = field(default_factory=dict)
def list_all(self) -> Dict[str, list[str]]:
return {
"agents": list(self.agents.keys()),
"adaptors": list(self.adaptors.keys()),
"evaluators": list(self.evaluators.keys()),
}
class Controller:
"""Coordinates agents, adaptors, and evaluators for simple task processing."""
def __init__(self) -> None:
self.registry = _RegistryState()
def register_agent(self, name: str, agent: Any) -> None:
self.registry.agents[name] = agent
def register_adaptor(self, name: str, adaptor: Any) -> None:
self.registry.adaptors[name] = adaptor
def register_evaluator(self, name: str, evaluator: Any) -> None:
self.registry.evaluators[name] = evaluator
def process_task(self, prompt: str, *, agent_name: str) -> str:
agent = self.registry.agents[agent_name]
result = agent.process_input(prompt)
for evaluator in self.registry.evaluators.values():
if hasattr(evaluator, "evaluate"):
evaluator.evaluate(result)
return result
__all__ = ["Controller"]
|