| """ |
| Base Agent class for all agents in the HelpScout processing workflow. |
| Provides a common interface and consistent error handling. |
| """ |
|
|
| from abc import ABC, abstractmethod |
| from typing import Dict, Any |
| import logging |
|
|
| logger = logging.getLogger(__name__) |
|
|
|
|
| class BaseAgent(ABC): |
| """ |
| Abstract base class for all agents in the agentic workflow. |
| Enforces a consistent interface and provides shared utilities. |
| """ |
|
|
| def __init__(self, name: str, config: Dict[str, Any]): |
| self.name = name |
| self.config = config |
| self.model = config.get("model", "gpt-5-nano") |
| self.temperature = config.get("temperature", 0.2) |
| self.max_retries = config.get("max_retries", 3) |
| logger.info(f"Initialized {self.name} with model {self.model}") |
|
|
| @abstractmethod |
| def process(self, input_data: Dict[str, Any]) -> Dict[str, Any]: |
| """ |
| Process input data and return results. |
| Must be implemented by all concrete agent classes. |
| """ |
| pass |
|
|
| @abstractmethod |
| def validate_input(self, input_data: Dict[str, Any]) -> bool: |
| """ |
| Validate input data before processing. |
| Returns True if input is valid, False otherwise. |
| """ |
| pass |
|
|
| def log_processing(self, message: str, level: str = "info"): |
| log_method = getattr(logger, level, logger.info) |
| log_method(f"[{self.name}] {message}") |
|
|
| def handle_error(self, error: Exception, context: str = "") -> Dict[str, Any]: |
| error_msg = f"Error in {self.name}" |
| if context: |
| error_msg += f" ({context})" |
| error_msg += f": {str(error)}" |
| logger.error(error_msg) |
| return { |
| "success": False, |
| "error": str(error), |
| "agent": self.name, |
| "context": context, |
| } |
|
|
