Менеджеры чат-ботов — NLP, intent routing и умная эскалация на человека

{
    "session_id": "uuid",
    "user_id": "telegram_chat_id",
    "history": [...],       # последние N сообщений
    "current_intent": "complaint",
    "entities": {"order_id": "12345"},
    "frustration_score": 0.3,
    "escalated": False,
    "turns_without_resolution": 1
}

[Telegram] → adapter → [Intent Engine] → action → [Telegram response]
[WhatsApp] → adapter → [Intent Engine] → action → [WhatsApp response]
[Web Widget] → adapter → [Intent Engine] → action → [Web response]

import os
import json
import redis
from fastapi import FastAPI, Request
from anthropic import Anthropic

app = FastAPI()
client = Anthropic()
r = redis.Redis(host='localhost', port=6379, decode_responses=True)

ESCALATION_PHRASES = [
    "хочу к менеджеру", "оператор", "живой человек",
    "переключи", "call me", "позвоните"
]

FRUSTRATION_WORDS = [
    "ужасно", "кошмар", "никогда", "мошенники",
    "верните деньги", "обман", "безобразие"
]


def get_session(session_id: str) -> dict:
    data = r.get(f"session:{session_id}")
    if data:
        return json.loads(data)
    return {
        "history": [],
        "frustration_score": 0.0,
        "turns_without_resolution": 0,
        "escalated": False,
        "current_intent": None
    }


def save_session(session_id: str, state: dict):
    r.setex(f"session:{session_id}", 3600, json.dumps(state))


def check_escalation_rules(message: str, state: dict) -> tuple[bool, str]:
    """Правила эскалации — быстрая проверка без AI."""
    msg_lower = message.lower()
    
    # Явный запрос оператора
    if any(phrase in msg_lower for phrase in ESCALATION_PHRASES):
        return True, "explicit_request"
    
    # Детектор фрустрации по словам
    frustration_hit = sum(1 for w in FRUSTRATION_WORDS if w in msg_lower)
    if frustration_hit >= 2:
        return True, "frustration_detected"
    
    # Накопленная фрустрация по сессии
    if state["frustration_score"] > 0.7:
        return True, "cumulative_frustration"
    
    # Бот не смог помочь 3 раза подряд
    if state["turns_without_resolution"] >= 3:
        return True, "repeated_fallback"
    
    return False, ""


def classify_intent(message: str, history: list) -> dict:
    """Claude Haiku классифицирует намерение и извлекает сущности."""
    history_text = "\n".join([
        f"{m['role']}: {m['content']}" for m in history[-4:]
    ])
    
    response = client.messages.create(
        model="claude-haiku-4-5",   # проверь, что модель ещё доступна в API; актуальные модели: страница «Актуальное сейчас»
        max_tokens=300,
        system="""Ты классификатор намерений для чат-бота интернет-магазина.
Верни JSON с полями:
- intent: одно из [faq_price, faq_delivery, faq_return, complaint, purchase_intent, order_status, out_of_scope]
- confidence: 0.0-1.0
- entities: объект с извлечёнными данными (order_id, amount, date, product)
- sentiment: positive/neutral/negative
- frustration_score: 0.0-1.0

Только JSON, без пояснений.""",
        messages=[{
            "role": "user",
            "content": f"История диалога:\n{history_text}\n\nНовое сообщение: {message}"
        }]
    )
    
    try:
        return json.loads("".join(b.text for b in response.content if b.type == "text"))
    except Exception:
        return {
            "intent": "out_of_scope",
            "confidence": 0.0,
            "entities": {},
            "sentiment": "neutral",
            "frustration_score": 0.3
        }


def generate_bot_response(intent: str, message: str, entities: dict, history: list) -> str:
    """Генерируем ответ на основе намерения."""
    
    intent_prompts = {
        "faq_price": "Ответь на вопрос о цене товара. Если нет конкретного товара — спроси уточнение.",
        "faq_delivery": "Ответь о доставке: срок 2-5 дней, бесплатно от 3000 руб.",
        "faq_return": "Ответь о возврате: 14 дней без объяснений, нужен чек.",
        "complaint": "Прими жалобу с сочувствием. Уточни детали для решения.",
        "purchase_intent": "Помоги с покупкой. Уточни детали, предложи добавить в корзину.",
        "order_status": "Запроси номер заказа если не указан.",
    }
    
    prompt = intent_prompts.get(intent, "Вежливо скажи, что не понял вопрос, и предложи варианты.")
    
    response = client.messages.create(
        model="claude-haiku-4-5",
        max_tokens=200,
        system=f"Ты вежливый помощник интернет-магазина. {prompt}. Отвечай коротко — 1-3 предложения.",
        messages=[{"role": "user", "content": message}]
    )
    return "".join(b.text for b in response.content if b.type == "text")


async def notify_operator(session_id: str, message: str, reason: str, state: dict):
    """Уведомление оператора в Telegram."""
    import httpx
    
    bot_token = os.environ.get("TELEGRAM_BOT_TOKEN")
    operator_chat_id = os.environ.get("OPERATOR_CHAT_ID")
    
    history_preview = "\n".join([
        f"{'👤' if m['role']=='user' else '🤖'} {m['content']}"
        for m in state["history"][-3:]
    ])
    
    text = (
        f"🚨 Эскалация | {reason}\n"
        f"Сессия: {session_id}\n\n"
        f"Последние сообщения:\n{history_preview}\n\n"
        f"Последнее: {message}"
    )
    
    async with httpx.AsyncClient() as http:
        await http.post(
            f"https://api.telegram.org/bot{bot_token}/sendMessage",
            json={"chat_id": operator_chat_id, "text": text}
        )


@app.post("/chat")
async def chat_endpoint(request: Request):
    body = await request.json()
    session_id = body["session_id"]
    message = body["message"]
    
    state = get_session(session_id)
    
    # Быстрая проверка правил эскалации
    should_escalate, escalation_reason = check_escalation_rules(message, state)
    
    if should_escalate and not state["escalated"]:
        state["escalated"] = True
        state["history"].append({"role": "user", "content": message})
        save_session(session_id, state)
        await notify_operator(session_id, message, escalation_reason, state)
        return {
            "response": "Понял вас. Переключаю на оператора — он ответит в течение 5 минут.",
            "escalated": True,
            "reason": escalation_reason
        }
    
    # AI-классификация намерения
    classification = classify_intent(message, state["history"])
    intent = classification.get("intent", "out_of_scope")
    confidence = classification.get("confidence", 0.0)
    
    # Обновляем state
    state["current_intent"] = intent
    state["frustration_score"] = max(
        state["frustration_score"],
        classification.get("frustration_score", 0.0)
    )
    
    # Низкая уверенность — считаем нерешённым
    if confidence < 0.5 or intent == "out_of_scope":
        state["turns_without_resolution"] += 1
    else:
        state["turns_without_resolution"] = 0
    
    # Генерируем ответ
    bot_response = generate_bot_response(
        intent, message,
        classification.get("entities", {}),
        state["history"]
    )
    
    # Сохраняем историю
    state["history"].append({"role": "user", "content": message})
    state["history"].append({"role": "assistant", "content": bot_response})
    state["history"] = state["history"][-20:]  # Храним последние 20 сообщений
    save_session(session_id, state)
    
    return {
        "response": bot_response,
        "intent": intent,
        "confidence": confidence,
        "escalated": False
    }

mkdir chatbot-manager && cd chatbot-manager
python -m venv venv && source venv/bin/activate
pip install fastapi uvicorn anthropic redis httpx python-dotenv

# .env файл
echo "ANTHROPIC_API_KEY=your_key" >> .env
echo "TELEGRAM_BOT_TOKEN=your_bot_token" >> .env
echo "OPERATOR_CHAT_ID=your_chat_id" >> .env

# Запускаем Redis локально (или используем Redis Cloud)
docker run -d -p 6379:6379 redis:alpine

{
  "faq_price": "Цены на наши товары начинаются от 500 руб. Актуальный каталог: shop.example.com/catalog",
  "faq_delivery": "Доставка 2-5 рабочих дней. Бесплатно при заказе от 3000 руб.",
  "faq_return": "Возврат в течение 14 дней. Нужен чек и оригинальная упаковка. Деньги возвращаем за 3-5 дней."
}

uvicorn main:app --reload --port 8000

# Обычный вопрос
curl -X POST http://localhost:8000/chat \
  -H "Content-Type: application/json" \
  -d '{"session_id": "test1", "message": "сколько стоит доставка?"}'

# Тест эскалации
curl -X POST http://localhost:8000/chat \
  -H "Content-Type: application/json" \
  -d '{"session_id": "test2", "message": "это УЖАСНО, верните деньги немедленно!!!"}'

# Явный запрос оператора
curl -X POST http://localhost:8000/chat \
  -H "Content-Type: application/json" \
  -d '{"session_id": "test3", "message": "хочу поговорить с живым человеком"}'

# telegram_adapter.py
from telegram.ext import Application, MessageHandler, filters
import httpx, os

async def handle_message(update, context):
    session_id = str(update.effective_chat.id)
    message = update.message.text
    
    async with httpx.AsyncClient() as client:
        resp = await client.post(
            "http://localhost:8000/chat",
            json={"session_id": session_id, "message": message}
        )
        data = resp.json()
    
    await update.message.reply_text(data["response"])

app = Application.builder().token(os.environ["TELEGRAM_BOT_TOKEN"]).build()
app.add_handler(MessageHandler(filters.TEXT, handle_message))
app.run_polling()

@app.get("/stats")
async def get_stats():
    keys = r.keys("session:*")
    total = len(keys)
    escalated = sum(1 for k in keys if json.loads(r.get(k)).get("escalated"))
    return {
        "total_sessions": total,
        "escalated": escalated,
        "escalation_rate": f"{escalated/total*100:.1f}%" if total else "0%",
        "active_last_hour": total  # упрощённо
    }