Надёжные ИИ-агенты: сохранение состояния с LangGraph, DynamoDB и Cosmos DB

Агент, который не помнит, где остановился, — это всего лишь демо. Когда идея сохранённого состояния понятна, начинаются практические вопросы: где это состояние хранится, как workflow снова его находит и что вообще стоит сохранять? Ниже ответы на них на примере LangGraph и двух крупных облаков.

План такой. Сначала — где место чекпоинту в продакшене: где-то надёжнее локального диска, а именно в сервисах AWS или Azure. Затем — как LangGraph возобновляет работу через thread_id и сменные бэкенды checkpointer. И наконец — какие части состояния стоит хранить, а какие дешевле собрать заново.

Чтобы применить это в проекте:

1. Выберите checkpointer под своё развёртывание: локальная разработка, один инстанс или несколько.
2. Передавайте thread_id в каждый вызов graph.invoke() с первого дня.
3. Разнесите каждое поле модели состояния по двум корзинам — «сохранять» или «пересчитывать» — по правилу ближе к концу статьи.

Модель WorkflowState полезна, только если её кто-то хранит там, где перезапуск пода её не сотрёт. Об этом «ком-то» и пойдёт речь дальше.

Где живёт чекпоинт

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

Локальный диск переживёт падение процесса. Но не переживёт переезд контейнера на другой узел, убитый и заменённый под или автоскейлер, запустивший второй инстанс, который ничего не знает о первом. Если ваш агент работает хоть где-то рядом с Kubernetes — а рано или поздно туда попадает большинство, — локальные файлы дают ложное чувство безопасности, а не стратегию.

Поэтому чекпоинты переезжают в облачное хранилище. В AWS обычная связка — быстрое key-value-хранилище для указателя и объектное хранилище для объёмных данных:

• DynamoDB хранит метаданные чекпоинтов и небольшие объекты состояния с ключом task_id или thread_id и быстро отдаёт их, когда задача возобновляется.
• S3 хранит крупные блобы — длинные истории диалогов или накопленный вывод инструментов, — которым не нужен доступ за миллисекунды.

В Azure есть аналогичное разделение:

• Cosmos DB или Azure Table Storage для метаданных и быстрых выборок.
• Azure Blob Storage для крупных данных.
• Entra ID, чтобы управлять тем, какие идентичности и сервисные принципалы могут читать или писать чекпоинт, — это становится важным, как только в состоянии появляется что-то чувствительное.

Абстракция выглядит одинаково на любом бэкенде, меняется только клиент:

from typing import Protocol

class Checkpointer(Protocol):
    def get(self, task_id: str) -> dict | None: ...
    def put(self, task_id: str, state: dict) -> None: ...class DynamoDBCheckpointer:
    def __init__(self, table):
        self.table = table    def get(self, task_id: str) -> dict | None:
        item = self.table.get_item(Key={"task_id": task_id}).get("Item")
        return item["state"] if item else None    def put(self, task_id: str, state: dict) -> None:
        self.table.put_item(Item={"task_id": task_id, "state": state})class AzureBlobCheckpointer:
    def __init__(self, container_client):
        self.container = container_client    def get(self, task_id: str) -> dict | None:
        blob = self.container.get_blob_client(f"{task_id}.json")
        if not blob.exists():
            return None
        return json.loads(blob.download_blob().readall())    def put(self, task_id: str, state: dict) -> None:
        self.container.upload_blob(
            f"{task_id}.json", json.dumps(state), overwrite=True
        )

Если писать код против протокола Checkpointer, а не против SDK конкретного вендора, функция load_or_initialize_state() остаётся независимой от хранилища. Вы меняете реализацию и не трогаете логику workflow.

Два продакшен-компромисса в туториалах обычно не упоминают:

• Частота записи против задержки. Сохранять после каждого узла надёжнее всего и медленнее всего, потому что каждая запись — это сетевой round trip. Реальные системы обычно сохраняются после значимых переходов, например завершённого вызова инструмента или одобрения человеком, и пропускают внутренние шаги между ними.
• Стоимость чекпоинта. Пайплайн из пяти узлов, сохраняющийся на каждом узле и запускаемый на тысячах задач в день, делает тысячи записей ещё до того, как появится хоть один результат. Закладывайте это в бюджет так же, как токены LLM: на масштабе это сопоставимая строка в счёте.

Как это сделать в LangGraph

До сих пор речь шла об архитектуре. В LangGraph она превращается в код на удивление малым числом строк.

Ключом для возобновления служат thread ID. LangGraph узнаёт конкретный запуск — диалог, отчёт или задачу — по его thread_id, а не по определению графа. Считайте граф чертежом, а поток — одним конкретным его экземпляром. Каждый вызов получает ID через config:

config = {"configurable": {"thread_id": "task_123"}}

# First call - no prior state for this thread_id, so LangGraph cold-starts it
result = graph.invoke({"task_id": "task_123"}, config)# Later call, same thread_id - LangGraph loads the last checkpoint
# and resumes instead of starting over
result = graph.invoke(None, config)

Второй вызов, с None на входе, делает то, что load_or_initialize_state() делала вручную: поиск и ветвление за вас выполняет checkpointer. Используйте тот же thread_id — и выполнение продолжится с середины графа, а новый ID запустит холодный старт.

Checkpointer — это подключаемый слой хранения. Он напрямую связан с облачными вариантами выше. LangGraph определяет интерфейс checkpointer, и переданный вами объект решает, где хранится состояние потока:

from langgraph.checkpoint.memory import MemorySaver
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.checkpoint.postgres import PostgresSaver

# Local dev / tests - gone when the process exits
checkpointer = MemorySaver()# Single-instance production - survives process restarts
checkpointer = SqliteSaver.from_conn_string("checkpoints.db")# Multi-instance production - survives pod restarts, works across replicas
checkpointer = PostgresSaver.from_conn_string(POSTGRES_URL)graph = builder.compile(checkpointer=checkpointer)

MemorySaver в точности воспроизводит проблему локального диска. Для демо он годится, но испаряется, когда процесс завершается. PostgresSaver или собственный класс поверх DynamoDB или Cosmos DB, устроенный как протокол Checkpointer выше, позволяет потоку пережить под, переехавший на другой узел.

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

Не сохраняйте всё подряд

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

Сохраняйте:

• результаты вызовов инструментов, которые стоят денег или расходуют rate limit
• решения об одобрении, принятые человеком
• текущий шаг или позицию узла
• всё, что человек ждёт от системы запомнить

Пересчитывайте:

• строки промптов, собранные из шаблонов и состояния
• производные значения вроде количества слов, которые можно получить из другого состояния
• всё, что свежий вызов функции восстановит за миллисекунды почти бесплатно
• найденные документы — при условии, что вы храните их ID, ведь именно ID нужны, чтобы получить их снова

Вопрос, по которому сортируется поле, простой. Если бы его не было в чекпоинте, стоило бы возобновление реальных денег, реальной задержки или повторного побочного эффекта? Или восстановить его — это один дешёвый вызов функции? Первое сохраняйте, второе пересобирайте. Чекпоинт, набитый пересчитываемыми данными, медленнее читается, медленнее пишется и дороже хранится, а устойчивости взамен не добавляет.