LangGraph или обычный цикл

Этот блок показывает, что LangGraph добавляет к уже написанному вами циклу, что оставляет вам и как понять, что он окупится.

Возвраты дороже €200 в Northwind теперь ждут менеджера, а он может ответить завтра утром. Сервер перезапускается при каждой выкатке. Цикл из прошлого блока держит messages в локальной переменной, и возврату, который ждёт до утра, негде жить.

Папка с делом в конторе

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

LangGraph и есть такая контора.

Ваш цикл из прошлого блока — это контора с двумя столами. Проверка причины остановки стала табличкой.

В таком простом цикле вы ничего не выигрываете: логика та же, а новых слов прибавилось. Выигрыш начинается с копий и шкафа.

Один возврат через контору

Вот папка для возврата. Часть с моделью опущена, чтобы был виден сам механизм.

class Case(TypedDict):
    order_id: str
    amount_cents: int
    approved: bool
    log: Annotated[list[str], operator.add]

Узел возвращает только то, что меняет. По умолчанию новое значение заменяет старое. У log есть reducer, operator.add, поэтому каждое изменение дописывается. Без него каждый стол выбрасывал бы листы всех предыдущих.

Стол, который ждёт, вызывает interrupt:

def ask_manager(case: Case):
    answer = interrupt({"order_id": case["order_id"],
                        "amount_cents": case["amount_cents"]})
    return {"approved": answer == "approve",
            "log": [f"manager said {answer}"]}

Таблички — это обычные функции, которые называют следующий узел:

def route(case: Case) -> Literal["ask_manager", "refund"]:
    return "ask_manager" if case["amount_cents"] > 20000 else "refund"

builder.add_conditional_edges("propose", route)

Ваш код запускает прогон с номером дела, и граф останавливается у стола менеджера:

config = {"configurable": {"thread_id": "case-4417"}}
result = graph.invoke({"order_id": "4417", "amount_cents": 34900,
                       "approved": False, "log": []}, config)
result["__interrupt__"]   # заказ и сумма для экрана менеджера

Теперь процесс может завершиться. Наутро новый процесс собирает тот же граф с тем же чекпоинтером и передаёт ответ:

graph.invoke(Command(resume="approve"), config)

Значение из Command(resume=...) становится тем, что вернёт interrupt. Документация LangGraph теперь советует вести такие прогоны через stream_events(..., version="v3"). Этот режим пока помечен как экспериментальный, а invoke по-прежнему работает, и здесь он короче.

На картинке пять чекпоинтов на три узла: один пустой в начале, один со входными данными и по одному после каждого узла. ask_manager выполнился дважды, потому что, как говорилось в блоке о надёжных прогонах, узел начинает работу с самого начала. SqliteSaver годится для локальной работы, а для продакшена документация LangGraph называет PostgresSaver. По умолчанию чекпоинт пишется в фоне, пока идёт следующий шаг (durability="async"), поэтому есть небольшой риск, что при падении последний чекпоинт потеряется. durability="sync" дожидается каждой записи и только потом идёт дальше.

Новый thread_id — это пустая папка. Повтор, который придумывает свежий id, заводит дело заново, а не продолжает его.

Та же пауза в обычном цикле

Обычный цикл справляется с этим лучше, чем принято думать, потому что messages и так простые данные. Сохранили, остановились, продолжили, когда пришёл ответ:

def pause(case_id, messages, results, block):
    db.save(case_id, {"messages": messages, "results": results,
                      "pending": block})

def resume(case_id, decision):
    run = db.load(case_id)
    answer = apply_decision(run["pending"], decision)   # tool_result
    run["messages"].append({"role": "user",
                            "content": run["results"] + [answer]})
    return run_agent(run["messages"])

Результаты остальных вызовов из того же ответа хранятся вместе с паузой, потому что в API Anthropic все они идут одним сообщением. Если точка паузы одна, вам хватит таблицы и двух функций, и все промпты остаются в вашем коде. Сравнение меняется, когда растут требования:

Последний вопрос на картинке — о движках надёжного выполнения вроде Temporal из блока о надёжных прогонах.

Чем вы платите и что остаётся за вами

Совет Anthropic 2024 года: начинайте с прямых вызовов API, а если берёте фреймворк, понимайте код под ним, потому что неверные предположения о нём — частая причина ошибок. LangGraph тут стоит низко: в его документации сказано, что он не прячет промпты и не навязывает архитектуру.

Цена всё равно есть. Придётся разобраться в его модели выполнения, состояние должно переживать сериализацию, а старые чекпоинты копятся, пока вы их не почистите или не зададите срок хранения. С 10 ноября 2025 по 10 ноября 2026 у него вышло 50 релизов на PyPI, считая патчи и 10 предварительных версий, так что фиксируйте версию.

Граф не делает возврат правильным. Проверка владельца, суммы и ключ идемпотентности из прошлых блоков по-прежнему живут в вашем узле refund.

Вывод

• LangGraph даёт состояние, узлы, рёбра, чекпоинт на каждый шаг и паузу с продолжением по thread_id.
• Одна пауза на одном пути в обычном цикле обходится дёшево. Граф окупается там, где есть несколько пауз, ветвления и восстановление.
• Промпты в LangGraph остаются вашими, и проверки того, что может инструмент, тоже.

Глубже