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

Task

Pianificazione & Coordinamento

Task Graph + Dependencies

207 LOC8 strumentiTaskManager 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

"Dividi obiettivi grandi in task piccoli, ordinali, persisti su disco" -- un grafo di task basato su file con dipendenze, gettando le basi per la collaborazione multi-agente.

Problema

Il TodoManager di s03 è una checklist piatta in memoria: nessun ordinamento, nessuna dipendenza, nessuno stato oltre fatto-o-no. Gli obiettivi reali hanno struttura -- il task B dipende dal task A, i task C e D possono eseguire in parallelo, il task E aspetta sia C che D.

Senza relazioni esplicite, l'agente non può dire cosa è pronto, cosa è bloccato, o cosa può eseguire concurrentemente. E poiché la lista vive solo in memoria, la compressione del contesto (s06) la cancella completamente.

Soluzione

Promuovi la checklist in un grafo di task persistito su disco. Ogni task è un file JSON con stato, dipendenze (blockedBy) e dipendenti (blocks). Il grafo risponde a tre domande in qualsiasi momento:

  • Cosa è pronto? -- task con stato pending e blockedBy vuoto.
  • Cosa è bloccato? -- task in attesa di dipendenze non completate.
  • Cosa è fatto? -- task completed, il cui completamento sblocca automaticamente i dipendenti.
.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"}

Grafo dei task (DAG):
                 +----------+
            +--> | task 2   | --+
            |    | pending  |   |
+----------+     +----------+    +--> +----------+
| task 1   |                          | task 4   |
| completed| --> +----------+    +--> | blocked  |
+----------+     | task 3   | --+     +----------+
                 | pending  |
                 +----------+

Ordinamento:  task 1 deve finire prima di 2 e 3
Parallelismo: task 2 e 3 possono eseguire contemporaneamente
Dipendenze:  task 4 aspetta sia 2 che 3
Stato:        pending -> in_progress -> completed

Questo grafo di task diventa la spina dorsale di coordinamento per tutto ciò che viene dopo s07: esecuzione in background (s08), team multi-agente (s09+), e isolamento worktree (s12) leggono e scrivono tutti nella stessa struttura.

Come Funziona

  1. TaskManager: un file JSON per task, CRUD con grafo delle dipendenze.
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. Risoluzione dipendenze: completare un task rimuove il suo ID dalla lista blockedBy di ogni altro task, sbloccando automaticamente i dipendenti.
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. Stato + cablaggio dipendenze: update gestisce transizioni e archi di dipendenza.
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. Quattro strumenti task vanno nella mappa di dispatch.
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"]),
}

Da s07 in poi, il grafo di task è il default per il lavoro multi-passo. Il Todo di s03 rimane per checklist rapide di una singola sessione.

Cosa è Cambiato da s06

ComponentePrima (s06)Dopo (s07)
Strumenti58 (task_create/update/list/get)
Modello pianificazioneChecklist piatta (in memoria)Grafo di task con dipendenze (su disco)
RelazioniNessunoArchi blockedBy + blocks
Tracciamento statoFatto o nopending -> in_progress -> completed
PersistenzaPerso su compressioneSopravvive a compressione e riavvii

Provalo

python agents/s07_task_system.py
  1. Crea 3 task: "Setup project", "Write code", "Write tests". Falli dipendere l'uno dall'altro in ordine.
  2. Elenca tutti i task e mostra il grafo delle dipendenze
  3. Completa il task 1 e poi elenca i task per vedere il task 2 sbloccato
  4. Crea una bacheca task per refactoring: parse -> transform -> emit -> test, dove transform e emit possono eseguire in parallelo dopo parse