Разработка автономной AI-системы коммуникации с клиентами
Автономная система коммуникации управляет полным циклом диалога с клиентом: инициирует коммуникацию по триггерам, поддерживает многоходовые разговоры с сохранением контекста, адаптирует тон под клиента, выполняет действия (бронирование, оплата, изменение заказа) внутри диалога и корректно передаёт разговор оператору при необходимости.
Отличие от FAQ-бота: система понимает intent, управляет состоянием диалога, помнит историю взаимодействий с клиентом и способна выполнять многошаговые сценарии.
Архитектура диалоговой системы
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.postgres import PostgresSaver
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from typing import TypedDict, Annotated, Optional
import operator
class ConversationState(TypedDict):
# Идентификация
session_id: str
customer_id: str
channel: str # "whatsapp", "telegram", "web_chat", "sms"
# История диалога
messages: Annotated[list, operator.add]
# Профиль клиента
customer_profile: Optional[dict]
customer_tier: str # "standard", "premium", "vip"
preferred_language: str # "ru", "en"
# Состояние диалога
current_intent: Optional[str]
dialog_context: dict # Накопленные данные текущего сценария
pending_confirmations: list[dict] # Ожидают подтверждения клиента
# Выполненные действия
completed_actions: Annotated[list, operator.add]
# Управление передачей
escalated: bool
escalation_reason: Optional[str]
escalation_priority: Optional[str]
Контекстный менеджер клиента
class CustomerContextManager:
"""Загружает и обновляет профиль клиента"""
async def load_context(self, customer_id: str) -> dict:
profile_task = crm.get_customer(customer_id)
orders_task = orders_db.get_recent(customer_id, limit=10)
preferences_task = preference_store.get(customer_id)
loyalty_task = loyalty_service.get_status(customer_id)
results = await asyncio.gather(
profile_task, orders_task, preferences_task, loyalty_task,
return_exceptions=True,
)
return {
"profile": results[0] if not isinstance(results[0], Exception) else {},
"recent_orders": results[1] if not isinstance(results[1], Exception) else [],
"preferences": results[2] if not isinstance(results[2], Exception) else {},
"loyalty": results[3] if not isinstance(results[3], Exception) else {},
}
def build_system_context(self, customer_context: dict) -> str:
profile = customer_context.get("profile", {})
loyalty = customer_context.get("loyalty", {})
context_parts = [
f"Клиент: {profile.get('name', 'клиент')}",
f"Статус: {loyalty.get('tier', 'standard')}",
f"Язык: {profile.get('preferred_language', 'ru')}",
]
orders = customer_context.get("recent_orders", [])
if orders:
last_order = orders[0]
context_parts.append(
f"Последний заказ: #{last_order['id']} от {last_order['date']}, "
f"статус: {last_order['status']}"
)
return "\n".join(context_parts)
Система обнаружения намерений и управления сценариями
from pydantic import BaseModel
from typing import Literal
class IntentDetection(BaseModel):
intent: Literal[
"order_status", "order_change", "order_cancel",
"delivery_issue", "return_request", "payment_issue",
"product_question", "complaint", "compliment",
"account_management", "general_question", "farewell",
]
confidence: float
entities: dict # Извлечённые сущности (номер заказа, дата и т.д.)
requires_action: bool # Нужно выполнить действие в системе
needs_clarification: bool # Нужны дополнительные данные от клиента
SYSTEM_PROMPT_TEMPLATE = """Ты — AI-ассистент клиентской службы компании "{company_name}".
Информация о клиенте:
{customer_context}
Правила общения:
- Обращайся к клиенту по имени
- Тон: {tone} (зависит от статуса клиента)
- Язык: {language}
- Никогда не обещай то, что не можешь выполнить
- При ошибках — признавай и предлагай решение
- Не раскрывай внутренние системы и базы данных
Доступные действия:
- Статус заказа
- Изменение адреса доставки (если заказ не передан курьеру)
- Инициирование возврата
- Перенос даты доставки
- Передача оператору
Если не знаешь ответа — честно скажи об этом и предложи передать специалисту."""
def build_system_prompt(state: ConversationState) -> str:
tone_map = {
"standard": "профессиональный, дружелюбный",
"premium": "персональный, внимательный",
"vip": "эксклюзивный, максимально персонализированный",
}
return SYSTEM_PROMPT_TEMPLATE.format(
company_name="RetailCo",
customer_context=state.get("customer_profile", {}).get("context", ""),
tone=tone_map.get(state["customer_tier"], "профессиональный"),
language=state["preferred_language"],
)
Исполнение действий внутри диалога
async def execute_dialog_action(action_name: str, params: dict, state: ConversationState) -> dict:
"""Выполняет действие и возвращает результат для включения в диалог"""
action_handlers = {
"get_order_status": lambda p: orders_api.get_status(p["order_id"]),
"change_delivery_address": lambda p: orders_api.update_address(
p["order_id"], p["new_address"]
),
"initiate_return": lambda p: returns_service.create_request(
order_id=p["order_id"],
reason=p["reason"],
customer_id=state["customer_id"],
),
"reschedule_delivery": lambda p: delivery_api.reschedule(
p["order_id"], p["new_date"]
),
}
handler = action_handlers.get(action_name)
if not handler:
return {"success": False, "error": f"Неизвестное действие: {action_name}"}
try:
result = await handler(params)
return {"success": True, "data": result}
except Exception as e:
return {"success": False, "error": str(e)}
Управление подтверждениями (Confirmation Flow)
def needs_confirmation(action_name: str) -> bool:
"""Действия, требующие подтверждения клиента"""
return action_name in {"cancel_order", "initiate_return", "change_payment_method"}
async def handle_pending_confirmation(state: ConversationState) -> ConversationState:
"""Обрабатывает ответ клиента на запрос подтверждения"""
if not state["pending_confirmations"]:
return state
last_message = state["messages"][-1].content.lower()
confirmation_words = {"да", "yes", "подтверждаю", "согласен", "ок", "ok"}
rejection_words = {"нет", "no", "отмена", "отказываюсь", "стоп"}
if any(word in last_message for word in confirmation_words):
# Выполняем ожидающее действие
pending = state["pending_confirmations"][0]
result = await execute_dialog_action(pending["action"], pending["params"], state)
return {
**state,
"pending_confirmations": state["pending_confirmations"][1:],
"completed_actions": [{"action": pending["action"], "result": result}],
}
elif any(word in last_message for word in rejection_words):
return {
**state,
"pending_confirmations": [],
"messages": [AIMessage("Хорошо, действие отменено. Чем ещё могу помочь?")],
}
# Неоднозначный ответ — переспрашиваем
return {
**state,
"messages": [AIMessage("Пожалуйста, ответьте 'да' для подтверждения или 'нет' для отмены.")],
}
Триггерная коммуникация
class OutboundCommunicationEngine:
"""Инициирует исходящую коммуникацию по бизнес-триггерам"""
TRIGGER_TEMPLATES = {
"order_shipped": {
"message": "Ваш заказ #{order_id} отправлен! Трекинг: {tracking_url}. Ожидаемая доставка: {eta}.",
"channel_priority": ["sms", "whatsapp", "email"],
},
"delivery_delay": {
"message": "Сообщаем о задержке доставки заказа #{order_id}. Новая дата: {new_eta}. Извините за неудобства.",
"channel_priority": ["whatsapp", "telegram", "sms"],
},
"return_approved": {
"message": "Ваш возврат по заказу #{order_id} одобрен. Средства вернутся в течение {refund_days} дней.",
"channel_priority": ["email", "whatsapp"],
},
}
async def send_trigger_message(self, customer_id: str, trigger: str, params: dict):
template_config = self.TRIGGER_TEMPLATES.get(trigger)
if not template_config:
return
# Персонализируем сообщение через LLM
customer = await crm.get_customer(customer_id)
base_message = template_config["message"].format(**params)
if customer.get("tier") in ("premium", "vip"):
# VIP получает более персонализированное сообщение
personalized = await personalize_message(base_message, customer)
else:
personalized = base_message
channel = await self.get_preferred_channel(customer_id, template_config["channel_priority"])
await channel_dispatcher.send(customer_id, channel, personalized)
Практический кейс: телеком-оператор, 18 000 диалогов/день
Компания: региональный телеком с 850 000 абонентов, контакт-центр 120 операторов.
Реализованные сценарии:
- Проверка баланса и трафика
- Подключение/отключение услуг
- Разблокировка номера (верификация по кодовому слову)
- Диагностика технических проблем (по алгоритму)
- Обработка жалоб на качество связи (создание тикета + компенсация автоматически по правилам)
- Прием платежей через dialog (redirect к платёжной форме)
Результаты:
- Автономное закрытие диалогов: 67%
- Среднее время ответа: 4.5 мин → 8 секунд
- CSAT (Customer Satisfaction Score): 3.8/5.0 → 4.2/5.0 (неожиданный рост за счёт скорости)
- Операторы сфокусировались на жалобах и сложных случаях
- NPS: +7 баллов за квартал
Сложности: настройка тональности для angry customers (4 недели итераций). VIP-клиенты ожидают «живого» общения — добавлена опция немедленной передачи человеку.
Сроки
- Архитектура и базовый диалоговый движок: 2–3 недели
- Реализация сценариев (каждый ~3–5 дней): от 3 недель
- Интеграция с каналами коммуникации: 1–2 недели
- Интеграция с CRM/ERP/API: 2–3 недели
- Обучение, тестирование, запуск: 2–3 недели
- Итого: 10–14 недель