Как LangGraph обрабатывает циклы и петли — как построить агента, который итерирует до выполнения условия?

Как LangGraph реализует циклы

В LangGraph цикл — это граф, в котором одно из условных рёбер ведёт назад к уже посещённому узлу. В отличие от DAG-фреймворков, LangGraph явно поддерживает такие петли: состояние сохраняется между итерациями и обновляется редукторами.

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

from typing import Annotated
from typing_extensions import TypedDict
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

MAX_ITERATIONS = 5

class State(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    iteration: int
    solved: bool

def solver_node(state: State) -> dict:
    """Пытается решить задачу. Имитирует постепенное улучшение."""
    iteration = state.get("iteration", 0) + 1
    if iteration >= 3:
        answer = AIMessage(content=f"[Итерация {iteration}] Нашёл решение!")
        return {"messages": [answer], "iteration": iteration, "solved": True}
    else:
        attempt = AIMessage(content=f"[Итерация {iteration}] Пробую ещё...")
        return {"messages": [attempt], "iteration": iteration, "solved": False}

def critic_node(state: State) -> dict:
    """Проверяет решение и добавляет обратную связь."""
    last = state["messages"][-1].content
    feedback = AIMessage(content=f"Критик: '{last}' — нужно доработать.")
    return {"messages": [feedback]}

def should_continue(state: State) -> str:
    """Маршрутизатор: продолжать цикл или выходить?"""
    if state.get("solved", False):
        return "done"
    if state.get("iteration", 0) >= MAX_ITERATIONS:
        return "done"  # защита от бесконечного цикла
    return "retry"

builder = StateGraph(State)
builder.add_node("solver", solver_node)
builder.add_node("critic", critic_node)

builder.add_edge(START, "solver")
builder.add_conditional_edges(
    "solver",
    should_continue,
    {"retry": "critic", "done": END}  # retry → цикл, done → выход
)
builder.add_edge("critic", "solver")  # петля: critic возвращает к solver

graph = builder.compile()

initial_state = {
    "messages": [HumanMessage(content="Реши сложную задачу")],
    "iteration": 0,
    "solved": False
}
result = graph.invoke(initial_state)
for msg in result["messages"]:
    print(f"{type(msg).__name__}: {msg.content}")

Паттерн ReAct (Reason + Act)

Классическая агентская петля строится по схеме: LLM решает, нужен ли инструмент → вызывается инструмент → результат возвращается LLM → повторить. Это стандартный цикл в LangGraph:

# llm_node → (tool_calls?) → tool_node → llm_node → ...
builder.add_conditional_edges("llm", route_after_llm, {"tools": "tools", "end": END})
builder.add_edge("tools", "llm")  # петля

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

Для отладки структуры цикла полезен метод get_graph().draw_mermaid_png() — он наглядно показывает петли и точки выхода.

Подводные камни

• Бесконечный цикл без safeguard. Если маршрутизатор никогда не вернёт "done", граф будет работать вечно. Всегда добавляйте лимит итераций как отдельное поле в состоянии.
• Состояние растёт с каждой итерацией. Если в цикле добавляются сообщения, через 100 итераций история может превысить контекстное окно LLM. Необходимо реализовать обрезку истории (trim_messages).
• Нет встроенного таймаута. LangGraph не ограничивает время выполнения графа по умолчанию. В production нужно оборачивать вызов в asyncio.wait_for или устанавливать лимит через RunnableConfig.
• Нелинейный порядок сообщений затрудняет отладку. При нескольких итерациях список сообщений становится длинным и запутанным. Рекомендуется хранить отдельное поле steps: list[str] для краткого лога итераций.
• Счётчик итераций нужно инициализировать явно. Если поле iteration не задано в начальном состоянии, state.get("iteration", 0) вернёт 0, но после первого обновления тип может конфликтовать с аннотацией TypedDict.
• Critic-узел не должен изменять solved. Если и solver, и critic обновляют одно поле без редуктора, значение от одного узла перезапишет значение от другого. Проектируйте поля ответственности узлов явно.
• Параллельные ветки и петли несовместимы без синхронизации. Смешение fan-out/fan-in топологии с циклами требует тщательной настройки редукторов — иначе состояние после слияния может быть некорректным.
• recursion_limit. LangGraph имеет защиту от слишком глубокой рекурсии (по умолчанию 25 шагов). Для длинных агентских петель нужно явно поднимать лимит: graph.invoke(state, {"recursion_limit": 100}).

• Объяснять cycles and loops только синтаксисом без shape, dtype, состояния или режима выполнения.
• Игнорировать leakage, воспроизводимость, пустые входы и скрытые копии данных.
• Не проверять production-симптомы: latency, память, ретраи, дрейф качества и несовпадение версий.

• Может ли связать cycles and loops с реальным контрактом входов и выходов.
• Упоминает ли тесты, метрики, reproducibility и диагностику ошибок.
• Видит ли различие между demo-кодом в ноутбуке и production-пайплайном.