День 3: Planner → Executor → Critic
Цель: Создать первую полноценную multi-agent систему с циклом самоулучшения
Время изучения: 2-3 часа
Код: src/day3_planner_executor_critic.py
Тесты: tests/test_day3_planner_executor_critic.py
🤖 Связь с агентами
Вчера: Научились создавать циклы и условную маршрутизацию
Сегодня: Создаём первую настоящую multi-agent систему!
Что мы строим?
Зачем это нужно?
Реальный сценарий: Написание статьи
Это паттерн самоулучшения — агенты работают в цикле, пока не достигнут качественного результата.
Архитектура системы
Три специализированных агента
class AgentState(TypedDict, total=False):
task: str # Исходная задача
plan: List[str] # План выполнения
current_step: int # Текущий шаг
step_result: str # Результат текущего шага
critique: str # Оценка критика
status: str # "planning", "executing", "critiquing", "done"
iteration: int # Счётчик итераций
max_iterations: int # Максимум итераций
final_result: str # Финальный результат
Поток выполнения
Часть 1: Planner Agent
Задача: Разбить задачу на шаги
def planner_agent(state: AgentState) -> AgentState:
"""
Planner Agent: создаёт план выполнения задачи.
В реальной системе здесь был бы вызов LLM:
plan = llm.invoke(f"Create a plan for: {task}")
"""
updated = dict(state)
task = state.get("task", "")
# Простая эмуляция планирования
# В реальности здесь LLM генерирует ��лан
plan = [
f"Step 1: Research about '{task}'",
f"Step 2: Analyze findings for '{task}'",
f"Step 3: Create summary for '{task}'"
]
updated["plan"] = plan
updated["current_step"] = 0
updated["status"] = "executing"
updated["iteration"] = updated.get("iteration", 0) + 1
print(f"\n📋 Planner: Created plan with {len(plan)} steps")
for i, step in enumerate(plan, 1):
print(f" {i}. {step}")
return updated
Ключевые моменты:
- Получает задачу из state
- Создаёт список шагов
- Устанавливает current_step = 0
- Переводит status в “executing”
Часть 2: Executor Agent
Задача: Выполнить текущий шаг плана
def executor_agent(state: AgentState) -> AgentState:
"""
Executor Agent: выполняет текущий шаг плана.
В реальной системе здесь был бы вызов LLM с tools:
result = llm_with_tools.invoke(current_step)
"""
updated = dict(state)
plan = state.get("plan", [])
current_step = state.get("current_step", 0)
iteration = updated.get("iteration", 0)
if current_step >= len(plan):
# Все шаги выполнены
updated["status"] = "done"
updated["final_result"] = state.get("step_result", "Task completed")
print(f"\n✅ Executor: All steps completed!")
return updated
step = plan[current_step]
# Простая эмуляция выполнения
# В реальности здесь LLM выполняет шаг с использованием tools
result = f"Completed: {step} (iteration {iteration})"
updated["step_result"] = result
updated["status"] = "critiquing"
updated["iteration"] = iteration + 1
print(f"\n⚙️ Executor: Executing step {current_step + 1}/{len(plan)}")
print(f" Step: {step}")
print(f" Result: {result}")
return updated
Ключевые моменты:
- Читает текущий шаг из плана
- Выполняет шаг (в реальности — вызов LLM с tools)
- Сохраняет результат в step_result
- Переводит status в “critiquing”
Часть 3: Critic Agent
Задача: Оценить результат и решить, что делать дальше
def critic_agent(state: AgentState) -> AgentState:
"""
Critic Agent: оценивает результат выполнения шага.
В реальной системе здесь был бы вызов LLM:
critique = llm.invoke(f"Evaluate: {step_result}")
"""
updated = dict(state)
step_result = state.get("step_result", "")
current_step = state.get("current_step", 0)
iteration = state.get("iteration", 0)
# Простая эмуляция критики
# В реальности здесь LLM оценивает качество результата
# Эмулируем: первая попытка всегда требует улучшения
if iteration % 2 == 1:
critique = "Good, but needs improvement. Try again."
decision = "retry"
else:
critique = "Excellent! Moving to next step."
decision = "next"
updated["current_step"] = current_step + 1
updated["critique"] = critique
updated["status"] = "executing" if decision == "retry" or updated["current_step"] < len(state.get("plan", [])) else "done"
print(f"\n🔍 Critic: Evaluating step {current_step + 1}")
print(f" Critique: {critique}")
print(f" Decision: {decision}")
return updated
Ключевые моменты:
- Оценивает step_result
- Решает: повторить (retry) или следующий шаг (next)
- Если retry — status остаётся “executing”
- Если next — увеличивает current_step
- Если все шаги выполнены — status = “done”
Часть 4: Роутер (Router)
Задача: Решить, какой агент вызвать следующим
def should_continue(state: AgentState) -> str:
"""
Роутер: решает, продолжать ли цикл.
"""
status = state.get("status", "")
iteration = state.get("iteration", 0)
max_iterations = state.get("max_iterations", 10)
# Защита от бесконечного цикла
if iteration >= max_iterations:
print(f"\n⚠️ Max iterations ({max_iterations}) reached!")
return "end"
# Проверяем статус
if status == "done":
print(f"\n✅ Task completed successfully!")
return "end"
elif status == "executing":
return "execute"
elif status == "critiquing":
return "critique"
else:
return "end"
Ключевые моменты:
- Проверяет max_iterations (защита)
- Маршрутизирует по status
- Возвращает имя следующего узла
Часть 5: Собираем граф
from langgraph.graph import StateGraph, END
def create_planner_executor_critic_graph():
"""Создаёт граф Planner → Executor → Critic."""
graph = StateGraph(AgentState)
# Добавляем агентов как узлы
graph.add_node("planner", planner_agent)
graph.add_node("executor", executor_agent)
graph.add_node("critic", critic_agent)
# Начинаем с планировщика
graph.set_entry_point("planner")
# Planner → Executor
graph.add_edge("planner", "executor")
# Executor → Critic
graph.add_edge("executor", "critic")
# Critic → (conditional) Executor или END
graph.add_conditional_edges(
"critic",
should_continue,
{
"execute": "executor", # Повторить или следующий шаг
"critique": "critic", # На случай, если нужно
"end": END # Завершить
}
)
return graph.compile()
Полный пример использования
def main():
"""Пример использо��ания Planner → Executor → Critic."""
workflow = create_planner_executor_critic_graph()
# Начальное состояние
initial_state: AgentState = {
"task": "Write a blog post about LangGraph",
"iteration": 0,
"max_iterations": 10,
"status": "planning"
}
# Запускаем граф
print("="*60)
print("🚀 Starting Planner → Executor → Critic workflow")
print("="*60)
result = workflow.invoke(initial_state)
# Выводим результат
print("\n" + "="*60)
print("📊 Final Result")
print("="*60)
print(f"Task: {result.get('task')}")
print(f"Plan steps: {len(result.get('plan', []))}")
print(f"Iterations: {result.get('iteration')}")
print(f"Status: {result.get('status')}")
print(f"Final result: {result.get('final_result')}")
print("="*60)
if __name__ == "__main__":
main()
Вывод:
Запуск
# Запустить пример
python src/day3_planner_executor_critic.py
# Запустить тесты
make test-day3
# Или напрямую
PYTHONPATH=. pytest tests/test_day3_planner_executor_critic.py -v
Улучшения для реальной системы
1. Интеграция с LLM
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4")
def planner_agent_with_llm(state: AgentState) -> AgentState:
task = state.get("task", "")
prompt = f"""Create a detailed plan to accomplish this task:
Task: {task}
Return a numbered list of steps."""
response = llm.invoke(prompt)
plan = response.content.split("\n")
updated = dict(state)
updated["plan"] = plan
updated["status"] = "executing"
return updated
2. Добавление логирования
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def executor_agent(state: AgentState) -> AgentState:
logger.info(f"Executing step {state.get('current_step')}")
# ... логика
logger.info(f"Step result: {result}")
return updated
3. Сохранение истории
class AgentState(TypedDict, total=False):
# ... существующие поля
history: List[Dict[str, str]] # История всех действий
def executor_agent(state: AgentState) -> AgentState:
updated = dict(state)
# ... выполнение
# Сохраняем в историю
history = updated.get("history", [])
history.append({
"agent": "executor",
"step": step,
"result": result,
"iteration": iteration
})
updated["history"] = history
return updated
Типичные ошибки
❌ Ошибка 1: Забыли увеличить current_step
def bad_critic(state: AgentState) -> AgentState:
updated = dict(state)
updated["status"] = "executing" # ❌ Не увеличили current_step!
return updated
# Результат: бесконечное выполнение одного шага
✅ Правильно:
def good_critic(state: AgentState) -> AgentState:
updated = dict(state)
if decision == "next":
updated["current_step"] = state.get("current_step", 0) + 1 # ✓
updated["status"] = "executing"
return updated
❌ Ошибка 2: Не проверили границы массива
def bad_executor(state: AgentState) -> AgentState:
plan = state.get("plan", [])
step = plan[state.get("current_step")] # ❌ IndexError!
# ...
✅ Правильно:
def good_executor(state: AgentState) -> AgentState:
plan = state.get("plan", [])
current_step = state.get("current_step", 0)
if current_step >= len(plan): # ✓ Проверка
updated["status"] = "done"
return updated
step = plan[current_step]
# ...
Ключевые выводы
✅ Planner Agent — разбивает задачу на шаги
✅ Executor Agent — выполняет шаги последовательно
✅ Critic Agent — оценивает результат и решает, что делать
✅ Цикл самоулучшения — повторяем, пока не достигнем качества
✅ MAX_ITERATIONS — защита от бесконечных циклов
✅ current_step — отслеживание прогресса по плану
🤖 Связь с агентами
Сегодня вы создали:
- ✅ Первую настоящую multi-agent систему
- ✅ Три специализированных агента
- ✅ Цикл самоулучшения с обратной связ��ю
- ✅ Паттерн, который используется в реальных системах
Следующий шаг: Добавим агентам инструменты (tools) — они смогут искать информацию, выполнять вычисления, работать с файлами!
Следующий шаг: День 4
В следующей статье мы изучим:
- Tool Calling — агенты используют инструменты
- Structured Outputs — строгая типизация ответов LLM
- Реальная интеграция с LLM — вместо эмуляции
👉 День 4: Tool Calling и Structured Outputs (скоро)
Ресурсы
Прогресс: 3/10 дней ✅