Основы LangGraph: состояние, узлы, рёбра

LangGraph моделирует рабочие процессы как графы. Узлы делают работу, рёбра определяют порядок.

#Графовый подход

LangGraph — низкоуровневый фреймворк оркестрации для долгоживущих, состоятельных агентов. Он позволяет смешивать детерминированные и агентивные шаги в одном графе, сохранять состояние, встраивать человека в цикл и передавать данные в реальном времени.

Граф состоит из трёх компонентов: состояние (общая структура данных), узлы (функции, выполняющие работу) и рёбра (связи, определяющие порядок).

#Состояние

Состояние определяется через TypedDict, dataclass или Pydantic BaseModel. Рекомендуется TypedDict — он быстрее и проще:

from typing_extensions import TypedDict

class State(TypedDict):
    messages: list[dict]
    step: int

#Редукторы

Редуктор определяет, как обновляется значение в состоянии. По умолчанию — замена. Для накопления используется add или специальный редуктор add_messages для работы с сообщениями:

from typing import Annotated
from langchain.messages import AnyMessage
from langgraph.graph.message import add_messages

class State(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]

add_messages добавляет новые сообщения к списку и обновляет существующие по идентификатору. Это стандартный редуктор для агентов с историей диалога.

Для удобства существует готовое состояние MessagesState с полем messages и редуктором add_messages. Можно расширить его дополнительными полями.

#Узлы

Узлы — это функции, которые принимают состояние и возвращают обновление. Узел может быть синхронным или асинхронным, может принимать дополнительные параметры config и runtime.

def my_node(state: State):
    return {"step": state["step"] + 1}

Узлы добавляются через builder.add_node("имя", функция). Если имя не указано, используется имя функции.

#Рёбра

Обычное ребро — всегда переход от одного узла к другому. Условное ребро — функция выбирает следующий узел на основе состояния.

from langgraph.graph import START, END

builder.add_edge(START, "node_a")        # точка входа
builder.add_edge("node_a", "node_b")     # обычное ребро
builder.add_edge("node_b", END)          # завершение

START и END — специальные узлы: точка входа и точка завершения графа.

#Компиляция

Перед использованием граф нужно скомпилировать через builder.compile(). Компиляция проверяет структуру и позволяет указать runtime-параметры (чекпоинтер, точки останова).

#Полный пример

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

class State(TypedDict):
    messages: list[dict]
    step: int

def greet(state: State):
    return {"messages": [{"role": "ai", "content": "Привет!"}], "step": state["step"] + 1}

def farewell(state: State):
    return {"messages": [{"role": "ai", "content": "До свидания!"}], "step": state["step"] + 1}

builder = StateGraph(State)
builder.add_node("greet", greet)
builder.add_node("farewell", farewell)
builder.add_edge(START, "greet")
builder.add_edge("greet", "farewell")
builder.add_edge("farewell", END)

graph = builder.compile()
result = graph.invoke({"messages": [], "step": 0})

#Супершаг

Одна итерация по узлам графа — супершаг. Узлы, работающие параллельно, принадлежат одному супершагу. Последовательные узлы — разным.