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()