LangGraph: как строить агентов с циклами и ветвлением

Введение: от линейных цепочек к управляемым графам

Эволюция AI-агентов за последние два года прошла путь от простых одношаговых вызовов LLM до сложных многоэтапных систем, способных планировать, выполнять инструменты, анализировать результаты и корректировать свои действия. Изначально разработчики опирались на линейные цепочки: запрос → обработка → ответ. Однако реальные бизнес-задачи редко укладываются в строго последовательный поток. Поиск информации требует повторных запросов при неполных данных, генерация кода подразумевает проверку синтаксиса и отладку, а обслуживание клиентов часто включает эскалацию или уточняющие вопросы. Именно для таких сценариев появился LangGraph — библиотека, расширяющая экосистему LangChain механизмами явного управления потоком выполнения.

В компании AST-SOFT мы регулярно сталкиваемся с проектами, где классические цепочки LangChain оказывались слишком жёсткими. Клиенты требовали агентов, способных самостоятельно принимать решения о продолжении работы, возвращаться к предыдущим шагам при ошибках или динамически менять стратегию в зависимости от промежуточных результатов. LangGraph решает эти задачи, предоставляя декларативный API для построения графов состояний с поддержкой циклов, условных переходов и чекпоинтов. В этой статье мы разберём архитектуру LangGraph, покажем, как реализовывать циклы и ветвления, настроим валидацию результатов и разберём практический пример агента с обратной связью.

1. Почему линейные цепочки больше не работают

Классический подход к построению AI-агентов в LangChain строится на концепции Chain: последовательное выполнение шагов, где вывод одного элемента становится входом следующего. Это отлично подходит для шаблонных задач: перевод текста, генерация отчёта по фиксированным данным, простой чат-бот с контекстом. Но как только задача требует неопределённого числа итераций, линейная модель ломается.

Основные ограничения последовательных цепочек:- Отсутствие состояния между итерациями: данные передаются только вперёд, что усложняет накопление контекста или историю попыток.- Невозможность возврата назад: если LLM выдал некорректный ответ, цепочка не может автоматически вернуться к шагу генерации с уточнением.- Жёсткая структура: добавление новых условий или параллельных ветвей требует переписывания всей логики.- Проблемы с отладкой: при длинных цепочках сложно отследить, на каком именно шаге возникла ошибка или деградация качества.

LangGraph устраняет эти ограничения, представляя агента как ориентированный граф. Каждый шаг — это узел, выполняющий конкретную функцию или LLM-вызов. Рёбра определяют переходы между узлами. Глобальное состояние хранится в объекте, который обновляется по мере прохождения графа. Это позволяет реализовывать циклы while, условные переходы if/else, параллельное выполнение и точки сохранения состояния.

2. Архитектура LangGraph: узлы, рёбра и состояние

В основе LangGraph лежит концепция StateGraph. Для работы необходимо определить схему состояния, создать граф, добавить узлы и рёбра, а затем скомпилировать его в исполняемый объект.

Схема состояния обычно описывается через TypedDict или Pydantic-модель. Она определяет, какие данные будут передаваться между узлами:

from typing import TypedDict, Annotated, List
from langgraph.graph import StateGraph, START, END
import operator

class AgentState(TypedDict):
    query: str
    steps: Annotated[List[str], operator.add]
    result: str
    validation_status: str
    retry_count: int

Поле Annotated[List[str], operator.add] указывает, что список шагов будет обновляться через операцию сложения (конкатенацию), а не перезаписываться. Это критически важно для накопления истории действий.

Создание графа выглядит следующим образом:

graph = StateGraph(AgentState)

# Добавление узлов
graph.add_node("researcher", research_node)
graph.add_node("validator", validate_node)
graph.add_node("refiner", refine_node)

# Определение переходов
graph.add_edge(START, "researcher")
graph.add_conditional_edges(
    "researcher",
    route_after_research,
    {"needs_refinement": "refiner", "ready": "validator"}
)
graph.add_edge("validator", END)
graph.add_edge("refiner", "researcher")  # Цикл обратной связи

Метод add_conditional_edges принимает функцию-маршрутизатор, которая получает текущее состояние и возвращает строку с именем следующего узла. Именно этот механизм позволяет реализовывать ветвление и циклы.

3. Реализация циклов и условных переходов

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

Рассмотрим функцию маршрутизации для условного перехода:

def route_after_research(state: AgentState) -> str:
    if state.get("validation_status") == "partial":
        if state["retry_count"] < 3:
            return "refiner"
        return "validator"
    return "validator"

В этом примере агент проверяет статус валидации. Если данные неполные и количество попыток не превысило три, выполняется переход к узлу refiner, который, в свою очередь, возвращает управление в researcher. Таким образом формируется цикл обратной связи.

Для защиты от бесконечных проходов LangGraph предоставляет параметр max_iterations при компиляции:

app = graph.compile(max_iterations=10)

При достижении лимита граф выбросит исключение GraphRecursionError, что позволяет безопасно обрабатывать сбои на уровне приложения.

Условные переходы также могут использоваться для параллельного выполнения. Например, узел parallel_processor может возвращать список имён узлов, которые будут запущены одновременно, а их результаты объединятся в состоянии перед следующим шагом.

4. Валидация результатов и циклы обратной связи

Одним из ключевых преимуществ графового подхода является возможность внедрения строгих проверок между шагами. Валидация может включать:- Проверку формата вывода (JSON, CSV, регулярные выражения)- Семантическую оценку качества (через отдельную LLM-модель или scorer)- Проверку наличия обязательных полей- Анализ тональности или соответствия бизнес-правилам

Пример узла валидации:

import json
import re

def validate_node(state: AgentState) -> dict:
    result = state["result"]
    status = "valid"

    # Проверка формата JSON
    try:
        data = json.loads(result)
        if not all(k in data for k in ["summary", "sources", "confidence"]):
            status = "missing_fields"
    except json.JSONDecodeError:
        status = "invalid_format"

    # Семантическая проверка через регулярное выражение
    if status == "valid" and not re.search(r"\d{4}\.\d{1,2}\.", result):
        status = "low_confidence"

    return {
        "validation_status": status,
        "retry_count": state.get("retry_count", 0) + 1
    }

Если статус не равен valid, маршрутизатор направляет выполнение в узел доработки (refiner). Этот узел может сгенерировать новый промпт с указанием ошибок, запросить дополнительные данные или изменить параметры генерации. Цикл продолжается до тех пор, пока валидатор не подтвердит соответствие требованиям или не исчерпается лимит попыток.

Такой подход значительно повышает надёжность агентов в production-среде, где некорректный вывод может привести к финансовым потерям или нарушению SLA.

5. Практический пример: агент-исследователь с рефлексией

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

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

class ResearchState(TypedDict):
    topic: str
    search_results: Annotated[list, operator.add]
    draft: str
    feedback: str
    attempts: int

# Узел поиска
def search_node(state: ResearchState):
    # Имитация API-вызова
    results = [f"Source {i}: {state['topic']} analysis part {i}" for i in range(2)]
    return {"search_results": results, "attempts": state.get("attempts", 0) + 1}

# Узел генерации черновика
def draft_node(state: ResearchState):
    draft = f"Draft based on: {', '.join(state['search_results'])}"
    return {"draft": draft}

# Узел рефлексии
def reflect_node(state: ResearchState):
    if len(state["search_results"]) < 3 and state["attempts"] < 4:
        return {"feedback": "Нужно больше источников. Уточните запрос."}
    return {"feedback": "Готово."}

# Маршрутизатор
def route_reflection(state: ResearchState):
    if state["feedback"] != "Готово.":
        return "search"
    return "draft"

# Сборка графа
builder = StateGraph(ResearchState)
builder.add_node("search", search_node)
builder.add_node("reflect", reflect_node)
builder.add_node("draft", draft_node)

builder.add_edge(START, "search")
builder.add_conditional_edges("search", route_reflection, {"search": "search", "draft": "draft"})
builder.add_edge("reflect", "search")
builder.add_edge("draft", END)

app = builder.compile()

Этот граф демонстрирует классическую схему с обратной связью: поиск → рефлексия → при необходимости повторный поиск → генерация черновика. Функция reflect_node играет роль внутреннего критика, оценивающего качество собранных данных.

6. Отладка, мониторинг и оптимизация графовых агентов

Разработка графовых агентов требует тщательного контроля за выполнением. LangGraph предоставляет несколько инструментов для отладки:

• Checkpointing: сохранение состояния на каждом шаге через Checkpointer. Позволяет возобновлять выполнение после сбоев или откатываться к предыдущей версии состояния.
• LangGraph Studio: визуальный интерфейс для трассировки выполнения, просмотра переходов и анализа содержимого состояния в реальном времени.
• Async-выполнение: поддержка асинхронных узлов через graph.ainvoke(), что критически важно для работы с внешними API и снижения времени отклика.

Рекомендации по оптимизации:1. Кэширование вызовов LLM: используйте get_cache() или внешние решения (Redis, Upstash) для повторяющихся запросов.2. Параллельная валидация: проверяйте разные аспекты вывода одновременно (формат, безопасность, семантика) через параллельные узлы.3. Динамическая маршрутизация: избегайте жёстко заданных лимитов; используйте метрики качества для принятия решений о продолжении цикла.4. Логирование состояний: сохраняйте snapshot состояния перед каждым переходом для постфактум-анализа.

В проектах, которые реализует AST-SOFT, мы обычно внедряем кастомные чекпоинтеры с интеграцией в PostgreSQL или Redis, что позволяет масштабировать агентов на тысячи одновременных сессий без потери контекста.

7. Когда выбирать LangGraph, а когда классические инструменты

Не каждая задача требует графовой архитектуры. Выбор подхода зависит от сложности workflows, необходимости повторных попыток и требований к контролю.

LangGraph идеален для задач, где требуется:- Итеративная доработка результатов- Динамическая маршрутизация на основе промежуточных данных- Чёткий контроль над порядком выполнения и точками отката- Интеграция с внешними системами, требующая обработки ошибок

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

Заключение

LangGraph кардинально меняет подход к разработке AI-агентов, перенося фокус с линейных пайплайнов на управляемые графы состояний. Возможность явно задавать циклы, условные переходы и механизмы обратной связи позволяет создавать системы, которые не просто генерируют ответы, а последовательно решают задачи, проверяют результаты и корректируют свою работу. Это особенно важно в enterprise-среде, где надёжность, предсказуемость и контролируемость стоят выше скорости прототипирования.

Команда AST-SOFT активно применяет LangGraph в проектах по автоматизации бизнес-процессов, разработке интеллектуальных ассистентов и построению сложных аналитических пайплайнов. Мы помогаем командам переходить от экспериментальных цепочек к production-ready архитектурам, внедряя чекпоинты, мониторинг, кэширование и интеграцию с корпоративными системами. Если ваш проект требует агентов с циклами, валидацией и адаптивным поведением, графовый подход станет следующим логическим шагом в развитии вашей AI-инфраструктуры.

Будущее AI-агентов — не в бесконечных диалогах с LLM, а в структурированных, проверяемых и управляемых workflows. LangGraph предоставляет необходимый инструментарий для этого перехода, а правильное проектирование графов состояния превращает экспериментальные модели в надёжные бизнес-инструменты.