95 lines
2.6 KiB
Python
95 lines
2.6 KiB
Python
"""
|
|
Base class for all AI functions
|
|
"""
|
|
from abc import ABC, abstractmethod
|
|
from typing import Dict, List, Any, Optional
|
|
|
|
|
|
class BaseAIFunction(ABC):
|
|
"""
|
|
Base class for all AI functions.
|
|
Each function only implements its specific logic.
|
|
"""
|
|
|
|
@abstractmethod
|
|
def get_name(self) -> str:
|
|
"""Return function name (e.g., 'auto_cluster')"""
|
|
pass
|
|
|
|
def get_metadata(self) -> Dict:
|
|
"""Return function metadata (display name, description, phases)"""
|
|
return {
|
|
'display_name': self.get_name().replace('_', ' ').title(),
|
|
'description': f'{self.get_name()} AI function',
|
|
'phases': {
|
|
'INIT': 'Initializing...',
|
|
'PREP': 'Preparing data...',
|
|
'AI_CALL': 'Processing with AI...',
|
|
'PARSE': 'Parsing response...',
|
|
'SAVE': 'Saving results...',
|
|
'DONE': 'Complete!'
|
|
}
|
|
}
|
|
|
|
def validate(self, payload: dict, account=None) -> Dict[str, Any]:
|
|
"""
|
|
Validate input payload.
|
|
Default: checks for 'ids' array.
|
|
Override for custom validation.
|
|
"""
|
|
ids = payload.get('ids', [])
|
|
if not ids:
|
|
return {'valid': False, 'error': 'No IDs provided'}
|
|
|
|
# Removed max_items limit check - no limits enforced
|
|
|
|
return {'valid': True}
|
|
|
|
def get_max_items(self) -> Optional[int]:
|
|
"""Override to set max items limit"""
|
|
return None
|
|
|
|
@abstractmethod
|
|
def prepare(self, payload: dict, account=None) -> Any:
|
|
"""
|
|
Load and prepare data for AI processing.
|
|
Returns: prepared data structure
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def build_prompt(self, data: Any, account=None) -> str:
|
|
"""
|
|
Build AI prompt from prepared data.
|
|
Returns: prompt string
|
|
"""
|
|
pass
|
|
|
|
def get_model(self, account=None) -> Optional[str]:
|
|
"""Override to specify model (defaults to account's default model)"""
|
|
return None # Uses account's default from AIProcessor
|
|
|
|
@abstractmethod
|
|
def parse_response(self, response: str, step_tracker=None) -> Any:
|
|
"""
|
|
Parse AI response into structured data.
|
|
Returns: parsed data structure
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def save_output(
|
|
self,
|
|
parsed: Any,
|
|
original_data: Any,
|
|
account=None,
|
|
progress_tracker=None,
|
|
step_tracker=None
|
|
) -> Dict:
|
|
"""
|
|
Save parsed results to database.
|
|
Returns: dict with 'count', 'items_created', etc.
|
|
"""
|
|
pass
|
|
|