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

Введение в LangGraph и построение AI-агентов

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

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

В этой статье рассмотрим, как использовать LangGraph для построения агентов с:

• условными переходами (ветвление)

• циклами обратной связи

• проверкой результатов выполнения шагов

• гибким управлением потоком через состояние и сообщения

Мы сосредоточимся на технических деталях реализации: структурах StateGraph, использовании messages/events, определении правил перехода, обработке ошибок. Цель — понять механизм управления сложным поведением агента без усложнения логики.

Основные концепции LangGraph

1. Граф состояний (StateGraph) как основа

В LangGraph поведение агента описывается через граф состояния, где:

• Узлы — это вызовы функций (LLM, API, действия)

• Рёбра — правила перехода между узлами на основе текущего state и результатов выполнения

Каждый шаг может возвращать не только ответ модели, но и метаданные: статус выполнения (success, retry, error) или новые данные для проверки.

Пример базового графа:

from langgraph.graph import StateGraph
from typing import TypedDict

class AgentState(TypedDict):
    input_data: dict
    result: str | None = None
    step_count: int = 0

def validate_input(state: AgentState) -> tuple:
    if not state['input_data'].get('required_field'):
        return 'error', {'message': 'Missing required field'}
    elif len(str(state['input_data']['value'])) > 100:
        return 'retry', {'message': 'Input too long, retrying with subset'}

    # Если всё в порядке — продолжить дальше
    return 'continue', state

Здесь validate_input возвращает тип перехода (error, retry, continue) и возможные данные для следующего шага. Это позволяет строить не линейные, а ветвящиеся потоки.

2. Управление переходами через условные правила

LangGraph поддерживает два способа определения путей:

1. Через явное перечисление всех возможных состояний (edges = {})

2. Через динамическое определение правил на основе результата выполнения (рекомендуется для сложных случаев)

Пример: динамический выбор следующего шага

from langgraph.checkpoint import MemorySaver
from langgraph.states import StateGraph
from typing import Literal, TypedDict
import operator

class AppState(TypedDict):
    query: str = ''
    results: list[str] | None = None
    error: bool = False
    stage: Literal['start', 'searching', 'processing', 'done'] = 'start'

def route_query(state: AppState) -> tuple:
    """
    Определяет следующий шаг на основе текущего состояния.
    """

    if state["stage"] == "start":
        return ("search", {})

    elif state.get("results") and not state.get("error"):  # Успешный поиск
        return ("process_results", {})

    elif state.get("error"):  # Ошибка — повторить запрос
        return ("retry_query", {})

    else:
        raise ValueError(f"Unknown stage: {state['stage']}")

Здесь route_query анализирует текущее состояние и возвращает имя следующего узла (search, process_results) вместе с данными для передачи. Это позволяет реализовать логическое ветвление без жёсткой привязки к порядку.

3. Реализация циклов обратной связи

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

Пример: сбор данных по частям (например, загрузка большого файла)

def fetch_chunk(state: AgentState) -> tuple[Literal["continue", "done"], dict]:
    chunk = download_next_part(state['input_data']['file_id'])
    if not chunk:
        return 'done', state

    # Обновляем состояние — добавляем новый фрагмент в result
    updated_state = {
        **state,
        'result': f"{state['result']} {chunk.text}"
    }

    return "continue", updated_state
def complete_task(state: AgentState) -> tuple:
    if len(str(state['result'])) > 500:
        return "success", state

    else:
        # Не хватает данных — повторить запрос на следующий фрагмент
        raise RuntimeError("Data not fully collected yet. Resuming from last point.")

В этом примере fetch_chunk работает в цикле: пока не загружен весь файл (или достигнут лимит), агент продолжает вызывать этот шаг. При достижении условия (len(result) > 501) — завершается успешно.

Важно: LangGraph сам управляет выполнением цикла, если вы возвращаете 'continue'. Это предотвращает бесконечные петли при неправильной логике.

4. Обработка ошибок и повторные попытки (retry logic)

LangGraph поддерживает механизм retry через специальное значение в результатах шагов ("retry", data). Это позволяет реализовать стратегию повтора без необходимости вручную управлять счётчиком попыток.

Пример: безопасный вызов API с ограничениями по частоте

def call_vulnerable_api(state: AgentState) -> tuple:
    try:
        response = api_request(state['input_data'])
        return 'success', {"result": response.data}

    except RateLimitError:
        # Ограничение частоты — повтор через 5 секунд
        import time
        time.sleep(5)
        return "retry", state

    except Exception as e:
        print(f"Unexpected error: {e}\")
        raise RuntimeError(f"Failed after retry logic. Error: {str(e)}") from e

Когда call_vulnerable_api возвращает 'retry', LangGraph автоматически повторит этот узел, пока не будет возвращён 'success'. Это работает только при использовании checkpointing (см. ниже).

⚠️ Без MemorySaver или аналогичного механизма состояние теряется между попытками — важно всегда использовать сохранение состояния.

Хранение и восстановление состояния (checkpointing)

Для работы с циклами, повторными попытками и ветвлением необходимо сохранять текущее состояние агента. LangGraph предоставляет встроенный механизм MemorySaver:

from langgraph.checkpoint import MemorySaver
memory_saver = MemorySaver()
graph = StateGraph(AgentState).add_conditional_edge(route_query, graph)
app = graph.compile(checkpointer=memory_saver)

Это позволяет:

• Восстановить процесс после прерывания (например, при тайм-ауте)

• Продолжить с того же места после ошибки или повторной попытки

• Реализовать долгосрочные агенты, которые работают часами/днями

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

Практический пример: агент для обработки пользовательского запроса

Рассмотрим задачу:

Пользователь отправляет запрос, содержащий имя файла и тип операции (upload, download). Агент должен: 1. Проверить наличие файла (если download) — если нет, вернуть ошибку 2. Если файл есть — загрузить его по частям до достижения нужного размера или полного завершения 3. При ошибке загрузки — повторить попытку (до 5 раз) 4. После успешной загрузки — отправить результат пользователю

Структура состояния:

class ProcessingState(TypedDict):
    user_id: str
    file_name: str
    operation_type: Literal["upload", "download"] = "download"
    current_chunk: int = 0
    total_size_required: int | None = None
    downloaded_size: int = 0
    error_count: int = 0
    status: Literal[
        "checking_existence", 
        "downloading_part",
        "processing_finished",
        "error" "] = "checking_existence"

Шаги агента:

1. check_file_exists — проверяет, есть ли файл на сервере

2. start_download_loop — запускает цикл загрузки по частям (до 5 попыток)

3. aggregate_results_and_send — собирает и отправляет финальный ответ

4. При ошибке — переходит в состояние error, где можно отправить уведомление о сбое

Реализация цикла с ограничением:

def download_part(state: ProcessingState) -> tuple[Literal, dict]:
    if state['error_count'] >= 5:
        return "finished_with_error", state

    try:
        chunk = fetch_file_chunk(state['file_name'], offset=state['current_chunk'])
        downloaded_size += len(chunk.text)
        # Обновляем состояние — добавляем новый фрагмент в result (если нужно)
        updated_state = {
            **state,
            "downloaded_size": downloaded_size
        }

        if chunk.is_last:
            return "finished_successfully", updated_state
        else:
            return "continue_downloading", updated_state
    except Exception as e:
        state["error_count"] += 1
        print(f"Retry {state['error_count']}: {e}")
        return "retry_download", state

В этом случае download_part возвращает 'continue_downloading', и LangGraph повторит шаг, пока не будет достигнут лимит попыток.

Заключение: ключевые преимущества подхода на основе графов

Использование LangGraph для построения AI-агентов даёт следующие преимущества: | Преимущество | Описание | |-------------|--------| | Гибкость потока | Возможность ветвления, циклов и повторных попыток без изменения кода основных узлов | | Ясность логики | Графовая структура делает поведение агента прозрачным — видно все возможные пути | | Устойчивость к ошибкам | Механизм retry и checkpointing обеспечивает надёжное восстановление после сбоев | | Масштабируемость | Легко добавлять новые узлы (шаги), правила перехода, условия остановки |

Ресурсы для дальнейшего изучения

• LangGraph Documentation — официальные туториалы и примеры

• langraph GitHub repository — исходный код с примерами использования

• LangChain + Graphs: сравнительный анализ подходов к построению агентов (блоговые статьи)

• Примеры в Colab: поиск по ключевым словам "LangGraph workflow control"

Контакты и поддержка

Если у вас возникли вопросы при реализации своих агентов, вы можете:

• Посмотреть примеры на GitHub

• Обратиться к сообществу LangChain/LangGraph (Discord, Stack Overflow)

• Изучить официальные туториалы по шагам: от простого графа до сложных бизнес-процессов

Лицензия и использование материалов

Данный материал подготовлен для образовательных целей. Все примеры кода можно использовать свободно при соблюдении условий лицензии LangChain/LangGraph (MIT).