How to Build a Proactive Pre-Emptive Churn Prevention Agent with Intelligent Observation and Strategy Formation

import os import time import json import random from datetime import datetime, timedelta from typing import List, Dict, Any import textwrap   try:    import google.generativeai as genai except ImportError:    !pip install -q -U google-generativeai    import google.generativeai as genai   from google.colab import userdata import getpass

def setup_gemini():    print("--- 🔐 Security Check ---")    try:        api_key = userdata.get('GEMINI_API_KEY')    except:        print("Please enter your Google Gemini API Key:")        api_key = getpass.getpass("API Key: ")    if not api_key:        raise ValueError("API Key is required to run the agent.")    genai.configure(api_key=api_key)    return genai.GenerativeModel('gemini-2.5-flash')   class MockCustomerDB:    def __init__(self):        self.today = datetime.now()        self.users = self._generate_mock_users()      def _generate_mock_users(self) -> List[Dict]:        profiles = [            {"id": "U001", "name": "Sarah Connor", "plan": "Enterprise",             "last_login_days_ago": 2, "top_features": ["Reports", "Admin Panel"], "total_spend": 5000},            {"id": "U002", "name": "John Smith", "plan": "Basic",             "last_login_days_ago": 25, "top_features": ["Image Editor"], "total_spend": 50},            {"id": "U003", "name": "Emily Chen", "plan": "Pro",             "last_login_days_ago": 16, "top_features": ["API Access", "Data Export"], "total_spend": 1200},            {"id": "U004", "name": "Marcus Aurelius", "plan": "Enterprise",             "last_login_days_ago": 45, "top_features": ["Team Management"], "total_spend": 8000}        ]        return profiles      def fetch_at_risk_users(self, threshold_days=14) -> List[Dict]:        return [u for u in self.users if u['last_login_days_ago'] >= threshold_days]

class ChurnPreventionAgent:    def __init__(self, model):        self.model = model      def analyze_and_strategize(self, user: Dict) -> Dict:        print(f"   ... 🧠 Analyzing strategy for {user['name']}...")        prompt = f"""        You are a Customer Success AI Specialist.        Analyze this user profile and determine the best 'Win-Back Strategy'.        USER PROFILE:        - Name: {user['name']}        - Plan: {user['plan']}        - Days Inactive: {user['last_login_days_ago']}        - Favorite Features: {', '.join(user['top_features'])}        - Total Spend: ${user['total_spend']}        TASK:        1. Determine the 'Churn Probability' (Medium/High/Critical).        2. Select a specific INCENTIVE.        3. Explain your reasoning briefly.        OUTPUT FORMAT:        {{            "risk_level": "High",            "incentive_type": "Specific Incentive",            "reasoning": "One sentence explanation."        }}        """        try:            response = self.model.generate_content(prompt)            clean_json = response.text.replace("```json", "").replace("```", "").strip()            return json.loads(clean_json)        except Exception as e:            return {                "risk_level": "Unknown",                "incentive_type": "General Check-in",                "reasoning": f"Analysis failed: {str(e)}"            }

def draft_engagement_email(self, user: Dict, strategy: Dict) -> str:        print(f"   ... ✍️  Drafting email for {user['name']} using '{strategy['incentive_type']}'...")        prompt = f"""        Write a short, empathetic, professional re-engagement email.        TO: {user['name']}        CONTEXT: They haven't logged in for {user['last_login_days_ago']} days.        STRATEGY: {strategy['incentive_type']}        REASONING: {strategy['reasoning']}        USER HISTORY: They love {', '.join(user['top_features'])}.        TONE: Helpful and concise.        """        response = self.model.generate_content(prompt)        return response.text

class ManagerDashboard:    def review_draft(self, user_name, strategy, draft_text):        print("n" + "="*60)        print(f"🚨 REVIEW REQUIRED: Re-engagement for {user_name}")        print(f"🎯 Strategy: {strategy['incentive_type']}")        print(f"📝 Risk Level: {strategy['risk_level']}")        print("-" * 60)        print("📨 DRAFT EMAIL:n")        print(textwrap.indent(draft_text, '    '))        print("-" * 60)        print("n[Auto-Simulation] Manager reviewing...")        time.sleep(1.5)        if strategy['risk_level'] == "Critical":            print("✅ MANAGER DECISION: Approved (Priority Send)")            return True        else:            print("✅ MANAGER DECISION: Approved")            return True

def main():    print("Initializing Agentic System...")    try:        model = setup_gemini()        db = MockCustomerDB()        agent = ChurnPreventionAgent(model)        manager = ManagerDashboard()    except Exception as e:        print(f"Setup failed: {e}")        return      print("n🔍 AGENT STATUS: Scanning Database for inactive users (>14 days)...")    at_risk_users = db.fetch_at_risk_users(threshold_days=14)    print(f"Found {len(at_risk_users)} at-risk users.n")      for user in at_risk_users:        print(f"--- Processing Case: {user['id']} ({user['name']}) ---")        strategy = agent.analyze_and_strategize(user)        email_draft = agent.draft_engagement_email(user, strategy)        approved = manager.review_draft(user['name'], strategy, email_draft)        if approved:            print(f"🚀 ACTION: Email queued for sending to {user['name']}.")        else:            print(f"🛑 ACTION: Email rejected.")        print("n")        time.sleep(1)   if __name__ == "__main__":    main()