LangGraph Time Travel

Time-travel — отладка через историю чекпоинтов: возможность вернуться в любую сохранённую точку исполнения и либо воспроизвести её (Replay), либо изменить состояние и запустить новую ветку (Fork). Существует только благодаря LangGraph Checkpointers.

Суть

Каждый шаг графа сохраняется как чекпоинт. Имея историю, можно откатиться к нужному снимку и: повторить тот же ход (Replay) или подменить состояние и пойти по альтернативному пути (Fork). Это превращает недетерминированного агента в воспроизводимый процесс.

Зачем это нужно

Из-за недетерминизма LLM классическая отладка почти невозможна: один и тот же вход даёт разные ответы, баг не воспроизвести как в обычном backend. Time-travel решает это: фиксируем состояние до ошибки и проигрываем его сколько угодно раз, проверяя гипотезы.

Как работает

  • Replay — воспроизведение state с прошлого чекпоинта (тот же thread_id).
  • Fork — ответвление от выбранного чекпоинта: конфиг наводится на конкретный снимок, aupdate_state возвращает конфиг новой ветки, по нему и запускается альтернативный ход.
  • История доступна через aget_state_history(thread, limit=...) — список снимков с checkpoint_id и значениями.

Где живёт checkpoint_id — место, на котором легко ошибиться. Это не аргумент методов, а ключ внутри configurable, рядом с thread_id и checkpoint_ns. Сигнатура aupdate_state — (config, values, as_node=None, task_id=None), и попытка передать checkpoint_id= отдельным аргументом кончается TypeError. Точку в истории задаёт конфиг, а не вызов.

  • Стоимость replay: если на пути дорогие LLM-узлы, повторный проход недёшев — их результаты кэшируют в state, чтобы переигрывать только дешёвую часть (см. LangGraph Reliability, Agent CostControl).

Пример

history = [s async for s in app.aget_state_history(thread)]
old = history[1]                                  # откат к checkpoint_2

# точку ветвления задаёт конфиг: thread_id + checkpoint_id
branch_point = old.config                         # уже содержит оба ключа

# update_state возвращает конфиг НОВОЙ ветки — по нему и запускаемся
fork_config = await app.aupdate_state(branch_point, {"messages": "другой ввод"})
await app.ainvoke(None, fork_config)              # новая ветка «а что если?»

ainvoke(None, ...) здесь означает «продолжить с сохранённого состояния, нового входа нет» — вход уже внесён предыдущим вызовом.

Связано с

  • LangGraph Checkpointers — time-travel физически опирается на чекпоинты
  • LangGraph Observability — дополняет трейсинг: не только смотреть, но и переигрывать
  • LangGraph — одно из ключевых production-преимуществ перед цепочками
  • LangGraph Reliability — кэширование результатов LLM-узлов удешевляет replay