Spaces:
Build error
Build error
File size: 3,272 Bytes
8a682b5 |
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 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 |
"""
Agent executor interface for executing agent tasks.
"""
from abc import ABC, abstractmethod
from typing import Dict, Any, Optional
from uuid import UUID
from src.core.entities.agent import Agent
from src.core.entities.message import Message
class AgentExecutor(ABC):
"""
Abstract interface for agent execution operations.
This interface defines the contract that all agent executor
implementations must follow, ensuring consistency across
different execution strategies.
"""
@abstractmethod
async def execute_agent(self, agent: Agent, message: Message) -> Dict[str, Any]:
"""
Execute an agent with a given message.
Args:
agent: The agent to execute
message: The message to process
Returns:
Dictionary containing the execution result
Raises:
ApplicationException: If execution fails
"""
pass
@abstractmethod
async def execute_agent_by_id(self, agent_id: UUID, message: Message) -> Dict[str, Any]:
"""
Execute an agent by ID with a given message.
Args:
agent_id: The agent's unique identifier
message: The message to process
Returns:
Dictionary containing the execution result
Raises:
ApplicationException: If execution fails
"""
pass
@abstractmethod
async def execute_agent_by_type(self, agent_type: str, message: Message) -> Dict[str, Any]:
"""
Execute an agent by type with a given message.
Args:
agent_type: The type of agent to execute
message: The message to process
Returns:
Dictionary containing the execution result
Raises:
ApplicationException: If execution fails
"""
pass
@abstractmethod
async def get_execution_status(self, execution_id: str) -> Dict[str, Any]:
"""
Get the status of an execution.
Args:
execution_id: The execution identifier
Returns:
Dictionary containing the execution status
Raises:
ApplicationException: If status retrieval fails
"""
pass
@abstractmethod
async def cancel_execution(self, execution_id: str) -> bool:
"""
Cancel an ongoing execution.
Args:
execution_id: The execution identifier
Returns:
True if cancellation was successful, False otherwise
Raises:
ApplicationException: If cancellation fails
"""
pass
@abstractmethod
async def get_execution_metrics(self, agent_id: Optional[UUID] = None) -> Dict[str, Any]:
"""
Get execution metrics for agents.
Args:
agent_id: Optional agent ID to filter metrics
Returns:
Dictionary containing execution metrics
Raises:
ApplicationException: If metrics retrieval fails
"""
pass |