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

Недельный дайджест «что я делал и читал» выглядит простой задачей для агента: выгрузить активность, отдать модели, получить связный текст. На практике такой текст быстро начинает врать в одном конкретном месте: он смешивает то, что человек написал сам, с тем, что он сохранил или лайкнул. В итоге в дайджесте появляется фраза «я пришёл к выводу, что…», хотя вывод принадлежит автору сохранённой статьи. Ссылки теряются, а повторный импорт той же выгрузки удваивает записи.
В этой статье мы соберём небольшой прототип, где каждое утверждение дайджеста можно проследить до исходной записи. Результат — недельный дайджест с журналом входных данных, идемпотентным импортом, ссылками на источники и автоматической проверкой происхождения каждого абзаца. Весь код ниже — учебный пример на Python и SQLite, а не готовый продукт; форматы выгрузок условные, и их нужно адаптировать под ваши реальные источники.
Что значит «проверяемый»
Мы опираемся на понятие провенанса (происхождения данных): у каждого абзаца дайджеста должен быть явный список записей, на которые он опирается, и явная пометка, от чьего лица он написан. Формально прототип должен соблюдать четыре инварианта:
- Нет абзаца без источника. Каждый абзац ссылается хотя бы на одну запись из выбранной недели.
- Голос соответствует типу источника. Абзац от первого лица («я написал», «я считаю») допустим, только если все его источники — публикации самого пользователя.
- Цитаты дословны. Если абзац цитирует источник, цитата буквально присутствует в тексте записи.
- Повторяемость. Повторный импорт той же выгрузки не меняет базу, а каждый запуск оставляет след в журнале.
Модель здесь — только автор черновика. Решение о том, попадёт ли черновик в дайджест, принимает детерминированный валидатор.
Шаг 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
Обратите внимание: валидатор проверяет форму атрибуции, а не смысл. Он гарантирует, что абзац от первого лица опирается только на авторские записи, но не может доказать, что пересказ точен. Для этого и нужны дословные цитаты и ссылки рядом с каждым абзацем — читатель проверяет смысл сам, за один клик.
Как проверить результат
Прогоните несколько контрольных сценариев на своих данных. Это проверки, которые вы выполняете сами, а не опубликованные результаты тестов.
- Идемпотентность. Второй запуск
import_events.pyна том же файле должен вывестиinserted=0. Убедитесь в журнале:sqlite3 digest.db "SELECT file, file_sha256, seen, inserted, skipped FROM ingest_log;" - Чужой вывод от первого лица. Поменяйте во втором абзаце
draft.jsonзначениеvoiceнаuser. Валидатор должен завершиться с кодом 1 и сообщением про чужие материалы. - Выдуманная цитата. Измените одно слово в поле
quotes[].text— проверка дословности должна упасть. - Абзац без опоры. Удалите
sourcesу любого абзаца — ошибка «нет источников». - Журнал решений. Каждая попытка, включая неудачные, видна в
digest_runs:sqlite3 digest.db "SELECT at, ok, errors FROM digest_runs ORDER BY at DESC LIMIT 5;"
Если все сценарии ведут себя так, как описано, у вас есть воспроизводимая цепочка: выгрузка → хеш файла → записи → хеш пакета → хеш черновика → решение валидатора → digest.html.
Типовые ошибки
- Ключ из хеша текста. Любая правка заметки превращается в «новую» запись, и дайджест пересказывает одно и то же дважды. Используйте внешний id источника.
- Определение
kindмоделью. Если тип записи угадывает LLM, вся проверка голоса теряет смысл.kindзадаётся адаптером источника на этапе импорта. - Нестабильный пакет. Без сортировки короткие id
S1…Snперетасовываются между запусками, и старые черновики начинают ссылаться не на те записи. Храните хеш пакета рядом с черновиком. - Свободный текст вместо JSON. Ссылки, вставленные моделью прямо в прозу, нельзя надёжно проверить. Источники — отдельным полем, текст — отдельным.
- Автоисправление валидатором. Соблазнительно «чинить» голос автоматически. Лучше отклонить черновик и перегенерировать: молчаливые правки скрывают системную проблему промпта.
- Сырые URL в HTML. Ссылки из внешних выгрузок экранируйте при рендере, как в примере, и не доверяйте схемам вроде
javascript:— добавьте проверку, что URL начинается сhttps://.
Ограничения прототипа
- Проверка цитат — точное совпадение подстроки. Разные кавычки, пробелы или регистр дадут ложный отказ; нормализацию текста добавляйте осознанно и одинаково для обеих сторон.
- Валидатор не ловит искажения смысла в пересказе без цитат. Абзац с верными источниками всё ещё может неверно их передавать.
- Репосты с комментарием — смешанный случай: комментарий авторский, исходный пост чужой. Их лучше разбивать на две записи при импорте.
- SQLite и локальные файлы подходят для одного пользователя. Многопользовательский сценарий потребует разграничения доступа и отдельного хранения журналов.
- Отправляя личную активность во внешнюю модель, вы передаёте ей эти данные. Проверьте условия обработки у выбранного провайдера или используйте локальную модель.
Что дальше
Следующий логичный шаг — хранить вместе с digest_runs версию промпта и идентификатор модели, чтобы различия между неделями можно было объяснить изменением входа, а не догадками. Другие практические разборы про проверку агентов собраны в разделе гайдов, а термины из этой статьи — в глоссарии.