Надійні ШІ-агенти: збереження стану з 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 потрібні, щоб отримати їх знову

Питання, за яким сортується поле, просте. Якби його не було в чекпоінті, чи коштувало б відновлення реальних грошей, реальної затримки або повторного побічного ефекту? Чи відтворити його — це один дешевий виклик функції? Перше зберігайте, друге перебудовуйте. Чекпоінт, напханий даними, які можна перерахувати, повільніше читається, повільніше пишеться й дорожче зберігається, а стійкості натомість не додає.