AI Customer Support — тикет-система, RAG по базе знаний, умная поддержка

Клиент пишет вопрос
    ↓
Классификация (Claude): типовой / нетипичный / критичный
    ↓
Типовой → RAG по базе знаний → автоответ (большая часть, ориентир 80%)
Нетипичный → AI-драфт для агента → агент редактирует → отправляет (ориентир 15%)
Критичный → немедленная эскалация + уведомление → агент сам отвечает (ориентир 5%)

knowledge-base/
  billing.md          — вопросы об оплате
  shipping.md         — доставка и возвраты
  integrations.md     — подключение интеграций
  plans.md            — тарифные планы
  account.md          — управление аккаунтом

## Как изменить тарифный план

Зайдите в личный кабинет → Настройки → Тарифный план → выберите новый план.
Изменение вступает в силу немедленно. Доплата рассчитывается пропорционально.

## Возврат средств

Возврат возможен в течение 14 дней после оплаты при первом заказе.
Для возврата напишите на billing@company.com с темой "Возврат" и номером заказа.
Срок обработки — 5 рабочих дней.

## Чем отличаются планы Basic и Pro

Basic: до 5 пользователей, 10 GB хранилища, email-поддержка.
Pro: неограниченно пользователей, 100 GB хранилища, приоритетная поддержка, API доступ.

import anthropic
import json
import time
from pathlib import Path
from datetime import datetime, timezone

client = anthropic.Anthropic()


def load_knowledge_base(kb_path: str) -> str:
    """Загружает базу знаний из MD файлов"""
    kb_content = []
    for md_file in Path(kb_path).glob("**/*.md"):
        kb_content.append(f"\n## {md_file.stem}\n{md_file.read_text()}")
    return "\n".join(kb_content)


KB = load_knowledge_base("knowledge-base/")


def answer_support_ticket(question: str) -> dict:
    """Классифицирует тикет и отвечает или эскалирует"""

    start_time = time.time()

    response = client.messages.create(
        model="claude-sonnet-5-5",
        max_tokens=800,
        system=f"""Ты AI-агент поддержки клиентов.

БАЗА ЗНАНИЙ КОМПАНИИ:
{KB}

Правила:
1. Если ответ есть в базе знаний — ответь точно используя её, не добавляй ничего от себя
2. В конце ответа укажи: Источник: [название раздела базы знаний]
3. Если ответа в базе знаний нет — напиши в первой строке ESCALATE и объясни почему нужен человек
4. Если вопрос касается возврата денег — ВСЕГДА пиши ESCALATE (даже если ответ в базе есть)
5. Если вопрос о технической неполадке у клиента — ESCALATE
6. Тон: дружелюбный, конкретный, без воды""",
        messages=[{"role": "user", "content": question}]
    )

    answer = "".join(b.text for b in response.content if b.type == "text")
    response_time = time.time() - start_time

    return {
        "answer": answer,
        "needs_human": answer.strip().startswith("ESCALATE"),
        "response_time_sec": round(response_time, 2),
        "confidence": "low" if answer.strip().startswith("ESCALATE") else "high"
    }


def track_metrics(result: dict, question: str):
    """Записывает метрики в лог"""
    log_entry = {
        "ts": datetime.now(timezone.utc).isoformat(),
        "escalated": result["needs_human"],
        "response_time_sec": result["response_time_sec"],
        "confidence": result["confidence"],
        "question_length": len(question)
    }
    with open("support-metrics.jsonl", "a") as f:
        f.write(json.dumps(log_entry, ensure_ascii=False) + "\n")


# Пример использования
if __name__ == "__main__":
    questions = [
        "Как изменить тарифный план?",
        "Хочу вернуть деньги за подписку",
        "У меня не работает интеграция с Slack, всё сломалось"
    ]

    for question in questions:
        print(f"\nВопрос: {question}")
        result = answer_support_ticket(question)
        track_metrics(result, question)

        if result["needs_human"]:
            print(f"ЭСКАЛАЦИЯ -> передаём живому агенту")
            print(f"Причина: {result['answer']}")
        else:
            print(f"Автоответ ({result['response_time_sec']} сек):")
            print(result["answer"])

from flask import Flask, request
import requests
import os

app = Flask(__name__)

INTERCOM_TOKEN = os.environ["INTERCOM_TOKEN"]
AI_BOT_ID = os.environ["INTERCOM_BOT_ID"]


def assign_to_human_agent(conversation_id: str, priority: str = "normal"):
    """Назначает тикет живому агенту и ставит метку"""
    requests.post(
        f"https://api.intercom.io/conversations/{conversation_id}/parts",
        headers={
            "Authorization": f"Bearer {INTERCOM_TOKEN}",
            "Content-Type": "application/json"
        },
        json={
            "type": "admin",
            "admin_id": AI_BOT_ID,
            "message_type": "assignment",
            "assignee_id": None  # назначит команде, не конкретному агенту
        }
    )

    # Ставим метку приоритета
    if priority == "high":
        requests.post(
            f"https://api.intercom.io/conversations/{conversation_id}/tags",
            headers={"Authorization": f"Bearer {INTERCOM_TOKEN}"},
            json={"id": os.environ["INTERCOM_HIGH_PRIORITY_TAG_ID"]}
        )


@app.post("/intercom-webhook")
def handle_message():
    data = request.json

    if data.get("type") != "conversation.user.created":
        return {"status": "ignored"}

    item = data["data"]["item"]
    conversation_id = item["id"]
    parts = item["conversation_parts"]["conversation_parts"]

    if not parts:
        return {"status": "no_message"}

    message = parts[0]["body"]

    # AI отвечает
    result = answer_support_ticket(message)
    track_metrics(result, message)

    if not result["needs_human"]:
        # Отправляем автоответ от имени бота
        requests.post(
            f"https://api.intercom.io/conversations/{conversation_id}/reply",
            headers={
                "Authorization": f"Bearer {INTERCOM_TOKEN}",
                "Content-Type": "application/json"
            },
            json={
                "type": "admin",
                "admin_id": AI_BOT_ID,
                "message_type": "comment",
                "body": result["answer"]
            }
        )
    else:
        # Сообщаем клиенту что подключаем человека
        requests.post(
            f"https://api.intercom.io/conversations/{conversation_id}/reply",
            headers={
                "Authorization": f"Bearer {INTERCOM_TOKEN}",
                "Content-Type": "application/json"
            },
            json={
                "type": "admin",
                "admin_id": AI_BOT_ID,
                "message_type": "comment",
                "body": "Ваш вопрос передан специалисту. Ответим в течение 2 часов."
            }
        )
        assign_to_human_agent(conversation_id, priority="high")

    return {"status": "ok"}


if __name__ == "__main__":
    app.run(port=5000)

import asyncio
import os
from telegram import Update, Bot
from telegram.ext import Application, MessageHandler, filters

TELEGRAM_BOT_TOKEN = os.environ["TELEGRAM_BOT_TOKEN"]
SUPPORT_TEAM_CHAT = os.environ["SUPPORT_TEAM_CHAT_ID"]

bot = Bot(token=TELEGRAM_BOT_TOKEN)


async def handle_support_message(update: Update, context):
    user_message = update.message.text
    user_id = update.effective_user.id
    username = update.effective_user.username or str(user_id)

    # Индикатор что бот работает
    await update.message.reply_text("Проверяю...")

    result = answer_support_ticket(user_message)
    track_metrics(result, user_message)

    if not result["needs_human"]:
        await update.message.reply_text(result["answer"])
    else:
        # Клиенту — сообщение что подключаем человека
        await update.message.reply_text(
            "Ваш вопрос требует внимания специалиста. "
            "Ответим в течение 2 часов рабочего времени."
        )
        # Команде — уведомление с полным контекстом
        escalation_text = (
            f"Эскалация от @{username} (id: {user_id})\n\n"
            f"Вопрос: {user_message}\n\n"
            f"Причина эскалации: {result['answer']}"
        )
        await bot.send_message(
            chat_id=SUPPORT_TEAM_CHAT,
            text=escalation_text
        )


def run_bot():
    application = Application.builder().token(TELEGRAM_BOT_TOKEN).build()
    application.add_handler(
        MessageHandler(filters.TEXT & ~filters.COMMAND, handle_support_message)
    )
    application.run_polling()


if __name__ == "__main__":
    run_bot()

def generate_support_report(metrics_file: str = "support-metrics.jsonl") -> dict:
    """Считает основные метрики за период"""
    entries = []
    with open(metrics_file) as f:
        for line in f:
            entries.append(json.loads(line))

    if not entries:
        return {"error": "нет данных"}

    total = len(entries)
    escalated = sum(1 for e in entries if e["escalated"])
    deflection_rate = round((total - escalated) / total * 100, 1)
    avg_response_time = round(
        sum(e["response_time_sec"] for e in entries) / total, 2
    )

    return {
        "total_tickets": total,
        "escalated": escalated,
        "auto_resolved": total - escalated,
        "deflection_rate_pct": deflection_rate,
        "avg_response_time_sec": avg_response_time
    }


# Пример вывода:
# {
#   "total_tickets": 150,
#   "escalated": 28,
#   "auto_resolved": 122,
#   "deflection_rate_pct": 81.3,
#   "avg_response_time_sec": 2.4
# }