65 lines
3.1 KiB
Python
65 lines
3.1 KiB
Python
from collections.abc import Callable
|
|
|
|
from app.core.contracts import AgentDefinition, RequestContext
|
|
from app.core.errors import AgentTypeNotFoundError, ValidationAgentError
|
|
from app.service.agent.authorizer import AgentAuthorizer
|
|
from app.service.agent.base import BaseAgent
|
|
from app.service.agent.governance import AgentGovernance, PlatformGovernance
|
|
from app.service.intent_classifier import IntentClassifier, IntentEndpointResolver
|
|
from app.service.model_gateway import ModelGenerationService
|
|
from app.service.tool_executor import ToolExecutor
|
|
|
|
AgentBuilder = Callable[[RequestContext], BaseAgent]
|
|
|
|
|
|
class AgentFactory:
|
|
def __init__(
|
|
self, governance: AgentGovernance | None = None,
|
|
model_service: ModelGenerationService | None = None,
|
|
tool_executor: ToolExecutor | None = None,
|
|
intent_classifier: IntentClassifier | None = None,
|
|
intent_endpoint_resolver: IntentEndpointResolver | None = None,
|
|
) -> None:
|
|
self._builders: dict[str, AgentBuilder] = {}
|
|
self._definitions: dict[str, AgentDefinition] = {}
|
|
self._governance = governance or PlatformGovernance()
|
|
self._model_service = model_service
|
|
self._tool_executor = tool_executor
|
|
self._intent_classifier = intent_classifier
|
|
self._intent_endpoint_resolver = intent_endpoint_resolver
|
|
|
|
def register(self, definition: AgentDefinition, builder: AgentBuilder) -> None:
|
|
if definition.agent_type in self._builders:
|
|
raise ValidationAgentError("agent_type already registered")
|
|
self._builders[definition.agent_type] = builder
|
|
self._definitions[definition.agent_type] = definition
|
|
|
|
def authorize(self, agent_type: str, context: RequestContext) -> None:
|
|
definition = self.definition(agent_type)
|
|
AgentAuthorizer.ensure_allowed(definition, context)
|
|
|
|
def definition(self, agent_type: str) -> AgentDefinition:
|
|
definition = self._definitions.get(agent_type)
|
|
if definition is None:
|
|
raise AgentTypeNotFoundError("agent_type is not registered")
|
|
return definition
|
|
|
|
def create(self, agent_type: str, context: RequestContext) -> BaseAgent:
|
|
self.authorize(agent_type, context)
|
|
builder = self._builders.get(agent_type)
|
|
if builder is None:
|
|
raise AgentTypeNotFoundError("agent_type is not registered")
|
|
agent = builder(context)
|
|
if not isinstance(agent, BaseAgent):
|
|
raise ValidationAgentError("registered builder must return BaseAgent")
|
|
if agent.definition != self._definitions[agent_type]:
|
|
raise ValidationAgentError("builder returned a different AgentDefinition")
|
|
agent.bind_governance(self._governance)
|
|
if self._model_service is not None:
|
|
agent.bind_model_service(self._model_service)
|
|
if self._tool_executor is not None:
|
|
agent.bind_tool_executor(self._tool_executor)
|
|
if self._intent_classifier is not None and self._intent_endpoint_resolver is not None:
|
|
agent.bind_intent_classifier(self._intent_classifier, self._intent_endpoint_resolver)
|
|
return agent
|