AI CSM Development — Digital Customer Success Manager
AI CSM (Customer Success Manager) automates key customer relationship management tasks: health score monitoring, proactive churn risk detection, personalized nurture campaigns, QBR (Quarterly Business Review) preparation, onboarding management. Goal — scale Customer Success without linear team growth while maintaining personalization.
Health Score and Churn Risk Detection
from pydantic import BaseModel
from typing import Literal, Optional
from openai import AsyncOpenAI
import pandas as pd
client = AsyncOpenAI()
class CustomerHealthScore(BaseModel):
customer_id: str
overall_score: int # 0-100
health_tier: Literal["healthy", "at_risk", "critical"]
score_components: dict # Component breakdown
churn_probability: float # 0-1
churn_signals: list[str] # Specific signals
recommended_actions: list[str]
priority_contact: bool
urgency: Literal["immediate", "this_week", "this_month", "monitoring"]
class HealthScoreCalculator:
WEIGHTS = {
"product_usage": 0.30, # Frequency and depth of product usage
"feature_adoption": 0.20, # Key feature adoption
"support_health": 0.15, # Ticket quantity/type
"engagement": 0.15, # Email opens, webinar participation
"nps_csat": 0.10, # NPS / CSAT scores
"contract_health": 0.10, # Payment timeliness, downgrade risk
}
def calculate_product_usage_score(self, customer: dict) -> float:
"""MAU, DAU, session duration vs baseline for plan"""
dau_ratio = customer.get("dau_30d_avg", 0) / customer.get("licensed_seats", 1)
sessions_per_user = customer.get("sessions_per_user_30d", 0)
# Normalize to 0-100
dau_score = min(dau_ratio * 100, 100)
session_score = min(sessions_per_user * 10, 100)
return (dau_score * 0.6 + session_score * 0.4)
def calculate_churn_signals(self, customer: dict) -> list[str]:
signals = []
if customer.get("logins_30d", 0) < customer.get("logins_prev_30d", 0) * 0.5:
signals.append(f"Sharp activity drop: -{int((1 - customer['logins_30d']/max(customer['logins_prev_30d'], 1)) * 100)}%")
if customer.get("open_critical_tickets", 0) >= 2:
signals.append(f"Open critical tickets: {customer['open_critical_tickets']}")
if customer.get("last_login_days_ago", 0) > 14:
signals.append(f"Last login: {customer['last_login_days_ago']} days ago")
if customer.get("nps_score") and customer["nps_score"] <= 6:
signals.append(f"Low NPS: {customer['nps_score']}/10")
if customer.get("payment_overdue_days", 0) > 0:
signals.append(f"Payment overdue: {customer['payment_overdue_days']} days")
if customer.get("contract_renewal_days", 365) < 90:
signals.append(f"Days to renewal: {customer['contract_renewal_days']}")
return signals
async def compute_health_score(self, customer: dict) -> CustomerHealthScore:
# Calculate numeric components
components = {
"product_usage": self.calculate_product_usage_score(customer),
"feature_adoption": customer.get("feature_adoption_pct", 0),
"support_health": max(0, 100 - customer.get("open_tickets", 0) * 15),
"engagement": customer.get("email_engagement_score", 50),
"nps_csat": (customer.get("nps_score", 7) - 1) / 9 * 100,
"contract_health": 100 - customer.get("payment_overdue_days", 0) * 2,
}
overall = sum(
components[k] * self.WEIGHTS[k] for k in self.WEIGHTS
)
signals = self.calculate_churn_signals(customer)
tier = "healthy" if overall >= 70 else ("at_risk" if overall >= 40 else "critical")
# LLM for action recommendations
actions_response = await client.chat.completions.create(
model="gpt-4o-mini",
messages=[{
"role": "user",
"content": f"""Customer: {customer['name']}, plan: {customer['plan']},
score: {overall:.0f}/100, signals: {signals}
Suggest 3 specific CSM actions. Return JSON: [{{"action": "...", "timeline": "..."}}]"""
}],
)
actions = json.loads(actions_response.choices[0].message.content)
return CustomerHealthScore(
customer_id=customer["id"],
overall_score=int(overall),
health_tier=tier,
score_components=components,
churn_probability=max(0, min(1, (100 - overall) / 100)),
churn_signals=signals,
recommended_actions=[a["action"] for a in actions],
priority_contact=tier == "critical" or len(signals) >= 3,
urgency="immediate" if tier == "critical" else "this_week" if tier == "at_risk" else "monitoring",
)
Proactive CSM Agent
from langgraph.graph import StateGraph, END
class CSMAgentState(TypedDict):
customer_id: str
health_score: CustomerHealthScore
customer_profile: dict
recent_interactions: list[dict]
action_plan: list[dict]
messages_sent: list[dict]
escalated: bool
async def analyze_and_plan(state: CSMAgentState) -> CSMAgentState:
"""Creates action plan for customer"""
plan_response = await client.chat.completions.create(
model="gpt-4o",
messages=[{
"role": "system",
"content": """You are an experienced CSM. Create a work plan for the customer for next 2 weeks.
Consider: health score, churn signals, interaction history.
Be specific: what exactly to say/write, when, through which channel."""
}, {
"role": "user",
"content": f"""
Customer: {state['customer_profile']['name']}, plan: {state['customer_profile']['plan']}
Health score: {state['health_score'].overall_score}/100
Signals: {state['health_score'].churn_signals}
Recent interactions: {state['recent_interactions'][-3:]}
Product usage: {state['customer_profile'].get('usage_summary')}
"""
}],
)
# Parse action plan
action_plan = parse_action_plan(plan_response.choices[0].message.content)
return {**state, "action_plan": action_plan}
async def execute_automated_actions(state: CSMAgentState) -> CSMAgentState:
"""Executes automatable actions"""
messages_sent = []
for action in state["action_plan"]:
if action["type"] == "send_email":
email = await generate_personalized_email(action, state)
await email_service.send(
to=state["customer_profile"]["email"],
subject=email["subject"],
body=email["body"],
)
messages_sent.append({"type": "email", "action": action["description"]})
elif action["type"] == "in_app_notification":
await notification_service.send_in_app(
customer_id=state["customer_id"],
message=action["message"],
)
elif action["type"] == "schedule_checkin":
await calendar.create_event(
title=f"Check-in: {state['customer_profile']['name']}",
date=action["date"],
description=action["context"],
)
return {**state, "messages_sent": messages_sent}
QBR Preparation
async def generate_qbr_preparation(customer_id: str) -> dict:
"""Automatic preparation for quarterly business review"""
customer_data = await fetch_customer_quarterly_data(customer_id)
qbr_content = await client.chat.completions.create(
model="gpt-4o",
messages=[{
"role": "system",
"content": """You are a CSM preparing QBR materials. Create:
1. Executive Summary (3-4 sentences about quarterly value)
2. Key achievements (measurable results)
3. Usage metrics (trends)
4. Resolved issues
5. Next quarter goals
6. Expansion recommendations (non-aggressive)"""
}, {
"role": "user",
"content": json.dumps(customer_data, ensure_ascii=False),
}],
)
return {
"qbr_deck_draft": qbr_content.choices[0].message.content,
"metrics_summary": customer_data["metrics"],
"renewal_signals": analyze_renewal_readiness(customer_data),
}
Case Study: SaaS Platform, 800 B2B Customers
Situation: 6 CSMs managing 800 customers = 133 customers/CSM. High-priority (enterprise) customers received adequate attention, mid-market customers had reactive service.
AI CSM covers mid-market segment (300 customers):
- Daily health score calculation for all
- Automatic emails on activity decline
- Weekly CSM digest: top-10 customers needing attention
- Automatic QBR preparation for all
- Monitor upsell signals (usage growth, approaching limits)
Results in 6 Months:
- NRR (Net Revenue Retention): 94% → 98%
- Churn in mid-market segment: 8.2% → 5.1%
- CSM time on administrative tasks: -45%
- QBR coverage: 40% of customers (only those managed) → 91%
- Mid-market upsell revenue: +34%
Timeline
- Health Score system: 2–3 weeks
- Proactive engagement engine: 2–3 weeks
- QBR automation: 1–2 weeks
- CRM and email platform integration: 1–2 weeks
- Team calibration: 2 weeks
- Total: 8–12 weeks