BabyAGI Implementation for Autonomous Task Execution

BabyAGI for Autonomous Task Execution

BabyAGI — концептуальный фреймворк автономного AI-агента, который сам генерирует задачи, приоритизирует их и выполняет в цикле. Оригинальный проект (Yohei Nakajima, 2023) демонстрирует паттерн: Task Creation → Prioritization → Execution → New Tasks. Этот архитектурный паттерн актуален для понимания принципов autonomous agents и реализации собственных систем.

Паттерн BabyAGI

from openai import OpenAI
from collections import deque
from typing import Optional
import json

client = OpenAI()

class BabyAGIAgent:
    """Реализация паттерна BabyAGI"""

    def __init__(
        self,
        objective: str,
        max_tasks: int = 20,
        execution_model: str = "gpt-4o",
        management_model: str = "gpt-4o-mini",
    ):
        self.objective = objective
        self.task_list = deque()
        self.completed_tasks = []
        self.results = {}
        self.task_id_counter = 1
        self.max_tasks = max_tasks
        self.execution_model = execution_model
        self.management_model = management_model

    def add_task(self, task_name: str, task_id: Optional[int] = None):
        task_id = task_id or self.task_id_counter
        self.task_list.append({"task_id": task_id, "task_name": task_name})
        self.task_id_counter += 1

    def task_creation(self, result: str, task: dict) -> list[dict]:
        """Генерирует новые задачи на основе результата выполненной"""

        response = client.chat.completions.create(
            model=self.management_model,
            messages=[{
                "role": "user",
                "content": f"""Цель: {self.objective}
Последняя выполненная задача: {task['task_name']}
Результат: {result[:500]}
Незавершённые задачи: {[t['task_name'] for t in self.task_list]}

Создай новые задачи для достижения цели на основе результата.
Не дублируй существующие задачи.
Верни JSON: [{{"task_name": "..."}}]
Если задач нет — верни [].""",
            }],
        )

        try:
            new_tasks = json.loads(response.choices[0].message.content)
            return new_tasks if isinstance(new_tasks, list) else []
        except Exception:
            return []

    def prioritization(self) -> list[dict]:
        """Переупорядочивает задачи по приоритету"""

        if not self.task_list:
            return []

        tasks_str = "\n".join(
            f"{t['task_id']}. {t['task_name']}" for t in self.task_list
        )

        response = client.chat.completions.create(
            model=self.management_model,
            messages=[{
                "role": "user",
                "content": f"""Цель: {self.objective}
Задачи для приоритизации:
{tasks_str}

Переупорядочи задачи по приоритету для достижения цели.
Верни JSON: [{{"task_id": N, "task_name": "..."}}]""",
            }],
        )

        try:
            return json.loads(response.choices[0].message.content)
        except Exception:
            return list(self.task_list)

    def execute_task(self, task: dict) -> str:
        """Выполняет задачу и возвращает результат"""

        context = "\n".join([
            f"Задача: {t}\nРезультат: {r[:200]}"
            for t, r in list(self.results.items())[-3:]  # Последние 3 результата как контекст
        ])

        response = client.chat.completions.create(
            model=self.execution_model,
            messages=[{
                "role": "system",
                "content": f"Выполняй задачи для достижения цели: {self.objective}",
            }, {
                "role": "user",
                "content": f"""Контекст предыдущих задач:
{context}

Выполни задачу: {task['task_name']}

Предоставь конкретный результат.""",
            }],
        )

        return response.choices[0].message.content

    def run(self, initial_task: str, max_iterations: int = 10):
        """Основной цикл выполнения"""

        # Инициализация
        self.add_task(initial_task)

        iteration = 0
        while self.task_list and iteration < max_iterations:
            print(f"\n--- Итерация {iteration + 1} ---")
            print(f"Задач в очереди: {len(self.task_list)}")

            # Выполняем первую задачу
            task = self.task_list.popleft()
            print(f"Выполняю: {task['task_name']}")

            result = self.execute_task(task)
            self.results[task["task_name"]] = result
            self.completed_tasks.append(task)
            print(f"Результат: {result[:200]}...")

            # Создаём новые задачи
            if iteration < max_iterations - 2:  # Не создаём задачи в последних итерациях
                new_tasks = self.task_creation(result, task)
                for nt in new_tasks[:3]:  # Ограничиваем рост задач
                    if len(self.task_list) < self.max_tasks:
                        self.add_task(nt["task_name"])

            # Приоритизируем
            prioritized = self.prioritization()
            if prioritized:
                self.task_list = deque(prioritized)

            iteration += 1

        return self.results

Практическое применение паттерна

# Исследование темы
agent = BabyAGIAgent(
    objective="Подготовь детальный анализ возможностей применения AI в логистике компании среднего бизнеса",
    max_tasks=15,
)

results = agent.run(
    initial_task="Определи основные болевые точки в логистических процессах среднего бизнеса",
    max_iterations=8,
)

# Финальный синтез
final_report = client.chat.completions.create(
    model="gpt-4o",
    messages=[{
        "role": "user",
        "content": f"""Синтезируй следующие результаты исследования в связный отчёт:
{json.dumps(results, ensure_ascii=False, indent=2)}""",
    }],
)

Современные альтернативы паттерна

Оригинальный BabyAGI — учебный пример. В production-системах используют:

• LangGraph — позволяет реализовать тот же цикл task creation/execution с персистентностью, человеческим контролем и надёжным state management
• Celery + Redis — для распределённого выполнения задач
• LlamaIndex Workflows — для document-oriented задач

# Более надёжная реализация паттерна через LangGraph
from langgraph.graph import StateGraph, END

class AGIState(TypedDict):
    objective: str
    task_queue: list[str]
    completed_tasks: list[dict]
    iteration: int
    max_iterations: int

graph = StateGraph(AGIState)
graph.add_node("execute_task", execute_current_task)
graph.add_node("create_tasks", create_new_tasks)
graph.add_node("prioritize", prioritize_task_queue)
graph.add_conditional_edges("prioritize", should_continue_or_stop)

Сроки

• Базовая реализация паттерна: 2–3 дня
• Production-реализация на LangGraph с персистентностью: 1–2 недели
• Интеграция инструментов для конкретной предметной области: 1–2 недели