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. Instead of waiting for churn to happen, we design an agentic loop in which we observe user inactivity, analyze behavior patterns, create promotion strategies, and generate human-crafted email drafts using Gemini. We organize the entire process step by step, ensuring that every component works together seamlessly, from data simulation to manager approval. check it out full code 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 necessary libraries, and make sure Gemini is available for use. We keep initialization to a minimum so that the rest of the system loads cleanly. As soon as we run it, we lay the foundation for an agent-driven workflow. check it out full code 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 build a fake customer database that behaves like a real system. We simulate users with different levels of inactivity to generate realistic churn scenarios. check it out full code 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 create the analytical core of our churn agent to evaluate user behavior and select win-win strategies. We let Gemini interpret signals like inactivity and usage patterns to determine risk and incentives. check it out full code 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 messages that are tailored to each user’s history. check it out full code 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 drafted emails. We keep the flow simple but realistic, ensuring that the agent’s actions remain consistent with human judgment. check it out full code 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 the entire system: scanning at-risk users, analyzing them, drafting messages, and routing everything for approval. We bring all the components together in a continuous loop.
Finally, we have completed a churn-prevention pipeline that observes, reasons, drafts, and engages a human reviewer before taking action. We see the agent detect risk patterns, formulate tailored strategies, and craft professional emails 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 that we can expand further by connecting to real databases, CRMs, web dashboards, or automation systems to create a truly production-ready churn prevention engine.
check it out full code hereAlso, feel free to follow us Twitter And don’t forget to join us 100k+ ml subreddit and subscribe our newsletterwait! Are you on Telegram? Now you can also connect with us on Telegram.
Asif Razzaq Marktechpost Media Inc. Is the CEO of. As a visionary entrepreneur and engineer, Asif is committed to harnessing the potential of Artificial Intelligence for social good. Their most recent endeavor is the launch of MarketTechPost, an Artificial Intelligence media platform, known for its in-depth coverage of Machine Learning and Deep Learning news that is technically robust and easily understood by a wide audience. The platform boasts of over 2 million monthly views, which shows its popularity among the audience.
