Разработка 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, # Открытие emails, участие в вебинарах
"nps_csat": 0.10, # NPS / CSAT оценки
"contract_health": 0.10, # Своевременность оплат, risk of 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 недели
- Чернила калибровка с командой CS: 2 недели
- Итого: 8–12 недель