AI-чатбот для клиентской поддержки: LLM + Telegram на Python

Telegram пользователь → aiogram бот → LLM API → Ответ
                    ↘ SQLite (история диалогов)
                    ↘ База знаний (FAQ, статьи)

pip install aiogram openai python-telegram-bot requests

# bot.py
import asyncio
import logging
from aiogram import Bot, Dispatcher, types
from aiogram.filters import Command
from aiogram.types import Message

API_TOKEN = "YOUR_BOT_TOKEN"
bot = Bot(token=API_TOKEN)
dp = Dispatcher()

logging.basicConfig(level=logging.INFO)

@dp.message(Command("start"))
async def cmd_start(message: Message):
    text = (
        "Привет! Я AI-ассистент поддержки.\n\n"
        "Напишите ваш вопрос, и я постараюсь помочь.\n"
        "Напишите /faq для списка частых вопросов.\n"
        "Напишите /operator если нужен живой оператор."
    )
    await message.answer(text)

# knowledge_base.py
import sqlite3
from pathlib import Path

DB_PATH = Path("support_bot.db")

def init_db():
    conn = sqlite3.connect(DB_PATH)
    c = conn.cursor()
    c.execute("""
        CREATE TABLE IF NOT EXISTS faq (
            id INTEGER PRIMARY KEY,
            question TEXT UNIQUE,
            answer TEXT,
            keywords TEXT,
            usage_count INTEGER DEFAULT 0
        )
    """)
    c.execute("""
        CREATE TABLE IF NOT EXISTS dialogues (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            user_id INTEGER,
            user_message TEXT,
            bot_response TEXT,
            timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
            escalated INTEGER DEFAULT 0
        )
    """)
    conn.commit()
    conn.close()

def add_faq(question: str, answer: str, keywords: str = ""):
    conn = sqlite3.connect(DB_PATH)
    c = conn.cursor()
    c.execute("INSERT OR REPLACE INTO faq (question, answer, keywords) VALUES (?, ?, ?)",
              (question, answer, keywords))
    conn.commit()
    conn.close()

def search_faq(user_message: str) -> tuple:
    """Ищем совпадение по ключевым словам. Возвращает (answer, confidence) или (None, 0)."""
    conn = sqlite3.connect(DB_PATH)
    c = conn.cursor()

    words = user_message.lower().split()
    best_match = None
    best_score = 0

    rows = c.execute("SELECT question, answer, keywords, usage_count FROM faq").fetchall()

    for question, answer, keywords, usage in rows:
        kw_lower = keywords.lower()
        score = sum(1 for w in words if w in kw_lower or w in question.lower())
        if score > best_score:
            best_score = score
            best_match = (answer, question)

    if best_match and best_score >= 2:
        confidence = min(best_score / len(words), 1.0)
        # увеличиваем счётчик использования
        c.execute("UPDATE faq SET usage_count = usage_count + 1 WHERE question = ?",
                  (best_match[1],))
        conn.commit()
    else:
        best_match = None
        confidence = 0

    conn.close()
    return best_match[0] if best_match else None, confidence

# llm_client.py
import openai

openai.api_key = "YOUR_API_KEY"

SYSTEM_PROMPT = """Ты — ассистент клиентской поддержки компании.
Отвечай кратко и по делу. Если не знаешь ответа — скажи об этом и предложи
передать вопрос оператору. Не выдумывай информацию.

Формат ответа:

- Краткий ответ (1-3 предложения)

- Если нужно — ссылка на документацию или FAQ

- Если не уверен — пометь ответ флагой ESCALATE
"""

def generate_response(user_message: str, history: list = None) -> str:
    messages = [{"role": "system", "content": SYSTEM_PROMPT}]

    if history:
        for h in history[-6:]:  # последние 3 диалога
            messages.append({"role": "user", "content": h["user"]})
            messages.append({"role": "assistant", "content": h["bot"]}")

    messages.append({"role": "user", "content": user_message})

    resp = openai.ChatCompletion.create(
        model="gpt-4",
        messages=messages,
        temperature=0.3,
        max_tokens=500
    )
    return resp.choices[0].message.content

def classify_confidence(user_message: str) -> float:
    """Оценивает, насколько бот уверен в ответе."""
    resp = openai.ChatCompletion.create(
        model="gpt-3.5-turbo",
        messages=[
            {"role": "system", "content": "Оцени уверенность ответа на вопрос поддержки. Верни число от 0 до 1."},
            {"role": "user", "content": f"Вопрос: {user_message}"}
        ],
        temperature=0,
        max_tokens=5
    )
    try:
        return float(resp.choices[0].message.content)
    except ValueError:
        return 0.5

# handlers.py
from aiogram.types import Message
from aiogram.dispatcher import Dispatcher
from knowledge_base import search_faq
from llm_client import generate_response, classify_confidence
import sqlite3
from pathlib import Path

dp = Dispatcher()
DB_PATH = Path("support_bot.db")

def save_dialogue(user_id: int, user_msg: str, bot_resp: str, escalated: bool = False):
    conn = sqlite3.connect(DB_PATH)
    c = conn.cursor()
    c.execute(
        "INSERT INTO dialogues (user_id, user_message, bot_response, escalated) VALUES (?, ?, ?, ?)",
        (user_id, user_msg, bot_resp, 1 if escalated else 0)
    )
    conn.commit()
    conn.close()

@dp.message()
async def process_message(message: Message):
    user_id = message.from_user.id
    user_text = message.text

    # 1. Пробуем базу знаний
    faq_answer, confidence = search_faq(user_text)

    if faq_answer and confidence > 0.5:
        await message.answer(faq_answer)
        save_dialogue(user_id, user_text, faq_answer)
        return

    # 2. Генерируем ответ через LLM
    await message.answer("Думаю...")
    llm_answer = generate_response(user_text)

    if "ESCALATE" in llm_answer:
        # 3. Эскалация оператору
        operator_text = f"⚠️ Нужна помощь оператора\n\nПользователь: {message.from_user.first_name}\nID: {user_id}\nВопрос: {user_text}"
        # здесь можно отправить уведомление оператору
        await message.answer("Передаю ваш вопрос оператору. Ответим в ближайшее время!")
        save_dialogue(user_id, user_text, "Эскалировано оператору", escalated=True)
    else:
        await message.answer(llm_answer)
        save_dialogue(user_id, user_text, llm_answer)

@dp.message(Command("faq"))
async def cmd_faq(message: Message):
    faq_list = [
        "💰 Цены и тарифы",
        "📦 Доставка и сроки",
        "🔄 Возврат и обмен",
        "🔧 Техническая поддержка",
        "📋 Статус заказа",
    ]
    text = "Частые вопросы:\n\n" + "\n".join(faq_list) + "\n\nНажмите на тему или напишите вопрос."
    await message.answer(text)

# main.py
import asyncio
from bot import bot, dp
from handlers import dp as handlers_dp
from knowledge_base import init_db

# инициализация базы данных и FAQ
init_db()
add_faq("Сколько стоит доставка?", "Доставка бесплатна при заказе от 5000₽. Иначе — 350₽.", "доставка цена стоимость")
add_faq("Как оформить возврат?", "Возврат в течение 14 дней. Напишите /return с номером заказа.", "возврат обмен деньги назад")

async def main():
    await dp.start_polling(bot)

if __name__ == "__main__":
    asyncio.run(main())

# analytics.py
import sqlite3
from pathlib import Path
from datetime import datetime, timedelta

DB_PATH = Path("support_bot.db")

def daily_stats(days: int = 7) -> dict:
    conn = sqlite3.connect(DB_PATH)
    c = conn.cursor()
    since = datetime.now() - timedelta(days=days)

    total = c.execute("SELECT COUNT(*) FROM dialogues WHERE timestamp > ?", (since,)).fetchone()[0]
    escalated = c.execute("SELECT COUNT(*) FROM dialogues WHERE escalated = 1 AND timestamp > ?", (since,)).fetchone()[0]
    auto_resolved = total - escalated

    return {
        "total_dialogues": total,
        "auto_resolved": auto_resolved,
        "escalated": escalated,
        "auto_rate": f"{auto_resolved / total * 100:.1f}%" if total else "0%"
    }

def top_faq():
    conn = sqlite3.connect(DB_PATH)
    c = conn.cursor()
    rows = c.execute("SELECT question, usage_count FROM faq ORDER BY usage_count DESC LIMIT 10").fetchall()
    conn.close()
    return rows