EVA uses an agent-based architecture where each agent specializes in a specific aspect of data analysis. This document provides detailed API reference for all available agents.
All agents inherit from the EVABaseAgent class, which defines the standard interface.
class EVABaseAgent:
"""
Base class for all EVA analysis agents.
Provides the standard interface that all agents must implement,
including execution, validation, and dependency management.
"""
def __init__(self, name: str):
"""
Initialize the agent.
Args:
name: Unique name for the agent
"""
self.name = name
self.logger = EVALogger(agent_name=name)
def execute(self, context: AnalysisContext) -> AgentResult:
"""
Execute the agent's main functionality.
Args:
context: Analysis context containing data and configuration
Returns:
AgentResult with success status, data, and metadata
Raises:
NotImplementedError: Must be implemented by subclasses
"""
raise NotImplementedError("Subclasses must implement execute method")
def validate_input(self, context: AnalysisContext) -> bool:
"""
Validate that the context contains required data for this agent.
Args:
context: Analysis context to validate
Returns:
True if context is valid for this agent, False otherwise
Raises:
NotImplementedError: Must be implemented by subclasses
"""
raise NotImplementedError("Subclasses must implement validate_input method")
def get_dependencies(self) -> List[str]:
"""
Get list of agent names that this agent depends on.
Returns:
List of agent names (empty list if no dependencies)
"""
return []
def get_name(self) -> str:
"""Get the agent's name."""
return self.nameHandles CSV file loading, validation, and initial data processing.
class CSVIngestorAgent(EVABaseAgent):
"""
Agent for ingesting and validating CSV files.
Handles file loading, encoding detection, data type inference,
and initial data validation.
"""
def __init__(self,
encoding_detection: bool = True,
delimiter_detection: bool = True,
max_sample_rows: int = 1000):
"""
Initialize CSV ingestor agent.
Args:
encoding_detection: Enable automatic encoding detection
delimiter_detection: Enable automatic delimiter detection
max_sample_rows: Maximum rows to sample for type inference
"""
super().__init__("CSVIngestorAgent")
self.encoding_detection = encoding_detection
self.delimiter_detection = delimiter_detection
self.max_sample_rows = max_sample_rows
def execute(self, context: AnalysisContext) -> CSVIngestorResult:
"""
Execute CSV ingestion and validation.
Expected context.metadata:
- file_path: Path to CSV file
- file_size_bytes: File size in bytes (optional)
Returns:
CSVIngestorResult containing:
- dataframe: Loaded pandas DataFrame
- column_info: Information about each column
- validation_report: File validation results
- preview_data: Sample of the data
"""
def validate_input(self, context: AnalysisContext) -> bool:
"""
Validate that context contains file path.
Args:
context: Analysis context
Returns:
True if file_path is present in metadata
"""
return (context.metadata is not None and
'file_path' in context.metadata)
def get_dependencies(self) -> List[str]:
"""CSV ingestor has no dependencies."""
return []@dataclass
class CSVIngestorResult(AgentResult):
"""Result from CSV ingestion agent."""
dataframe: pd.DataFrame
column_info: Dict[str, Dict[str, Any]]
validation_report: Dict[str, Any]
preview_data: Dict[str, Any]
# Example column_info structure:
# {
# 'column_name': {
# 'type': 'int64',
# 'non_null_count': 1000,
# 'unique_count': 500,
# 'sample_values': [1, 2, 3, 4, 5]
# }
# }
# Example validation_report structure:
# {
# 'is_valid': True,
# 'row_count': 1000,
# 'column_count': 5,
# 'encoding': 'utf-8',
# 'delimiter': ',',
# 'issues': []
# }Performs comprehensive exploratory data analysis.
class EDAGeneratorAgent(EVABaseAgent):
"""
Agent for generating exploratory data analysis.
Computes statistics, correlations, missing value analysis,
outlier detection, and data quality assessment.
"""
def __init__(self,
include_correlations: bool = True,
outlier_method: str = 'iqr',
outlier_threshold: float = 1.5,
max_categorical_unique: int = 50):
"""
Initialize EDA generator agent.
Args:
include_correlations: Whether to compute correlation matrices
outlier_method: Method for outlier detection ('iqr', 'zscore')
outlier_threshold: Threshold for outlier detection
max_categorical_unique: Max unique values to treat as categorical
"""
super().__init__("EDAGeneratorAgent")
self.include_correlations = include_correlations
self.outlier_method = outlier_method
self.outlier_threshold = outlier_threshold
self.max_categorical_unique = max_categorical_unique
def execute(self, context: AnalysisContext) -> EDAResult:
"""
Execute exploratory data analysis.
Requires:
- context.dataset: pandas DataFrame
Returns:
EDAResult containing:
- statistics: Summary statistics for all columns
- correlations: Correlation matrix for numeric columns
- missing_values: Missing value counts and percentages
- outliers: Outlier indices for numeric columns
- data_quality_score: Overall data quality score (0-1)
"""
def validate_input(self, context: AnalysisContext) -> bool:
"""Validate that context contains a dataset."""
return context.dataset is not None
def get_dependencies(self) -> List[str]:
"""EDA depends on CSV ingestion."""
return ["CSVIngestorAgent"]@dataclass
class EDAResult(AgentResult):
"""Result from EDA generator agent."""
statistics: Dict[str, Dict[str, Any]]
correlations: pd.DataFrame
missing_values: Dict[str, float]
outliers: Dict[str, List[int]]
data_quality_score: float
# Example statistics structure:
# {
# 'numeric_column': {
# 'count': 1000,
# 'mean': 50.5,
# 'std': 15.2,
# 'min': 10.0,
# 'max': 90.0,
# 'skewness': 0.1,
# 'kurtosis': -0.5
# },
# 'categorical_column': {
# 'count': 1000,
# 'unique': 5,
# 'top': 'Category A',
# 'freq': 300,
# 'distribution': {'A': 300, 'B': 250, ...}
# }
# }Creates comprehensive visualizations for the dataset.
class VisualizerAgent(EVABaseAgent):
"""
Agent for generating data visualizations.
Creates appropriate plots based on data types and relationships,
including histograms, scatter plots, correlation heatmaps, etc.
"""
def __init__(self,
max_plots_per_column: int = 3,
plot_style: str = 'seaborn',
color_palette: str = 'viridis',
figure_size: Tuple[int, int] = (10, 6),
dpi: int = 300,
include_interactive: bool = True):
"""
Initialize visualizer agent.
Args:
max_plots_per_column: Maximum plots to generate per column
plot_style: Matplotlib/seaborn style to use
color_palette: Color palette for plots
figure_size: Default figure size (width, height)
dpi: Resolution for saved plots
include_interactive: Whether to generate interactive plots
"""
super().__init__("VisualizerAgent")
self.max_plots_per_column = max_plots_per_column
self.plot_style = plot_style
self.color_palette = color_palette
self.figure_size = figure_size
self.dpi = dpi
self.include_interactive = include_interactive
def execute(self, context: AnalysisContext) -> VisualizationResult:
"""
Execute visualization generation.
Requires:
- context.dataset: pandas DataFrame
Returns:
VisualizationResult containing:
- plots: Dictionary of generated plot information
- interactive_plots: HTML content for interactive plots
- plot_metadata: Metadata about generated plots
"""
def validate_input(self, context: AnalysisContext) -> bool:
"""Validate that context contains a dataset."""
return context.dataset is not None
def get_dependencies(self) -> List[str]:
"""Visualizer depends on CSV ingestion."""
return ["CSVIngestorAgent"]@dataclass
class VisualizationResult(AgentResult):
"""Result from visualizer agent."""
plots: Dict[str, PlotInfo]
interactive_plots: Dict[str, str]
plot_metadata: Dict[str, Any]
@dataclass
class PlotInfo:
"""Information about a generated plot."""
plot_type: str
file_path: str
interactive_html: Optional[str] = None
metadata: Dict[str, Any] = field(default_factory=dict)
# Example plot types:
# - 'histogram'
# - 'boxplot'
# - 'scatter'
# - 'correlation_heatmap'
# - 'bar_chart'
# - 'line_plot'Provides AI-powered insights and data cleaning suggestions.
class InsightSuggesterAgent(EVABaseAgent):
"""
Agent for generating AI-powered insights and suggestions.
Uses AI services to analyze data patterns and provide
recommendations for data cleaning and feature engineering.
"""
def __init__(self,
ai_provider: str = 'openai',
model: str = 'gpt-4',
max_suggestions: int = 10,
include_explanations: bool = True):
"""
Initialize insight suggester agent.
Args:
ai_provider: AI service provider ('openai', 'gemini')
model: Model to use for generating insights
max_suggestions: Maximum number of suggestions to generate
include_explanations: Whether to include explanations
"""
super().__init__("InsightSuggesterAgent")
self.ai_provider = ai_provider
self.model = model
self.max_suggestions = max_suggestions
self.include_explanations = include_explanations
def execute(self, context: AnalysisContext) -> InsightResult:
"""
Execute insight generation.
Requires:
- context.dataset: pandas DataFrame
- context.results['EDAGeneratorAgent']: EDA results (optional)
Returns:
InsightResult containing:
- suggestions: List of actionable suggestions
- explanations: Explanations for each suggestion
- priority_scores: Priority scores for suggestions
"""
def validate_input(self, context: AnalysisContext) -> bool:
"""Validate that context contains a dataset."""
return context.dataset is not None
def get_dependencies(self) -> List[str]:
"""Insight suggester depends on CSV ingestion and optionally EDA."""
return ["CSVIngestorAgent"]@dataclass
class InsightResult(AgentResult):
"""Result from insight suggester agent."""
suggestions: List[Suggestion]
explanations: Dict[str, str]
priority_scores: Dict[str, float]
@dataclass
class Suggestion:
"""A single insight suggestion."""
id: str
type: str # 'cleaning', 'feature_engineering', 'modeling'
description: str
action: str
priority: float
confidence: float
metadata: Dict[str, Any] = field(default_factory=dict)
# Example suggestion types:
# - 'handle_missing_values'
# - 'remove_outliers'
# - 'encode_categorical'
# - 'scale_features'
# - 'create_derived_features'Recommends machine learning models and creates baseline pipelines.
class ModelRecommenderAgent(EVABaseAgent):
"""
Agent for recommending machine learning models.
Analyzes the dataset to determine problem type and recommends
appropriate models with baseline implementations.
"""
def __init__(self,
max_models: int = 5,
cv_folds: int = 5,
test_size: float = 0.2,
random_state: int = 42,
include_ensemble: bool = True):
"""
Initialize model recommender agent.
Args:
max_models: Maximum number of models to recommend
cv_folds: Number of cross-validation folds
test_size: Proportion of data for testing
random_state: Random state for reproducibility
include_ensemble: Whether to include ensemble methods
"""
super().__init__("ModelRecommenderAgent")
self.max_models = max_models
self.cv_folds = cv_folds
self.test_size = test_size
self.random_state = random_state
self.include_ensemble = include_ensemble
def execute(self, context: AnalysisContext) -> ModelResult:
"""
Execute model recommendation.
Requires:
- context.dataset: pandas DataFrame
- Target variable identification (automatic or specified)
Returns:
ModelResult containing:
- problem_type: 'classification' or 'regression'
- recommendations: List of recommended models
- baselines: Baseline model performance
- feature_importance: Feature importance scores
"""
def validate_input(self, context: AnalysisContext) -> bool:
"""Validate that context contains a dataset."""
return context.dataset is not None
def get_dependencies(self) -> List[str]:
"""Model recommender depends on CSV ingestion."""
return ["CSVIngestorAgent"]@dataclass
class ModelResult(AgentResult):
"""Result from model recommender agent."""
problem_type: str
recommendations: List[ModelRecommendation]
baselines: Dict[str, Dict[str, float]]
feature_importance: Dict[str, float]
@dataclass
class ModelRecommendation:
"""A single model recommendation."""
name: str
algorithm: str
score: float
hyperparameters: Dict[str, Any]
pros: List[str]
cons: List[str]
use_cases: List[str]
# Example algorithms:
# - 'RandomForestClassifier'
# - 'LogisticRegression'
# - 'XGBRegressor'
# - 'SVM'Exports analysis results to Jupyter notebooks and Python scripts.
class NotebookExporterAgent(EVABaseAgent):
"""
Agent for exporting analysis results to executable code.
Generates Jupyter notebooks and Python scripts that reproduce
the analysis pipeline with all results and visualizations.
"""
def __init__(self,
export_formats: List[str] = None,
include_markdown: bool = True,
include_comments: bool = True,
template_style: str = 'standard'):
"""
Initialize notebook exporter agent.
Args:
export_formats: List of formats to export ('ipynb', 'py', 'html')
include_markdown: Whether to include markdown explanations
include_comments: Whether to include code comments
template_style: Template style to use
"""
super().__init__("NotebookExporterAgent")
self.export_formats = export_formats or ['ipynb']
self.include_markdown = include_markdown
self.include_comments = include_comments
self.template_style = template_style
def execute(self, context: AnalysisContext) -> ExportResult:
"""
Execute notebook export.
Requires:
- context.results: Results from other agents
- context.dataset: pandas DataFrame
Returns:
ExportResult containing:
- notebook_path: Path to generated notebook
- script_path: Path to generated script
- validation_report: Code validation results
"""
def validate_input(self, context: AnalysisContext) -> bool:
"""Validate that context contains results to export."""
return (context.dataset is not None and
len(context.results) > 0)
def get_dependencies(self) -> List[str]:
"""Exporter depends on all other agents."""
return ["CSVIngestorAgent", "EDAGeneratorAgent", "VisualizerAgent"]@dataclass
class ExportResult(AgentResult):
"""Result from notebook exporter agent."""
notebook_path: Optional[str]
script_path: Optional[str]
html_path: Optional[str]
validation_report: Dict[str, Any]
# Example validation_report:
# {
# 'syntax_valid': True,
# 'imports_valid': True,
# 'executable': True,
# 'issues': []
# }To create a custom agent, inherit from EVABaseAgent and implement the required methods:
class CustomAnalysisAgent(EVABaseAgent):
"""Example custom agent."""
def __init__(self, custom_param: str = "default"):
super().__init__("CustomAnalysisAgent")
self.custom_param = custom_param
def validate_input(self, context: AnalysisContext) -> bool:
"""Validate input requirements."""
return context.dataset is not None
def get_dependencies(self) -> List[str]:
"""Define dependencies."""
return ["CSVIngestorAgent"]
def execute(self, context: AnalysisContext) -> AgentResult:
"""Implement custom analysis logic."""
try:
# Your custom analysis logic here
result_data = self._perform_custom_analysis(context.dataset)
return AgentResult(
success=True,
data=result_data,
agent_name=self.name
)
except Exception as e:
return AgentResult(
success=False,
errors=[str(e)],
agent_name=self.name
)
def _perform_custom_analysis(self, df: pd.DataFrame) -> Dict[str, Any]:
"""Custom analysis implementation."""
# Implement your analysis logic
return {"custom_result": "value"}- Input Validation: Always validate inputs in
validate_input() - Error Handling: Use try-catch blocks and return meaningful errors
- Dependencies: Declare dependencies accurately in
get_dependencies() - Logging: Use
self.loggerfor debugging and monitoring - Result Structure: Return structured data in
AgentResult - Documentation: Include comprehensive docstrings
Agents can be configured through the AnalysisConfig.agent_configs dictionary:
config = AnalysisConfig(
agent_configs={
'CustomAnalysisAgent': {
'custom_param': 'custom_value',
'another_param': 42
}
}
)All agents should handle errors gracefully and return AgentResult with appropriate error information:
def execute(self, context: AnalysisContext) -> AgentResult:
try:
# Agent logic here
return AgentResult(success=True, data=result_data, agent_name=self.name)
except ValidationError as e:
return AgentResult(
success=False,
errors=[f"Validation failed: {str(e)}"],
agent_name=self.name
)
except Exception as e:
return AgentResult(
success=False,
errors=[f"Unexpected error: {str(e)}"],
agent_name=self.name
)Use the provided test utilities to test custom agents:
from tests.fixtures.agent_fixtures import sample_analysis_context
def test_custom_agent():
agent = CustomAnalysisAgent()
context = sample_analysis_context()
# Test validation
assert agent.validate_input(context) == True
# Test execution
result = agent.execute(context)
assert result.success == True
assert 'custom_result' in result.data