How to Build an AI Agent
How to Build an AI Agent
s07

Задачи

Планирование и координация

Task Graph + Dependencies

207 LOC8 инструментовTaskManager with file-based state + dependency graph
A file-based task graph with ordering, parallelism, and dependencies -- the coordination backbone for multi-agent work

s01 > s02 > s03 > s04 > s05 > s06 | [ s07 ] s08 > s09 > s10 > s11 > s12

"Разбивайте большие цели на маленькие задачи, организуйте их, персистируйте на диск" -- граф задач на основе файлов с зависимостями, закладывая основы мультиагентного сотрудничества.

Проблема

TodoManager из s03 — это плоский список в памяти: нет планирования, нет зависимостей, нет статуса помимо сделано/не сделано. У настоящих целей есть структура — задача B зависит от задачи A, задачи C и D могут выполняться параллельно, задача E ждёт и C, и D.

Без явных отношений агент не может сказать, что готово, что заблокировано, что может выполняться параллельно. И поскольку список живёт только в памяти, сжатие контекста (s06) полностью его стирает.

Решение

Продвигаем список в граф задач, персистируемый на диск. Каждая задача — это JSON-файл со статусом, зависимостями (blockedBy) и зависимыми (blocks). Граф отвечает на три вопроса в любой момент:

  • Что готово? — задачи со статусом pending и пустым blockedBy.
  • Что заблокировано? — задачи, ожидающие незавершённые зависимости.
  • Что сделано? — задачи completed, чьё завершение автоматически разблокирует зависимые.
.tasks/
  task_1.json  {"id":1, "status":"completed"}
  task_2.json  {"id":2, "blockedBy":[1], "status":"pending"}
  task_3.json  {"id":3, "blockedBy":[1], "status":"pending"}
  task_4.json  {"id":4, "blockedBy":[2,3], "status":"pending"}

Граф задач (DAG):
                 +----------+
            +--> | task 2   | --+
            |    | pending  |   |
+----------+     +----------+    +--> +----------+
| task 1   |                          | task 4   |
| completed| --> +----------+    +--> | blocked  |
+----------+     | task 3   | --+     +----------+
                 | pending  |
                 +----------+

Ordonnancement:  task 1 doit finir avant 2 et 3
Parallélisme:    tâches 2 et 3 peuvent s'exécuter en même temps
Dépendances:     task 4 attend à la fois 2 et 3
Statut:          pending -> in_progress -> completed

Этот граф задач становится основой координации для всего, что следует далее: фоновое выполнение (s08), мультиагентные команды (s09+), и изоляция worktree (s12) — все читают и пишут в эту же структуру.

Как Это Работает

  1. TaskManager: один JSON-файл на задачу, CRUD с графом зависимостей.
class TaskManager:
    def __init__(self, tasks_dir: Path):
        self.dir = tasks_dir
        self.dir.mkdir(exist_ok=True)
        self._next_id = self._max_id() + 1

    def create(self, subject, description=""):
        task = {"id": self._next_id, "subject": subject,
                "status": "pending", "blockedBy": [],
                "blocks": [], "owner": ""}
        self._save(task)
        self._next_id += 1
        return json.dumps(task, indent=2)
  1. Разрешение зависимостей: завершение задачи удаляет её ID из списка blockedBy каждой другой задачи, автоматически разблокируя зависимых.
def _clear_dependency(self, completed_id):
    for f in self.dir.glob("task_*.json"):
        task = json.loads(f.read_text())
        if completed_id in task.get("blockedBy", []):
            task["blockedBy"].remove(completed_id)
            self._save(task)
  1. Статус + связывание зависимостей: update обрабатывает переходы и рёбра зависимостей.
def update(self, task_id, status=None,
           add_blocked_by=None, add_blocks=None):
    task = self._load(task_id)
    if status:
        task["status"] = status
        if status == "completed":
            self._clear_dependency(task_id)
    self._save(task)
  1. Четыре инструмента задач идут в карту диспетчеризации.
TOOL_HANDLERS = {
    # ...base tools...
    "task_create": lambda **kw: TASKS.create(kw["subject"]),
    "task_update": lambda **kw: TASKS.update(kw["task_id"], kw.get("status")),
    "task_list":   lambda **kw: TASKS.list_all(),
    "task_get":    lambda **kw: TASKS.get(kw["task_id"]),
}

Начиная с s07, граф задач — умолчание для многоэтапной работы. Todo из s03 остаётся для быстрых чеклистов одной сессии.

Что Изменилось по Сравнению с s06

КомпонентДо (s06)После (s07)
Инструменты58 (task_create/update/list/get)
Модель планированияПлоский чеклист (в памяти)Граф задач с зависимостями (на диске)
ОтношенияНетРёбра blockedBy + blocks
Отслеживание статусаСделано или нетpending -> in_progress -> completed
ПерсистентностьТеряется при сжатииПереживает сжатие и перезапуски

Попробуйте

python agents/s07_task_system.py
  1. Создать 3 задачи: "Setup project", "Write code", "Write tests". Сделать их зависимыми друг от друга по порядку.
  2. Перечислить все задачи и показать граф зависимостей
  3. Завершить задачу 1, затем перечислить задачи, чтобы увидеть задачу 2 разблокированной
  4. Создать доску задач для рефакторинга: parse -> transform -> emit -> test, где transform и emit могут выполняться параллельно после parse