Долговременная память для ИИ-ассистента: как превратить переписку в Telegram в структурированную базу знаний

У больших языковых моделей нет памяти, есть контекстное окно: сегодня это десятки-сотни тысяч токенов, но как только диалог выходит за его пределы, модель «забывает» всё. Это серьезное ограничение для ассистента, который должен помнить ваши дела, людей и договорённости месяцами.

Я делаю персонального ассистента, который читает переписку пользователя в Telegram и отвечает на вопросы вида «какой бюджет мы обсуждали на поездку в Турцию?», «что решили по договору с подрядчиком?», «что я обещал Ане?». То есть строю долговременную память.

Казалось бы, задача решается стандартно: заливаем все сообщения в векторную базу, на каждый вопрос делаем top-k поиск и подкладываем найденные чанки в промпт. Я попробовал этот путь и довольно быстро отказался от него в чистом виде. В этой статье я расскажу почему наивный RAG по переписке ломается и какую архитектуру я построил вместо него.

Почему обычный RAG по чатам не работает

Векторный поиск хорош, когда есть корпус независимых документов: статьи, документация, тикеты. Переписка устроена иначе.

1. Смысл размазан по десяткам сообщений. Один факт («согласовали бюджет 350 000 ₽») собирается из реплик в трёх чатах за две недели. Нарезка на чанки по 512 токенов рвёт его на куски, и ни один чанк не содержит ответа целиком.

2. Нет дедупликации. Один и тот же человек, проект или тема упоминаются сотни раз под разными именами («Саша», «Александр», «Sasha»). Векторный индекс хранит все вхождения, и top-k возвращает пять почти одинаковых фрагментов вместо одного связного знания.

3. Нет обновления и актуальности. В марте «решили брать квартиру», в июне «передумали». Оба фрагмента лежат в индексе с одинаковым весом, и модель не знает, что второе отменяет первое. Вектор — это снимок, а не журнал изменений.

4. Нет провенанса. Непонятно, из какого сообщения взялся факт. Пользователь не может проверить, а мы — отладить.

5. Плохо с временем и фильтрами. «Что обсуждали в марте?», «только по работе» — для чанков это неудобно.

6. Поиск ≠ ответ. Top-k чанков — это не ответ на вопрос, а сырьё, которое каждый раз приходится заново синтезировать.

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

Архитектура

Общая схема конвейера:

Разберём слои по порядку.

Слой 1. Инкрементальный сбор и состояние обработки

Сообщения забираются через Telethon по выбранным пользователем чатам и папкам. Ключевой момент — не обрабатывать одно и то же дважды. Для каждого сырого сообщения есть запись в memory_message_processing_state:

create table memory_message_processing_state (
  telegram_message_raw_id bigint primary key
    references telegram_messages_raw(id) on delete cascade,
  user_id uuid not null,
  last_job_id uuid,
  processing_status text not null
    check (processing_status in ('pending','processing','processed','failed')),
  processed_at timestamptz,
  error text,
  updated_at timestamptz not null default now()
);

Планировщик выбирает только сообщения, у которых состояния нет или оно pending/failed. Ночью по расписанию сначала синхронизируется Telegram, затем запускается обработка памяти. Это даёт предсказуемую стоимость: плачу только за новый трафик, а не за всю историю при каждом прогоне.

Слой 2. Чанки с сохранением контекста

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

Из сообщений собирается транскрипт вида [дата] Отправитель: текст, с жёстким лимитом по символам (у меня — 30 000), чтобы не вылетать за контекст и не раздувать счёт за токены.

def build_transcript(messages):
    lines = []
    for m in messages:
        text = (m["message_text"] or "").strip()
        if not text:
            continue
        lines.append(f"[{m['message_date']}] {m['sender_name']}: {text}")
    transcript = "n".join(lines)
    return transcript[:MAX_TRANSCRIPT_CHARS]

Слой 3. Структурированное извлечение

На каждый чанк делается один вызов LLM с требованием вернуть строго валидный JSON по схеме:

{
  "daily_note": { "title": "...", "summary": "..." },
  "topics":   [{ "name": "...", "summary": "...", "facts": ["..."] }],
  "people":   [{ "name": "...", "summary": "...", "action_items": ["..."] }],
  "projects": [{ "name": "...", "summary": "...", "action_items": ["..."] }],
  "links":    [{ "from": "...", "to": "...", "type": "related_to" }]
}

Два приёма, которые заметно повышают качество:

• Передаём модели уже существующие сущности пользователя (topics/people/projects) и просим переиспользовать стандартные имена. Это снижает создание множества сущностей, когда один и тот же проект каждый раз называется по-новому.

• Низкая температура и response_format: json_object.

payload = {
    "model": model,
    "response_format": {"type": "json_object"},
    "messages": [
        {"role": "system", "content": prompt},   # схема + существующие сущности
        {"role": "user",   "content": transcript},
    ],
    "temperature": 0.1,
}

Слой 4. Merge: дедупликация и граф

Когда все чанки задания завершились (chunks_completed == chunks_total), запускается merge. Здесь сырой JSON превращается в нормализованные записи:

• Сущности вставляются по ключу (user_id, entity_type, canonical_name) — повторные упоминания не создают дубликаты, а обновляют summary.

• Факты и action items привязываются к сущностям.

• Связи складываются в memory_links — получается граф, по которому можно ходить («этот человек → этот проект → эти задачи»).

• Каждая сущность, факт и задача ссылаются на конкретные telegram_message_raw_id через memory_source_refs. Это позволяет показать пользователю исходное сообщение и разобрать любой баг до первоисточника.

Если часть чанков упала, задание получает статус partial, а не success — данные не теряются.

Слой 5. Экспорт в markdown-vault

Структурированные знания выгружаются в обычное файловое хранилище в формате Markdown:

memory/
├── Daily/       # ежедневные заметки
├── Knowledge/   # темы и факты
├── People/      # люди
└── Projects/    # проекты

Экспорт идемпотентен: перед записью блока считается его sha256, и если такой блок уже выгружался, он пропускается.

def append_block_if_new(cur, user_id, ..., path, block):
    content_hash = sha256_text(block)
    if already_exported(cur, user_id, item_kind, item_ref_id, content_hash):
        return False
    with open(path, "a", encoding="utf-8") as f:
        f.write(block.rstrip() + "n")
    mark_exported(cur, user_id, ..., path, content_hash)
    return True

Почему Markdown, а не проприетарная БД:

• знания прозрачны — пользователь может открыть файлы и прочитать, что о нём «помнит» система;

• связи оформлены как [[wiki-links]], поэтому vault совместим с Obsidian и подобными инструментами;

• данные переносимы и удаляются одной командой — это важно для приватности;

• отладка сводится к cat файла, а не к запросу по векторной базе.

Слой 6. Агент и поиск

Каждому пользователю генерируется изолированный конфиг агента (Jinja-шаблоны, атомарная запись). Агенту разрешены только безопасные инструменты:

tools: {
  allow: ["read", "sessions_list", "sessions_send",
          "sessions_history", "session_status",
          "memory_search", "memory_get"],
  deny:  ["exec", "write", "edit", "apply_patch",
          "browser", "canvas", "cron", "process"]
}

Поиск по памяти указывает на vault конкретного пользователя (memorySearch.extraPaths), поэтому агент физически не видит чужие данные. Сообщение из Telegram приходит в bot-relay, тот проверяет регистрацию и подписку и передаёт текст в сессию нужного агента. Ответ возвращается в чат.

Эксплуатация: очереди, метрики, стоимость

Всё работает в Docker, фоновая обработка — на Celery с разными очередями под разные профили нагрузки:

• default — планирование и периодические задачи (плюс beat);

• telegram — сетевые операции с Telegram;

• memory — вызовы LLM (самые долгие и дорогие);

• memory_export — выгрузка Markdown.

Разделение очередей не даёт тяжёлым LLM-задачам блокировать сбор сообщений. У воркеров acks_late и prefetch_multiplier=1, у задач — ретраи с экспоненциальным backoff.

Я сразу заложил наблюдаемость, потому что без неё такой конвейер плохо отлаживается:

• токены и стоимость LLM в разрезе пользователя и модели;

• латентность вызовов и длительность заданий;

• бэклог упавших чанков и число необработанных сообщений;

• размер vault на диске и число файлов.

Отдельная метрика «сколько сообщений ещё не превращено в память» по каждому пользователю оказалась самой полезной: она мгновенно показывает, где что-то пошло не так.

Что я понял

• Для персональных фактов структура важнее вектора. Наивный RAG возвращает только фрагменты; структурированная память же возвращает нужные нам данные. Векторный поиск я оставил как дополнительный инструмент, но не как основу.

• Размер чанка — это компромисс. Маленькие чанки теряют контекст, большие размывают смысл и дорожают. 100 сообщений / 30k символов оказались рабочей серединой.

• Множество сущностей — самая сложная часть. Помогают канонические имена, передача уже известных сущностей в промпт и стабильная схема.

• Идемпотентность обязательна. Ночные перезапуски не должны дублировать заметки — хеши контента решают это.

• Прозрачность — это фича. Markdown-vault, ссылка на исходное сообщение и удаление одной кнопкой дают пользователю контроль над данными.

• Стоимость надо видеть. Когда на каждого пользователя приходится свой поток LLM-вызовов, учёт токенов становится важной частью продукта.

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

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

Я применяю эту архитектуру в проекте memory.tg — второй памяти для Telegram. Если тема интересна, в следующих статьях могу подробнее разобрать доступ к Telegram в условиях блокировок или устройство мультитенантного слоя агентов.