How to Build an Effective Pre-Emptive Churn Prevention Agent with Smart Vision and Strategy
In this tutorial, we build a fully functional Pre-Emptive Churn agent that identifies vulnerable users and writes personalized re-engagement emails before they cancel. Rather than waiting for churn to happen, we design an agent loop where we see user inactivity, analyze behavioral patterns, create motivational strategies, and generate personalized email drafts using Gemini. We organize the entire process step by step, ensuring that each part, from data simulation to manager approval, works together seamlessly. Check it out FULL CODES here.
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 getpassWe set up our environment, import all the required libraries, and make sure that Gemini is available for use. We keep startup to a minimum so that the entire program loads cleanly. As we run it, we prepare the basis for the following agent-driven workflow. Check it out FULL CODES here.
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]We configure Gemini authentication and create a fake customer database that behaves like a real system. We simulate users with various levels of inactivity to generate realistic churn scenarios. Check it out FULL CODES here.
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)}"
}We are building the analytics core of our churn agent to evaluate user behavior and select retargeting strategies. We let Gemini interpret signals, such as inactivity and usage patterns, to determine risks and incentives. Check it out FULL CODES here.
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.textWe create personalized re-engagement emails based on the strategy output from the previous step. We use Gemini to create short, sensitive messages that match each user's history. Check it out FULL CODES here.
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 TrueWe simulate a manager's dashboard where a person's supervision approves or rejects a written email. We keep the flow simple but realistic, ensuring that the agent's actions are always consistent with human judgment. Check it out FULL CODES here.
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()We organize a full system: scanning vulnerable users, analyzing them, writing messages, and submitting everything for approval. We combine all the parts into one continuous loop.
In conclusion, we have completed a churn prevention pipeline that checks, reasons, drafts, and involves a human reviewer before action. We watch the agent discover risk patterns, draw up tailored strategies, and produce expert emails, all while monitoring the person to make final decisions. This implementation demonstrates how agent workflow can transform customer success operations by enabling timely, personalized, and scalable interventions. We now have a modular foundation that can be further extended, connecting it to real databases, CRMs, web dashboards, or automation systems, to create a churn prevention engine that is truly production-ready.
Check it out FULL CODES here. Also, feel free to follow us Twitter and don't forget to join our 100k+ ML SubReddit and Subscribe to Our newspaper. Wait! are you on telegram? now you can join us on telegram too.
Asif Razzaq is the CEO of Marktechpost Media Inc. As a visionary entrepreneur and engineer, Asif is committed to harnessing the power of Artificial Intelligence for the benefit of society. His latest endeavor is the launch of Artificial Intelligence Media Platform, Marktechpost, which stands out for its extensive coverage of machine learning and deep learning stories that sound technically sound and easily understood by a wide audience. The platform boasts of more than 2 million monthly views, which shows its popularity among the audience.
