Вступление

Многие из нас состоят в десятках групповых чатов Telegram, которые заменили нам форумы из нулевых. Это могут быть как чаты по интересам, так и тематические площадки, на которых люди обмениваются опытом в формате вопрос-ответ. Последние особенно важны, так как такие групповые чаты дают возможность получать актуальную информацию из первых рук практически в реальном времени (оставим тему достоверности за кадром для этой статьи): во многих случаях даже самые важные новости могут появляться в таких чатах намного раньше, чем в СМИ, а в ряде случаев информация может быть вообще уникальной, и в других источниках ее не найти. Однако, если чатов становится слишком много или если они генерируют слишком много сообщений, мониторить их становится слишком затратно, а то и невозможно. Как итог, получаем такую картину:

На дворе 2024 год, поэтому решение напрашивается само собой: нужно написать бота, который будет делать это за нас с помощью ChatGPT или другой вашей любимой LLM! Статья рассказывает о тонкостях реализации такого бота на Python с нуля.

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

Идея и архитектура решения

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

Несмотря на кажущуюся простоту, есть несколько моментов, которые все усложняют:

1. Telegram Bot API не позволяет добавлять ботов в чат никому, кроме админов чата. Так как нас не интересует мониторинг собственных чатов, это ставит крест на простой реализации, построенной исключительно на базе Bot API.

2. Вместо этого нам придется использовать Telegram API и наш личный аккаунт Telegram для скраппинга сообщений из чатов. Однако, кроме скраппинга сообщений и их суммаризации нам нужно еще как-то общаться с конечным пользователем-получателем выжимок. Telegram API для этого в теории подходит, но мы не хотим, чтобы для общения использовался наш личный аккаунт. Тут то на помощь к нам и приходит Bot API.

3. Мы не хотим завязываться на конкретную LLM, так как ряд моделей платные, часть не работает из определенных стран, а часть дает слишком плохое качество. Поэтому мы хотим спрятать функцию суммаризации за неким фасадом, который позволит нам менять LLM, не переписывая половину приложения. Для этого мы используем библиотеку LangChain.

4. Те, кто знаком с LLM, знают, что чтобы получить от модели желаемый ответ, нужно немного попотеть с промптом. Наш случай не исключение, поэтому мы должны, во-первых, аккуратно составить промпт для суммаризации чата, а во-вторых, предусмотреть возможность добавления пользовательских промптов отдельно для каждого чата.

С учетом соображений выше вырисовывается четыре основных компонента реализации:

1. Компонент для скраппинга групповых чатов Telegram посредством Telegram API.

2. Компонент для суммаризации истории чата.

3. Компонент для общения с пользователем.

4. Компонент, который свяжет первые три и реализует выполнение суммаризации по расписанию.

Реализация

Скраппинг групповых чатов

Для общения с Telegram API нам понадобятся так называемые api_id и api_hash - это параметры для доступа к API. Получить их можно, следуя документации Telegram. Здесь важно отметить, что получая таким образом доступ к API, мы создаем Telegram App, то есть запрашиваем доступ к Telegram API для работы с ним через приложение, отличное от официального клиента. Это потенциально рискованный шаг потому что:

1. Нарушение ToS Telegram API приведет к блокировке аккаунта.

2. Выданные API ID и API Hash потом нельзя ни удалить, ни изменить. В целом это не должно влиять на безопасность аккаунта, так как даже при наличии этих значений для использования Telegram API нужно по-честному входить в аккаунт, используя номер телефона и OTP.

С учетом вышесказанного, читатель может действовать на свой страх и риск.

Так или иначе, предположим, что api_id и api_hash нами получены, и можно приступать к реализации. Для общения с Telegram API будем использовать библиотеку telethon. Напишем простой класс GroupChatScrapper, который будет выгружать историю сообщений чата, получая на вход ID чата и период, за который нужно выгрузить сообщения:

import atexit
from datetime import datetime, timedelta, timezone
from telethon.sync import TelegramClient
from telethon.tl.types import User, Channel


class GroupChatScrapper:
    def __init__(self, telegram_api_id, telegram_api_hash):
        self.client = TelegramClient("session", api_id=telegram_api_id, api_hash=telegram_api_hash)
        # Первый запуск клиента попросит нас залогиниться в Telegram аккаунт в терминале
        self.client.start()
        # telethon по умолчанию хранит данные сессии в БД на диске, поэтому работу с клиентом
        # нужно завершать корректно, чтобы не сломать БД
        atexit.register(self.client.disconnect)

    @staticmethod
    def get_telegram_user_name(sender):
        # Для выжимки нам нужны имена отправителей сообщений (позже увидим, зачем именно)
        if type(sender) is User:
            if sender.first_name and sender.last_name:
                return sender.first_name + " " + sender.last_name
            elif sender.first_name:
                return sender.first_name
            elif sender.last_name:
                return sender.last_name
            else:
                return "<unknown>"
        else:
            if type(sender) is Channel:
                return sender.title

    @staticmethod
    def get_datetime_from(lookback_period):
        return (datetime.utcnow() - timedelta(seconds=lookback_period)).replace(tzinfo=timezone.utc)

    def get_message_history(self, chat_id, lookback_period):
        history = []
        datetime_from = self.get_datetime_from(lookback_period)
        for message in self.client.iter_messages(chat_id):
            if message.date < datetime_from:
                break
            if not message.text:
                # Пропускаем не-текстовые сообщения
                continue
            sender = message.get_sender()
            data = {
                "id": message.id,
                "datetime": str(message.date),
                "text": message.text,
                "sender_user_name": self.get_telegram_user_name(sender),
                "sender_user_id": sender.id,
                "is_reply": message.is_reply
            }
            if message.is_reply:
                data["reply_to_message_id"] = message.reply_to.reply_to_msg_id
            history.append(data)
        return list(reversed(history))

Итак, у нас есть класс, который может логиниться в аккаунт Telegram и выгружать историю сообщений c интересующими нас полями. Вот пример индивидуального сообщения из выгруженной истории:

{
  "id": 12345,
  "datetime": "2024-03-29 20:34:47+00:00",
  "text": "Как мне пропатчить KDE2 под FreeBSD?",
  "sender_user_name": "Sashok",
  "sender_user_id": 12345,
  "is_reply": True,
  "reply_to_message_id": 12345
}

Здесь есть бóльшая часть полей, которые нужны для суммаризации: текст сообщения, имя отправителя и ID, нужный для связи цепочки ответов на сообщения. Однако, наш класс имеет один серьезный недостаток: не-текстовые сообщения и вложения к сообщениям не выгружаются. Это значит, что изображения, видео и голосовые сообщения не попадают во входные данные для суммаризации. Тем не менее, для простоты оставим как есть - все же большая часть контекста в чатах Telegram передается текстом.

Суммаризация истории сообщений

Как мы обсуждали выше, мы не хотим жестко привязываться к конкретной LLM, поэтому используем библиотеку LangChain в качестве фасада. Здесь нужно оговориться, что "гениальная" архитектура LangChain все равно допускает утечку реализации в пользовательский API, но большую часть кода все равно можно будет переиспольовать.

Для простоты будем использовать модель gpt-4-turbo-preview от OpenAI. Для этого идем на platform.openai.com, заносим денег и получаем API ключ. Далее реализуем логику суммаризации в классе Summarizer:

from langchain.prompts import (
    ChatPromptTemplate,
    HumanMessagePromptTemplate,
    MessagesPlaceholder,
)
from langchain_core.messages import SystemMessage
from langchain_openai import ChatOpenAI
from langchain.chains import LLMChain
from langchain.memory import ConversationBufferMemory


class Summarizer:
    def __init__(self, openai_api_key):
        self.openai_api_key = openai_api_key
        self.openai_model = "gpt-4-turbo-preview"

        # Конструкция промпта для хранения истории запросов к LLM (UX чат-бота),
        # которые делает пользователь для уточнениня выжимки из истории чата
        self.persistent_prompt = ChatPromptTemplate.from_messages(
            [
                SystemMessage(
                    content="You are a chatbot having a conversation with a human."),
                MessagesPlaceholder(variable_name="chat_history"),
                HumanMessagePromptTemplate.from_template("{human_input}")
            ]
        )

    # Подаем на вход историю чата Telegram в формате JSON и промпт-запрос к LLM
    def summarize(self, text_to_summarize, summarization_prompt):
        llm = ChatOpenAI(model_name=self.openai_model, openai_api_key=self.openai_api_key)
        # Здесь будем хранить историю запросов/промптов
        memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True)
        chat_llm_chain = LLMChain(
            llm=llm,
            prompt=self.persistent_prompt,
            verbose=False,
            memory=memory,
        )
        init_prompt = summarization_prompt.format(text_to_summarize=text_to_summarize)
        return chat_llm_chain.predict(human_input=init_prompt), chat_llm_chain

    @staticmethod
    def validate_summarization_prompt(summarization_prompt):
        if not "{text_to_summarize}" in summarization_prompt:
            raise RuntimeError("Summarization prompt should include \"{ text_to_summarize }\"")

Описанная реализация использует API OpenAI, доступ к которому затруднен из России. Прошу понять, простить, и иметь в виду, что реализацию достаточно легко переключить на тот же GigaChat:

import os
from langchain_community.chat_models import GigaChat

...

class Summarizer:
    def __init__(self, gigachat_api_key):
        self.gigachat_api_key = gigachat_api_key
        ...

    def summarize(self, text_to_summarize, summarization_prompt):
        llm = GigaChat(verify_ssl_certs=False, scope="GIGACHAT_API_PERS", credentials=self.gigachat_api_key)
        ...

Промпт для суммаризации

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

• Расскажем LLM про контекст - мы хотим суммаризировать историю чата Telegram в формате JSON.

• Попросим выделить 5 основным тем обсуждения.

• Попросим снабдить каждую из обозначенных тем примерами 2-3 наиболее показательных диалогов из истории чата на данную тему. Это нужно, чтобы добавить конкретики в описание основных тем обсуждения.

• Чтобы мы могли после прочтения выжимки найти конкретные сообщения в чате, попросим для каждого примера диалога давать ключевые слова для поиска.

• Мы пишем промпт для GPT-4, для которой "родной" язык - английский, поэтому и промпт на всякий случай напишем на английском, отметив, что ответ мы хотим получить на языке общения в поданной на вход истории сообщений чата.

• По заветам промпт-инжиниринга поднапряжемся и напишем подробный пример ожидаемого ответа.

Таким образом, получаем что-то вроде (прим.: промпт предназначен для суммаризации чата по вопросам налогов и банкинга для русскоязычных экспатов в Испании, темы и имена пользователей выдуманные):

The JSON document below is a message history from a Telegram group chat.
I need you to summarize this chat history and yield 5 primary conversation topics.
Each conversation topic mentioned should be accompanied by one-sentence summaries of 2-3 most representative dialogs (not single messages) from the conversation on the given topic including usernames.
For each dialog summary provide the exact keywords with which the message can be found in the history using text search.

IMPORTANT: The output should be provided in the language which prevails in the messages text.

Here's an example of desired output in Russian language (follow the exact structure):

1. <b>Изменения в политике открытия счетов в Испании для россиян по паспорту гражданина РФ без получения ВНЖ. Обсуждаются новые ограничения, введенные в начале 2024</b>.
Примеры сообщений:
- <b>Bolzhedor рассказывает</b> о неудачной попытке открытия счета по паспорту РФ в банке Caixa. Ключевые слова: "<i>завернули с паспортом</i>", "<i>больше никому не открывают :(</i>".
- <b>Александр Сергеевич</b> отмечает, что единственный банк, до сих пор открывающий счета россиянам по паспорту - это BBVA. Ключевые слова: "<i>BBVA пока разрешает</i>", "<i>главное дружить с хестором</i>".

2. <b>Изменения в политике налогообложения России и Испании в 2024 году</b>.
Примеры сообщений:
- <b>Себастьян Перейро</b> и <b>Max</b> обсуждают изменения в налоговом законодательстве и влияние налогового резидентства на обязательства. Ключевые слова: "<i>нерезидентам сейчас хуже всего</i>", "<i>кто попался на непредоставлении?</i>", "<i>зачем вообще об этом сообщать/<i>".
- <b>Akakij M</b> и <b>Олег</b> делятся опытом и советами по вопросам налогообложения и требованиям налоговых органов. Ключевые слова: "<i>Будут спрашивать - скажете</i>", "<i>у меня пока ничего не просили</i>".

3. <b>Судебные приставы и исполнение налоговых требований: пользователи делились опытом взаимодействия с судебными приставами и налоговыми органами, включая случаи неправомерного списания средств</b>.
Примеры сообщений:
- <b>Маша К</b> рассказывает о своем опыте с неправомерным списанием средств и последующим взысканием через суд. Ключевые слова: "<i>по судам затаскают</i>".
- <b>Любитель Бокса</b> упоминает о списании штрафов с нескольких счетов одновременно. Ключевые слова: "<i>уж не знаю как, но нашли</i>".

4. <b>Вопросы по открытию и пополнению счета для получения студенческой визы</b>.
Примеры сообщений:
- <b>Родион Раскольников</b> ищет информацию о том, как показать на счете 1337€ для студенческой визы, учитывая ограничения на пополнение счета в Nickel. Ключевые слова: "<i>студенческая виза</i>", "<i>leet</i>", "<i>1337€</i>".
- <b>Kusswurm</b> предлагает пополнение через Bank для обхода лимитов Nickel. Ключевые слова: "<i>пополнение через Bank</i>", "<i>обход лимитов Nickel</i>".

5. <b>Обсуждение возможности использования банковских услуг для нерезидентов и резидентов с TIE</b>.
Примеры сообщений:
- <b>Александр</b> спрашивает о переводе средств из РФ в Испанию, будучи нерезидентом без резиденции. Ключевые слова: "<i>вывод средств</i>", "<i>РФ в Испанию</i>", "<i>нерезидент</i>".
- <b>Жулик Обманщик</b> предлагает привезти наличные, а также упоминает о наличии людей, заинтересованных в обмене рублей на евро. Ключевые слова: "<i>привезти наличные</i>", "<i>рубли на евро</i>".

Here's the JSON document:
{text_to_summarize}

Обратите внимание: в конце идет шаблонное поле text_to_summarize для вставки истории сообщений в промпт. Для проверки его наличия мы написали метод Summarizer.validate_summarization_prompt. Кроме того, мы используем в примере выжимки HTML теги: таким образом LLM будет выдавать отформатированный ответ, который Telegram сможет корректно отобразить.

Общение с пользователем

Теперь нам нужно реализовать рассылку выжимки истории чата и функциональность чат-бота для ответов на вопросы по ней.

Первым шагом будет создание бота через Telegram Bot Father согласно официальной инструкции. Как только токен для бота получен, приступим к реализации логики общения с пользователем, используя библиотеку telebot. Реализацию определят два важных аспекта:

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

2. Поскольку бот будет одновременно общаться с несколькими пользователями (бот регулярно присылает выжимку списку пользователей, который мы задали в конфигурации заранее) и на тему нескольких чатов (список чатов для суммаризации также задается в конфигурации), нам нужно реализовать хранение состояния чата в зависимости от пользователя и выбранного чата, и переключение между ними.

С учетом таких вводных, примерно так будет выглядеть реализация класса EnvoyBot:

import threading
import time
import telebot


class EnvoyBot:
    def __init__(self, telegram_bot_auth_token, telegram_summary_receivers, allowed_contexts, chat_callback):
        # Предопределнный список пользователей, которые могут получать выжимки
        self.telegram_summary_receivers = telegram_summary_receivers
        
        # Словарь верифицировавшихся пользователей (пуст при запуске приложения)
        # Здесь ключ - имя (никнейм) пользвателя, значение - chat_id
        self.verified_receivers = dict()

        # Набор команд для переключения между контекстами (чатами для суммаризации)
        self.allowed_commands = ["/" + c for c in allowed_contexts]
        
        # Текущие контексты пользователей
        self.current_user_contexts = dict()

        # Callback для обработки уточняющих запросов пользователей
        # Сигнатура: chat_callback(input_message_text, sender, context_name, send_message_func)
        self.chat_callback = chat_callback

        # Запуск бота в фоновом потоке, чтобы не блокировать текущий поток скрипта
        self.bot = telebot.TeleBot(telegram_bot_auth_token)
        self.bot.set_update_listener(self.__handle_messages)
        self.bot_thread = threading.Thread(target=self.bot.infinity_polling)
        self.bot_thread.start()

    # Метод для отправки выжимки заданному пользователю
    def send_summary(self, username, text, chat_id):
        # Мы не можем отправить сообщение пользователю, который не верифицировался, так как
        # не знаем его chat_id
        if not username in self.verified_receivers:
            return
        self.bot.send_message(self.verified_receivers[username], text, parse_mode="HTML")
        # Контекст общения с пользователем всегда стартует с последней отправлленой выжимки
        self.set_current_user_context(username, chat_id)

    # Метод для выставления статуса "печатает" в момент, когда бот составляет сообщение
    def set_typing_status(self, users, predicate):
        # self.bot.send_chat_action(user, "typing") выставляет статус на <= 5 секунд или до тех пор, пока не будет отправлено сообщение
        # Это ограничение Bot API, поэтому костыльнем, чтобы можно было выставлять статус на заданное время
        def f():
            while predicate():
                for u in users:
                    if u in self.verified_receivers:
                        self.bot.send_chat_action(self.verified_receivers[u], "typing")
                time.sleep(5)

        threading.Thread(target=f).start()

    def set_current_user_context(self, username, context):
        self.current_user_contexts[username] = context

    # Главный метод, который обрабатывает входящие сообщеня
    def __handle_messages(self, messages):
        for message in messages:
            if not message.text:
                return
            sender = message.from_user.username
            
            # Сесурити: игнорируем сообщения от пользователей, которых мы не указали в конфиге
            if not sender or not sender in self.telegram_summary_receivers:
                return

            # Обработка команд
            if message.text.startswith("/"):
                if message.text == "/verify":
                    # Самостоятельная верификация пользователя (регистрация его chat_id для последущего общения)
                    self.verified_receivers[sender] = message.chat.id
                    self.bot.send_message(message.chat.id, "You are now verified and will receive generated summaries")
                    return
                else:
                    # Команды для переключения между контекстами (чатами)
                    if not message.text in self.allowed_commands:
                        self.bot.send_message(message.chat.id,
                                              "Invalid command, valid commands are: " + ", ".join(
                                                  self.allowed_commands))
                        return
                    self.set_current_user_context(sender, message.text[1:])
                    self.bot.send_message(message.chat.id, f"Switched context to {self.current_user_contexts[sender]}")
            else:
                # Обработка уточняющих запросов пользователя
                if not sender in self.current_user_contexts:
                    self.bot.send_message(message.chat.id,
                                          "Select context first, valid commands are: " + ", ".join(
                                              self.allowed_commands))
                    return
                self.chat_callback(message.text, sender, self.current_user_contexts[sender],
                                   lambda x: self.bot.send_message(message.chat.id, x))

Итого, мы получаем бота, который:

• Может отправлять пользователю выжимку, храня при этом контекст диалога. Чтобы бот мог взаимодействовать с пользователем, пользователь должен сам инициировать диалог с ботом, отправив команду /verify.

• Может переключать контексты для каждого пользователя по команде. Для этого пользователь может отправить команду /<chat_name>, где chat_name - ID чата (t.me/<chat_name>), чтобы бот переключил контекст диалога на выбранный чат.

Описание выглядит немного запутанно, но пазл сложится, когда мы увидим бота в действии.

Склеваем все вместе

Осталась заключительная часть: склеить все компоненты вместе, добавить чтение конфигурации и запуск суммаризации по расписанию.

Начнем с чтения конфигурации, и для этого используем библиотеку pydantic:

from typing import List, Union
from pydantic import BaseModel, Field
import argparse
from summarization import Summarizer


class SummarizationConfig(BaseModel):
    id: Union[str, int]
    lookback_period_seconds: int
    summarization_prompt_path: str


class AppConfig(BaseModel):
    telegram_api_id: int
    telegram_api_hash: str
    telegram_bot_auth_token: str
    openai_api_key: str
    chats_to_summarize: List[SummarizationConfig]
    telegram_summary_receivers: List[str]


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("path_to_config")
    args = parser.parse_args()

    app_config = AppConfig.parse_file(args.path_to_config)

    for c in app_config.chats_to_summarize:
        with open(c.summarization_prompt_path, "r") as f:
            Summarizer.validate_summarization_prompt(f.read())

С таким форматом файл конфигурации для двух чатов (используем один и тот же промпт для обоих) и одним получателем выжимок каждые 24 часа будет выглядеть так:

{
    "telegram_app_id": <id>, // Telegram API ID, который мы получили в самом первом шаге
    "telegram_app_hash": "<hash>", // Telegram API Hash
    "openai_api_key": "<key>", // Ключ для OpenAI API
    "telegram_bot_auth_token": "<token>", // Токен дл бота
    "chats_to_summarize": [
        {
            "id": "chat_name_1",
            "lookback_period_seconds": 86400, // 24 часа
            "summarization_prompt_path": "summarization_prompt.txt"
        },
        {
            "id": "chat_name_2",
            "lookback_period_seconds": 86400, // 24 часа
            "summarization_prompt_path": "summarization_prompt.txt"
        }
    ],
    "telegram_summary_receivers": [
        "my_telegram_nickname"
    ]
}

Далее добавим запуск суммаризации по расписанию с помощью библиотеки schedule:

import argparse
import threading
from collections import defaultdict
import schedule
import time
import json

from communication import GroupChatScrapper, EnvoyBot


if __name__ == "__main__":
    ...
  
    # Здесь мы будем хранить контекст диалога с LLM (вложенный словарь chat_name -> username -> context)
    llm_contexts = defaultdict(dict)
    # Вспомним, что бот работает в отдельном потоке, и добавим синхронизацию
    llm_contexts_lock = threading.Lock()


    # Callback для EnvoyBot, в котором мы будем обрабатывать уточняющие вопросы пользователя
    def chat_callback(input_message_text, sender, context_name, send_message_func):
        with llm_contexts_lock:
            envoy_bot.set_typing_status([sender], llm_contexts_lock.locked)
            if not context_name in llm_contexts or not sender in llm_contexts[context_name]:
                send_message_func(f"No context is available for {context_name} yet")
                return
            response = llm_contexts[context_name][sender].predict(human_input=input_message_text)
            send_message_func(response)


    summarizer = Summarizer(app_config.openai_api_key)
    group_chat_scrapper = GroupChatScrapper(app_config.telegram_api_id, app_config.telegram_api_hash)
    envoy_bot = EnvoyBot(
        app_config.telegram_bot_auth_token,
        app_config.telegram_summary_receivers,
        [c.id for c in app_config.chats_to_summarize],
        chat_callback
    )


    # Функция для суммаризация чата
    def summarization_job(chat_cfg, summarization_prompt, summary_receivers):
        with llm_contexts_lock:
            envoy_bot.set_typing_status(summary_receivers, llm_contexts_lock.locked)

            # Выгрузим историю сообщений за период chat_cfg.lookback_period_seconds
            messages, chat_title= group_chat_scrapper.get_message_history(chat_cfg.id, chat_cfg.lookback_period_seconds)
            serialized_messages = json.dumps({"messages": messages}, ensure_ascii=False)

            # Суммаризируем историю сообщений
            summary, context = summarizer.summarize(serialized_messages, summarization_prompt)

            # Отправим выжимки указанным получателям и обновим контекст LLM, чтобы относить уточняющие вопросы
            # к последней полученной выжимке
            for u in summary_receivers:
                llm_contexts[chat_cfg.id][u] = context
                chat_lookback_period_hours = int(chat_cfg.lookback_period_seconds / 60 / 60)
                envoy_bot.send_summary(
                    u,
                    f"Summary for <b>{chat_cfg.id}</b> for the last {chat_lookback_period_hours} hours:\n\n{summary}",
                    chat_cfg.id
                )


    # Добавим джобы для суммаризации в расписание
    for chat_config in app_config.chats_to_summarize:
        with open(chat_config.summarization_prompt_path, "r") as f:
            chat_summarization_prompt = f.read()
        schedule.every(chat_config.lookback_period_seconds).seconds.do(
            job_func=summarization_job,
            chat_cfg=chat_config,
            summarization_prompt=chat_summarization_prompt,
            summary_receivers=app_config.telegram_summary_receivers
        )

    # Запустим первую суммаризацию сразу же
    schedule.run_all()

    # Цикл для суммаризации по расписанию
    while True:
        schedule.run_pending()
        time.sleep(1)

Вот и все! Приложение готово к работе, давайте теперь посмотрим на него в действии.

Работающий код приложения целиком, инструкции по установке и запуску и примеры доступны в этом GitHub репозитории.

Демо

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

Вместо заключения

Хоть наша реализация и решает поставленную задачу, но все же она может быть улучшена в ряде аспектов:

1. Контекст пользователей и их chat_id хранятся в памяти, а значит при перезапуске приложения эти данные будут потеряны, что не очень удобно. Вместо этого можно реализовать хранение этой информации в БД.

2. Механзим самостоятельной верификации пользователей в целом не очень удобен. Скорее всего есть способ реализовать это менее костыльно.

3. Сейчас управление набором чатов, промптами и пользователями происходит через конфиг файл. Удобнее было бы делать это на лету, через того же бота, реализовав операции CRUD.

4. Текущая реализация ограничена анализом текстовых сообщений. Поддержка фото, видео, файлов и голосовых сообщений в суммаризации улучшит ее качество, но эта задача уже менее тривиальна.