In this tutorial, we build a fully functional Pre-Emptive Churn Agent that proactively identifies at-risk users and drafts personalized re-engagement emails before they cancel. Rather than waiting for churn to occur, we design an agentic loop in which we observe user inactivity, analyze behavioral patterns, strategize incentives, and generate human-ready email drafts using Gemini. We orchestrate the entire process step by step, ensuring each component, from data simulation to manager approval, works seamlessly together. Check out theFULL 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 required libraries, and ensure Gemini is available for use. We keep the initialization minimal so the rest of the system loads cleanly. As we run it, we prepare the foundation for the agent-driven workflow that follows. Check out theFULL 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 authentication for Gemini and construct a mock customer database that behaves like a real system. We simulate users with varying levels of inactivity to generate realistic churn scenarios. Check out theFULL 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 build the analytical core of our churn agent to evaluate user behavior and select win-back strategies. We let Gemini interpret signals, such as inactivity and usage patterns, to determine risk and incentives. Check out theFULL 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 generate personalized re-engagement emails based on the strategy output from the previous step. We use Gemini to craft concise, empathetic messaging that aligns with each user’s history. Check out theFULL 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 dashboard where human oversight approves or rejects the drafted email. We keep the flow simple but realistic, ensuring the agent’s actions remain aligned with human judgment. Check out theFULL 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 orchestrate the full system: scanning for at-risk users, analyzing them, drafting messages, and routing everything for approval. We bring all components together into one continuous loop.
In conclusion, we have completed a churn-prevention pipeline that observes, reasons, drafts, and involves a human reviewer before action. We watch the agent detect risk patterns, craft tailored strategies, and generate professional emails, all while maintaining human oversight for final decisions. This implementation demonstrates how agentic workflows can transform customer success operations by enabling timely, personalized, and scalable interventions. We now have a modular foundation we can expand further, connecting it to real databases, CRMs, web dashboards, or automation systems, to build a truly production-ready churn prevention engine.
Check out theFULL CODES here.Also,feel free to follow us onTwitterand don’t forget to join our100k+ ML SubRedditand Subscribe toour Newsletter. Wait! are you on telegram?now you can join us on telegram as well.
Asif Razzaq is the CEO of Marktechpost Media Inc.. As a visionary entrepreneur and engineer, Asif is committed to harnessing the potential of Artificial Intelligence for social good. His most recent endeavor is the launch of an Artificial Intelligence Media Platform, Marktechpost, which stands out for its in-depth coverage of machine learning and deep learning news that is both technically sound and easily understandable by a wide audience. The platform boasts of over 2 million monthly views, illustrating its popularity among audiences.
