3 июн в 00:27

Как интегрировать GPT-4 Omni в Телеграм Бот на Python без танцев с бубном?

Простой

Ожидает приглашения

Создаем файл запуска нашего бота

Используем библиотеку Aiogram 3.х, в общем все как у всех, ничего необычного в рамках этого пункта. Назовем его app.py.

import os
import asyncio
import logging
import dotenv

from aiogram import Bot, Dispatcher
from aiogram.enums import ParseMode
from aiogram.client.default import DefaultBotProperties
from aiogram.fsm.storage.memory import MemoryStorage

from core.handlers import start, error, generate
from core.middlewares.access import AccessMiddleware


dotenv.load_dotenv()


async def main() -> None:
    bot = Bot(
        token=os.getenv("TG_TOKEN"),
        default=DefaultBotProperties(parse_mode=ParseMode.MARKDOWN),
    )

    storage = MemoryStorage()

    dp = Dispatcher(storage=storage)

    dp.include_routers(start.start_router, error.error_router, generate.generate_router)
    dp.message.middleware(AccessMiddleware())

    await bot.delete_webhook(drop_pending_updates=True)

    await dp.start_polling(bot, polling_timeout=10)


if __name__ == "__main__":
    try:
        logging.basicConfig(level=logging.INFO)
        asyncio.run(main())
    except KeyboardInterrupt:
        print("Exit")

Прописываем хендлеры

Их у нас не много. Достаточно команды старт, генерации ответа ИИ и обработки дублирования запроса.

Начнем с start.py:

from aiogram import Router
from aiogram.types import Message
from aiogram.filters import CommandStart
from aiogram.fsm.context import FSMContext


start_router = Router()


@start_router.message(CommandStart())
async def command_start(message: Message, state: FSMContext) -> None:
    await message.answer("Авторизация прошла успешно! Какой у вас вопрос?")
    await state.update_data(prompt=[])

Авторизация тут упоминается, потому что в дальнейшем напишем middleware для проверки ID пользователя. Ну и машина состояний, обновляем "память" нашей модели при запуске команды /start.

Теперь реализуем самую объемную и важную функцию проекта generate.py:

from aiogram import Router, F
from aiogram.types import Message
from aiogram.fsm.context import FSMContext

from core.utils.ai import ai_omni
from core.states.user import Generate


generate_router = Router()


@generate_router.message(F.text)
async def generate(message: Message, state: FSMContext) -> None:
    processing = await message.answer("Обрабатка запроса...")

    await state.set_state(Generate.text)

    data = await state.get_data()

    prompt = data.get("prompt", [])
    prompt.append({"role": "user", "content": message.text})

    completion = await ai_omni(prompt)

    prompt.append({"role": "assistant", "content": completion})

    await state.update_data(prompt)

    try:
        await message.answer(completion)
    finally:
        await processing.delete()

    await state.set_state(None)

Что из этого хочется подчеркнуть? Наверное удаление сообщений, как только ИИ сгенерировал ответ. На мобильных устройствах это выглядит интересно. Я думаю, все знают про разлетающиеся песчинки в момент удаления. Ну и конечно же машина состояний для того, чтобы наш GPT помнил, что ему писал пользователь, и даже что он сам мне писал. Так как весь контент мы сохраняем в виде словарей в наш массив prompt.

Ну и последний хендлер обработки дублирования запроса. Когда наш пользователь случайно ввел что-то 2 раза или более, и нам бы не хотелось тратить на это токены, вашему вниманию error.py:

from aiogram import Router
from aiogram.types import Message

from core.states.user import Generate


error_router = Router()


@error_router.message(Generate.text)
async def generate_error(message: Message) -> None:
    await message.answer(
        "Подождите, пока обрабатывается предыдущий запрос или нажмите /start чтобы начать сначала"
    )

И да, тут ничего криминального. Все максимально упрощенно. Обработчик срабатывает в том случае, если у нас идет генерация ответа на прошлый запрос, а именно существует активное состояние.

Подключение к OpenAI

Тут конечно, первостепенно, стоит сказать про сервис, который стал основой для этого поста и роликов на моем YouTube канале.

Сервис называется ProxyAPI.

Благодаря этим ребятам, можно использовать API OpenAI только тогда, когда тебе нужно и платить ровно столько, сколько ты спросил и сколько тебе ответили. Никаких иностранных карт, верификаций, VPN и прочего. Минимальный платеж 200 рублей - СБП, карты РФ и СНГ. Никаких проблем. Все максимально прозрачно.

После оплаты ты создаешь API токен и интегрируешь его в своего бота, больше ничего делать не нужно. Он будет работать как швейцарские часы.

Давайте же настроим подключение в нашем боте, создаем файл ai.py:

import os
import dotenv

from openai import AsyncOpenAI


dotenv.load_dotenv()


client = AsyncOpenAI(api_key=os.getenv("AI_TOKEN"), base_url=os.getenv("AI_BASE_URL"))


async def ai_omni(prompt: str) -> str:
    completion = await client.chat.completions.create(model="gpt-4o", messages=prompt)

    return completion.choices[0].message.content

Заметьте, что работаем с оригинальной библиотекой OpenAI, ничего необычного.

Кстати, у них есть разные модели, не только GPT-4 Omni. Со всем списком можете ознакомиться у них на сайте, вот ссылка на их документацию.

Забыл про состояния

Тут не будет задерживаться. Просто создаем состояние фиксации запросов states.py.

from aiogram.fsm.state import State, StatesGroup


class Generate(StatesGroup):
    text = State()

Я думаю комментарии излишни.

Авторизация

Ну и последнее, что хотелось бы сделать - это не дать доступ неавторизованным пользователям тратить твои деньги с аккаунта ProxyAPI, пишем access.py.

import os
import logging
import dotenv

from typing import Callable, Awaitable, Dict, Any

from aiogram import BaseMiddleware
from aiogram.types import Message


dotenv.load_dotenv()


AUTHORIZED_USERS = [int(user) for user in os.getenv("USERS").split(",")]


class AccessMiddleware(BaseMiddleware):
    async def __call__(
        self,
        handler: Callable[[Message, Dict[str, Any]], Awaitable[Any]],
        event: Message,
        data: Dict[str, Any],
    ) -> Any:

        if event.from_user.id not in AUTHORIZED_USERS:
            logging.warning(
                f"Попытка входа неавторизованного пользователя: {event.from_user.id}"
            )
            return
        return await handler(event, data)

Переменные окружения

Файл .env тоже следует показать, вот он (пример):

# Telegram
TG_TOKEN=43537456387:AAGYr_JwQFBQd9fbmQZQ_1zoV92AoqeqrzY

# OpenAI
AI_TOKEN=sk-JJHHJHJKjjlkl3NYsYy2EOVrRegxaTwijJ
AI_BASE_URL=https://api.proxyapi.ru/openai/v1

# Users
USERS=1234567,7654321,272727272

Ну вот и все друзья, надеюсь я помог кому-то данной статьей. Я не разработчик, по крайней мере пока. Поэтому мог допускать какие-то ошибки. Если вы их видите, буду рад почитать комментарии и внести правки.

Если вы новичок, и хотите разобрать работу бота более подробно, у меня есть ряд видео на YouTube канале, вот ссылка.

Теги:

Хабы:

Python