Семантический поиск в Telegram: разбираем кейс с Habr и строим свой аналог

Зачем искать в Telegram семантически? Потому что обычный поиск там сломан

Помните тот кейс на Хабре? Автор уволился, заскучал и решил построить поиск по куче своих Telegram-каналов. Встроенный поиск в мессенджере — это катастрофа. Он ищет по точным совпадениям слов, игнорируя смысл. Запросите "как настроить VPN" и получите ноль результатов, хотя в канале есть десять постов про WireGuard и OpenVPN. Проблема очевидна.

Что было в том кейсе? Краткий разбор на костях

Автор брал ссылки на каналы, логинился через MTProto, выкачивал историю сообщений. Потом текст чистил, разбивал на куски (чанковал), преобразовывал в векторы с помощью какой-то модели от Sentence Transformers и складывал в FAISS — локальную векторную базу от Facebook. Поисковый бот принимал запрос, превращал его в вектор, искал ближайшие соседи в индексе и выдавал ссылки на сообщения.

Работало. Но сейчас можно сделать лучше, надежнее и с учетом новых ограничений. Давайте не просто повторим, а сделаем улучшенную версию.

Современный стек: что взять в 2026 году вместо устаревших деталей

• Для работы с Telegram API: Telethon все еще актуален, но Pyrogram (версия 2.0+) часто оказывается проще и имеет более чистый асинхронный API. Выбор за вами.
• Для эмбеддингов (векторизации текста): Модели из семейства text-embedding-3-large от OpenAI задают высокую планку качества. Но если нужна полная локальность и бесплатность — берите BAAI/bge-m3 или Snowflake/snowflake-arctic-embed-l. Они показывают SOTA результаты в открытых бенчмарках на начало 2026.
• Для векторной базы: FAISS — быстро, но только для индекса. Для продакшена с персистентностью и метаданными смотрим в сторону Qdrant 1.9.x или Weaviate 1.24+. Они умеют работать и в памяти, и на диске, и даже как облачный сервис.
• Для чанкинга (разбиения текста): Не режьте просто по символам. Используйте семантическое разбиение. Библиотека LangChain TextSplitter умеет делить по токенам и с учетом разделителей, но для русского лучше написать свой сплиттер, ориентируясь на абзацы и точки.

1Шаг 1: Получение доступа и настройка окружения

Первое — получаем api_id и api_hash на my.telegram.org. Это стандартно. Создаем виртуальное окружение Python 3.11+ и ставим зависимости.

pip install pyrogram==2.0.5 sentence-transformers qdrant-client

Pyrogram выбран для примера — его синтаксис проще для новичков.

2Шаг 2: Скачивание истории канала

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

from pyrogram import Client
import asyncio
import json

async def main():
    client = Client("my_session", api_id=YOUR_API_ID, api_hash=YOUR_API_HASH)
    await client.start()
    
    channel = await client.get_chat("@channel_username")
    messages = []
    
    async for message in client.get_chat_history(channel.id, limit=1000):
        if message.text:
            messages.append({
                "id": message.id,
                "date": message.date.isoformat(),
                "text": message.text,
                "link": f"https://t.me/{channel.username}/{message.id}"
            })
    
    with open("messages.json", "w", encoding="utf-8") as f:
        json.dump(messages, f, ensure_ascii=False, indent=2)
    
    await client.stop()

if __name__ == "__main__":
    asyncio.run(main())

3Шаг 3: Подготовка текста и чанкинг

Длинные посты нужно резать. Простое разбиение по 500 токенов может разрушить смысл. Лучшая эвристика для Telegram: режем по двойному переносу строки (абзацы), а если абзац слишком длинный — делим по предложениям.

from typing import List
import re

def smart_chunker(text: str, max_tokens: int = 500) -> List[str]:
    """Грубый, но работающий сплиттер для русского текста."""
    # Сначала делим на абзацы
    paragraphs = [p.strip() for p in text.split('\n\n') if p.strip()]
    chunks = []
    current_chunk = []
    current_length = 0
    
    for para in paragraphs:
        # Оцениваем длину абзаца в словах (грубая оценка токенов)
        para_len = len(para.split())
        if current_length + para_len > max_tokens and current_chunk:
            chunks.append(' '.join(current_chunk))
            current_chunk = [para]
            current_length = para_len
        else:
            current_chunk.append(para)
            current_length += para_len
    
    if current_chunk:
        chunks.append(' '.join(current_chunk))
    return chunks

4Шаг 4: Векторизация — сердце системы

Здесь выбираем модель. Для локального запуска BAAI/bge-m3 — отличный баланс. Устанавливаем через sentence-transformers.

from sentence_transformers import SentenceTransformer
import torch

# Убедитесь, что у вас PyTorch 2.3+ и CUDA 12.1 если есть GPU
device = 'cuda' if torch.cuda.is_available() else 'cpu'
model = SentenceTransformer('BAAI/bge-m3', device=device)

# Кодируем один чанк
embedding = model.encode("Привет, мир!", normalize_embeddings=True)
print(f"Размерность вектора: {embedding.shape}")  # Должно быть 1024

Нормализация (normalize_embeddings=True) критически важна для косинусного сходства, которое использует большинство векторных баз.

5Шаг 5: Индексация в Qdrant

Поднимаем локальный Qdrant через Docker (последняя стабильная версия на 27.03.2026 — 1.9.2).

docker pull qdrant/qdrant:1.9.2
docker run -p 6333:6333 -p 6334:6334 \
    -v $(pwd)/qdrant_storage:/qdrant/storage \
    qdrant/qdrant:1.9.2

Теперь заполняем коллекцию.

from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct
import uuid

client = QdrantClient(host="localhost", port=6333)
collection_name = "telegram_messages"

# Создаем коллекцию, если ее нет
if not client.collection_exists(collection_name):
    client.create_collection(
        collection_name=collection_name,
        vectors_config=VectorParams(size=1024, distance=Distance.COSINE)
    )

points = []
for msg in messages:  # messages - наши данные из JSON
    chunks = smart_chunker(msg['text'])
    for i, chunk in enumerate(chunks):
        vector = model.encode(chunk, normalize_embeddings=True).tolist()
        point_id = str(uuid.uuid4())
        points.append(
            PointStruct(
                id=point_id,
                vector=vector,
                payload={
                    "original_text": chunk,
                    "message_id": msg['id'],
                    "link": msg['link'],
                    "date": msg['date'],
                    "chunk_index": i
                }
            )
        )
# Пакетная загрузка по 100 векторов
for i in range(0, len(points), 100):
    client.upsert(collection_name=collection_name, points=points[i:i+100])

6Шаг 6: Telegram-бот для поиска

Создаем бота через @BotFather и используем библиотеку python-telegram-bot версии 21.x. Бот будет принимать запрос, векторизовать его и искать в Qdrant.

from telegram import Update
from telegram.ext import ApplicationBuilder, CommandHandler, MessageHandler, filters, ContextTypes
from qdrant_client import QdrantClient
from sentence_transformers import SentenceTransformer
import asyncio

# Инициализация
qdrant_client = QdrantClient(host="localhost", port=6333)
model = SentenceTransformer('BAAI/bge-m3')

async def search_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    query = update.message.text
    query_vector = model.encode(query, normalize_embeddings=True).tolist()
    
    search_result = qdrant_client.search(
        collection_name="telegram_messages",
        query_vector=query_vector,
        limit=5
    )
    
    response = "Результаты поиска:\n\n"
    for hit in search_result:
        payload = hit.payload
        response += f"• {payload['original_text'][:200]}...\n"
        response += f"  Ссылка: {payload['link']}\n\n"
    
    await update.message.reply_text(response)

app = ApplicationBuilder().token("YOUR_BOT_TOKEN").build()
app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, search_command))
app.run_polling()

Это минимальный рабочий вариант. В реальности нужно добавить обработку ошибок, кэширование, возможно, пересказ результатов через LLM для красоты — но это уже агентный RAG.

Где споткнуться? Ошибки, которые почти гарантированы

Частые вопросы от тех, кто уже начал

Можно ли искать по картинкам или документам? Можно, но сложнее. Для картинок нужны vision-модели (например, CLIP) для получения эмбеддингов изображений. Для PDF/docx — извлекать текст библиотекой типа pypdf или python-docx, затем стандартный пайплайн. Это тема для отдельной статьи, но наш гайд по скрапингу и векторизации покрывает базовые принципы.

Как масштабировать на сотни каналов? Нужна распределенная очередь задач (Celery или RQ), чтобы парсить каналы параллельно с разными аккаунтами. И векторную базу перенести в кластерный режим Qdrant или в облако.

А если Telegram изменит API? Они меняют его постоянно. Поэтому не завязывайтесь на одну библиотеку. Заключите логику работы с API в отдельный модуль-адаптер, чтобы в случае чего заменить реализацию.

Есть ли готовые аналоги? Есть коммерческие сервисы вроде Telemetr.io (партнерская ссылка), но они не дадут вам полного контроля и кастомного поиска под ваши нужды. Своя система — это боль, но и полная власть.

И последнее. Тот самый автор кейса с Habr в итоге нашел работу? Не знаю. Но его проект живет в десятках форков. Ваш может оказаться лучше, потому что вы строите его сейчас, с новыми инструментами. Главное — не забросьте на этапе, когда Qdrant откажется запускаться из-за нехватки памяти. Такое бывает. Увеличьте swap-файл и попробуйте снова.

Подписаться на канал