We automate the search for valuable information in Telegram group chats using LLM
Introduction
Many of us are members of dozens of Telegram group chats, which have replaced forums from the 2000s. These can be either chats based on interests, or thematic platforms where people exchange experiences in a question-and-answer format. The latter are especially important, since such group chats make it possible to receive up-to-date information first-hand in almost real time (Let's leave the topic of reliability behind the scenes for this article.): in many cases, even the most important news can appear in such chats much earlier than in the media, and in some cases the information may be completely unique and cannot be found in other sources. However, if there are too many chats or if they generate too many messages, monitoring them becomes too expensive, if not impossible. As a result, we get the following picture:
It's 2024, so the solution comes naturally: we need to write a bot that will do this for us using ChatGPT or your other favorite LLM! The article talks about the intricacies of implementing such a bot in Python from scratch.
While writing the bot and the text of the article, a similar publication was published on the Selectel blog called “The simplest guide to creating a bot for analyzing messages in Telegram”. Despite the similarity of approaches, both solutions solve the problem in different ways, so it was decided to publish this article, despite an earlier publication on the Selectel blog.
Idea and architecture of the solution
We want to make a Telegram bot that, with a given period, will summarize given Telegram group chats and send a summary to specified users for selected dates. We want to receive the extract in a predetermined format, and the user should be able to find messages in the chat history that correspond to the provisions of the extract. In addition, we want to receive answers to our clarifying questions about this extract, that is, in addition to sending out summarization results, we want the bot to also be a chat bot.
Despite its apparent simplicity, there are several points that complicate everything:
Telegram Bot API does not allow anyone except chat admins to add bots to the chat. Since we are not interested in monitoring our own chats, this puts an end to a simple implementation built solely on the Bot API.
Instead we will have to use Telegram API and our personal Telegram account for scraping messages from chats. However, in addition to scraping messages and summarizing them, we also need to somehow communicate with the end user who is the recipient of the scraps. The Telegram API is suitable for this in theory, but we don’t want our personal account to be used for communication. This is where the Bot API comes to our aid.
We don’t want to be tied to a specific LLM, since a number of models are paid, some do not work from certain countries, and some provide too poor quality. Therefore, we want to hide the summarization function behind some kind of facade that will allow us to change the LLM without rewriting half of the application. For this we use the library LangChain.
Those who are familiar with LLM know that in order to get the desired answer from the model, you need to sweat a little with the prompt. Our case is no exception, so we must, firstly, carefully compose a prompt to summarize the chat, and secondly, provide for the possibility of adding custom prompts separately for each chat.
Taking into account the above considerations, four main components of the implementation emerge:
Component for scraping Telegram group chats using the Telegram API.
Component for summarizing chat history.
Component for communicating with the user.
A component that connects the first three and implements the execution of summarization on a schedule.
Implementation
Scraping group chats
To communicate with the Telegram API we will need the so-called api_id And api_hash – these are the parameters for accessing the API. You can get them by following Telegram documentation. It is important to note here that by gaining access to the API in this way, we are creating a Telegram App, that is, we are requesting access to the Telegram API to work with it through an application other than the official client. This is a potentially risky move because:
Violation ToS Telegram API will lead to account blocking.
The issued API ID and API Hash cannot then be deleted or changed. In general, this should not affect the security of your account, since even with these values, to use the Telegram API you need to honestly log into your account using your phone number and OTP.
Subject to the foregoing, the reader proceeds at his own risk.
Anyway, let's assume that api_id And api_hash We have received them and we can begin implementation. To communicate with the Telegram API we will use the library telethon. Let's write a simple class GroupChatScrapperwhich will download the history of chat messages, receiving as input the chat ID and the period for which messages need to be downloaded:
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))So, we have a class that can log into a Telegram account and upload message history with the fields of interest to us. Here is an example of an individual message from an uploaded history:
{
"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
}Here you will find most of the fields that are needed for summarization: the text of the message, the sender's name and the ID needed to link the chain of responses to messages. However, our class has one serious drawback: non-text messages and message attachments are not unloaded. This means that images, videos and voice messages are not included in the input data for summarization. However, for the sake of simplicity, let's leave it as it is – still, most of the context in Telegram chats is transmitted by text.
Summarizing message history
As we discussed above, we don't want to be rigidly tied to a specific LLM, so we use the library LangChain as a facade. Here it is necessary to make a reservation that the “brilliant” architecture of LangChain still allows the implementation to leak into the user API, but most of the code can still be reused.
For simplicity, we will use the model gpt-4-turbo-preview from OpenAI. To do this we go to platform.openai.com, deposit money and receive an API key. Next, we implement the summarization logic in the class 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 }\"")The described implementation uses the OpenAI API, which is difficult to access from Russia. Please understand, forgive, and keep in mind that the implementation is quite easy to switch to the same 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)
...Prompt for summarization
So, the code for summarization has been written, all that remains is to compose the prompt. As stated earlier, we want to receive a summary of the chat history in a predefined format, and the user should be able to find messages in the chat history that match the provisions of the summary. To do this, in the prompt:
Let's tell LLM about the context – we want to summarize the Telegram chat history in JSON format.
We ask you to highlight 5 main topics of discussion.
We ask you to provide each of the identified topics with examples of 2-3 of the most indicative dialogues from the chat history on this topic. This is necessary to add specifics to the description of the main topics of discussion.
So that after reading the extract we can find specific messages in the chat, we will ask you to provide search keywords for each example of dialogue.
We are writing a prompt for GPT-4, for which the “native” language is English, so, just in case, we will write the prompt in English, noting that we want to receive an answer in the language of communication in the chat message history submitted as input.
According to the precepts of prompt engineering, let’s push ourselves and write a detailed example of the expected answer.
So we get something like (note: the prompt is intended to summarize a chat on tax and banking issues for Russian-speaking expats in Spain, topics and user names are fictitious):
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}
Please note: there is a template field at the end text_to_summarize to insert message history into prompt. To check its presence, we wrote a method Summarizer.validate_summarization_prompt. In addition, we use HTML tags in the example squeeze: this way LLM will produce a formatted response that Telegram can display correctly.
Communication with the user
Now we need to implement the distribution of a summary of the chat history and the chatbot functionality for answering questions about it.
The first step is to create a bot via Telegram Bot Father according to official instructions. As soon as the token for the bot is received, we will begin to implement the logic of communication with the user using the library telebot. Implementation will be determined by two important aspects:
The Telegram bot cannot independently initiate a dialogue with the user. At the API level, this means that messages can only be sent using the parameter
chat_id, which cannot be determined by the user's nickname. Thus, we need to provide a mechanism for self-verification of the user by sending the appropriate command to the bot.Since the bot will simultaneously communicate with several users (the bot regularly sends a summary to the list of users that we set in the configuration in advance) and on the topic of several chats (the list of chats for summarization is also set in the configuration), we need to implement storage of the chat state depending on the user and selected chat, and switch between them.
Taking into account these inputs, the implementation of the class will look something like this: 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))In total, we get a bot that:
Can send a summary to the user while storing the context of the dialogue. In order for the bot to interact with the user, the user must initiate a dialogue with the bot by sending a command
/verify.Can switch contexts for each user on command. To do this, the user can send the command
/<chat_name>Wherechat_name– Chat ID (t.me/), so that the bot switches the conversation context to the selected chat.
The description looks a little confusing, but the puzzle will come together once we see the bot in action.
Let's glue everything together
The final part remains: gluing all the components together, adding configuration reading and running summarization on a schedule.
Let's start by reading the configuration, and for this we use the library 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())With this format, the configuration file for two chats (we use the same prompt for both) and one recipient of extracts every 24 hours will look like this:
{
"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"
]
}Next, let’s add a scheduled run for summarization using the library 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)That's all! The application is ready to go, let's now see it in action.
The entire working code of the application, installation and launch instructions and examples are available in this GitHub repository.
Demo
The video below shows an example of a bot configured to summarize two Russian-language Telegram group chats every 24 hours: one dedicated to tax and banking issues for Russian-speaking expats in Spain, the other to speech recognition. Usernames and chat names are hidden for privacy reasons.
Instead of a conclusion
Although our implementation solves the problem, it can still be improved in a number of aspects:
The user context and their chat_id are stored in memory, which means that when the application is restarted, this data will be lost, which is not very convenient. Instead, you can implement storing this information in a database.
The mechanism for self-verification of users in general is not very convenient. Most likely there is a way to implement this in a less crutch.
Currently, a set of chats, prompts and users are managed through a config file. It would be more convenient to do this on the fly, through the same bot, implementing CRUD operations.
The current implementation is limited to text message analysis. Supporting photos, videos, files and voice messages in summarization will improve its quality, but this task is less trivial.
