День 2: State и Conditional Edges

Цель: Научиться работать с условной маршрутизацией и создавать безопасные циклы

Время изучения: 1-2 часа
Код: src/day2_state_conditional.py
Тесты: tests/test_day2_state_conditional.py

🤖 Связь с агентами

Вчера: Создали простой граф A → B (линейный поток)
Сегодня: Добавляем интеллект — агенты принимают решения!

Что мы строим?

В A ( П ч в р е → с о р е с а B г т : д о а й B п ) о т о к У С A ( м е з н г → а ы о в й д ( и н р с а я е и г : ш т е е н н о т и т е ) с и → т у B а ц и и л и и ) C

Зачем это нужно для агентов?

Реальный сценарий:

S П Е Е Е u р с с с p о л л л e в и и и r ↓ е ↓ v р з з о i я а а ш s е д д и o т а а б r ч ч к з а а а A а g д п с → e а р л n ч о о E t у с ж r т н r а а o я я r → → H a Q E n u x d i p l c e e k r r t A g A e g n e t n t

Conditional Edges = способность агента принимать решения!

Что мы изучим

В День 1 мы создали простой линейный граф A → B. Сегодня усложним:

  1. Расширенная state-схема — больше полей, Optional типы
  2. Conditional Edges — выбор следующего узла на основе state
  3. Циклы — узел может вызывать сам себя
  4. Защита от бесконечности — MAX_ITERATIONS

Часть 1: Расширенная State-схема

Простая схема (День 1)

class SimpleState(TypedDict, total=False):
    message: str
    count: int

Расширенная схема (День 2)

from typing import TypedDict, List, Optional

class AgentState(TypedDict, total=False):
    messages: List[str]      # История всех сообщений
    plan: Optional[str]       # Текущий план действий
    result: Optional[str]     # Результат выполнения
    iteration: int            # Счётчик итераций
    status: str               # Статус: "planning", "executing", "done", "error"

Зачем больше полей?

  • messages — накапливаем историю для отладки
  • plan / result — разделяем планирование и выполнение
  • iteration — защита от бесконечных циклов
  • status — управление потоком выполнения

Часть 2: Граф с тремя узлами

Создадим граф: Planner → Executor → Finalizer

def node_planner(state: AgentState) -> AgentState:
    """Узел планирования: создаёт план."""
    updated = dict(state)
    updated["plan"] = "Execute task X"
    updated["status"] = "planning"
    updated["iteration"] = updated.get("iteration", 0) + 1
    updated["messages"] = updated.get("messages", []) + ["Planning"]
    print(f"Planner: iteration={updated['iteration']}")
    return updated

def node_executor(state: AgentState) -> AgentState:
    """Узел выполнения: выполняет план."""
    updated = dict(state)
    plan = state.get("plan", "No plan")
    updated["result"] = f"Executed: {plan}"
    updated["status"] = "executing"
    updated["iteration"] = updated.get("iteration", 0) + 1
    updated["messages"] = updated.get("messages", []) + ["Executing"]
    print(f"Executor: iteration={updated['iteration']}")
    return updated

def node_finalizer(state: AgentState) -> AgentState:
    """Узел финализации: завершает работу."""
    updated = dict(state)
    updated["status"] = "done"
    updated["iteration"] = updated.get("iteration", 0) + 1
    updated["messages"] = updated.get("messages", []) + ["Done"]
    print(f"Finalizer: iteration={updated['iteration']}")
    return updated

Собираем граф

from langgraph.graph import StateGraph

def create_simple_graph():
    graph = StateGraph(AgentState)
    graph.add_node("planner", node_planner)
    graph.add_node("executor", node_executor)
    graph.add_node("finalizer", node_finalizer)
    
    graph.add_edge("planner", "executor")
    graph.add_edge("executor", "finalizer")
    
    graph.set_entry_point("planner")
    graph.set_finish_point("finalizer")
    
    return graph.compile()

Запуск

workflow = create_simple_graph()
result = workflow.invoke({"messages": [], "iteration": 0})

print(f"Status: {result['status']}")        # done
print(f"Iterations: {result['iteration']}")  # 3
print(f"Messages: {result['messages']}")     # ['Planning', 'Executing', 'Done']

Часть 3: Conditional Edges (условная маршрутизация)

Теперь добавим выбор пути на основе состояния.

Сценарий

После планирования проверяем status:

  • Если status == "error" → идём в error_handler
  • Если status == "done" → завершаем
  • Иначе → продолжаем выполнение

Функция-роутер

def router_function(state: AgentState) -> str:
    """Выбирает следующий узел на основе status."""
    status = state.get("status", "")
    
    if status == "error":
        return "error"
    elif status == "done":
        return "finish"
    else:
        return "continue"

Важно: Функция возвращает строку — имя следующего пути

Обработчики

def error_handler(state: AgentState) -> AgentState:
    """Обрабатывает ошибки."""
    updated = dict(state)
    updated["messages"] = updated.get("messages", []) + ["Error handled"]
    updated["status"] = "recovered"
    print("Error Handler: recovered")
    return updated

def success_handler(state: AgentState) -> AgentState:
    """Обрабатывает успех."""
    updated = dict(state)
    updated["messages"] = updated.get("messages", []) + ["Success!"]
    updated["status"] = "done"
    print("Success Handler: done")
    return updated

Граф с conditional edge

from langgraph.graph import StateGraph, END

def create_conditional_graph():
    graph = StateGraph(AgentState)
    
    graph.add_node("start", node_planner)
    graph.add_node("error_handler", error_handler)
    graph.add_node("success_handler", success_handler)
    
    graph.set_entry_point("start")
    
    # Conditional edge: выбор пути на основе router_function
    graph.add_conditional_edges(
        "start",                    # Из какого узла
        router_function,            # Функция-роутер
        {
            "error": "error_handler",      # Если вернула "error"
            "finish": END,                 # Если вернула "finish"
            "continue": "success_handler"  # Если вернула "continue"
        }
    )
    
    graph.add_edge("error_handler", END)
    graph.add_edge("success_handler", END)
    
    return graph.compile()

Тестируем разные пути

workflow = create_conditional_graph()

# Тест 1: Успешный путь
result = workflow.invoke({
    "messages": [],
    "iteration": 0,
    "status": "start"
})
print(result["status"])  # done
print(result["messages"])  # ['Planning', 'Success!']

# Тест 2: Путь с ошибкой
result = workflow.invoke({
    "messages": [],
    "iteration": 0,
    "status": "error"
})
print(result["status"])  # recovered
print(result["messages"])  # ['Planning', 'Error handled']

Часть 4: Циклы с защитой

Самое интересное — создадим узел, который может вызывать сам себя.

Проблема: бесконечный цикл

# ❌ Опасно!
graph.add_conditional_edges(
    "loop_node",
    lambda state: "continue",  # Всегда продолжаем
    {"continue": "loop_node"}  # Возврат к себе
)
# Это зациклится навсегда!

Решение: MAX_ITERATIONS

MAX_ITERATIONS = 10

def loop_node(state: AgentState) -> AgentState:
    """Узел, который выполняется в цикле."""
    updated = dict(state)
    iteration = updated.get("iteration", 0) + 1
    updated["iteration"] = iteration
    updated["messages"] = updated.get("messages", []) + [f"Loop {iteration}"]
    
    # Условие завершения: после 5 итераций
    if iteration >= 5:
        updated["status"] = "done"
    else:
        updated["status"] = "continue"
    
    print(f"Loop iteration {iteration}: status={updated['status']}")
    return updated

def should_continue(state: AgentState) -> str:
    """Решает, продолжать ли цикл."""
    iteration = state.get("iteration", 0)
    status = state.get("status", "")
    
    # Защита 1: Максимум итераций
    if iteration >= MAX_ITERATIONS:
        print(f"⚠️  Max iterations ({MAX_ITERATIONS}) reached!")
        return "stop"
    
    # Защита 2: Проверка статуса
    if status == "done":
        print("✓ Task completed!")
        return "stop"
    
    return "continue"

Граф с циклом

from langgraph.graph import StateGraph, END

def create_loop_graph():
    graph = StateGraph(AgentState)
    graph.add_node("loop", loop_node)
    graph.set_entry_point("loop")
    
    graph.add_conditional_edges(
        "loop",
        should_continue,
        {
            "continue": "loop",  # Возврат к самому себе
            "stop": END          # Завершение
        }
    )
    
    return graph.compile()

Запуск

workflow = create_loop_graph()
result = workflow.invoke({
    "messages": [],
    "iteration": 0,
    "status": "start"
})

print(f"Iterations: {result['iteration']}")  # 5
print(f"Status: {result['status']}")         # done
print(f"Messages: {result['messages']}")     # ['Loop 1', 'Loop 2', ..., 'Loop 5']

Вывод:

L L L L L ✓ o o o o o o o o o o T p p p p p a s i i i i i k t t t t t e e e e e c r r r r r o a a a a a m t t t t t p i i i i i l o o o o o e n n n n n t e 1 2 3 4 5 d : : : : : ! s s s s s t t t t t a a a a a t t t t t u u u u u s s s s s = = = = = c c c c d o o o o o n n n n n t t t t e i i i i n n n n u u u u e e e e

Полный пример: Все вместе

from typing import TypedDict, List, Optional
from langgraph.graph import StateGraph, END

MAX_ITERATIONS = 10

class AgentState(TypedDict, total=False):
    messages: List[str]
    plan: Optional[str]
    result: Optional[str]
    iteration: int
    status: str

def loop_node(state: AgentState) -> AgentState:
    updated = dict(state)
    iteration = updated.get("iteration", 0) + 1
    updated["iteration"] = iteration
    updated["messages"] = updated.get("messages", []) + [f"Loop {iteration}"]
    
    if iteration >= 5:
        updated["status"] = "done"
    else:
        updated["status"] = "continue"
    
    return updated

def should_continue(state: AgentState) -> str:
    if state.get("iteration", 0) >= MAX_ITERATIONS:
        return "stop"
    if state.get("status") == "done":
        return "stop"
    return "continue"

def create_loop_graph():
    graph = StateGraph(AgentState)
    graph.add_node("loop", loop_node)
    graph.set_entry_point("loop")
    graph.add_conditional_edges(
        "loop",
        should_continue,
        {"continue": "loop", "stop": END}
    )
    return graph.compile()

# Запуск
workflow = create_loop_graph()
result = workflow.invoke({"messages": [], "iteration": 0, "status": "start"})
print(result)

Запуск ��римеров

# Запустить все примеры дня 2
python src/day2_state_conditional.py

# Запустить тесты
make test-day2

# Или напрямую
PYTHONPATH=. pytest tests/test_day2_state_conditional.py -v

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

❌ Ошибка 1: Забыли проверить MAX_ITERATIONS

def bad_should_continue(state: AgentState) -> str:
    if state.get("status") == "done":
        return "stop"
    return "continue"  # ❌ Может зациклиться!

✅ Правильно:

def good_should_continue(state: AgentState) -> str:
    if state.get("iteration", 0) >= MAX_ITERATIONS:  # ✓ Защита
        return "stop"
    if state.get("status") == "done":
        return "stop"
    return "continue"

❌ Ошибка 2: Роутер возвращает несуществующий путь

def bad_router(state: AgentState) -> str:
    return "unknown_path"  # ❌ Нет в маппинге!

graph.add_conditional_edges(
    "node",
    bad_router,
    {"path1": "node1", "path2": "node2"}  # "unknown_path" отсутствует!
)

✅ Правильно:

def good_router(state: AgentState) -> str:
    status = state.get("status", "")
    if status == "error":
        return "error"
    return "continue"  # ✓ Дефолтный путь

graph.add_conditional_edges(
    "node",
    good_router,
    {"error": "error_node", "continue": "next_node"}  # Все пути покрыты
)

Ключевые выводы

✅ TypedDict с Optional — гибкая типизация для сложных состояний
✅ Conditional Edges — выбор пути на основе state через функцию-роутер
✅ Циклы �� узел может вызывать сам себя через conditional edge
✅ MAX_ITERATIONS — обязательная защита от бесконечных циклов
✅ END — специальный маркер завершения графа
✅ Логирование — добавляйте print-ы для отладки

🤖 Связь с агентами

Сегодня вы научились:

  • ✅ Агенты принимают решения (conditional routing)
  • ✅ Агенты могут повторять действия (циклы)
  • ✅ Агенты защищены от зацикливания (MAX_ITERATIONS)

Следующий шаг: Создадим первую полноценную multi-agent систему с тремя специализированными агентами!

Следующий шаг: День 3

В следующей статье мы создадим полноценный цикл Planner → Executor → Critic:

  • Planner генерирует план из задачи
  • Executor выполняет шаги плана
  • Critic оценивает результат и решает, продолжать или завершить
  • Цикл с обратной связью и самоул��чшением

👉 День 3: Planner → Executor → Critic (скоро)

Ресурсы

Прогресс: 2/10 дней ✅