Кейс: Создание маркетингового ассистента с использованием Telegram бота и ChatGPT

В современном мире искусственный интеллект становится неотъемлемой частью маркетинговых стратегий. В этом кейсе расскажу, как был создан бот-ассистент маркетолога, использующий возможности ChatGPT для взаимодействия с клиентами через Telegram.

Содержание

Основные компоненты системы
Структура и логика работы бота
1. Импорт необходимых библиотек
2. Настройка логирования
3. Подключение к базе данных
4. Обработка команд и сообщений
5. Интеграция с ChatGPT
6. Сохранение данных пользователей и сообщений
7. Функционал рассылки сообщений
Ключевые особенности и преимущества
Код бота на Python

Основные компоненты системы

Telegram Bot API
OpenAI API (ChatGPT)
База данных PostgreSQL
Python с асинхронными библиотеками

Структура и логика работы бота

1. Импорт необходимых библиотек

import asyncio
from telegram import Update, Bot
from telegram.ext import Application, MessageHandler, filters, ContextTypes, CommandHandler
from openai import AsyncOpenAI
import asyncpg
from datetime import datetime
# ... другие импорты

Этот блок кода импортирует все необходимые библиотеки для работы с Telegram API, базой данных PostgreSQL и OpenAI API.

2. Настройка логирования

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

Логирование важно для отслеживания работы бота и диагностики возможных проблем.

3. Подключение к базе данных

async def db_connect():
    try:
        connection = await asyncpg.connect(user='...', password='...', database='...', host='...')
        logger.info("Успешное подключение к базе данных")
        return connection
    except Exception as e:
        logger.error(f"Ошибка подключения к базе данных: {e}")
        raise

Эта функция устанавливает асинхронное соединение с базой данных PostgreSQL. Важно использовать асинхронные операции для эффективной работы бота.

4. Обработка команд и сообщений

async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
    # ... код обработки команды /start

async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
    # ... код обработки входящих сообщений

Эти функции обрабатывают команду /start и входящие сообщения от пользователей. Они сохраняют информацию о пользователе, обновляют историю чата и генерируют ответы с помощью ChatGPT.

5. Интеграция с ChatGPT

client = AsyncOpenAI(api_key=openai_api_key)

async def update_chat_history(chat_id, role, content, username):
    # ... код обновления истории чата

chat_completion = await client.chat.completions.create(
    model="gpt-3.5-turbo",
    messages=messages_with_context,
)
answer = chat_completion.choices[0].message.content

Этот блок кода отвечает за взаимодействие с API OpenAI, отправку запросов к ChatGPT и получение ответов.

6. Сохранение данных пользователей и сообщений

async def save_user_data_if_not_exists(context, telegram_id, username, full_name):
    # ... код сохранения данных пользователя

async def save_message_to_db(telegram_id, username, content, role):
    # ... код сохранения сообщения в базу данных

Эти функции обеспечивают сохранение информации о пользователях и их сообщениях в базе данных для дальнейшего анализа и улучшения работы бота.

7. Функционал рассылки сообщений

async def send_broadcast_message(bot, message_text):
    # ... код для массовой рассылки сообщений

async def sendall_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    # ... код обработки команды рассылки

Эти функции позволяют администратору отправлять массовые сообщения всем пользователям бота.

Ключевые особенности и преимущества

Асинхронность: Использование асинхронного программирования позволяет боту эффективно обрабатывать множество запросов одновременно.
Интеграция с ChatGPT: Бот использует передовую модель языкового ИИ для генерации релевантных и персонализированных ответов.
Сохранение контекста: Бот хранит историю диалогов, что позволяет ChatGPT генерировать более контекстно-зависимые ответы.
База данных: Использование PostgreSQL для хранения данных о пользователях и сообщениях позволяет анализировать взаимодействия и улучшать работу бота.
Функционал рассылки: Возможность массовой рассылки сообщений полезна для маркетинговых кампаний и информирования пользователей.

Этот бот представляет собой мощный инструмент для маркетологов, сочетающий удобство Telegram с возможностями искусственного интеллекта ChatGPT. Он может эффективно отвечать на вопросы клиентов, предоставлять информацию о услугах и продуктах, а также помогать в различных маркетинговых задачах.

Для дальнейшего развития проекта можно рассмотреть следующие направления:

Интеграция с CRM-системами
Добавление функций аналитики и отчетности
Расширение возможностей персонализации ответов
Внедрение механизмов обучения на основе обратной связи от пользователей

Код бота на Python

import asyncio
from telegram import Update, Bot
from telegram.ext import Application, MessageHandler, filters, ContextTypes, CommandHandler
from openai import AsyncOpenAI
import csv
import logging
from telegram import InlineKeyboardButton, InlineKeyboardMarkup
from functools import partial
import asyncpg
from datetime import datetime


logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

admin_state = {}

async def db_connect():
    try:
        connection = await asyncpg.connect(user='имя пользователя', password='пароль', database='название базы данных', host='localhost')
        logger.info("Успешное подключение к базе данных")
        return connection
    except Exception as e:
        logger.error(f"Ошибка подключения к базе данных: {e}")
        raise

# Словарь, сопоставляющий команды с URL каналов. Можно заменить или удалить
channels = {
    'channel1': ('https://t.me/+JsobbT1ArkY3M2Zi', 'Эй-Ай 🤖 Искусственный интеллект'),
    'channel2': ('https://t.me/+oWkVbBbGQtFiYTAy', 'ИИ инструменты в маркетинге'),
    'channel3': ('https://t.me/+Z_20lkPs2ItlNDcy', 'Богатый ремесленник')
}

async def channel_command(update: Update, context: ContextTypes.DEFAULT_TYPE, channel_url, channel_name):
    keyboard = [[InlineKeyboardButton("Перейти в канал", url=channel_url)]]
    reply_markup = InlineKeyboardMarkup(keyboard)
    await context.bot.send_message(chat_id=update.effective_chat.id, text=f"Подписаться на канал: {channel_name}", reply_markup=reply_markup)

def get_main_menu():
    keyboard = [
        [InlineKeyboardButton("Эй-Ай 🤖 Искусственный интеллект", url="https://t.me/+JsobbT1ArkY3M2Zi")],
        [InlineKeyboardButton("ИИ инструменты в маркетинге", url="https://t.me/+oWkVbBbGQtFiYTAy")],
        [InlineKeyboardButton("Богатый ремесленник", url="https://t.me/+Z_20lkPs2ItlNDcy")]
    ]
    return InlineKeyboardMarkup(keyboard)

async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
    full_name = update.effective_user.full_name
    telegram_id = update.effective_user.id
    username = update.effective_user.username
    welcome_message = f"""Привет, {full_name}. Этот бот открывает доступ к ChatGPT 3.5 Turbo - БЕСПЛАТНО🚀 - без ограничений по количеству запросов.

Работу бота настроил и оплачивает маркетолог <a href="https://t.me/aglamov">Динар Аглямов</a>. Так же, бот может вас проконсультировать какие маркетинговые услуги можно приобрести у Динара.

✅ Чат-бот умеет:

- Ответы на вопросы (для быстрого поиска информации)
- Генерация текста (от SMM до сказок детям)
- Создание кода (например, этого бота написал при поддержке ChatGPT 3.5)
- Помощь с математическими задачами
- Генерация идей и контента (создание изображений будет в другом боте)
- Обработка текстов на разных языках
- Интерактивный диалог (просто поговорить =))
- И многое другое

👉🏻 Для запуска бота, напишите в чат вопрос на любом языке

Приятного пользования

💭 Теперь, можно задать свой вопрос исскуственному интеллекту!"""

    # Отправляем приветственное сообщение с использованием HTML
    await context.bot.send_message(
        chat_id=update.effective_chat.id,
        text=welcome_message,
        parse_mode='HTML',
        disable_web_page_preview=True  # Отключаем предварительный просмотр веб-страницы
    )
    
    # Отправляем меню с кнопками
    menu = get_main_menu()
    await context.bot.send_message(
        chat_id=update.effective_chat.id,
        text="Выберите один из каналов, чтобы подписаться:",
        reply_markup=menu
    )
    await save_user_data_if_not_exists(context, telegram_id, username, full_name)

async def notify_admin(context: ContextTypes.DEFAULT_TYPE, message: str):
    admin_chat_id = 'chat_id администратора'  # Замените на chat_id администратора или пользователя
    try:
        await context.bot.send_message(chat_id=admin_chat_id, text=message)
    except Exception as e:
        logger.error(f"Ошибка при отправке сообщения администратору: {e}")

async def save_user_data_if_not_exists(context, telegram_id, username, full_name):
    conn = await db_connect()
    try:
        user_exists = await conn.fetchval('SELECT EXISTS(SELECT 1 FROM users WHERE telegram_id = $1)', telegram_id)
        if not user_exists:
            await conn.execute('INSERT INTO users(telegram_id, username, full_name) VALUES ($1, $2, $3)', telegram_id, username, full_name)
            logger.info(f"Новый пользователь сохранен: {full_name} (@{username})")
            await notify_admin(context, f"Новый пользователь: {full_name} (@{username})")
    except Exception as e:
        logger.error(f"Ошибка при сохранении данных пользователя: {e}")
    finally:
        await conn.close()


async def save_message_to_db(telegram_id, username, content, role):
    conn = await db_connect()
    try:
        await conn.execute(
            '''
            INSERT INTO messages(telegram_id, username, content, role, timestamp)
            VALUES($1, $2, $3, $4, $5)
            ''',
            telegram_id, username, content, role, datetime.now()
        )
        logger.info("Сообщение успешно сохранено в базу данных.")
    except Exception as e:
        logger.error(f"Ошибка при сохранении сообщения в базу данных: {e}")
    finally:
        await conn.close()


async def send_broadcast_message(bot, message_text):
    conn = await db_connect()
    try:
        subscribers = await conn.fetch('SELECT telegram_id FROM users')
        logger.info("Получен список подписчиков для рассылки")
    except Exception as e:
        logger.error(f"Ошибка при получении списка подписчиков: {e}")
        return 0, 0, []
    finally:
        await conn.close()

    success_count = 0
    failed_count = 0
    failed_subscribers = []

    for record in subscribers:
        subscriber_id = record['telegram_id']
        try:
            await bot.send_message(chat_id=subscriber_id, text=message_text)
            success_count += 1
        except Exception as e:
            logger.error(f"Не удалось отправить сообщение {subscriber_id}: {e}")
            failed_count += 1
            failed_subscribers.append(subscriber_id)

    logger.info(f"Рассылка завершена. Успешно: {success_count}, Не удалось: {failed_count}")
    return success_count, failed_count, failed_subscribers



openai_api_key = "API ключ"
telegram_token = "токен бота"
bot = Bot(token=telegram_token)
client = AsyncOpenAI(api_key=openai_api_key)

chat_histories = {}

# Функция для добавления сообщения в историю чата
async def update_chat_history(chat_id, role, content, username):
    if chat_id not in chat_histories:
        chat_histories[chat_id] = []
    if len(chat_histories[chat_id]) > 10:
        chat_histories[chat_id].pop(0)  # Удаляем самое старое сообщение
    chat_histories[chat_id].append({"role": role, "content": content})
    await save_message_to_db(chat_id, username, content, role)

async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    telegram_id = update.effective_user.id
    user_message = update.message.text
    chat_id = update.effective_chat.id
    username = update.effective_user.username
    full_name = update.effective_user.full_name
    logger.info(f"Получено сообщение от {username}: {user_message}")
    # Логика для обработки текста рассылки от администратора
    if admin_state.get(telegram_id) == "awaiting_message_text":
        if telegram_id == телеграм id админа:
            success_count, failed_count, failed_subscribers = await send_broadcast_message(context.bot, user_message)
            admin_state[telegram_id] = None  # Сброс состояния
            report_message = f"Рассылка выполнена.\nУспешно отправлено сообщений: {success_count}.\nНе удалось отправить сообщений: {failed_count}."
            if failed_count > 0:
                report_message += f"\nНе удалось отправить следующим пользователям: {', '.join(failed_subscribers)}"
            await update.message.reply_text(report_message)
            return

    # Сохраняем данные пользователя, если они еще не были сохранены
    await save_user_data_if_not_exists(chat_id, telegram_id, username, full_name)

    # Обновляем историю диалога для текущего чата
    await update_chat_history(chat_id, "user", user_message, username)

    # Предустановленный контекст и т.д.
    personal_context = [
        {"role": "system", "content": "Ты ассистент маркетолога Динара Аглямова, твоя задача нативно и предлагать консультации у Динара, предлагать пройти мини курсы и подписываться на его телеграм каналы и отвечать на всевозможные вопросы пользователей. Динар Аглямов - маркетолог со стажем 13 лет. Проходил обучение у Максима Белоусова по разработке маркетинговых стратегий. Динар оказывает услуги разработки. маркетинговых стратегий, консультирует по вопросам запуска лидогенерации для B2B, имеет большой опыт (более 40 запусков) лидогенерации на франшизы. Работает с нишами в B2B секторе. Для связи с Динаром контакт в телеграм @aglamov. Динар записывает минин-курсы по маркетингку и внедрению искусственного интеллекта в бизнес процессы. Мини курсы можно посмотреть по ссылке: http://aglamovcourses.ru. Так же, если пользователь хочет больше узнать про маркетинг и задает вопросы касательно маркетиннка, нативно предлагай подписаться на телеграм канал Динара про маркетинг. Вот ссылка https://t.me/+oWkVbBbGQtFiYTAy. Если пользователи интересуются искусственным интелектом, нативно предлагай им подписаться на телеграм канал Динара Эй-Ай 🤖 Искусственный интеллект по ссылке https://t.me/+JsobbT1ArkY3M2Zi. Будь не навязчив. Выполняй роль Ассистента Динара Аглямова, который умеет отвечает на все вопросы. Общайся только на русском языке. Коротко о проектах: с 2011 по 2015 съемка видео. Реклама и семейная съемка, 2014 год - аренда камер Go Pro, 2014 год - продажа игрушек BRUDER, с 2013 по 2014 купил пилораму занимался лесопереработкой, с 2013 по 2017 кондитерская Мистер Макарун. С кухни съемной квартиры до производства в 90м2, в мае 2016 году - пробую собирать и продавать Дизайнерскую мебель, с декабря 2017 по июнь 2017 открываю под ключ пекарню по франшизе Хлеб из Тандыра, в мае 2017 продажа мягких игрушек ЖДУН, в июне 2017 регистрирую домен aglamov.biz, где выкладываю свои статьи про бизнес, фриланс и продвижение. Начинаю углубленное изучение SEO продвижения. августе 2018 зарегистрировал ИП в третий раз. Начал официально заниматься продвижением бизнеса в интернете. в 2022 прошел курсы у Максима Белоусова. в 2024 погрузился в изучение искуссвенного интеллекта. Сейчас занимается разработкой программы для автоматизированного анализа звонков отдела продаж и обслуживания на основе ии. Часть кейсов и портфолио Динара можно найти по ссылке: https://vc.ru/u/1047023-dinar-aglyamov-pro-marketing, "},
    ]

    messages_with_context = personal_context + chat_histories.get(chat_id, [])

    chat_completion = await client.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=messages_with_context,
    )
    answer = chat_completion.choices[0].message.content
    logger.info(f"Ответ от ChatGPT: {answer}")
    await update_chat_history(chat_id, "assistant", answer, username)

    # Отправляем ответ пользователю
    await context.bot.send_message(chat_id=chat_id, text=answer)

async def sendall_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    telegram_id = update.effective_user.id
    if telegram_id == Telegram ID администратора:  # Замените на ваш Telegram ID администратора
        admin_state[telegram_id] = "awaiting_message_text"
        await update.message.reply_text("Пожалуйста, отправьте текст для рассылки.")
    else:
        await update.message.reply_text("У вас нет прав для выполнения этой команды.")



if __name__ == '__main__':
    application = Application.builder().token(telegram_token).build()

    application.add_handler(CommandHandler('start', start))
    application.add_handler(CommandHandler('sendall', sendall_command))

    for command, (url, name) in channels.items():
        handler = CommandHandler(command, partial(channel_command, channel_url=url, channel_name=name))
        application.add_handler(handler)

    application.add_handler(MessageHandler(filters.TEXT, handle_message))

    loop = asyncio.get_event_loop()
    loop.run_until_complete(application.run_polling())