Entry point, END
и компиляция графа

Две фазы жизни графа

Главная идея, которую стоит усвоить: описание графа и его выполнение — это разные сущности. Объект StateGraph (builder) — это всего лишь чертёж: список узлов и рёбер, который можно дополнять. Пока ты вызываешь add_node и add_edge, ничего не выполняется. Чтобы получить то, что умеет считать, чертёж нужно скомпилировать.

Зачем разделять? Потому что компиляция — это момент, когда LangGraph проверяет граф на целостность и «запекает» в него инфраструктуру: правила слияния состояния, сохранение (checkpointer), точки прерывания. Скомпилированный граф неизменяем — это надёжный, провалидированный артефакт, который можно переиспользовать и запускать многократно.

Точка входа: START и set_entry_point

Точка входа (entry point) — узел, в который попадает стартовое состояние, переданное в invoke. Мы уже задавали её ребром от служебного псевдоузла START:

from langgraph.graph import StateGraph, START, END

builder = StateGraph(State)
builder.add_node("agent", agent_node)

# Способ 1: ребро от START (мы так и делали)
builder.add_edge(START, "agent")

# Способ 2: явный метод — то же самое, читается чуть короче
builder.set_entry_point("agent")

Оба варианта идентичны: set_entry_point("agent") — это просто синтаксический сахар над add_edge(START, "agent"). START — не настоящий узел и не выполняет код; он лишь маркирует, куда подать вход.

Условная точка входа

Иногда нужно выбрать первый узел в зависимости от входа — например, направить запрос в разные ветки сразу. Для этого есть условная точка входа: маршрутизатор работает прямо от START.

def pick_start(state: State) -> str:
    return "vip_flow" if state["is_vip"] else "normal_flow"

# Эквивалент add_conditional_edges, но из START
builder.set_conditional_entry_point(
    pick_start,
    {"vip_flow": "vip_node", "normal_flow": "normal_node"},
)

END и точки выхода

END — зеркальный START служебный псевдоузел: он отмечает, что выполнение на этой ветке завершено. Когда поток управления доходит до END, граф останавливается и invoke возвращает текущее состояние. Как и со входом, есть два способа задать выход:

# Способ 1: ребро в END
builder.add_edge("polish", END)

# Способ 2: явный метод — то же самое
builder.set_finish_point("polish")

Важные свойства END:

• Точек выхода может быть несколько. Разные ветки могут вести в END независимо — граф завершится по той, что сработала.
• END — это не «return». Он не выбирает, что вернуть: invoke всегда отдаёт всё состояние целиком (или поля выходной схемы, если ты её задал — см. урок про State).
• До END дойти обязательно. Если ни одна ветка не ведёт в END, граф либо зациклится, либо упрётся в тупиковый узел — это поймает валидация при компиляции.

Что делает compile()

compile() превращает builder в исполняемый CompiledStateGraph. За этим вызовом стоит больше, чем кажется:

1. Валидация структуры — проверка, что граф связный и корректный (подробно ниже).
2. Сборка движка — построение внутреннего представления (каналы состояния, reducers по аннотациям полей).
3. Подключение инфраструктуры — checkpointer для памяти и точки прерывания для human-in-the-loop передаются именно сюда.

from langgraph.checkpoint.memory import MemorySaver

graph = builder.compile(
    checkpointer=MemorySaver(),        # сохранение состояния (урок про checkpointing)
    interrupt_before=["tools"],        # пауза ПЕРЕД узлом (human-in-the-loop)
    interrupt_after=[],                # пауза ПОСЛЕ узла
    debug=False,                       # подробный лог выполнения
)

Все параметры опциональны — в простейшем случае хватает голого builder.compile(). Но запомни главное: память и точки прерывания — это свойства скомпилированного графа, а не builder'а. Их задают здесь, и мы вернёмся к ним в разделе «Продвинутые возможности».

Валидация графа при компиляции

Главная польза отдельной фазы компиляции — ошибки структуры ловятся сразу, а не в середине прогона на проде. compile() проверяет, в частности:

builder = StateGraph(State)
builder.add_node("agent", agent_node)
builder.add_node("tools", tools_node)
builder.add_edge(START, "agent")
builder.add_edge("agent", "tools")
# забыли: builder.add_edge("tools", END)  и  выход из tools

graph = builder.compile()
# ValueError: узел 'tools' — тупик: нет исходящего ребра и пути к END
# Ошибка возникает СЕЙЧАС (при компиляции), а не на первом запросе

Скомпилированный граф — это Runnable

CompiledStateGraph реализует интерфейс Runnable из LangChain. Это значит, что у графа единый, предсказуемый набор методов запуска — те же, что у любой LangChain-цепочки:

# Синхронно, целиком → финальное состояние
state = graph.invoke({"query": "..."}, {"recursion_limit": 25})

# Стриминг по узлам (что вернул каждый шаг)
for chunk in graph.stream({"query": "..."}):
    print(chunk)

# Пакетная обработка нескольких входов
states = graph.batch([{"query": "A"}, {"query": "B"}])

# Асинхронные версии — для async-приложений
state = await graph.ainvoke({"query": "..."})
async for chunk in graph.astream({"query": "..."}):
    print(chunk)

Второй аргумент всех методов — config: словарь рантайм-настроек. Самые частые ключи — recursion_limit (лимит шагов) и configurable (например, {"thread_id": "user-42"} для выбора сессии при включённом checkpointer).

Визуализация графа

Раз граф — это данные, его можно нарисовать. У скомпилированного графа есть метод get_graph(), а у результата — экспорт в Mermaid, ASCII или PNG. Это лучший способ проверить, что структура получилась той, что задумана.

# Текстовая Mermaid-схема — вставляется в Markdown/доку
print(graph.get_graph().draw_mermaid())
# %%{init: ...}%%
# graph TD;
#   __start__ --> agent;
#   agent -.-> tools;
#   agent -.-> __end__;
#   tools --> agent;

# ASCII прямо в терминал
graph.get_graph().print_ascii()

# PNG-картинка (нужны доп. зависимости)
png_bytes = graph.get_graph().draw_mermaid_png()
open("graph.png", "wb").write(png_bytes)

Типичные ошибки

Шпаргалка

from langgraph.graph import StateGraph, START, END

# ФАЗА 1 — описание (builder)
builder = StateGraph(State)
builder.add_node("a", node_a)

# точка входа: два эквивалентных способа
builder.add_edge(START, "a")          # = builder.set_entry_point("a")
# точка выхода: два эквивалентных способа
builder.add_edge("a", END)            # = builder.set_finish_point("a")
# условный вход
builder.set_conditional_entry_point(router, {"x": "a", "y": "b"})

# ФАЗА 2 — компиляция (один раз!)
graph = builder.compile(
    checkpointer=MemorySaver(),       # память (опционально)
    interrupt_before=["a"],           # пауза (опционально)
)

# Запуск — граф это Runnable
graph.invoke(state, {"recursion_limit": 25})
graph.stream(state);  graph.batch([s1, s2])
await graph.ainvoke(state)

# Визуализация
print(graph.get_graph().draw_mermaid())
graph.get_graph().print_ascii()

# Правила:
#  • builder описывает, compile() проверяет и собирает
#  • compile() ловит тупики/опечатки/недостижимые узлы
#  • компилируй один раз, переиспользуй объект
#  • checkpointer/interrupts → в compile(), не в StateGraph()

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

Поработай с жизненным циклом и валидацией графа:

Что дальше

Поздравляю — это последний урок раздела «Основы LangGraph». Ты умеешь собирать состояние, узлы, рёбра, ветвление и превращать всё это в исполняемый граф. Дальше — продвинутые возможности, которые делают агента по-настоящему рабочим.