Самое важное из мира AI — в канале MAX AgentLabОтдельные разборы, инструменты и практические схемы Читайте Agent Lab в Telegram Разборы, кейсы и новости о практических AI-агентах без лишней воды.

Собираем проверяемый AI-дайджест из личной активности

Собираем проверяемый AI-дайджест из личной активности
Temporary fallback cover; replace in editorial pass.

· Уровень: продвинутый · Чтение: около 11 минут

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

В этой статье мы соберём небольшой прототип, где каждое утверждение дайджеста можно проследить до исходной записи. Результат — недельный дайджест с журналом входных данных, идемпотентным импортом, ссылками на источники и автоматической проверкой происхождения каждого абзаца. Весь код ниже — учебный пример на Python и SQLite, а не готовый продукт; форматы выгрузок условные, и их нужно адаптировать под ваши реальные источники.

Что значит «проверяемый»

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

  1. Нет абзаца без источника. Каждый абзац ссылается хотя бы на одну запись из выбранной недели.
  2. Голос соответствует типу источника. Абзац от первого лица («я написал», «я считаю») допустим, только если все его источники — публикации самого пользователя.
  3. Цитаты дословны. Если абзац цитирует источник, цитата буквально присутствует в тексте записи.
  4. Повторяемость. Повторный импорт той же выгрузки не меняет базу, а каждый запуск оставляет след в журнале.

Модель здесь — только автор черновика. Решение о том, попадёт ли черновик в дайджест, принимает детерминированный валидатор.

Шаг 1. Подготовка окружения

Нужны Python 3.10+ и SQLite (модуль sqlite3 входит в стандартную библиотеку). Сторонних зависимостей нет.

mkdir digest-lab && cd digest-lab
python3 -m venv .venv
. .venv/bin/activate
python3 --version

Шаг 2. Единый формат входных записей

Какие бы источники вы ни подключали (экспорт заметок, сохранённые ссылки, посты), сначала приведите их к одному формату JSON Lines. Главное поле — kind: оно разделяет авторское и чужое ещё до того, как текст увидит модель.

{"source": "notes", "external_id": "n-101", "kind": "authored", "url": "https://example.com/notes/101", "author": "me", "created_at": "2026-10-20T09:12:00Z", "text": "Перевёл импорт на upsert по внешнему id, дубли исчезли."}
{"source": "bookmarks", "external_id": "b-77", "kind": "saved", "url": "https://example.com/article/77", "author": "Другой автор", "created_at": "2026-10-21T18:40:00Z", "text": "Автор утверждает, что журнал запусков важнее метрик качества."}

Это иллюстративный файл events.jsonl, домены — заглушки. Правило: если вы не можете уверенно определить kind для записи, помечайте её как saved. Ошибка в сторону «чужое» безопаснее, чем приписывание человеку чужих мыслей.

Шаг 3. Идемпотентный импорт и журнал входных данных

Ключ записи строится из пары «источник + внешний id», а не из хеша текста: если пользователь отредактировал заметку, это та же запись, а не новая. Каждый запуск импорта пишет строку в ingest_log с хешем файла и счётчиками.

# import_events.py
import datetime, hashlib, json, sqlite3, sys, uuid

db = sqlite3.connect("digest.db")
db.executescript("""
CREATE TABLE IF NOT EXISTS items(
  key TEXT PRIMARY KEY,
  kind TEXT NOT NULL CHECK (kind IN ('authored', 'saved')),
  url TEXT NOT NULL,
  author TEXT,
  created_at TEXT NOT NULL,
  text TEXT NOT NULL);
CREATE TABLE IF NOT EXISTS ingest_log(
  run_id TEXT, file TEXT, file_sha256 TEXT,
  seen INT, inserted INT, skipped INT, at TEXT);
""")

path = sys.argv[1]
raw = open(path, "rb").read()
seen = inserted = 0
for line in raw.decode("utf-8").splitlines():
    if not line.strip():
        continue
    e = json.loads(line)
    seen += 1
    key = hashlib.sha256(f"{e['source']}:{e['external_id']}".encode()).hexdigest()
    cur = db.execute(
        "INSERT OR IGNORE INTO items VALUES (?, ?, ?, ?, ?, ?)",
        (key, e["kind"], e["url"], e.get("author"), e["created_at"], e["text"]))
    inserted += cur.rowcount

db.execute("INSERT INTO ingest_log VALUES (?, ?, ?, ?, ?, ?, ?)", (
    str(uuid.uuid4()), path, hashlib.sha256(raw).hexdigest(),
    seen, inserted, seen - inserted,
    datetime.datetime.now(datetime.timezone.utc).isoformat()))
db.commit()
print(f"seen={seen} inserted={inserted} skipped={seen - inserted}")
python3 import_events.py events.jsonl
python3 import_events.py events.jsonl   # второй запуск: inserted=0

INSERT OR IGNORE сознательно не обновляет отредактированные записи. Если вам нужны правки, замените его на INSERT ... ON CONFLICT(key) DO UPDATE и добавьте колонку с хешем текста, чтобы журнал показывал, что именно изменилось.

Шаг 4. Пакет недели с короткими идентификаторами

Модели не нужны длинные хеши: она будет путать их при копировании. Выдаём записи недели под короткими id S1, S2… и сохраняем соответствие в файле пакета.

# build_packet.py
import hashlib, json, sqlite3, sys

start, end = sys.argv[1], sys.argv[2]   # например 2026-10-19 2026-10-26
db = sqlite3.connect("digest.db")
rows = db.execute(
    "SELECT key, kind, url, author, created_at, text FROM items "
    "WHERE created_at >= ? AND created_at < ? ORDER BY created_at, key",
    (start, end)).fetchall()
sources = [
    {"sid": f"S{i}", "key": r[0], "kind": r[1], "url": r[2],
     "author": r[3], "created_at": r[4], "text": r[5]}
    for i, r in enumerate(rows, 1)]
packet = {"week": [start, end], "sources": sources}
body = json.dumps(packet, ensure_ascii=False, indent=2)
open("week.json", "w", encoding="utf-8").write(body)
print("packet_sha256", hashlib.sha256(body.encode()).hexdigest(), "sources", len(sources))

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

Шаг 5. Черновик от модели в структурированном виде

Отдайте week.json любой модели, которой вы доверяете работу с личными данными, и потребуйте ответ строго в JSON. Примерная инструкция:

Составь недельный дайджест по записям из week.json.
Верни только JSON вида:
{"paragraphs": [{"voice": "user" | "external",
                 "text": "...",
                 "sources": ["S1", ...],
                 "quotes": [{"sid": "S1", "text": "точная подстрока"}]}]}
Правила:
- voice "user" — только если все sources имеют kind "authored".
- Записи kind "saved" пересказывай от третьего лица: "в сохранённой статье автор пишет...".
- Не объединяй в одном абзаце авторские и сохранённые записи с voice "user".
- Не добавляй фактов, которых нет в источниках.

Пример того, как может выглядеть ответ (сохраните его как draft.json):

{"paragraphs": [
  {"voice": "user",
   "text": "На этой неделе я перевёл импорт на upsert по внешнему id.",
   "sources": ["S1"],
   "quotes": []},
  {"voice": "external",
   "text": "В сохранённой статье автор утверждает, что журнал запусков важнее метрик качества.",
   "sources": ["S2"],
   "quotes": [{"sid": "S2", "text": "журнал запусков важнее метрик качества"}]}
]}

Правила в промпте полезны, но не являются гарантией. Их соблюдение проверяет следующий шаг.

Шаг 6. Проверка происхождения каждого абзаца

Валидатор проверяет все четыре инварианта, записывает результат в журнал digest_runs и только при успехе рендерит HTML со ссылками на источники.

# validate_digest.py
import datetime, hashlib, html, json, sqlite3, sys

packet_raw = open("week.json", "rb").read()
draft_raw = open("draft.json", "rb").read()
packet = json.loads(packet_raw)
draft = json.loads(draft_raw)
src = {s["sid"]: s for s in packet["sources"]}
errors = []

for i, p in enumerate(draft.get("paragraphs", []), 1):
    sids = p.get("sources") or []
    if not sids:
        errors.append(f"#{i}: нет источников")
        continue
    unknown = [s for s in sids if s not in src]
    if unknown:
        errors.append(f"#{i}: неизвестные источники {unknown}")
        continue
    kinds = {src[s]["kind"] for s in sids}
    if p.get("voice") == "user" and kinds != {"authored"}:
        errors.append(f"#{i}: голос пользователя опирается на чужие материалы")
    if p.get("voice") not in ("user", "external"):
        errors.append(f"#{i}: неизвестный voice")
    for q in p.get("quotes", []):
        if q.get("sid") not in sids:
            errors.append(f"#{i}: цитата из источника вне списка sources")
        elif q.get("text", "") not in src[q["sid"]]["text"]:
            errors.append(f"#{i}: цитата не найдена дословно в {q['sid']}")

db = sqlite3.connect("digest.db")
db.execute("""CREATE TABLE IF NOT EXISTS digest_runs(
  at TEXT, packet_sha256 TEXT, draft_sha256 TEXT, ok INT, errors TEXT)""")
db.execute("INSERT INTO digest_runs VALUES (?, ?, ?, ?, ?)", (
    datetime.datetime.now(datetime.timezone.utc).isoformat(),
    hashlib.sha256(packet_raw).hexdigest(), hashlib.sha256(draft_raw).hexdigest(),
    int(not errors), json.dumps(errors, ensure_ascii=False)))
db.commit()

if errors:
    print("\n".join(errors))
    sys.exit(1)

out = ["<article class='digest'>"]
for p in draft["paragraphs"]:
    label = "Моё" if p["voice"] == "user" else "Из сохранённого"
    links = ", ".join(
        f"<a href='{html.escape(src[s]['url'])}'>{s}</a>" for s in p["sources"])
    out.append(f"<p data-voice='{p['voice']}'><strong>{label}:</strong> "
               f"{html.escape(p['text'])} <small>[{links}]</small></p>")
out.append("</article>")
open("digest.html", "w", encoding="utf-8").write("\n".join(out))
print("ok: digest.html")
python3 build_packet.py 2026-10-19 2026-10-26
python3 validate_digest.py

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

Как проверить результат

Прогоните несколько контрольных сценариев на своих данных. Это проверки, которые вы выполняете сами, а не опубликованные результаты тестов.

Если все сценарии ведут себя так, как описано, у вас есть воспроизводимая цепочка: выгрузка → хеш файла → записи → хеш пакета → хеш черновика → решение валидатора → digest.html.

Типовые ошибки

Ограничения прототипа

Что дальше

Следующий логичный шаг — хранить вместе с digest_runs версию промпта и идентификатор модели, чтобы различия между неделями можно было объяснить изменением входа, а не догадками. Другие практические разборы про проверку агентов собраны в разделе гайдов, а термины из этой статьи — в глоссарии.