Память ИИ-агента: состояние под управлением схемы и provenance

Viacheslav Dubrov

2026-06-20 · Обновлено 2026-07-22 · 14 мин · ИИ-агенты

Автоматический перевод Эта статья была автоматически переведена с оригинальной английской версии.

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

Если пользователь меняет дедлайн по паспорту с 15 июля на 30 июня, vector search может извлечь оба утверждения. Слой памяти с поддержкой состояния должен зафиксировать, что значение «30 июня» вытесняет более раннее.

Для инженеров, создающих долгоживущих или мультиарендных агентов, такая проблема с устаревшими фактами объясняет, почему память должна сохраняться между отдельными запусками, одновременно обеспечивая актуальность состояния и границы арендаторов. Ниже показано, как записывать типизированные записи и тестировать чтение текущего состояния и состояния на определённый момент времени, не считая vector recall источником истины.

Ловушка контекстного окна

Контекстное окно — это входные данные, доступные для одного вызова модели. Приложение может переносить сообщения в последующие вызовы, но именно политика приложения должна определять, какие старые факты всё ещё актуальны и кто может их видеть.

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

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

Контракт памяти должен отвечать на конкретные вопросы:

• Что актуально сейчас?
• Что было актуально 2 июня?
• Кто это сообщил?
• Какому tenant это принадлежит?
• Какой более старый факт был заменён этим?
• Можно ли его удалить или пометить как истёкший?

Schema-Guided Agent Memory (SGAM) хранит эти ответы в виде полей и связей, а не оставляет их неявными в прозе.

Что означает SGAM

Три близких по названию концепции задают область применения SGAM.

Schema-Guided Dialogue (SGD) — это датасет Google для task-oriented dialogue 2019 года. Его схема описывает сервисные API, интенты и слоты, чтобы диалоговая модель могла отслеживать состояние сервисов, с которыми она раньше не работала. Это полезный пример трекинга на основе схемы, но его область ограничена диалоговыми сервисами.

Schema-Guided Memory (SGM) — исследовательский термин, который Mei и соавторы используют в статье According to Me: Long-Term Personalized Referential Memory QA. В статье свободнотекстовая Descriptive Memory (DM) сравнивается с элементами key-value-памяти фиксированной схемы. Обе репрезентации содержат одну и ту же исходную информацию, но имеют разную структуру.

В этой статье под Schema-Guided Agent Memory (SGAM) я понимаю инженерный паттерн, в котором схемы управляют записью, обновлением, извлечением и удалением. Схема определяет состояние приложения и его жизненный цикл.

ATM-Bench показывает, почему репрезентация имеет значение. В нём используются примерно четыре года персональных данных из email, изображений и видео. Вопросы требуют персональных отсылок, локации, нескольких фрагментов доказательств и учёта изменений во времени. Для сложного сплита авторы сообщают, что в протестированной конфигурации SGM превосходит DM в извлечении и question answering. SGM предоставляет в фиксированной репрезентации такие поля, как время, источник, локация, сущности и теги. Я считаю эту репрезентацию и заявленные результаты подтверждённым выводом; статья не доказывает наличие прямого механизма адресации полей.

Сравнение SGM и DM отвечает на вопрос хранения: должна ли память оставаться свободным текстом или использовать именованные поля? У production-агента до хранения возникает ещё одна проблема. Нужно превратить неструктурированный разговор в предлагаемое обновление памяти. Schema-Guided Reasoning (SGR) делает предполагаемый путь принятия решения наблюдаемым: ответ может содержать доказательства, субъект и атрибут, сравнение с текущим состоянием и кандидата на запись. Ответ модели не обеспечивает зависимости или порядок между этими полями. Эти правила обеспечиваются отдельными вызовами, валидаторами и политикой приложения; SGAM применяет правила хранения и жизненного цикла после вызова модели.

Разделяйте извлечение моделью и владение памятью

Запись в память проходит через три слоя. Structured Output (SO) задаёт форму объекта-кандидата. Schema-Guided Reasoning (SGR) делает предполагаемые поля и намерение решения модели наблюдаемыми в структурированном ответе. Schema-Guided Agent Memory (SGAM) управляет кандидатом как долговременным состоянием после вызова модели. Зависимости, порядок и жизненный цикл обеспечиваются отдельными вызовами, валидаторами и политикой приложения.

Вердикт, роутинг или план обычно истекают вместе с текущим запросом. Другой запуск может прочитать кандидата на запись через несколько дней или использовать его для выбора tool call. Такой длительный срок жизни требует правил хранения, которых SGR не предоставляет.

SGR структурирует один вызов модели вокруг предполагаемой топологии ризонинга. При записи в память ответ может включать исходные доказательства, нормализованные субъект и атрибут, сравнение с текущим состоянием и предлагаемое обновление. Pydantic или JSON Schema описывают эти поля. Provider-native Structured Output или рантайм guided decoding, например XGrammar, удерживает ответ в этой форме.

Такая форма делает заявленные моделью поля и намерение наблюдаемыми. Но она не гарантирует корректность вывода и не доказывает, что модель использовала эти поля в правильном порядке. Граница enforcement для зависимостей, порядка и жизненного цикла проходит через отдельные вызовы, валидаторы и политику приложения.

SGAM решает, что делать после создания объекта. Нужно ли его сохранить? Вытесняет ли он более старый факт? Какой tenant может его видеть? Является ли он текущим или историческим? Какой исходный эпизод его подтверждает?

В таблице показано, за какой аспект отвечает каждый слой и какие сбои для него характерны:

На пути записи последовательность выглядит так:

SGR reasoning schema + Structured Output -> candidate write -> SGAM policy and persistence

Следующий иллюстративный фрагмент из memory_models.py определяет объект, передаваемый от экстрактора в сервис записи SGAM. Для этого блока нужен Pydantic, поэтому в проверках окружений без этой опциональной зависимости он помечен как no-run:

from datetime import datetime
from pydantic import BaseModel, Field

class MemoryDelta(BaseModel):
    tenant_id: str = Field(description="Isolation boundary, e.g. acme")
    subject: str = Field(description="Normalized entity ID, e.g. mira")
    attribute: str = Field(description="Property being updated")
    value: str = Field(description="New value")
    valid_from: datetime
    source_episode_id: str

MemoryDelta фиксирует, что именно извлекла модель. Сервис записи SGAM по-прежнему сам решает, отклонить, объединить или сохранить объект.

У путей записи и чтения разные задачи

Только путь записи изменяет сохранённое состояние. Путь чтения выбирает записи для текущего запроса.

Поток ingestion — это путь записи:

1. Зафиксировать сырой эпизод из сообщений, результатов tool call или бизнес-событий.
2. Извлечь типизированных кандидатов через structured output.
3. Провалидировать схему и отклонить некорректные записи.
4. Разрешить конфликты, закрыть устаревшие факты и сохранить provenance.
5. Записать запись в SGAM-хранилище.

Поток запроса — это путь чтения:

1. Начать с вопроса пользователя.
2. Определить, требуется ли текущий state или state на определённый момент времени.
3. Отфильтровать по tenant, типу памяти, субъекту, атрибуту и окну валидности.
4. Добавить vector- или graph-expansion, только если точного поиска состояния недостаточно.
5. Собрать минимальный контекст с цитатами для модели.

Читайте эту диаграмму слева направо по двум дорожкам. Верхняя дорожка записывает память, нижняя читает её. Обе используют одно и то же хранилище.

Что должно входить в схему памяти

Минимальной записи SGAM нужно больше, чем text.

tenant_id
memory_id
subject
attribute
value
memory_type
schema_version
valid_from
valid_to
supersedes_memory_id
source_episode_id
confidence
retention_policy

С этими полями новый дедлайн по паспорту может закрыть предыдущий, не стирая историю. Та же таблица может отвечать на текущие вопросы и вопросы на определённый момент времени, а затем связывать результат с исходным эпизодом. schema_version поддерживает миграции, а retention_policy сообщает задачам удаления, что ещё нужно удалить.

Используйте RAG для извлечения документов, а SGAM — для поддержки изменяемого состояния. Vector search по-прежнему нужен в системе для нечёткого recall, кластеризации и expansion. Текущее значение mira.passport_deadline должно извлекаться из записи памяти с заданной областью видимости, а не из чанка, который случайно занял первое место в ранжировании.

Пример с устаревшим фактом

Рассмотрим синтетический трейс из двух эпизодов, представленный в виде baseline DM и журнала SGAM:

e1: Mira prefers concise answers. Her passport deadline is 2026-07-15.
e2: Mira corrected the deadline. It is now 2026-06-30.

Baseline DM хранит оба эпизода как свободный текст, поэтому text search может вернуть e1, поскольку там есть нужные слова. SGAM извлекает из каждого эпизода типизированный факт, ключом которого служат tenant, субъект и атрибут. Он должен вернуть e2 как текущее состояние и сохранить e1 для исторического запроса.

Ниже приведён самодостаточный путь записи в SQLite. Он использует строки времени в формате ISO-8601 UTC, настраивает sqlite3.Row перед чтением по имени столбца и представляет факты полуоткрытыми интервалами: [valid_from, valid_to). Таблица отклоняет некорректные интервалы и содержит частичный уникальный индекс, допускающий одну открытую запись на tenant, субъект и атрибут. replace_fact владеет одной транзакцией; SQLite сериализует конкурентные записи, поэтому после ошибки блокировки или нарушения уникальности вызывающим сторонам следует повторить всю транзакцию.

import sqlite3
from dataclasses import dataclass

@dataclass(frozen=True)
class MemoryFact:
    tenant_id: str
    subject: str
    attribute: str
    value: str
    valid_from: str
    source_episode_id: str

def open_db() -> sqlite3.Connection:
    db = sqlite3.connect(":memory:")
    db.row_factory = sqlite3.Row
    db.executescript(
        """
        create table memory_facts (
            fact_id integer primary key,
            tenant_id text not null,
            subject text not null,
            attribute text not null,
            value text not null,
            valid_from text not null,
            valid_to text,
            source_episode_id text not null,
            check (valid_to is null or valid_from < valid_to)
        );
        create unique index one_open_fact
        on memory_facts (tenant_id, subject, attribute)
        where valid_to is null;
        """
    )
    return db

def replace_fact(db: sqlite3.Connection, fact: MemoryFact) -> None:
    with db:
        exact = db.execute(
            """
            select fact_id
            from memory_facts
            where tenant_id = ? and subject = ? and attribute = ? and valid_from = ?
            """,
            (fact.tenant_id, fact.subject, fact.attribute, fact.valid_from),
        ).fetchone()
        if exact:
            raise ValueError("equal valid_from requires an application conflict policy")

        containing = db.execute(
            """
            select fact_id, valid_to
            from memory_facts
            where tenant_id = ?
              and subject = ?
              and attribute = ?
              and valid_from < ?
              and (valid_to is null or valid_to > ?)
            limit 1
            """,
            (
                fact.tenant_id,
                fact.subject,
                fact.attribute,
                fact.valid_from,
                fact.valid_from,
            ),
        ).fetchone()
        if containing:
            db.execute(
                "update memory_facts set valid_to = ? where fact_id = ?",
                (fact.valid_from, containing["fact_id"]),
            )
            successor_boundary = containing["valid_to"]
        else:
            successor = db.execute(
                """
                select valid_from
                from memory_facts
                where tenant_id = ? and subject = ? and attribute = ? and valid_from > ?
                order by valid_from
                limit 1
                """,
                (fact.tenant_id, fact.subject, fact.attribute, fact.valid_from),
            ).fetchone()
            successor_boundary = successor["valid_from"] if successor else None

        db.execute(
            """
            insert into memory_facts
                (tenant_id, subject, attribute, value, valid_from, valid_to, source_episode_id)
            values (?, ?, ?, ?, ?, ?, ?)
            """,
            (
                fact.tenant_id,
                fact.subject,
                fact.attribute,
                fact.value,
                fact.valid_from,
                successor_boundary,
                fact.source_episode_id,
            ),
        )

def fact_at(db: sqlite3.Connection, timestamp: str) -> sqlite3.Row:
    return db.execute(
        """
        select value, source_episode_id
        from memory_facts
        where tenant_id = ?
          and subject = ?
          and attribute = ?
          and valid_from <= ?
          and (valid_to is null or valid_to > ?)
        limit 1
        """,
        ("acme", "mira", "passport_deadline", timestamp, timestamp),
    ).fetchone()

db = open_db()
replace_fact(
    db,
    MemoryFact("acme", "mira", "passport_deadline", "2026-07-15", "2026-06-01T09:00:00Z", "e1"),
)
replace_fact(
    db,
    MemoryFact("acme", "mira", "passport_deadline", "2026-06-30", "2026-06-03T10:00:00Z", "e2"),
)
current = fact_at(db, "2026-06-04T00:00:00Z")
assert (current["value"], current["source_episode_id"]) == ("2026-06-30", "e2")

historical = fact_at(db, "2026-06-02T00:00:00Z")
assert (historical["value"], historical["source_episode_id"]) == ("2026-07-15", "e1")

# A delayed extraction predates e1. It ends at e1's existing boundary,
# rather than closing e2, which remains current.
replace_fact(
    db,
    MemoryFact("acme", "mira", "passport_deadline", "2026-08-01", "2026-05-30T08:00:00Z", "e0"),
)
delayed = fact_at(db, "2026-05-31T00:00:00Z")
assert (delayed["value"], delayed["source_episode_id"]) == ("2026-08-01", "e0")
current = fact_at(db, "2026-06-04T00:00:00Z")
assert (current["value"], current["source_episode_id"]) == ("2026-06-30", "e2")
intervals = db.execute("select valid_from, valid_to from memory_facts").fetchall()
assert all(row["valid_to"] is None or row["valid_from"] < row["valid_to"] for row in intervals)

episodes = (
    "e1: Mira prefers concise answers. Her passport deadline is 2026-07-15.",
    "e2: Mira corrected the deadline. It is now 2026-06-30.",
)
naive_match = next(episode for episode in episodes if "passport deadline" in episode)
assert naive_match.startswith("e1:")

replace_fact сначала отклоняет совпадающий valid_from: такой конфликт требует политики приложения — например, приоритета источников или подтверждённой пользователем правки, — а не молчаливой перезаписи. Затем он ищет интервал, содержащий timestamp входящей записи. Если такой интервал найден, он закрывает предыдущий интервал на границе входящей записи и присваивает вставляемой строке бывший конец интервала предшественника. Если timestamp находится раньше всех сохранённых интервалов, в качестве конца новой строки используется valid_from следующего интервала. Поэтому отложенный факт, расположенный раньше e1, становится [2026-05-30, 2026-06-01) и сохраняет e1 и текущее e2 без изменений. Фабрика строк делает row["fact_id"] и именованные поля результатов доступными в созданном здесь соединении по умолчанию. Частичный уникальный индекс служит ограничением базы данных для одной открытой строки; при развёртывании с несколькими писателями нужно выбрать БД и политику повторных попыток, соответствующие нагрузке. У DM нет эквивалентного шага обновления, поэтому старый текст всё ещё может опередить исправление в ранжировании.

Проверки подтверждают текущую строку e2, историческую строку e1, отложенную вставку перед e1, инвариант интервала и первый совпавший эпизод наивного text search:

Naive text memory:
  returned episode: e1 -> passport deadline is 2026-07-15

SGAM current state:
  mira.passport_deadline = 2026-06-30
  valid_from=2026-06-03T10:00:00Z, source=e2

SGAM point-in-time state:
  on 2026-06-02, mira.passport_deadline = 2026-07-15

В production к этой транзакции следует добавить structured extraction на пути записи. Транзакция базы данных обновляет временную валидность. Модель извлекает факт-кандидат, но не решает, какая сохранённая строка останется текущей.

Выбор хранилища определяется паттерном извлечения

В проектах используются разные названия для частей этого паттерна: хранилища памяти, контекстные графы, профили, долгосрочные хранилища, graph RAG и stateful-агенты.

Graphiti — конкретная open-source реализация реляционной временной памяти. Она отслеживает изменения фактов, хранит указатели на исходные эпизоды и поддерживает hybrid retrieval. LangGraph разделяет чекпоинты тредов и хранилища между тредами. Mem0 упаковывает операции с памятью в управляемый сервис. Letta использует редактируемые блоки контекста вместо SGAM на уровне полей, но всё равно рассматривает состояние агента как персистентные данные.

Начинайте с модели данных. Если основная операция — точный поиск факта, обычно достаточно реляционной таблицы с JSON payloads, колонками валидности, индексами по tenant и vector sidecar. Добавляйте граф, когда обход связей является частью продукта, а не потому, что графовая демоверсия выглядит впечатляюще.

Сначала создайте путь записи, а потом граф

Сначала решите, что продукту разрешено запоминать. Выбор между графом и вектором идёт позже.

Support-агент может помнить тариф аккаунта, открытые обращения и постоянные предпочтения по контактам. Но он не должен превращать каждую эмоциональную реплику в состояние профиля. Coding-агент может помнить соглашения репозитория и нерешённые задачи. Но он не должен хранить приватную заметку вечно только потому, что однажды она попалась в retrieval.

Начинайте с пути записи и рассматривайте память как небольшую мутацию состояния:

1. Назовите тип памяти, субъект, область tenant и класс хранения.
2. Извлеките записи-кандидаты с помощью structured output.
3. Провалидируйте payload с помощью Pydantic или уже используемого в вашем стеке слоя схем.
4. Разрешите конфликты до вставки, включая проверку, вытесняет ли новая запись старую.
5. Сохраните указатель на сырой эпизод, результат tool call, файл, тикет или подтверждение пользователя, породившие запись.
6. Записывайте версию схемы вместе с каждой записью, а не оставляйте её только в коде приложения.

Первым SGAM-хранилищем может стать реляционная таблица с JSON-колонкой и несколькими индексами. Граф становится полезным, когда продукт должен обходить связи вроде customer-to-account, account-to-policy, task-to-artifact или project-to-decision.

Hot path и фоновые записи

Немедленное извлечение оправдано, когда следующий ход зависит от новой памяти. Если пользователь говорит «запомни, что я предпочитаю короткие ответы», системе не следует ждать ночного job, прежде чем начать вести себя иначе.

Большинству ходов немедленная запись не нужна. Сохраните сырой эпизод с метаданными tenant, сессии и инструментов, а затем передайте извлечение кандидатов фоновому worker. При консолидации на основе recurrence worker буферизует слабые сигналы и повышает факт до памяти только после повторения похожих свидетельств или подтверждения пользователя. Это добавляет лаг свежести. Для факта «пользователь часто просит CSV-экспорты» это приемлемо, а для «клиент изменил адрес доставки» — рискованно.

Делайте путь чтения детерминированным. Сначала применяйте область tenant и валидность, а fuzzy retrieval используйте только тогда, когда он может добавить полезный контекст.

1. Отфильтруйте по tenant, типу памяти и окну валидности.
2. Сначала извлеките точное структурированное состояние, а уже потом семантических соседей.
3. Используйте vector- или graph-expansion для подтверждающих доказательств, связанных сущностей и примеров, но не как источник текущих фактов.
4. Соберите минимальный контекст с цитатами, достаточный для ответа на вопрос.

Относитесь к миграции схемы как к изменению продукта, потому что она меняет то, что агент может вспомнить, процитировать или удалить. Она также может изменить набор исторических фактов, считающихся текущими. Планируйте migration scripts, backfills, dual-read windows и поведение удаления в рамках того же релиза.

Когда SGAM оправдывает сложность

Используйте SGAM, когда факты могут изменяться со временем:

• пользовательские предпочтения, которые можно обновить или отозвать
• факты о клиентах или аккаунтах с требованиями аудита
• состояние задач для долгоживущих ассистентов
• память coding-агента о проекте
• общее состояние мультиагентной системы
• compliance-заметки, для которых важен provenance
• временные вопросы вроде «что мы считали истинным до миграции?»

SGAM избыточен, если память краткоживущая, исследовательская или её дёшево пересчитать. Если агенту нужна лишь связность на несколько ходов, достаточно чекпоинта и сокращённой истории сообщений. Для статического QA по документам может хватить RAG. А если домен настолько нестабилен, что схема меняется каждый день, типизированная память будет замедлять команду.

Чеклист оценки

Оценивайте жизненный цикл памяти, а не только финальный ответ. Система может выдать правдоподобный ответ после того, как записала неправильный факт, извлекла устаревший или пересекла границу tenant.

Я использую то же поэтапное разделение, что и в своей статье об оценке RAG. Измеряйте этап, на котором может возникнуть сбой, а не ограничивайтесь оценкой сгенерированного текста. Дисциплина трейсов из статьи об оценке агентов также применима: ошибка памяти часто появляется в истории запуска ещё до того, как попадает в ответ.

Я бы тестировал SGAM с replay. Подавайте фиксированную последовательность эпизодов в writer памяти и после каждого значимого хода проверяйте журнал. Затем задавайте текущие вопросы и вопросы на определённый момент времени к получившемуся хранилищу.

Бенчмарки, такие как LoCoMo, LongMemEval и ATM-Bench, предоставляют публичные тестовые случаи. Они не заменяют доменный тестовый набор. Для coding-ассистента, бота поддержки клиентов и compliance-copilot нужны разные схемы, фильтры, правила хранения и тесты отказов.

Ограничения

SGAM — моё обозначение паттерна, а не стандарт. Существующие проекты по-разному разделяют эту задачу. Память LangGraph и LangMem описывают краткосрочные и долгосрочные хранилища, профили, коллекции, записи на hot path и фоновые memory managers. Zep Graphiti использует термин temporal Context Graph. Letta сохраняет редактируемые блоки памяти, а Mem0 предоставляет управляемый слой памяти. Microsoft GraphRAG, property graphs LlamaIndex и Cognee рассматривают связанные части задачи как графы знаний.

Профиль пользователя, журнал эпизодов, граф документов и редактируемый агентом блок памяти решают разные задачи retrieval и обновления. Я использую SGAM для обозначения долговременной памяти, представляющей текущее состояние приложения и потому требующей схемы, валидности, provenance, обработки конфликтов, retention и миграций.

Типизированная память всё равно может быть неправильной. Схема упрощает проверку плохих записей, но не делает их заслуживающими доверия. По-прежнему нужны доверие к источникам, подтверждение пользователем чувствительных фактов, политика конфликтов, удаление и мониторинг.

Миграция схемы требует работы. Как только память становится состоянием, вы отвечаете за версионирование, backfills, старые записи и поведение удаления. Если пропустить эту работу, старые записи переживут семантику или retention-политику, в рамках которых они были созданы.

Источники

• According to Me: Long-Term Personalized Referential Memory QA — статья Mei и соавторов, вводящая ATM-Bench и Schema-Guided Memory.
• Towards Scalable Multi-Domain Conversational Agents: The Schema-Guided Dialogue Dataset — статья Rastogi и соавторов о датасете SGD.
• LongMemEval: Benchmarking Chat Assistants on Long-Term Interactive Memory — бенчмарк Wu и соавторов для оценки возможностей долгосрочной памяти.
• Graphiti: Build Temporal Context Graphs for AI Agents
• Zep: A Temporal Knowledge Graph Architecture for Agent Memory
• Mem0 Platform Overview
• Mem0: Building Production-Ready AI Agents with Scalable Long-Term Memory
• LangGraph Memory Overview
• LangGraph Persistence
• LangMem documentation
• Letta: Introduction to Stateful Agents
• Microsoft GraphRAG documentation
• LlamaIndex: Using a Property Graph Index
• Cognee Documentation
• Pydantic model validation docs
• Python sqlite3 documentation