Checkpointer: сохранение, продолжение и возврат назад

Цель урока: сохранить состояние запуска графа, продолжить запуск в новом процессе после сбоя и вернуться к любому прошлому шагу.

Необходимые знания:

1) урок 12 курса по LangChain: параметр checkpointer, ключ thread_id, поля StateSnapshot, режимы durability, SqliteSaver.from_conn_string

2) урок 1 этого курса: шаг выполнения и граница шага

3) урок 3 этого курса: Send и копии узла

4) урок 4 этого курса: обход стенда циклом из pick_file и scan_file, повторное выполнение узла с начала

5) Python на уровне джуниора: subprocess, try/except, функция next с генераторным выражением

Чем закончился урок 4

Обход стенда из урока 4 идёт циклом и останавливается по признаку. Состояние обхода при этом хранится только в памяти процесса. Допустим, процесс оборвался на третьем файле. Тогда разбор первых двух файлов пропал, и обход начнётся с начала.

В уроке 1 checkpoint назван состоянием запуска на границе шага. Сохраняет его чекпойнтер, переданный при компиляции. В этом уроке граф обхода получает чекпойнтер. С сохранёнными checkpoints запуск продолжается в новом процессе после сбоя. Запуск можно повторить от прошлого шага. Можно и поправить состояние в прошлой точке и продолжить запуск от неё.

Примеры работают на стенде из урока 1: положите папку mini_repo с файлами app.py, config.py, db.py и utils.py рядом с примерами, листинги файлов есть в уроке 1. Пакет langgraph-checkpoint-sqlite для примера 2 указан в requirements.txt из урока 1. Модель и ключ API в этом уроке не нужны.

Thread: адрес сохранённого состояния

Чекпойнтер передаётся в compile параметром checkpointer. Для опытов подходит InMemorySaver из langgraph.checkpoint.memory. Он держит checkpoints в памяти процесса, и при перезапуске процесса они пропадают.

Checkpoints собираются в thread, набор записей под одним ключом. Ключ называется thread_id и стоит в разделе configurable конфигурации вызова. По этому ключу чекпойнтер ищет сохранённое состояние. Поэтому граф с чекпойнтером без thread_id не запускается.

В части примеров документации импортируется MemorySaver. Это тот же класс под прежним именем, разбор есть в уроке 12 курса по LangChain.

Пример 01_thread.py

"""Пример 1 урока 5: чекпойнтер, thread и история checkpoints.

Обход стенда из урока 4 скомпилирован с InMemorySaver. Вызов без thread_id
падает. Вызов с thread_id оставляет в чекпойнтере историю: checkpoint на
каждом шаге. Второй вызов с тем же thread_id продолжает сохранённое состояние,
вызов с другим thread_id начинает с пустого.
"""

import operator
from pathlib import Path
from typing import Annotated, Literal

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from typing_extensions import TypedDict

REPO = Path(__file__).parent / "mini_repo"


class WalkState(TypedDict):
    root: str
    queue: list[str]
    current: str
    findings: Annotated[list[str], operator.add]


def collect_files(state: WalkState) -> dict:
    """Складывает имена файлов стенда в очередь обхода."""
    names = sorted(path.name for path in Path(state["root"]).glob("*.py"))
    return {"queue": names, "current": ""}


def pick_file(state: WalkState) -> dict:
    """Берёт следующее имя из очереди и убирает его оттуда."""
    queue = list(state["queue"])
    if not queue:
        return {"current": ""}
    return {"current": queue.pop(0), "queue": queue}


def scan_file(state: WalkState) -> dict:
    """Считает строки в выбранном файле."""
    path = Path(state["root"]) / state["current"]
    lines = len(path.read_text(encoding="utf-8").splitlines())
    return {"findings": [f"{state['current']}: строк {lines}"]}


def route_next(state: WalkState) -> Literal["scan_file", "__end__"]:
    """Пустое имя означает, что очередь закончилась."""
    if not state["current"]:
        return END
    return "scan_file"


builder = StateGraph(WalkState)
builder.add_node("collect_files", collect_files)
builder.add_node("pick_file", pick_file)
builder.add_node("scan_file", scan_file)
builder.add_edge(START, "collect_files")
builder.add_edge("collect_files", "pick_file")
builder.add_conditional_edges("pick_file", route_next)
builder.add_edge("scan_file", "pick_file")
graph = builder.compile(checkpointer=InMemorySaver())

START_STATE = {"root": str(REPO), "queue": [], "current": "", "findings": []}
WALK_1 = {"configurable": {"thread_id": "walk-1"}}
WALK_2 = {"configurable": {"thread_id": "walk-2"}}

# 1. С чекпойнтером вызов без thread_id не проходит.
try:
    graph.invoke(START_STATE)
except ValueError as error:
    print("1. без thread_id:", error)

# 2. Обход в thread walk-1 и последний checkpoint этого thread.
result = graph.invoke(START_STATE, WALK_1)
snapshot = graph.get_state(WALK_1)
print("2. находок:", len(result["findings"]))
print("   последний checkpoint: шаг", snapshot.metadata["step"], "next", snapshot.next)

# 3. Вся история thread, сверху самый свежий checkpoint.
print("3. история walk-1:")
print("   шаг  источник  next              current     находок")
for item in graph.get_state_history(WALK_1):
    following = ", ".join(item.next) or "-"
    current = item.values.get("current") or "-"
    print(
        f"   {item.metadata['step']:>3}  {item.metadata['source']:<8}  "
        f"{following:<16}  {current:<10}  {len(item.values.get('findings', []))}"
    )

# 4. Тот же вход в тот же thread и в новый thread.
again = graph.invoke(START_STATE, WALK_1)
fresh = graph.invoke(START_STATE, WALK_2)
print("4. повтор в walk-1, находок:", len(again["findings"]))
print("   первый вызов в walk-2, находок:", len(fresh["findings"]))
print("   checkpoints в walk-1:", len(list(graph.get_state_history(WALK_1))))

# Вывод:
# 1. без thread_id: Checkpointer requires one or more of the following 'configurable' keys: thread_id, checkpoint_ns, checkpoint_id
# 2. находок: 4
#    последний checkpoint: шаг 10 next ()
# 3. история walk-1:
#    шаг  источник  next              current     находок
#     10  loop      -                 -           4
#      9  loop      pick_file         utils.py    4
#      8  loop      scan_file         utils.py    3
#      7  loop      pick_file         db.py       3
#      6  loop      scan_file         db.py       2
#      5  loop      pick_file         config.py   2
#      4  loop      scan_file         config.py   1
#      3  loop      pick_file         app.py      1
#      2  loop      scan_file         app.py      0
#      1  loop      pick_file         -           0
#      0  loop      collect_files     -           0
#     -1  input     __start__         -           0
# 4. повтор в walk-1, находок: 8
#    первый вызов в walk-2, находок: 4
#    checkpoints в walk-1: 24

Пункт 1: вызов без thread_id падает с ValueError до работы узлов.

Пункт 2: get_state отдаёт последний checkpoint thread. Поле next у checkpoint, это кортеж имён узлов, которые граф выполнит, если продолжить запуск с этого checkpoint. Обход закончен, выполнять нечего, поэтому next пустой: ().

Пункт 3: история идёт от свежего checkpoint к старому: двенадцать записей на десять шагов обхода. Нижняя запись, шаг -1 с источником input, хранит вход. Запись шага 0 стоит перед первым узлом, collect_files. Дальше стоят десять записей, по одной после каждого шага обхода из урока 4.

Поле next есть у каждой записи истории. В этом обходе каждый шаг выполняет один узел, поэтому в next у записи одно имя. По next и значениям состояния находят нужную точку. Например, запись шага 6 стоит перед разбором db.py: в next у неё scan_file, а current равен db.py. У верхней записи next пустой, в таблице это прочерк.

В пункте 4 второй вызов с тем же входом в thread walk-1 дал восемь находок, а первый вызов в walk-2 четыре. Вход нового вызова применяется к сохранённому состоянию так же, как правка узла. Редьюсер operator.add дописал в findings четыре новые находки к четырём прежним. Ключи queue и current без редьюсера перезаписаны. В walk-1 теперь 24 checkpoints, по двенадцать от каждого вызова.

Поэтому новый обход начинайте с нового thread_id. Вызов с тем же thread_id работает с прежним состоянием.

Продолжение после падения процесса

Чтобы продолжить обход на следующий день, пишите checkpoints на диск. SqliteSaver из langgraph.checkpoint.sqlite пишет их в файл SQLite. Метод from_conn_string работает как контекстный менеджер: открывает соединение с файлом и закрывает его на выходе из блока with. Для сервера подходит PostgresSaver из пакета langgraph-checkpoint-postgres. Этот пакет ставится отдельно.

Параметр durability у invoke и stream задаёт, когда checkpoint уходит в файл:

1) "sync": checkpoint записывается до начала следующего шага

2) "async", значение по умолчанию: checkpoint записывается в фоне, пока идёт следующий шаг

3) "exit": состояние записывается при выходе из вызова после успешного окончания, после ошибки или на паузе для человека

Пример 02_crash_resume.py

"""Пример 2 урока 5: падение процесса и продолжение по файлу SQLite.

Обход стенда пишет checkpoints в файл walk.sqlite. Первый день идёт отдельным
процессом: на файле db.py узел scan_file обрывает процесс вызовом os._exit,
как при отключении питания. Второй день, это новый процесс с новым соединением и
новым графом: он читает из файла, что успело сохраниться, и продолжает обход
вызовом invoke(None, CONFIG). Если thread в файле пуст, обход начинается заново
с входа. Так повторяется для трёх режимов durability.
"""

import operator
import os
import subprocess
import sys
from pathlib import Path
from typing import Annotated, Literal

from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.graph import END, START, StateGraph
from typing_extensions import TypedDict

HERE = Path(__file__).parent
REPO = HERE / "mini_repo"
DB_PATH = HERE / "walk.sqlite"
CONFIG = {"configurable": {"thread_id": "walk-1"}}
START_STATE = {"root": str(REPO), "queue": [], "current": "", "findings": []}

# Признак первого дня: процесс запущен с режимом durability в аргументе.
DAY_ONE = len(sys.argv) > 1


class WalkState(TypedDict):
    root: str
    queue: list[str]
    current: str
    findings: Annotated[list[str], operator.add]


def collect_files(state: WalkState) -> dict:
    """Складывает имена файлов стенда в очередь обхода."""
    names = sorted(path.name for path in Path(state["root"]).glob("*.py"))
    return {"queue": names, "current": ""}


def pick_file(state: WalkState) -> dict:
    """Берёт следующее имя из очереди и убирает его оттуда."""
    queue = list(state["queue"])
    if not queue:
        return {"current": ""}
    return {"current": queue.pop(0), "queue": queue}


def scan_file(state: WalkState) -> dict:
    """Считает строки. В первый день на db.py процесс обрывается."""
    if DAY_ONE and state["current"] == "db.py":
        print("     сбой процесса на db.py", flush=True)
        os._exit(1)
    print("     разбор", state["current"], flush=True)
    path = Path(state["root"]) / state["current"]
    lines = len(path.read_text(encoding="utf-8").splitlines())
    return {"findings": [f"{state['current']}: строк {lines}"]}


def route_next(state: WalkState) -> Literal["scan_file", "__end__"]:
    """Пустое имя означает, что очередь закончилась."""
    if not state["current"]:
        return END
    return "scan_file"


def build(checkpointer: SqliteSaver):
    """Собирает обход с переданным чекпойнтером."""
    builder = StateGraph(WalkState)
    builder.add_node("collect_files", collect_files)
    builder.add_node("pick_file", pick_file)
    builder.add_node("scan_file", scan_file)
    builder.add_edge(START, "collect_files")
    builder.add_edge("collect_files", "pick_file")
    builder.add_conditional_edges("pick_file", route_next)
    builder.add_edge("scan_file", "pick_file")
    return builder.compile(checkpointer=checkpointer)


def remove_db() -> None:
    """Удаляет файл базы вместе со служебными файлами SQLite."""
    for suffix in ("", "-wal", "-shm"):
        Path(f"{DB_PATH}{suffix}").unlink(missing_ok=True)


def day_one(durability: str) -> None:
    """Первый день: обход до сбоя на db.py."""
    with SqliteSaver.from_conn_string(str(DB_PATH)) as checkpointer:
        build(checkpointer).invoke(START_STATE, CONFIG, durability=durability)


def day_two() -> None:
    """Второй день: новый процесс читает файл и продолжает обход."""
    with SqliteSaver.from_conn_string(str(DB_PATH)) as checkpointer:
        graph = build(checkpointer)
        saved = graph.get_state(CONFIG)
        history = len(list(graph.get_state_history(CONFIG)))
        print("   в файле checkpoints:", history)
        print("   находки в файле:", saved.values.get("findings"))
        if saved.metadata is None:
            print("   в thread пусто, обход заново:")
            result = graph.invoke(START_STATE, CONFIG)
        else:
            print("   продолжение invoke(None, CONFIG):")
            result = graph.invoke(None, CONFIG)
        print("   итог, находок:", len(result["findings"]))


if DAY_ONE:
    day_one(sys.argv[1])
else:
    for number, mode in enumerate(("sync", "async", "exit"), start=1):
        remove_db()
        print(f"{number}. durability={mode}", flush=True)
        print("   первый день:", flush=True)
        subprocess.run([sys.executable, __file__, mode])
        print("   второй день:")
        day_two()
    remove_db()

# Вывод:
# 1. durability=sync
#    первый день:
#      разбор app.py
#      разбор config.py
#      сбой процесса на db.py
#    второй день:
#    в файле checkpoints: 8
#    находки в файле: ['app.py: строк 13', 'config.py: строк 6']
#    продолжение invoke(None, CONFIG):
#      разбор db.py
#      разбор utils.py
#    итог, находок: 4
# 2. durability=async
#    первый день:
#      разбор app.py
#      разбор config.py
#      сбой процесса на db.py
#    второй день:
#    в файле checkpoints: 2
#    находки в файле: []
#    продолжение invoke(None, CONFIG):
#      разбор app.py
#      разбор config.py
#      разбор db.py
#      разбор utils.py
#    итог, находок: 4
# 3. durability=exit
#    первый день:
#      разбор app.py
#      разбор config.py
#      сбой процесса на db.py
#    второй день:
#    в файле checkpoints: 0
#    находки в файле: None
#    в thread пусто, обход заново:
#      разбор app.py
#      разбор config.py
#      разбор db.py
#      разбор utils.py
#    итог, находок: 4
#
# Строки durability=async зависят от скорости диска, у вас числа могут быть другими.

Первый день во всех трёх режимах закончился одинаково: разобраны app.py и config.py, на db.py процесс оборвался. Второй день, это новый процесс с новым графом и новым соединением. Общий у двух дней только файл walk.sqlite.

Продолжение запуска, это вызов invoke(None, CONFIG). Вход None означает: взять последний checkpoint thread и выполнить то, что от него осталось. Результат зависит от режима:

1) При "sync" в файле восемь checkpoints, последний стоит перед разбором db.py. Во второй день разобраны только db.py и utils.py.

2) При "async" в файле два checkpoints, а при "sync" к тому же месту их было восемь. Узлы стенда работают быстрее, чем SQLite записывает checkpoints. Остальные шесть ждали записи в фоне и пропали вместе с процессом. Документация называет риск такой потери небольшим, а на стенде потеря случилась в каждом из четырёх запусков подряд. Во второй день app.py и config.py разобраны заново.

3) При "exit" в файле нет ни одного checkpoint: процесс оборвался до выхода из вызова. Обход начался заново с входа.

Вывод для "async" зависит от скорости диска и у вас может быть другим.

При "sync" каждый шаг ждёт записи на диск. При "async" шаги записи не ждут, зато при сбое процесса пропадают последние записи. Режим "exit" обращается к диску реже всех. Но при сбое процесса пропадает всё, что сделал этот вызов. Если после сбоя запуск должен продолжиться с последнего законченного шага, ставьте "sync": только этот режим записывает каждый checkpoint до начала следующего шага.

Исключение в узле отличается от сбоя процесса. Исключение заканчивает вызов ошибкой, а это тоже выход из вызова. Поэтому даже в режиме "exit" последний checkpoint перед упавшим шагом записывается, это проверяет пункт 6 задания.

Отличить пустой thread от сохранённого помогает get_state: у пустого thread поле metadata равно None. Вызов invoke(None, CONFIG) для пустого thread падает с EmptyInputError. В thread законченного запуска тот же вызов возвращает итоговое состояние и не выполняет ни одного узла.

Пустой next в ответе get_state ещё не значит, что запуск закончен. В запуске примера 2 при "async" верхний checkpoint в файле стоял перед collect_files, а правка этого узла уже была сохранена отдельно от checkpoint. Такие правки разобраны в следующем разделе. Узел с сохранённой правкой в next не попадает, а его задача с правкой стоит в поле tasks. Запуск закончен, когда у get_state поле metadata не равно None, а поля next и tasks пустые.

Допустим, после сбоя граф вызвали с прежним входом вместо None. Тогда незаконченный шаг отбрасывается, а вход применяется к сохранённому состоянию, как в пункте 4 примера 1. Обход пойдёт с начала, и находки по app.py и config.py задвоятся.

Во втором процессе граф собирается из текущего кода. Между днями узлы и рёбра можно поменять: при продолжении берётся сохранённое состояние и выполняется граф, собранный сейчас. Предел шагов из урока 4 действует на один вызов. Поэтому при продолжении он отсчитывается заново.

Правки задач упавшего шага: pending writes

Шаг может состоять из нескольких задач, задача это одно выполнение узла. Например, копии узла из Send выполняются одним шагом, по задаче на копию. Checkpoint пишется, когда шаг закончен целиком. В режимах "sync" и "async" правку каждой законченной задачи чекпойнтер записывает сразу, отдельно от checkpoint, с привязкой к checkpoint начала шага. В режиме "exit" эти правки уходят в чекпойнтер при выходе из вызова. Документация называет такие правки pending writes.

Пример 03_pending_writes.py

"""Пример 3 урока 5: правки задач упавшего шага (pending writes).

Fan-out из урока 3: Send запускает по копии узла scan_file на каждый файл
стенда, все копии выполняются одним шагом. Копия для db.py падает с OSError.
Остальные три копии успевают закончить работу, и их правки чекпойнтер хранит
отдельно от checkpoint. После починки вызов invoke(None, CONFIG) выполняет
заново только упавшую копию.
"""

import operator
from pathlib import Path
from typing import Annotated

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Send
from typing_extensions import TypedDict

REPO = Path(__file__).parent / "mini_repo"
CONFIG = {"configurable": {"thread_id": "scan-1"}}
CALLS: list[str] = []
BROKEN = {"db.py"}


class ScanState(TypedDict):
    root: str
    files: list[str]
    findings: Annotated[list[str], operator.add]
    card: str


class FileState(TypedDict):
    """Состояние одной копии узла scan_file: свой файл и корень стенда."""

    root: str
    name: str


def collect_files(state: ScanState) -> dict:
    """Собирает имена файлов стенда."""
    names = sorted(path.name for path in Path(state["root"]).glob("*.py"))
    return {"files": names}


def scan_file(state: FileState) -> dict:
    """Разбирает один файл. Файл из BROKEN читается с ошибкой."""
    CALLS.append(state["name"])
    if state["name"] in BROKEN:
        raise OSError(f"не удалось прочитать {state['name']}")
    path = Path(state["root"]) / state["name"]
    lines = len(path.read_text(encoding="utf-8").splitlines())
    return {"findings": [f"{state['name']}: строк {lines}"]}


def make_card(state: ScanState) -> dict:
    """Сводит находки в одну карточку."""
    return {"card": f"разобрано файлов {len(state['findings'])}"}


def fan_out(state: ScanState) -> list[Send]:
    """По одному Send на файл: имя узла и состояние для этой копии."""
    return [
        Send("scan_file", {"root": state["root"], "name": name})
        for name in state["files"]
    ]


builder = StateGraph(ScanState)
builder.add_node("collect_files", collect_files)
builder.add_node("scan_file", scan_file, input_schema=FileState)
builder.add_node("make_card", make_card)
builder.add_edge(START, "collect_files")
builder.add_conditional_edges("collect_files", fan_out, ["scan_file"])
builder.add_edge("scan_file", "make_card")
builder.add_edge("make_card", END)
graph = builder.compile(checkpointer=InMemorySaver())

# 1. Одна копия из четырёх падает, invoke бросает её исключение.
try:
    graph.invoke({"root": str(REPO)}, CONFIG)
except OSError as error:
    print("1. запуск упал:", error)
print("   выполнены копии:", sorted(CALLS))

# 2. Что сохранил чекпойнтер: задачи шага, их результат и ошибка.
snapshot = graph.get_state(CONFIG)
print("2. next:", snapshot.next)
print("   находки:", snapshot.values["findings"])
for task in snapshot.tasks:
    outcome = repr(task.error) if task.error else task.result
    print("   задача", task.name, "->", outcome)
pinned = graph.get_state(snapshot.config)
print("   находки в самом checkpoint:", pinned.values["findings"])

# 3. Файл починили, запуск продолжается тем же thread.
CALLS.clear()
BROKEN.clear()
result = graph.invoke(None, CONFIG)
print("3. после продолжения выполнены копии:", CALLS)
print("   карточка:", result["card"])

# Вывод:
# 1. запуск упал: не удалось прочитать db.py
#    выполнены копии: ['app.py', 'config.py', 'db.py', 'utils.py']
# 2. next: ('scan_file',)
#    находки: ['app.py: строк 13', 'config.py: строк 6', 'utils.py: строк 10']
#    задача scan_file -> {'findings': ['app.py: строк 13']}
#    задача scan_file -> {'findings': ['config.py: строк 6']}
#    задача scan_file -> "OSError('не удалось прочитать db.py')"
#    задача scan_file -> {'findings': ['utils.py: строк 10']}
#    находки в самом checkpoint: []
# 3. после продолжения выполнены копии: ['db.py']
#    карточка: разобрано файлов 4

В пункте 1 копия для db.py упала, и invoke бросил OSError. Три другие копии выполнились до конца.

Пункт 2 показывает, что сохранил чекпойнтер. В next попадают только задачи без сохранённой правки. Такая задача одна, упавшая копия, поэтому в next один scan_file. В tasks у трёх задач есть result с правкой, у четвёртой error с исключением.

Метод get_state без checkpoint_id добавляет правки законченных задач к значениям состояния, поэтому в находках три строки. В самом checkpoint, прочитанном по snapshot.config с его checkpoint_id, находок нет.

В пункте 3 после починки invoke(None, CONFIG) выполнил одну копию, для db.py. Правки трёх других копий взяты из чекпойнтера, и карточка собрана по четырём находкам.

Допустим, в копии стоит вызов модели. Если копия закончила работу до падения соседней, при продолжении вызов модели не повторится. Повторные попытки упавшего узла без участия человека разберу в уроке 8.

Повтор от прошлого checkpoint

Вызов invoke(None, config) с config старой записи из истории повторяет запуск от этой точки. Документация называет такой повтор replay. Узлы до точки не выполняются, их результат уже сохранён. Узлы после точки выполняются заново, вместе с вызовами модели и запросами к API, и результат у них может быть другим.

Пример 04_replay.py

"""Пример 4 урока 5: повтор запуска от прошлого checkpoint (replay).

Обход стенда проходит целиком. По истории находится checkpoint перед разбором
db.py, и граф вызывается с его config. Узлы до этой точки не выполняются, узлы
после неё выполняются заново. Старые checkpoints остаются в истории, новые
идут от выбранной точки отдельной цепочкой.
"""

import operator
from pathlib import Path
from typing import Annotated, Literal

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from typing_extensions import TypedDict

REPO = Path(__file__).parent / "mini_repo"
CONFIG = {"configurable": {"thread_id": "walk-1"}}
CALLS: list[str] = []


class WalkState(TypedDict):
    root: str
    queue: list[str]
    current: str
    findings: Annotated[list[str], operator.add]


def collect_files(state: WalkState) -> dict:
    """Складывает имена файлов стенда в очередь обхода."""
    names = sorted(path.name for path in Path(state["root"]).glob("*.py"))
    return {"queue": names, "current": ""}


def pick_file(state: WalkState) -> dict:
    """Берёт следующее имя из очереди и убирает его оттуда."""
    queue = list(state["queue"])
    if not queue:
        return {"current": ""}
    return {"current": queue.pop(0), "queue": queue}


def scan_file(state: WalkState) -> dict:
    """Считает строки в выбранном файле и отмечается в счётчике вызовов."""
    CALLS.append(state["current"])
    path = Path(state["root"]) / state["current"]
    lines = len(path.read_text(encoding="utf-8").splitlines())
    return {"findings": [f"{state['current']}: строк {lines}"]}


def route_next(state: WalkState) -> Literal["scan_file", "__end__"]:
    """Пустое имя означает, что очередь закончилась."""
    if not state["current"]:
        return END
    return "scan_file"


def checkpoint_id(config: dict) -> str:
    """Достаёт checkpoint_id из config снимка."""
    return config["configurable"]["checkpoint_id"]


builder = StateGraph(WalkState)
builder.add_node("collect_files", collect_files)
builder.add_node("pick_file", pick_file)
builder.add_node("scan_file", scan_file)
builder.add_edge(START, "collect_files")
builder.add_edge("collect_files", "pick_file")
builder.add_conditional_edges("pick_file", route_next)
builder.add_edge("scan_file", "pick_file")
graph = builder.compile(checkpointer=InMemorySaver())

graph.invoke({"root": str(REPO), "queue": [], "current": "", "findings": []}, CONFIG)
history = list(graph.get_state_history(CONFIG))
print("1. первый запуск: разборов", len(CALLS), "checkpoints", len(history))

# 2. Поиск точки: следующим идёт разбор, текущий файл db.py.
before_db = next(
    item
    for item in history
    if item.next == ("scan_file",) and item.values["current"] == "db.py"
)
print("2. точка повтора: шаг", before_db.metadata["step"], "next", before_db.next)
print("   находки в точке:", before_db.values["findings"])

# 3. Повтор от этой точки: вход None, config выбранного checkpoint.
CALLS.clear()
result = graph.invoke(None, before_db.config)
print("3. повтор выполнил разборы:", CALLS)
print("   находок в итоге:", len(result["findings"]))

# 4. Новые записи истории идут от точки повтора.
fresh = list(graph.get_state_history(CONFIG))
print("4. checkpoints в thread:", len(fresh), "из них новых:", len(fresh) - len(history))
for item in fresh[: len(fresh) - len(history)]:
    following = ", ".join(item.next) or "-"
    print(f"   шаг {item.metadata['step']:>2}  {item.metadata['source']:<5}  next {following}")
fork = fresh[len(fresh) - len(history) - 1]
print(
    "   родитель записи fork, это точка повтора:",
    checkpoint_id(fork.parent_config) == checkpoint_id(before_db.config),
)

# 5. Старая точка не изменилась, повтор от последнего checkpoint не выполняет узлов.
print("5. находки в точке повтора:", graph.get_state(before_db.config).values["findings"])
CALLS.clear()
graph.invoke(None, history[0].config)
print("   повтор от конца первого запуска, разборов:", len(CALLS))

# Вывод:
# 1. первый запуск: разборов 4 checkpoints 12
# 2. точка повтора: шаг 6 next ('scan_file',)
#    находки в точке: ['app.py: строк 13', 'config.py: строк 6']
# 3. повтор выполнил разборы: ['db.py', 'utils.py']
#    находок в итоге: 4
# 4. checkpoints в thread: 17 из них новых: 5
#    шаг 11  loop   next -
#    шаг 10  loop   next pick_file
#    шаг  9  loop   next scan_file
#    шаг  8  loop   next pick_file
#    шаг  7  fork   next scan_file
#    родитель записи fork, это точка повтора: True
# 5. находки в точке повтора: ['app.py: строк 13', 'config.py: строк 6']
#    повтор от конца первого запуска, разборов: 0

В пункте 3 повтор выполнил два разбора из четырёх: db.py и utils.py. Находки по app.py и config.py уже лежали в точке повтора.

Повтор начинается записью шага 7 с источником fork. Так LangGraph помечает запись, с которой повтор отходит от старой цепочки. Значения в ней те же, что в точке повтора. Родителем у неё записана сама точка, поэтому по этой записи видно, откуда начался повтор: это показывает последняя строка пункта 4. От записи fork идут четыре записи с источником loop. Пять новых записей встали в историю рядом со старыми двенадцатью, всего их 17. Старая запись шага 6 не изменилась, это видно в пункте 5.

У записи шага 6 теперь два продолжения: первый запуск и повтор. Поэтому история thread из цепочки превратилась в дерево. get_state_history отдаёт все записи thread, свежие сверху. get_state(CONFIG) без checkpoint_id отдаёт самую свежую, то есть конец повтора.

Повтор от последней записи первого запуска узлов не выполнил: у этой записи пустой next.

Повтор выполняет узлы вместе с их побочными действиями. Поэтому требование урока 4 к идемпотентности узлов действует и здесь.

Правка состояния и развилка

Метод update_state принимает config записи истории и словарь значений. Правку он записывает новым checkpoint, следующим за этой записью, и возвращает config нового checkpoint. Сама выбранная запись не меняется.

Редьюсер сводит правку со значением ключа так же, как правку узла. Параметр as_node задаёт, от имени какого узла записана правка. От него зависит, какой узел выполнится следующим.

Пример 05_fork.py

"""Пример 5 урока 5: правка состояния и развилка (fork).

Обход стенда проходит целиком. От checkpoint перед разбором db.py делаются две
правки методом update_state: одна от имени узла scan_file, другая без
as_node. Редьюсер сводит правку со значением ключа, update_state записывает
её новым checkpoint, а старый не меняется. От правки с as_node="scan_file"
обход продолжается без разбора db.py.
"""

import operator
from pathlib import Path
from typing import Annotated, Literal

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from typing_extensions import TypedDict

REPO = Path(__file__).parent / "mini_repo"
CONFIG = {"configurable": {"thread_id": "walk-1"}}
CALLS: list[str] = []
SKIPPED = {"findings": ["db.py: пропущен вручную"]}


class WalkState(TypedDict):
    root: str
    queue: list[str]
    current: str
    findings: Annotated[list[str], operator.add]


def collect_files(state: WalkState) -> dict:
    """Складывает имена файлов стенда в очередь обхода."""
    names = sorted(path.name for path in Path(state["root"]).glob("*.py"))
    return {"queue": names, "current": ""}


def pick_file(state: WalkState) -> dict:
    """Берёт следующее имя из очереди и убирает его оттуда."""
    queue = list(state["queue"])
    if not queue:
        return {"current": ""}
    return {"current": queue.pop(0), "queue": queue}


def scan_file(state: WalkState) -> dict:
    """Считает строки в выбранном файле и отмечается в счётчике вызовов."""
    CALLS.append(state["current"])
    path = Path(state["root"]) / state["current"]
    lines = len(path.read_text(encoding="utf-8").splitlines())
    return {"findings": [f"{state['current']}: строк {lines}"]}


def route_next(state: WalkState) -> Literal["scan_file", "__end__"]:
    """Пустое имя означает, что очередь закончилась."""
    if not state["current"]:
        return END
    return "scan_file"


builder = StateGraph(WalkState)
builder.add_node("collect_files", collect_files)
builder.add_node("pick_file", pick_file)
builder.add_node("scan_file", scan_file)
builder.add_edge(START, "collect_files")
builder.add_edge("collect_files", "pick_file")
builder.add_conditional_edges("pick_file", route_next)
builder.add_edge("scan_file", "pick_file")
graph = builder.compile(checkpointer=InMemorySaver())

graph.invoke({"root": str(REPO), "queue": [], "current": "", "findings": []}, CONFIG)
before_db = next(
    item
    for item in graph.get_state_history(CONFIG)
    if item.next == ("scan_file",) and item.values["current"] == "db.py"
)

# 1. Правка от имени scan_file: граф считает, что разбор db.py уже был.
as_scan = graph.update_state(before_db.config, SKIPPED, as_node="scan_file")
state = graph.get_state(as_scan)
print("1. as_node='scan_file': next", state.next, "источник", state.metadata["source"])
print("   находки:", state.values["findings"])

# 2. Та же правка без as_node: узел выводится из истории, это pick_file.
inferred = graph.update_state(before_db.config, SKIPPED)
state = graph.get_state(inferred)
print("2. без as_node: next", state.next)

# 3. Правка не трогает checkpoint, от которого сделана.
print("3. находки в исходной точке:", graph.get_state(before_db.config).values["findings"])

# 4. Развилка: продолжение от правки с as_node="scan_file".
CALLS.clear()
result = graph.invoke(None, as_scan)
print("4. разборы после развилки:", CALLS)
for line in result["findings"]:
    print("   ", line)

# Вывод:
# 1. as_node='scan_file': next ('pick_file',) источник update
#    находки: ['app.py: строк 13', 'config.py: строк 6', 'db.py: пропущен вручную']
# 2. без as_node: next ('scan_file',)
# 3. находки в исходной точке: ['app.py: строк 13', 'config.py: строк 6']
# 4. разборы после развилки: ['utils.py']
#     app.py: строк 13
#     config.py: строк 6
#     db.py: пропущен вручную
#     utils.py: строк 10

В пункте 1 правка записана от имени scan_file. Редьюсер дописал находку к двум прежним. Ребро из scan_file ведёт в pick_file, поэтому следующим стал pick_file, и разбор db.py не выполнится. Новая запись получила в metadata источник update.

В пункте 2 as_node не задан. Тогда LangGraph берёт узел, который выполнился перед точкой последним, здесь это pick_file. Поэтому следующим остался scan_file. Продолжение от этой правки разобрало бы db.py, и в находках стояли бы обе строки: пометка о пропуске и число строк db.py. Если перед точкой одним шагом выполнялись несколько узлов, LangGraph не может выбрать один, и update_state без as_node падает с InvalidUpdateError.

В пункте 3 запись шага 6, от которой сделана правка, по-прежнему хранит две находки. update_state старый checkpoint не меняет: правка становится новой записью истории.

В пункте 4 вызов invoke(None, as_scan) продолжил запуск от правки, и выполнился один разбор, utils.py. Такое продолжение от исправленного состояния документация называет fork, развилка. В истории у развилки другой источник. Её первая запись получает источник update, а fork ставит повтор из прошлого раздела.

Повтор и развилка отличаются началом: повтор идёт от точки как есть, развилка от правки. В обоих случаях узлы после точки выполняются заново, а в thread добавляется новая цепочка записей.

Практическое задание

Соберите обход стенда, который переживает ошибку узла и продолжается по файлу SQLite.

Требования:

1) граф обхода из этого урока скомпилирован с SqliteSaver, открытым методом from_conn_string

2) узел scan_file бросает OSError на файле db.py

3) первый вызов идёт с durability="sync", исключение перехватывается

4) после ошибки скрипт открывает файл базы новым соединением, собирает граф заново и по get_state проверяет, что следующим идёт scan_file, а текущий файл db.py

5) db.py пропускается правкой update_state от имени scan_file с находкой "db.py: пропущен", обход продолжается вызовом invoke(None, config)

Как проверить результат:

1) итог содержит четыре находки, одна из них "db.py: пропущен"

2) узел scan_file вызывался для app.py, config.py и db.py в первом вызове и только для utils.py после продолжения

3) в истории thread есть checkpoint с источником update

4) checkpoint, от которого сделана правка, по-прежнему хранит две находки

5) граф, собранный заново на третьем соединении с тем же файлом, отдаёт из get_state те же четыре находки

6) если первый вызов запустить с durability="exit" вместо "sync", проверки 1-5 тоже сходятся

Подсказка: config для правки возьмите из get_state, он указывает на последний checkpoint thread.

Итоги урока

Чекпойнтер, переданный в compile, сохраняет checkpoints в thread с ключом thread_id. Вызов с тем же thread_id работает с сохранённым состоянием, вызов с новым thread_id начинается с пустого состояния.

get_state отдаёт последний checkpoint thread вместе с правками законченных задач, get_state_history все checkpoints, свежие сверху. Нужную точку находят по next и значениям состояния.

invoke(None, config) продолжает запуск с последнего checkpoint, в том числе в новом процессе по файлу SqliteSaver. Что успеет попасть на диск до сбоя процесса, задаёт durability. "sync" пишет каждый шаг до начала следующего, "async" (по умолчанию) пишет в фоне, "exit" только при выходе из вызова.

Правки законченных задач шага чекпойнтер хранит отдельно от checkpoint. При продолжении выполняются только задачи шага, у которых правки нет.

Вызов invoke(None, config) с config старого checkpoint повторяет запуск от этой точки, это replay. Развилка, fork, начинается с update_state: метод записывает правку новым checkpoint, а исходный не меняется. Редьюсеры сводят правку так же, как правку узла. От as_node зависит, какой узел выполнится следующим. В обоих случаях узлы после точки выполняются заново, а старые записи остаются в истории.

В уроке 6 разберу interrupt(): остановку графа внутри узла до ответа человека и возобновление запуска вызовом с Command(resume=...) по тому же thread_id.

Код урока

Примеры этого урока лежат в репозитории курса, папка lesson_05. Закреплённые версии, на которых получен вывод в тексте, лежат в requirements.txt в корне репозитория.

Предыдущий урок: Циклы и остановка

Следующий урок: Пауза и человек в цикле

Подписывайтесь на мой Telegram канал

Если вам нужен ментор и вы хотите научиться разрабатывать AI агентов, пишите, обсудим условия