Розробка AI CSM — цифрового менеджера з обслуговування клієнтів
AI CSM (Customer Success Manager) автоматизує ключові завдання управління взаємовідносинами з клієнтами: моніторинг health score, проактивне виявлення ризиків оттоку, персоналізовані nurture-кампанії, підготовка до QBR (Quarterly Business Review), управління онбордингом. Мета — масштабувати Customer Success без лінійного зростання команди, зберігаючи персоналізацію.
Health Score та 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 # Розбивка за компонентами
churn_probability: float # 0-1
churn_signals: list[str] # Конкретні сигнали
recommended_actions: list[str]
priority_contact: bool
urgency: Literal["immediate", "this_week", "this_month", "monitoring"]
class HealthScoreCalculator:
WEIGHTS = {
"product_usage": 0.30, # Частота та глибина використання продукту
"feature_adoption": 0.20, # Освоєння ключових функцій
"support_health": 0.15, # Кількість/тип тікетів
"engagement": 0.15, # Відкриття email, участь у вебінарах
"nps_csat": 0.10, # NPS / CSAT оцінки
"contract_health": 0.10, # Своєчасність платежів, ризик downgrade
}
def calculate_product_usage_score(self, customer: dict) -> float:
"""MAU, DAU, session duration vs baseline для тарифу"""
dau_ratio = customer.get("dau_30d_avg", 0) / customer.get("licensed_seats", 1)
sessions_per_user = customer.get("sessions_per_user_30d", 0)
# Нормалізуємо до 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"Різке падіння активності: -{int((1 - customer['logins_30d']/max(customer['logins_prev_30d'], 1)) * 100)}%")
if customer.get("open_critical_tickets", 0) >= 2:
signals.append(f"Відкритих критичних тікетів: {customer['open_critical_tickets']}")
if customer.get("last_login_days_ago", 0) > 14:
signals.append(f"Останній вхід: {customer['last_login_days_ago']} днів тому")
if customer.get("nps_score") and customer["nps_score"] <= 6:
signals.append(f"Низький NPS: {customer['nps_score']}/10")
if customer.get("payment_overdue_days", 0) > 0:
signals.append(f"Прострочка платежу: {customer['payment_overdue_days']} днів")
if customer.get("contract_renewal_days", 365) < 90:
signals.append(f"До продовження: {customer['contract_renewal_days']} днів")
return signals
async def compute_health_score(self, customer: dict) -> CustomerHealthScore:
# Розрахунок числових компонентів
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 для рекомендацій дій
actions_response = await client.chat.completions.create(
model="gpt-4o-mini",
messages=[{
"role": "user",
"content": f"""Клієнт: {customer['name']}, тариф: {customer['plan']},
score: {overall:.0f}/100, сигнали: {signals}
Запропонуй 3 конкретні дії CSM. Поверни 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",
)
Проактивний CSM-агент
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:
"""Складає план дій для клієнта"""
plan_response = await client.chat.completions.create(
model="gpt-4o",
messages=[{
"role": "system",
"content": """Ти — досвідчений CSM. Створи план роботи з клієнтом на наступні 2 тижні.
Враховуй: health score, сигнали оттоку, історію взаємодій.
Будь конкретним: що саме сказати/написати, коли, через який канал."""
}, {
"role": "user",
"content": f"""
Клієнт: {state['customer_profile']['name']}, тариф: {state['customer_profile']['plan']}
Health score: {state['health_score'].overall_score}/100
Сигнали: {state['health_score'].churn_signals}
Останні взаємодії: {state['recent_interactions'][-3:]}
Використання продукту: {state['customer_profile'].get('usage_summary')}
"""
}],
)
# Парсимо план дій
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:
"""Виконує автоматизовані дії"""
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-підготовка
async def generate_qbr_preparation(customer_id: str) -> dict:
"""Автоматична підготовка до квартального бізнес-огляду"""
customer_data = await fetch_customer_quarterly_data(customer_id)
qbr_content = await client.chat.completions.create(
model="gpt-4o",
messages=[{
"role": "system",
"content": """Ти — CSM, готуєш матеріали до QBR. Створи:
1. Executive Summary (3-4 речення про вартість за квартал)
2. Ключові досягнення (вимірні результати)
3. Метрики використання (в динаміці)
4. Розв'язані проблеми
5. Цілі на наступний квартал
6. Рекомендації з розширення (не агресивні продажі)"""
}, {
"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),
}
Практичний кейс: SaaS-платформа, 800 B2B клієнтів
Ситуація: 6 CSM на 800 клієнтів = 133 клієнта/CSM. Високопріоритетні (enterprise) отримували достатньо уваги, mid-market клієнти — реактивне обслуговування.
AI CSM охоплює mid-market сегмент (300 клієнтів):
- Щоденний розрахунок health score для всіх
- Автоматичні листи при зниженні активності
- Щотижневий дайджест CSM: топ-10 клієнтів, які потребують уваги
- Автоматична підготовка QBR для всіх
- Моніторинг сигналів upsell (зростання використання, наближення до лімітів)
Результати за 6 місяців:
- NRR (Net Revenue Retention): 94% → 98%
- Churn у mid-market сегменті: 8.2% → 5.1%
- CSM час на адміністративні завдання: -45%
- QBR coverage: 40% клієнтів (тільки ті, кого встигали) → 91%
- Upsell виручка з mid-market: +34%
Часова шкала
- Health Score система: 2–3 тижні
- Proactive engagement engine: 2–3 тижні
- QBR автоматизація: 1–2 тижні
- Інтеграція з CRM та email-платформою: 1–2 тижні
- Калібрування з командою: 2 тижні
- Разом: 8–12 тижнів