Spaces:
Build error
Build error
File size: 8,087 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 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 |
"""
Custom exceptions for the AI Agent system.
This module defines custom exception classes that are used throughout
the application to provide meaningful error handling and debugging.
"""
from typing import Optional, Dict, Any
class AIAgentException(Exception):
"""Base exception for all AI Agent system exceptions."""
def __init__(self, message: str, error_code: Optional[str] = None,
details: Optional[Dict[str, Any]] = None):
super().__init__(message)
self.message = message
self.error_code = error_code
self.details = details or {}
def __str__(self) -> str:
if self.error_code:
return f"[{self.error_code}] {self.message}"
return self.message
class DomainException(AIAgentException):
"""Exception raised for domain/business logic errors."""
def __init__(self, message: str, error_code: Optional[str] = None,
details: Optional[Dict[str, Any]] = None):
super().__init__(message, error_code or "DOMAIN_ERROR", details)
class ValidationException(AIAgentException):
"""Exception raised for validation errors."""
def __init__(self, message: str, field: Optional[str] = None,
value: Optional[Any] = None, details: Optional[Dict[str, Any]] = None):
super().__init__(message, "VALIDATION_ERROR", details)
self.field = field
self.value = value
class InfrastructureException(AIAgentException):
"""Exception raised for infrastructure/technical errors."""
def __init__(self, message: str, component: Optional[str] = None,
details: Optional[Dict[str, Any]] = None):
super().__init__(message, "INFRASTRUCTURE_ERROR", details)
self.component = component
class ApplicationException(AIAgentException):
"""Exception raised for application layer errors."""
def __init__(self, message: str, use_case: Optional[str] = None,
details: Optional[Dict[str, Any]] = None):
super().__init__(message, "APPLICATION_ERROR", details)
self.use_case = use_case
class ConfigurationException(AIAgentException):
"""Exception raised for configuration errors."""
def __init__(self, message: str, config_key: Optional[str] = None,
details: Optional[Dict[str, Any]] = None):
super().__init__(message, "CONFIGURATION_ERROR", details)
self.config_key = config_key
class AgentException(AIAgentException):
"""Exception raised for agent-related errors."""
def __init__(self, message: str, agent_id: Optional[str] = None,
agent_type: Optional[str] = None, details: Optional[Dict[str, Any]] = None):
super().__init__(message, "AGENT_ERROR", details)
self.agent_id = agent_id
self.agent_type = agent_type
class TaskException(AIAgentException):
"""Exception raised for task-related errors."""
def __init__(self, message: str, task_id: Optional[str] = None,
task_type: Optional[str] = None, details: Optional[Dict[str, Any]] = None):
super().__init__(message, "TASK_ERROR", details)
self.task_id = task_id
self.task_type = task_type
class ToolException(AIAgentException):
"""Exception raised for tool-related errors."""
def __init__(self, message: str, tool_name: Optional[str] = None,
tool_type: Optional[str] = None, details: Optional[Dict[str, Any]] = None):
super().__init__(message, "TOOL_ERROR", details)
self.tool_name = tool_name
self.tool_type = tool_type
class CommunicationException(AIAgentException):
"""Exception raised for communication/API errors."""
def __init__(self, message: str, endpoint: Optional[str] = None,
status_code: Optional[int] = None, details: Optional[Dict[str, Any]] = None):
super().__init__(message, "COMMUNICATION_ERROR", details)
self.endpoint = endpoint
self.status_code = status_code
class ResourceException(AIAgentException):
"""Exception raised for resource-related errors."""
def __init__(self, message: str, resource_type: Optional[str] = None,
resource_id: Optional[str] = None, details: Optional[Dict[str, Any]] = None):
super().__init__(message, "RESOURCE_ERROR", details)
self.resource_type = resource_type
self.resource_id = resource_id
class TimeoutException(AIAgentException):
"""Exception raised for timeout errors."""
def __init__(self, message: str, timeout_duration: Optional[float] = None,
operation: Optional[str] = None, details: Optional[Dict[str, Any]] = None):
super().__init__(message, "TIMEOUT_ERROR", details)
self.timeout_duration = timeout_duration
self.operation = operation
class RateLimitException(AIAgentException):
"""Exception raised for rate limiting errors."""
def __init__(self, message: str, rate_limit: Optional[int] = None,
retry_after: Optional[float] = None, details: Optional[Dict[str, Any]] = None):
super().__init__(message, "RATE_LIMIT_ERROR", details)
self.rate_limit = rate_limit
self.retry_after = retry_after
class AuthenticationException(AIAgentException):
"""Exception raised for authentication errors."""
def __init__(self, message: str, auth_method: Optional[str] = None,
details: Optional[Dict[str, Any]] = None):
super().__init__(message, "AUTHENTICATION_ERROR", details)
self.auth_method = auth_method
class AuthorizationException(AIAgentException):
"""Exception raised for authorization errors."""
def __init__(self, message: str, required_permissions: Optional[list] = None,
details: Optional[Dict[str, Any]] = None):
super().__init__(message, "AUTHORIZATION_ERROR", details)
self.required_permissions = required_permissions
class DataException(AIAgentException):
"""Exception raised for data-related errors."""
def __init__(self, message: str, data_type: Optional[str] = None,
data_source: Optional[str] = None, details: Optional[Dict[str, Any]] = None):
super().__init__(message, "DATA_ERROR", details)
self.data_type = data_type
self.data_source = data_source
class StateException(AIAgentException):
"""Exception raised for state management errors."""
def __init__(self, message: str, current_state: Optional[str] = None,
expected_state: Optional[str] = None, details: Optional[Dict[str, Any]] = None):
super().__init__(message, "STATE_ERROR", details)
self.current_state = current_state
self.expected_state = expected_state
# Convenience functions for common error patterns
def raise_validation_error(message: str, field: Optional[str] = None,
value: Optional[Any] = None) -> None:
"""Raise a validation exception with the given details."""
raise ValidationException(message, field, value)
def raise_infrastructure_error(message: str, component: Optional[str] = None) -> None:
"""Raise an infrastructure exception with the given details."""
raise InfrastructureException(message, component)
def raise_agent_error(message: str, agent_id: Optional[str] = None,
agent_type: Optional[str] = None) -> None:
"""Raise an agent exception with the given details."""
raise AgentException(message, agent_id, agent_type)
def raise_task_error(message: str, task_id: Optional[str] = None,
task_type: Optional[str] = None) -> None:
"""Raise a task exception with the given details."""
raise TaskException(message, task_id, task_type)
def raise_tool_error(message: str, tool_name: Optional[str] = None,
tool_type: Optional[str] = None) -> None:
"""Raise a tool exception with the given details."""
raise ToolException(message, tool_name, tool_type) |