File size: 2,889 Bytes
9858829 | 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 | """
Base Agent class for all agents in the workflow
This provides a common interface and structure for extensibility
"""
from abc import ABC, abstractmethod
from typing import Dict, Any, Optional
import json
import logging
logger = logging.getLogger(__name__)
class BaseAgent(ABC):
"""
Abstract base class for all agents in the agentic workflow.
Provides common functionality and enforces consistent interface.
"""
def __init__(self, name: str, config: Dict[str, Any]):
"""
Initialize the base agent.
Args:
name: Name of the agent
config: Configuration dictionary for the agent
"""
self.name = name
self.config = config
self.model = config.get("model", "gpt-4o-mini")
self.temperature = config.get("temperature", 0.7)
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.
This method must be implemented by all concrete agent classes.
Args:
input_data: Dictionary containing input data for processing
Returns:
Dictionary containing processing results
"""
pass
@abstractmethod
def validate_input(self, input_data: Dict[str, Any]) -> bool:
"""
Validate input data before processing.
Args:
input_data: Dictionary containing input data
Returns:
True if input is valid, False otherwise
"""
pass
def get_name(self) -> str:
"""Get the agent name."""
return self.name
def get_config(self) -> Dict[str, Any]:
"""Get the agent configuration."""
return self.config
def log_processing(self, message: str, level: str = "info"):
"""
Log processing information.
Args:
message: Log message
level: Log level (info, warning, error, debug)
"""
log_method = getattr(logger, level, logger.info)
log_method(f"[{self.name}] {message}")
def handle_error(self, error: Exception, context: str = "") -> Dict[str, Any]:
"""
Handle errors consistently across all agents.
Args:
error: The exception that occurred
context: Additional context about the error
Returns:
Error dictionary with details
"""
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
} |