Spaces:
Sleeping
Sleeping
| """ | |
| Cadence Agent - AI-driven multi-channel outreach orchestration | |
| Handles intelligent sequencing, timing optimization, engagement-based branching, | |
| personalization, compliance, rate limiting, and comprehensive error handling. | |
| Author: StrategyStack Team | |
| Date: November 2025 | |
| Version: 1.0.0 | |
| """ | |
| import logging | |
| import json | |
| import csv | |
| from datetime import datetime, timedelta | |
| from typing import Dict, List, Optional, Tuple, Any | |
| from enum import Enum | |
| import pytz | |
| from dataclasses import dataclass, field, asdict | |
| import yaml | |
| import time | |
| from collections import defaultdict | |
| from pathlib import Path | |
| # Configure logging | |
| logging.basicConfig( | |
| level=logging.INFO, | |
| format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' | |
| ) | |
| logger = logging.getLogger(__name__) | |
| class ChannelType(Enum): | |
| """Enumeration of supported outreach channels""" | |
| EMAIL = "email" | |
| LINKEDIN = "linkedin" | |
| SMS = "sms" | |
| PHONE = "phone" | |
| DIRECT_MAIL = "direct_mail" | |
| class TouchpointStatus(Enum): | |
| """Enumeration of touchpoint execution statuses""" | |
| PENDING = "pending" | |
| SCHEDULED = "scheduled" | |
| SENT = "sent" | |
| DELIVERED = "delivered" | |
| OPENED = "opened" | |
| CLICKED = "clicked" | |
| REPLIED = "replied" | |
| FAILED = "failed" | |
| SKIPPED = "skipped" | |
| CANCELLED = "cancelled" | |
| class EngagementLevel(Enum): | |
| """Enumeration of engagement levels for branching logic""" | |
| HIGH = "high" | |
| MEDIUM = "medium" | |
| LOW = "low" | |
| NEGATIVE = "negative" | |
| UNKNOWN = "unknown" | |
| class CadenceError(Exception): | |
| """Base exception class for cadence-related errors""" | |
| pass | |
| class RateLimitError(CadenceError): | |
| """Exception raised when rate limits are exceeded""" | |
| pass | |
| class ComplianceError(CadenceError): | |
| """Exception raised for compliance violations""" | |
| pass | |
| class DataValidationError(CadenceError): | |
| """Exception raised for data validation failures""" | |
| pass | |
| class Touchpoint: | |
| """Represents a single touchpoint in a cadence sequence""" | |
| step: int | |
| day: int | |
| channel: ChannelType | |
| template: str | |
| priority: str | |
| conditions: Dict[str, Any] = field(default_factory=dict) | |
| status: TouchpointStatus = TouchpointStatus.PENDING | |
| scheduled_time: Optional[datetime] = None | |
| sent_time: Optional[datetime] = None | |
| engagement_data: Dict[str, Any] = field(default_factory=dict) | |
| error_log: List[Dict[str, Any]] = field(default_factory=list) | |
| retry_count: int = 0 | |
| max_retries: int = 3 | |
| class CadenceState: | |
| """Represents the current state of a lead's cadence journey""" | |
| lead_id: str | |
| cadence_template: str | |
| start_date: datetime | |
| current_step: int = 0 | |
| touchpoints_completed: List[int] = field(default_factory=list) | |
| touchpoints_pending: List[int] = field(default_factory=list) | |
| engagement_level: EngagementLevel = EngagementLevel.UNKNOWN | |
| engagement_score: float = 0.0 | |
| total_opens: int = 0 | |
| total_clicks: int = 0 | |
| total_replies: int = 0 | |
| opted_out: bool = False | |
| suppressed: bool = False | |
| paused: bool = False | |
| completed: bool = False | |
| last_touchpoint_date: Optional[datetime] = None | |
| next_touchpoint_date: Optional[datetime] = None | |
| metadata: Dict[str, Any] = field(default_factory=dict) | |
| error_history: List[Dict[str, Any]] = field(default_factory=list) | |
| class RateLimitTracker: | |
| """Tracks rate limits per channel""" | |
| channel: ChannelType | |
| hourly_count: int = 0 | |
| daily_count: int = 0 | |
| hourly_reset_time: datetime = field(default_factory=datetime.now) | |
| daily_reset_time: datetime = field(default_factory=datetime.now) | |
| domain_counts: Dict[str, int] = field(default_factory=dict) | |
| contact_counts: Dict[str, int] = field(default_factory=dict) | |
| class CadenceAgent: | |
| """ | |
| Main Cadence Agent class for orchestrating multi-channel outreach sequences. | |
| Handles: | |
| - Intelligent sequencing and timing optimization | |
| - Engagement-based branching logic | |
| - Personalization and context-aware messaging | |
| - Compliance and consent management | |
| - Rate limiting and throttling | |
| - Comprehensive error handling and recovery | |
| """ | |
| def __init__(self, config_path: str = "docs/configs/cadence_config.yaml"): | |
| """ | |
| Initialize the Cadence Agent. | |
| Args: | |
| config_path: Path to cadence configuration YAML file | |
| Raises: | |
| FileNotFoundError: If config file not found | |
| DataValidationError: If config validation fails | |
| """ | |
| try: | |
| self.config = self._load_config(config_path) | |
| self.cadence_states: Dict[str, CadenceState] = {} | |
| self.rate_limiters: Dict[ChannelType, RateLimitTracker] = { | |
| channel: RateLimitTracker(channel=channel) | |
| for channel in ChannelType | |
| } | |
| self.suppression_lists: Dict[str, set] = self._load_suppression_lists() | |
| # State persistence files | |
| self.state_file = Path("cadence_states.json") | |
| self.activity_log = Path("cadence_activity_log.csv") | |
| self.error_log = Path("cadence_error_log.csv") | |
| # Load existing states if available | |
| self._load_cadence_states() | |
| # Initialize activity logging | |
| self._initialize_activity_log() | |
| logger.info("CadenceAgent initialized successfully") | |
| except Exception as e: | |
| logger.error(f"Failed to initialize CadenceAgent: {str(e)}") | |
| raise DataValidationError(f"Initialization failed: {str(e)}") | |
| def _load_config(self, config_path: str) -> Dict[str, Any]: | |
| """ | |
| Load and validate cadence configuration. | |
| Args: | |
| config_path: Path to configuration file | |
| Returns: | |
| Validated configuration dictionary | |
| Raises: | |
| FileNotFoundError: If config file not found | |
| DataValidationError: If config validation fails | |
| """ | |
| try: | |
| with open(config_path, 'r') as f: | |
| config = yaml.safe_load(f) | |
| # Validate required config sections | |
| required_sections = [ | |
| 'cadence_templates', | |
| 'touchpoint_sequence', | |
| 'timing_optimization', | |
| 'engagement_branching', | |
| 'rate_limits', | |
| 'compliance' | |
| ] | |
| for section in required_sections: | |
| if section not in config: | |
| raise DataValidationError( | |
| f"Missing required config section: {section}" | |
| ) | |
| logger.info(f"Configuration loaded from {config_path}") | |
| return config | |
| except FileNotFoundError: | |
| logger.error(f"Configuration file not found: {config_path}") | |
| raise | |
| except yaml.YAMLError as e: | |
| logger.error(f"YAML parsing error: {str(e)}") | |
| raise DataValidationError(f"Invalid YAML configuration: {str(e)}") | |
| except Exception as e: | |
| logger.error(f"Unexpected error loading config: {str(e)}") | |
| raise DataValidationError(f"Config load failed: {str(e)}") | |
| def _load_suppression_lists(self) -> Dict[str, set]: | |
| """ | |
| Load suppression lists from configuration. | |
| Returns: | |
| Dictionary of suppression lists | |
| Note: | |
| In production, these would be loaded from database or files. | |
| This implementation uses placeholder data. | |
| """ | |
| suppression_lists = { | |
| 'global_unsubscribe': set(), | |
| 'domain_blocklist': set(), | |
| 'competitor_list': set(), | |
| 'do_not_contact': set() | |
| } | |
| # TODO: Load from actual data sources | |
| logger.info("Suppression lists loaded") | |
| return suppression_lists | |
| def _load_cadence_states(self) -> None: | |
| """ | |
| Load persisted cadence states from disk. | |
| Error handling: | |
| - FileNotFoundError: Log and continue (first run) | |
| - JSONDecodeError: Log error, backup corrupt file, continue | |
| - Exception: Log error and continue with empty state | |
| """ | |
| try: | |
| if self.state_file.exists(): | |
| with open(self.state_file, 'r') as f: | |
| states_data = json.load(f) | |
| # Reconstruct CadenceState objects | |
| for lead_id, state_dict in states_data.items(): | |
| # Convert string dates back to datetime objects | |
| state_dict['start_date'] = datetime.fromisoformat( | |
| state_dict['start_date'] | |
| ) | |
| if state_dict.get('last_touchpoint_date'): | |
| state_dict['last_touchpoint_date'] = datetime.fromisoformat( | |
| state_dict['last_touchpoint_date'] | |
| ) | |
| if state_dict.get('next_touchpoint_date'): | |
| state_dict['next_touchpoint_date'] = datetime.fromisoformat( | |
| state_dict['next_touchpoint_date'] | |
| ) | |
| # Reconstruct EngagementLevel enum | |
| state_dict['engagement_level'] = EngagementLevel( | |
| state_dict['engagement_level'] | |
| ) | |
| self.cadence_states[lead_id] = CadenceState(**state_dict) | |
| logger.info(f"Loaded {len(self.cadence_states)} cadence states") | |
| except FileNotFoundError: | |
| logger.info("No existing cadence states found (first run)") | |
| except json.JSONDecodeError as e: | |
| logger.error(f"Corrupt state file detected: {str(e)}") | |
| # Backup corrupt file | |
| backup_file = self.state_file.with_suffix('.json.corrupt') | |
| self.state_file.rename(backup_file) | |
| logger.info(f"Backed up corrupt state file to {backup_file}") | |
| except Exception as e: | |
| logger.error(f"Unexpected error loading states: {str(e)}") | |
| def _initialize_activity_log(self) -> None: | |
| """ | |
| Initialize activity log CSV file. | |
| Error handling: | |
| - PermissionError: Log error and raise (critical) | |
| - OSError: Log error and raise (critical) | |
| """ | |
| try: | |
| if not self.activity_log.exists(): | |
| with open(self.activity_log, 'w', newline='') as f: | |
| writer = csv.writer(f) | |
| writer.writerow([ | |
| 'timestamp', 'lead_id', 'step', 'channel', 'template', | |
| 'status', 'engagement_level', 'error_message' | |
| ]) | |
| logger.info(f"Activity log initialized at {self.activity_log}") | |
| except (PermissionError, OSError) as e: | |
| logger.error(f"Failed to initialize activity log: {str(e)}") | |
| raise | |
| except Exception as e: | |
| logger.error(f"Unexpected error initializing activity log: {str(e)}") | |
| raise |