LangGraph с нуля:
StateGraph, узлы и рёбра

Зачем агенту граф?

Типичный агент из прошлого модуля — это цикл: спросили LLM → она попросила вызвать инструмент → вызвали → вернули результат → снова спросили LLM → ... → пока не получили финальный ответ. На голом Python это выглядит так:

messages = [{"role": "user", "content": query}]

while True:
    response = llm.invoke(messages)
    messages.append(response)

    if not response.tool_calls:        # LLM решила, что готова ответить
        break

    for call in response.tool_calls:   # надо вызвать инструменты
        result = run_tool(call)
        messages.append(result)

    if len(messages) > 50:             # а вдруг зациклились?
        break
    # ... а здесь хочется логировать, делать паузу на подтверждение,
    #     откатываться к прошлому шагу, ветвиться по типу ответа ...

Пока шагов мало — терпимо. Но как только появляются ветвления, паузы на подтверждение человеком, откаты к прошлому шагу и параллельные ветки — управляющая логика расползается по if-ам, и её невозможно ни нарисовать, ни отладить.

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

Модель вычислений: узлы, рёбра, состояние

LangGraph заимствует модель из теории графов и из таких систем, как Apache Beam / Pregel: вычисление описывается как направленный граф. У него три сущности, которые нужно понять до единой строчки кода.

• Node (узел) — обычная Python-функция. Получает текущее состояние, возвращает изменения к нему. Здесь живёт вся работа: вызов LLM, инструмента, парсинг, любая логика шага.
• Edge (ребро) — связь «после узла A иди в узел B». Рёбра задают порядок выполнения.
• State (состояние) — общая «доска», которую видят все узлы: единственный способ передавать данные между шагами. В этом уроке относимся к нему просто как к общему словарю; детально схему состояния разберём в следующем уроке.

Граф всегда стартует из служебного узла START и завершается в END. Выполнение идёт по рёбрам: узел отрабатывает, его изменения вливаются в общее состояние, движок смотрит на исходящее ребро и шагает в следующий узел — и так до END.

Это самый простой граф — линейный конвейер: START → draft → polish → END. Состояние течёт слева направо, каждый узел дописывает в него свой результат. Именно такой граф мы и соберём в этом уроке. Ветвление (когда из узла можно пойти в разные стороны) появится позже — в уроке про conditional edges.

Узлы: функции, делающие работу

Узел — это просто функция state → изменения состояния. Никакой магии: на вход приходит текущее состояние (словарь), на выходе — словарь с теми полями, которые надо обновить. Вся «работа» агента живёт именно в узлах.

Чтобы граф знал, какие поля вообще есть в состоянии, его схему описывают типом. По умолчанию для этого берут TypedDict — обычный словарь с объявленными полями. Подробно про схему состояния — в следующем уроке; сейчас достаточно объявить пару полей и идти дальше.

from typing import TypedDict

# Схема состояния — пока просто общий словарь с двумя полями
class State(TypedDict):
    text: str      # текущий текст
    steps: int     # сколько шагов сделали

def draft_node(state: State) -> dict:
    """Первый шаг: пишет черновик."""
    draft = f"Черновик: {state['text']}"
    # Возвращаем ТОЛЬКО изменённые ключи
    return {"text": draft, "steps": state["steps"] + 1}

def polish_node(state: State) -> dict:
    """Второй шаг: дорабатывает текст."""
    polished = state["text"].replace("Черновик", "Готово")
    return {"text": polished, "steps": state["steps"] + 1}

Ключевая деталь: узел возвращает не всё состояние целиком, а только те ключи, которые меняет. LangGraph сам вольёт их в общую «доску». Поэтому узлы получаются маленькими и независимыми — каждый отвечает за свой кусочек работы и ничего не знает про соседей.

Рёбра: START, END и связи между узлами

Узлы есть — теперь надо задать порядок. За это отвечают рёбра. Простейшее ребро — безусловное: «после A всегда иди в B». Граф собирается через объект StateGraph: добавляем узлы, добавляем рёбра, компилируем.

from langgraph.graph import StateGraph, START, END

builder = StateGraph(State)              # граф знает схему состояния

builder.add_node("draft", draft_node)    # имя узла → функция
builder.add_node("polish", polish_node)

builder.add_edge(START, "draft")         # вход графа → первый узел
builder.add_edge("draft", "polish")      # draft → polish
builder.add_edge("polish", END)          # последний узел → выход графа

START и END — служебные псевдоузлы. START отмечает, куда подаётся входное состояние, END — где выполнение останавливается. Между ними — твои узлы, соединённые рёбрами. Ребро ссылается на узлы по имени (строке), которое ты задал в add_node — поэтому важно не опечататься.

Компиляция и запуск: invoke и stream

Описание графа само по себе ничего не выполняет. Чтобы получить исполняемый объект, граф нужно скомпилировать методом compile(). Он проверяет граф (например, что нет узлов-сирот, в которые нельзя попасть) и возвращает объект, который уже умеет запускаться.

graph = builder.compile()      # получаем исполняемый граф

# 1. Прогон целиком — возвращает финальное состояние
result = graph.invoke({"text": "привет, мир", "steps": 0})
print(result)
# {'text': 'Готово: привет, мир', 'steps': 2}

# 2. Стриминг по узлам — видно, что вернул каждый шаг
for chunk in graph.stream({"text": "привет, мир", "steps": 0}):
    print(chunk)
# {'draft':  {'text': 'Черновик: привет, мир', 'steps': 1}}
# {'polish': {'text': 'Готово: привет, мир',   'steps': 2}}

invoke() прогоняет весь поток от START до END и отдаёт финальное состояние. stream() отдаёт результат каждого узла по мере выполнения — незаменимо для отладки и для показа прогресса пользователю. Ключ в каждом чанке — имя сработавшего узла, значение — что он вернул.

Соберём всё в один файл — это и есть твой первый полноценный граф:

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    text: str
    steps: int

def draft_node(state: State) -> dict:
    return {"text": f"Черновик: {state['text']}", "steps": state["steps"] + 1}

def polish_node(state: State) -> dict:
    return {"text": state["text"].replace("Черновик", "Готово"),
            "steps": state["steps"] + 1}

builder = StateGraph(State)
builder.add_node("draft", draft_node)
builder.add_node("polish", polish_node)
builder.add_edge(START, "draft")
builder.add_edge("draft", "polish")
builder.add_edge("polish", END)

graph = builder.compile()

print(graph.invoke({"text": "привет, мир", "steps": 0}))
# {'text': 'Готово: привет, мир', 'steps': 2}

Это та же последовательность шагов, что и в while-агенте, но выраженная явной структурой. Захочешь вставить шаг между draft и polish — добавляешь узел и переподключаешь два ребра, а не правишь логику в середине цикла.

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

Шпаргалка

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

# 1. Схема состояния (детально — в следующем уроке)
class State(TypedDict):
    text: str
    steps: int

# 2. Узел: state -> dict с изменёнными ключами
def my_node(state: State) -> dict:
    return {"steps": state["steps"] + 1}

# 3. Сборка графа
builder = StateGraph(State)
builder.add_node("a", my_node)      # имя -> функция
builder.add_node("b", my_node)
builder.add_edge(START, "a")        # вход графа
builder.add_edge("a", "b")          # порядок шагов
builder.add_edge("b", END)          # выход графа

# 4. Компиляция и запуск
graph = builder.compile()                  # builder сам не запускается!
graph.invoke({"text": "...", "steps": 0})  # весь поток -> финальное состояние
for step in graph.stream({"text": "...", "steps": 0}):
    print(step)                            # по одному узлу за раз

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

Закрепи материал, собрав граф своими руками:

Что дальше