LangGraph Checkpoints и State Persistence: Устойчиви AI агенти за 2026

Актуализирано: 16 юли 2026 г.

LangGraph checkpoints са механизмът, чрез който агент, изграден с LangGraph, запазва цялото си състояние след всеки node. При рестарт, срив или намеса от човек, графът продължава точно оттам, откъдето е спрял. В продукция това означава три конкретни неща: устойчивост на прекъсвания (rate limit в 2 през нощта, срив на контейнер), възможност за human-in-the-loop през дни, не секунди, и „time travel" (връщане към предишен snapshot, за да пуснеш агента по различен клон). Тази статия покрива всичко това с работещ код върху PostgresSaver.

• LangGraph чекпойнтерите (MemorySaver, SqliteSaver, PostgresSaver, AsyncPostgresSaver) сериализират state след всеки node, не само в края на графа.
• thread_id е основната единица за изолация: един потребител, един разговор, една задача, един thread. Не смесвайте thread-ове през различни клиенти.
• PostgresSaver е стандартът за production от версия 0.2.30 нататък. SqliteSaver е за локална разработка, а MemorySaver е само за тестове.
• Interrupts дават true human-in-the-loop: агентът замразява state, връща контрола, а вие продължавате с Command(resume=...) часове или дни по-късно.
• Time travel през graph.get_state_history(config) ви позволява да пуснете агента отново от произволен checkpoint с различен вход. Задължително е за eval harness-и.
• Retention на checkpoints е ваша отговорност. Изтривайте thread-ове след завършване или пишете cron job, защото таблиците растат бързо.

Какъв реален проблем решава state persistence?

Всеки път, когато изграждам агент за клиент (брокер за товари, платформа за клинични изпитания, дори прост Slack bot), първата седмица минава без чекпойнтер. Вторият спринт започва с обаждане: „агентът забрави какво обсъждахме преди 3 минути" или „когато Anthropic ни rate-лимитира в 2 през нощта, всички in-flight разговори се сринаха и клиентите повториха диалозите наново."

Честно казано, това е един и същ сценарий, който съм виждал може би десетина пъти. State persistence в LangGraph решава четири конкретни production-проблема:

• Устойчивост на срив: контейнерът се рестартира, но графът продължава от последния успешен node, не отначало.
• Дълги human-in-the-loop цикли: агентът чака одобрение 4 часа. State се пази безопасно в Postgres, а не в RAM.
• Auditability: всеки checkpoint е snapshot със timestamp. Това е регулаторно изискване за clinical trials и финтех.
• Time travel за evals: вземаш checkpoint от продукция, пускаш го отново с различен prompt, сравняваш резултатите. Това е основата на eval harness, който не лъже.

Ако не решавате поне два от тези проблема, честно казано, чекпойнтер не ви е нужен. Карайте с in-memory state и си спестете една зависимост. Но щом стигнете production, липсата на persistence се превръща в тих пожар: агентите работят, докато не спрат, и никой не знае защо.

Видове checkpointer-и в LangGraph 0.6

От версия 0.6 на langgraph (юни 2026) checkpointer-ите са разделени в отделни пакети: langgraph-checkpoint-sqlite, langgraph-checkpoint-postgres, плюс базовият MemorySaver, който идва с ядрото. Вижте официалната документация за checkpointing за пълния списък.

Checkpointer	Използване	Плюсове	Минуси
MemorySaver	Unit тестове, notebooks	Нула setup	Изгубва state при рестарт
SqliteSaver	Локална разработка, single-node CLI	Един файл, лесно git-ване на fixtures	Няма concurrent writers, single-machine
PostgresSaver	Production, single или multi-node	ACID, connection pool, migrations built-in	Изисква Postgres 14+, малко по-висока латентност
AsyncPostgresSaver	Async приложения (FastAPI, aiohttp)	Non-blocking I/O за high-throughput API	Изисква asyncpg

Моята практическа препоръка е простичка. Започвайте с SqliteSaver, преминете към PostgresSaver в момента, в който имате повече от един процес, обслужващ агента. Не се опитвайте да си пишете custom Redis checkpointer в първия месец, защото интерфейсът е обширен (get, put, list, put_writes, get_tuple) и грешки в сериализацията стават много тихо.

Setup на PostgresSaver стъпка по стъпка

Ето минимален, но production-ready setup с Postgres. Всичко е тествано срещу langgraph==0.6.3, langgraph-checkpoint-postgres==2.1.0, Postgres 16.

# pip install langgraph langgraph-checkpoint-postgres psycopg[binary,pool]

from typing import TypedDict, Annotated
from operator import add

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.postgres import PostgresSaver
from langchain_anthropic import ChatAnthropic

# 1. State schema. Annotated + operator.add прави messages "append-only".
class AgentState(TypedDict):
    messages: Annotated[list, add]
    approved: bool

# 2. Nodes. Всеки връща частичен state, LangGraph го мърджва.
llm = ChatAnthropic(model="claude-opus-4-8", max_retries=3)

def draft_reply(state: AgentState) -> dict:
    reply = llm.invoke(state["messages"])
    return {"messages": [reply]}

def send_reply(state: AgentState) -> dict:
    # В production: тук изпращате отговора към Slack/email/CRM.
    print("Изпращаме:", state["messages"][-1].content[:80])
    return {}

# 3. Postgres connection string. В production идва от secrets manager.
DB_URI = "postgresql://agent:secret@localhost:5432/agents?sslmode=require"

# 4. Context manager отваря connection pool и го затваря чисто.
with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    # setup() е идемпотентна, безопасно се вика при всеки boot.
    checkpointer.setup()

    graph = (
        StateGraph(AgentState)
        .add_node("draft", draft_reply)
        .add_node("send", send_reply)
        .add_edge(START, "draft")
        .add_edge("draft", "send")
        .add_edge("send", END)
        .compile(checkpointer=checkpointer)
    )

    # thread_id = уникален за (потребител, разговор). Не преизползвайте!
    config = {"configurable": {"thread_id": "user-42-conv-7"}}
    result = graph.invoke(
        {"messages": [("user", "Обобщи ми този тикет.")]},
        config=config,
    )

Три неща, които правя винаги в реални проекти и които не са очевидни от туториалите:

1. Използвайте connection pool. PostgresSaver.from_conn_string отваря ConnectionPool под капака. Ако инстанцирате чекпойнтера ръчно с Connection, ще стигнете лимита на Postgres при 20 concurrent requests и всичко ще спре.
2. Отделен database user с ограничени права. Агентът пише в 3 таблици: checkpoints, checkpoint_blobs, checkpoint_writes. Не му давайте CREATE права след първоначалния setup().
3. Sanitizирайте thread_id. Не приемайте raw user input. Hash-нете го или проверете срещу whitelist, иначе злонамерен клиент може да чете чужди thread-ове.

Human-in-the-loop с interrupt() и Command

Това е причината повечето мои клиенти изобщо инвестират в LangGraph: възможността да замразиш агент по средата, да поискаш човешко одобрение, и часове по-късно да продължиш от същата точка. В един проект за freight broker точно този pattern спести на клиента около $40k от грешно резервирани товари в първия месец. Ето кода, който диспечер използва, за да потвърди резервация над $10,000:

from langgraph.types import interrupt, Command
from langgraph.graph import StateGraph, START, END

def propose_booking(state: AgentState) -> dict:
    booking = llm_choose_carrier(state)
    return {"proposed_booking": booking}

def human_approval(state: AgentState) -> dict:
    booking = state["proposed_booking"]
    if booking["total_usd"] < 10_000:
        return {"approved": True}

    # interrupt() спира графа; всичко в state е персистирано.
    decision = interrupt({
        "action": "approve_booking",
        "carrier": booking["carrier"],
        "amount_usd": booking["total_usd"],
    })
    return {"approved": decision["approved"], "notes": decision.get("notes", "")}

def execute(state: AgentState) -> dict:
    if not state["approved"]:
        return {"messages": [("system", "Резервацията беше отказана от диспечера.")]}
    api.book(state["proposed_booking"])
    return {"messages": [("system", "Booked.")]}

graph = (
    StateGraph(AgentState)
    .add_node("propose", propose_booking)
    .add_node("approve", human_approval)
    .add_node("execute", execute)
    .add_edge(START, "propose")
    .add_edge("propose", "approve")
    .add_edge("approve", "execute")
    .add_edge("execute", END)
    .compile(checkpointer=checkpointer)
)

# Първо викане: стига до interrupt, после спира.
config = {"configurable": {"thread_id": "booking-2026-07-16-42"}}
result = graph.invoke({"messages": [...]}, config=config)
# result съдържа __interrupt__ ключ с данни за UI-я.

# ...диспечерът вижда prompt в интерфейса, натиска "Одобри"...
# ...след 45 минути, друг процес продължава графа:
graph.invoke(
    Command(resume={"approved": True, "notes": "Проверих rating на carrier-а."}),
    config=config,
)

Няколко ключови детайла. Command(resume=...) не преиграва изпълнените вече nodes, а продължава точно от interrupt()-а. State се пази в Postgres, така че може да са минали часове, процесът може да е бил рестартиран, и въпреки това ще продължи вярно. Комбинирайте това с LLM observability с Langfuse, за да имате пълна audit trail: кой node е взел решение, с какъв prompt, кой човек го е одобрил.

Time travel и разклоняване на state

Един от най-подценяваните feature-и на LangGraph е възможността да разгледате цялата история на checkpoint-ите и да „превъртите" графа назад. Това е злато за две неща: eval harness-и и debug на incidents.

# Всички snapshots за даден thread, от най-нов към най-стар.
config = {"configurable": {"thread_id": "user-42-conv-7"}}
history = list(graph.get_state_history(config))

for snap in history:
    print(snap.config["configurable"]["checkpoint_id"], "->", snap.next)

# Избираме snapshot преди агентът да е избрал tool.
target = history[3]

# update_state пише нова версия на state, разклонена от target-а.
new_cfg = graph.update_state(
    target.config,
    values={"messages": [("user", "Обясни по-подробно, моля.")]},
)

# Продължаваме графа от разклонението, оригиналната линия остава.
graph.invoke(None, config=new_cfg)

Аз лично използвам това по два начина. Първо, при incident: клиент праща screenshot от странен отговор, аз изтеглям thread_id, извиквам get_state_history, и виждам точно кой node е взел кое решение. Второ, за regression tests: съхранявам „golden" checkpoints, срещу които пускам новия prompt или модел, и сравнявам изходите. Без това всеки refactor е риск. Ако още не сте изградили такъв eval harness, вижте моята статия за мулти-агентна оркестрация, където описвам подобен подход при по-сложни топологии.

Изолация на потребители чрез thread_id

thread_id е основната единица за изолация в LangGraph. Един thread = един непрекъснат разговор или задача. Но какво правите, когато имате много потребители, много организации, регулаторни изисквания за data residency?

Модел 1: thread_id като композитен ключ

thread_id = f"{org_id}:{user_id}:{conversation_uuid}"
config = {"configurable": {"thread_id": thread_id}}

Просто, работи за 90% от случаите. Комбинирайте с row-level security в Postgres, ако имате strict изисквания.

Модел 2: отделен checkpointer per tenant

За SaaS с dedicated schemas на клиент: инстанцирайте по един PostgresSaver per tenant с различен schema_search_path. По-скъпо в connection pool-а, но абсолютна изолация.

Модел 3: subgraphs с cross-thread state

Ако имате множество агенти, които трябва да си споделят част от знанието (например „профил на потребителя"), използвайте Store API-то (introduced в 0.5). То живее извън thread-a и е достъпно от всички графове. Прочетете официалната документация за Memory Store преди да го използвате в продукция, защото семантиката около `namespace` е специфична.

Production съвети: retention, migrations, latency

Ето какво съм научил след деплой на LangGraph checkpoints в 6 клиентски проекта през последната година. Ако имате нужда от инфраструктура за инструменти извън checkpoint-ите, вижте моята статия за изграждане на MCP сървър с Python, тя допълва добре тази.

Retention: изтривайте thread-ове

Таблицата checkpoint_blobs расте ~2-10 KB на всеки node execution. При агент с 50k разговора седмично, за 3 месеца може да стигне 40 GB. LangGraph 0.6 предостави checkpointer.delete_thread(thread_id). Използвайте го агресивно.

# Cron job, който изтрива завършени thread-ове по-стари от 30 дни.
import psycopg
from datetime import datetime, timedelta, timezone

cutoff = datetime.now(timezone.utc) - timedelta(days=30)

with psycopg.connect(DB_URI) as conn, conn.cursor() as cur:
    cur.execute(
        "SELECT DISTINCT thread_id FROM checkpoints "
        "WHERE checkpoint_ns = '' AND created_at < %s",
        (cutoff,),
    )
    old_threads = [r[0] for r in cur.fetchall()]

for tid in old_threads:
    checkpointer.delete_thread(tid)

Migrations и версии

Всяка нова minor версия на langgraph-checkpoint-postgres потенциално добавя columns или indexes. Винаги пускайте await checkpointer.setup() при boot. Той е идемпотентен. Но тествайте upgrade-а в staging. Веднъж загубих 2 часа, защото 0.5 → 0.6 добави индекс, който rebuild-ваше таблица от 8 GB под load.

Latency: batching и async

Всеки node в графа прави поне 2 записа в Postgres (checkpoint + writes). За high-throughput API preferвайте AsyncPostgresSaver, тя не блокира event loop-а. При тежки state обекти (например голям messages list), обмислете да пазите само референции и да зареждате пълния контекст lazy.

Чести грешки при работа с checkpoints

• Преизползване на thread_id: ако викате graph.invoke два пъти с един и същ thread_id, вторият call продължава първия, а не го стартира отначало. Искате ли нов разговор — нов UUID.
• Non-serializable state: LangGraph използва msgpack (или JsonSerializer) под капака. Ако сложите numpy array или отворен file handle в state, ще получите тиха грешка при put. Дръжте state „plain data": dict, list, str, int, Pydantic BaseModel.
• Забравено compile(checkpointer=...): ако не подадете чекпойнтер при compile, графът работи, но не персистира. Типичен bug, който откриваш едва след първия рестарт в продукция.
• Race conditions при concurrent invoke на един thread: LangGraph не сериализира invocations на един thread_id. Ако два процеса викат едновременно същия thread, ще имате corrupt state. Слагайте distributed lock (Redis, advisory lock в Postgres).
• Игнориране на __interrupt__: когато граф достигне interrupt(), invoke връща state с ключ __interrupt__. Ако вашият код не го проверява и тихо праща отговор на потребителя, ще пропуснете human-in-the-loop и графът ще увисне.

Често задавани въпроси

Каква е разликата между MemorySaver и PostgresSaver в LangGraph?

MemorySaver пази checkpoint-ите в паметта на процеса. Те се губят при рестарт и не работят между процеси. PostgresSaver ги персистира в Postgres таблици, което ги прави устойчиви, споделяеми между инстанции и подходящи за production. За локална разработка използвайте SqliteSaver като компромис.

Може ли един LangGraph агент да поддържа memory между сесии?

Да, това е точно за какво служат checkpoint-ите. Използвайте един и същ thread_id при следващ разговор със същия потребител, и графът ще започне с целия предишен state. За cross-thread memory (например профил на потребителя, споделен между всички разговори), използвайте Store API-то, добавено в 0.5.

Как да направя human-in-the-loop в LangGraph без да блокирам HTTP request?

Извикайте interrupt() вътре в node, върнете response към UI с данни за одобрение, и когато човек одобри, извикайте graph.invoke(Command(resume=...)) от background worker (Celery, RQ, SQS). HTTP endpoint-ът не трябва да чака resume-а, защото interrupt-ите могат да траят часове.

Как да ограничавам размера на checkpoint таблиците в Postgres?

Пуснете cron job, който извиква checkpointer.delete_thread(thread_id) за завършени разговори по-стари от определен период (обикновено 30-90 дни). Освен това добавете VACUUM FULL периодично на checkpoint_blobs, защото при масово изтриване dead tuples заемат място.

Работят ли LangGraph checkpoints с async FastAPI приложение?

Да, но задължително използвайте AsyncPostgresSaver вместо синхронния PostgresSaver. Той работи с asyncpg и не блокира event loop-а. Синхронният вариант ще ви убие throughput-а на FastAPI при повече от 10-20 concurrent requests.