• redis:

  • Директория Redis контейнера содержит файл конфигурации redis.conf

• youtubot:

  • Директория Python контейнера содержит файлы Telegram бота:

    • Dockerfile: Файл, необходимый для сборки Docker-образа Python.

    • requirements.txt: Файл, содержащий зависимости Python проекта.

    • tests/: Директория, в которой расположены модульные тесты для проверки функциональности кода.

    • bot/: Основная директория бота, содержащая следующие файлы и поддиректории:

      • handlers/: Директория с обработчиками запросов пользователей.

      • llm.py: Файл, в котором реализованы методы для работы с Large Language Models.

    • config.py: Файл с настройками проекта.

    • main.py: Основной файл, запускающий бота.

• docker-compose.yml:

  • Файл конфигурации Docker Compose, находится в корневой директории проекта. Этот файл определяет конфигурацию проекта и обеспечивает удобный запуск и управление контейнерами Redis и Python.

import os
import asyncio
import logging
import redis.asyncio as redis
import aiogram as ag
import config as cf
from bot.handlers import users

# Настройка логирования
logger = logging.getLogger(__name__)

async def main():
    # Установка уровня логирования в зависимости от переменной окружения DEBUG
    if os.getenv('DEBUG'):
        logging.basicConfig(level=logging.DEBUG, format=cf.logging_format)

    # Подключение к Redis
    db = redis.Redis(
        host=os.getenv('REDIS_HOST', 'localhost'),
        port=os.environ['REDIS_PORT'],
        password=os.getenv('REDIS_PASSWORD'),
        decode_responses=True  # Автоматическое преобразование ответов из байтов в строки
    )
    logger.info('Redis connected.')

    try:
        # Инициализация бота и диспетчера
        bot = ag.Bot(token=os.environ['TELEGRAM_BOT_TOKEN'])
        dp = ag.Dispatcher(bot=bot, storage=ag.fsm.storage.memory.MemoryStorage())

        # Настройка обработчиков
        await users.setup(bot, dp, db)
        logger.info('users.setup is done.')

        # Удаление webhook и запуск поллинга
        await bot.delete_webhook(drop_pending_updates=True)
        logger.info('Starting polling...')
        await dp.start_polling(bot, allowed_updates=dp.resolve_used_update_types())
    finally:
        # Закрытие соединения с Redis
        await db.aclose()

if __name__ == '__main__':
    # Загрузка переменных окружения из файла .env и запуск основной функции
    import dotenv
    dotenv.load_dotenv()
    asyncio.run(main())

import asyncio
import aiogram as ag
from typing import Any, Callable, Dict, Awaitable
from aiogram.types import Message, TelegramObject
from aiogram.utils.chat_action import ChatActionSender
from .. import llm

# Создание роутера для обработки сообщений
router = ag.Router()
router.message.filter(ag.F.chat.type == 'private')  # Фильтр для приватных чатов

# Обработчик команды '/start'
@router.message(ag.F.text, ag.filters.Command("start"))
async def start_handler(msg: Message):
    await msg.answer("Здравствуйте! Я могу посмотреть любой ролик из YouTube и рассказать вам о нем. "
                     "Пришлите мне ссылку на ролик и задавайте любые вопросы.")

# Обработчик всех прочих сообщений
@router.message(ag.F.text)
async def message_handler(msg: Message, llm_answer):
    # Передача пользовательского запроса в отдельную задачу
    await llm_answer.user_query(msg)

# Middleware для пользовательских запросов
class UsersMiddleware(ag.BaseMiddleware):
    def __init__(self, llm_answer):
        super().__init__()
        self.llm_answer = llm_answer
    
    async def __call__(
        self,
        handler: Callable[[ag.types.TelegramObject, Dict[str, Any]], Awaitable[Any]],
        event: ag.types.TelegramObject,
        data: Dict[str, Any],
    ) -> Any:
        # Передача ссылки на задачу обработки запросов в каждый пользовательский запрос
        data['llm_answer'] = self.llm_answer
        # Создание статуса "...печатает"
        async with ChatActionSender(bot=data['bot'], chat_id=event.chat.id, action='typing'):
            # Вызов соответствующего обработчика сообщений
            return await handler(event, data)

# Функция настройки обработчиков и маршрутизатора
async def setup(bot, dispatcher, db):
    router.message.middleware(UsersMiddleware(llm.UsersLLMAnswer(db)))
    dispatcher.include_routers(router)

    async def user_query(self, msg):

        # получить ссылку (если она есть) на ютуб видео из запроса
        query, ids = self.get_video_id(msg.text)
        #~ print(query, ids)
        if ids:

            # если запрос содержит ссылки на видео
            if len(ids) > 1:
                await self.a_post(msg, 'Я запомнил видео по последней ссылке, потому что '
                                                        'могу просмотреть только одно видео за раз.')
            else:
                await self.a_post(msg, 'Я запомнил это видео.')

            # получить контекст пользователя по id
            user = self.user_info.setdefault(msg.from_user.id, {})

            # получить ссылку на faiss db и объект langchain
            user['chain'], faiss = await self.create_chain(ids[-1])
            if not query:
                # если в запросе была только ссылка
                # сумаризовать весь ролик
                await self.summarize(msg, ids[-1], faiss)
                await self.a_post(msg, 'Я могу рассказать еще что-нибудь по этому видео.')
                return

        if (user := self.user_info.get(msg.from_user.id)) is None:
            # если контекст пользователя еще не создан
            await self.a_post(msg, 'Я не знаю к какому видео относится запрос.\n'
                                                    'Пришлите пожалуйста ссылку на видео из YouTube.')
        else:
            await self.a_query_answer(msg, user['chain'], query)

    async def create_chain(self, id):

        faiss = await self.create_faiss(id)
        faiss_cell = faiss.as_retriever(k=cf.chunks_by_query_from_faiss) | self.joiner
        retrieval_pass = RunnableParallel( {
            "context": faiss_cell,
            "query": RunnablePassthrough() })
        chain = retrieval_pass | self.prompt | self.llm | StrOutputParser()
        return chain, faiss


    def joiner(self, docs):
        """ Объеденить список чанков из векторного хранилища """
        sorted_docs = list(sorted(docs, key=lambda x: x.metadata['start']))
        return ' '.join(d.page_content for d in sorted_docs)


    async def create_faiss(self, video_id):

        redis_key = f'faiss_db/{video_id}'
        faiss_db = await self.redis.get(redis_key)
        if faiss_db:
            # Загрузить векторное хранилище из файла 
            return FAISS.deserialize_from_bytes( embeddings=self.embeddinger, serialized=faiss_db)
            
        # Загрузить субтитры из видео 
        transcript, language_code = self.load_youtube_transcript(video_id)

        # Создать список документов из чанков субтитров
        docs = self.join_transcript_to_docs(transcript, cf.max_chunk_size)

        # Создать векторное хранилище
        search_index = await FAISS.afrom_documents( docs, self.embeddinger )

        # Сохранить в файл векторное хранилище
        await self.redis.set(redis_key, search_index.serialize_to_bytes())      # serializes the faiss

        return search_index

    async def a_query_answer(self, msg, chain, query):
        try:
            # получить ответ от LLM на запрос полльзователя
            text = await chain.ainvoke(query)
        except GigaChatException:
            # Запись в лог всех ошибок aiogram при отправке в чат
            logger.exception('GigaChatException!')
        else:
            await self.a_post(msg, text)
            return text


    async def a_post(self, msg, text):
        """ Отправить текст в Telegram чат """
        try:
            await msg.answer(text)
        except AiogramError:
            # Запись в лог всех ошибок aiogram при отправке в чат
            logger.exception('AiogramError!')

    async def summarize(self, msg, id, faiss):

        redis_key = f'summarizing/{id}'
        
        # Получить сохраненную сумаризацию из БД
        text = await self.redis.get(redis_key)
        if text:
            for chunk in text.split(b'\n'):
                await self.a_post(msg, chunk.decode())
            return

        # Получить список, сортированых по времени, чанков текста субтитров из Faiss хранилища
        docs = list(sorted(faiss.docstore._dict.values(), key=lambda x: x.metadata['start']))
        join_docs = []

        # Сделать каждый блок для сумаризации из `cf.chunks_by_query_from_faiss` штук Faiss чанков 
        for i in range(0, len(docs), cf.chunks_by_query_from_faiss):
            chunks = '\n'.join(d.page_content for d in docs[i:i+cf.chunks_by_query_from_faiss])
            join_docs.append( chunks )

        # LLM цепочка для сумаризации
        chain = self.prompt_sum | self.llm | StrOutputParser()  
        chunks = []
        for doc in join_docs:
            # Сумаризовать блок текста до 25 слов приблизительно
            text = await self.a_query_answer(msg, chain, {'max_words': '25', 'text': doc})
            if text:
                chunks.append(text.encode())
        if chunks:
            # Объеденить сумаризованные блоки и отправить в БД
            await self.redis.set(redis_key, b'\n'.join(chunks))

    def load_youtube_transcript(self, video_id):

        # Загрузить метаданные субтитров из youtube видео
        transcript_list = yta.YouTubeTranscriptApi.list_transcripts(video_id)
        generated = set()
        manual = set()

        for tra in transcript_list:
            if tra.is_generated:
                # Создать множество доступных языков текста субтитров, созданных youtube сервисом
                generated.add(tra.language_code)
            else:
                # Создать множество доступных языков текста субтитров, созданных автором
                manual.add(tra.language_code)
        if manual:
            language_code = 'en' if 'en' in manual else manual.pop()
        else:
            language_code = 'en' if 'en' in generated else generated.pop()
        
        # Загрузить субтитры из youtube видео
        transcript = yta.YouTubeTranscriptApi.get_transcript(video_id, languages=[language_code]) # , proxies={"https": "https://user:pass@domain:port"} # format used by the requests library
        return transcript, language_code