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

Tareas

Planificación y Coordinación

Task Graph + Dependencies

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

"Divide grandes objetivos en tareas pequeñas, organízalas, persístelas en disco" -- un grafo de tareas basado en archivos con dependencias, estableciendo las bases para la colaboración multi-agente.

Problema

El TodoManager de s03 es una lista de verificación plana en memoria: sin orden, sin dependencias, sin estado más allá de hecho o no. Los objetivos reales tienen estructura -- la tarea B depende de la tarea A, las tareas C y D pueden ejecutarse en paralelo, la tarea E espera tanto C como D.

Sin relaciones explícitas, el agente no puede decir qué está listo, qué está bloqueado, o qué puede ejecutarse concurrentemente. Y porque la lista vive solo en memoria, la compresión de contexto (s06) la borra completamente.

Solución

Promueve la lista de verificación a un grafo de tareas persistido en disco. Cada tarea es un archivo JSON con estado, dependencias (blockedBy) y dependientes (blocks). El grafo responde tres preguntas en cualquier momento:

  • ¿Qué está listo? -- tareas con estado pending y blockedBy vacío.
  • ¿Qué está bloqueado? -- tareas esperando dependencias inconclusas.
  • ¿Qué está hecho? -- tareas completed, cuya completación automáticamente desbloquea dependientes.
.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 de tareas (DAG):
                 +----------+
            +--> | task 2   | --+
            |    | pending  |   |
+----------+     +----------+    +--> +----------+
| task 1   |                          | task 4   |
| completed| --> +----------+    +--> | blocked  |
+----------+     | task 3   | --+     +----------+
                 | pending  |
                 +----------+

Ordenamiento:  task 1 debe terminar antes que 2 y 3
Paralelismo:    tareas 2 y 3 pueden ejecutarse al mismo tiempo
Dependencias:   task 4 espera tanto 2 como 3
Estado:         pending -> in_progress -> completed

Este grafo de tareas se convierte en la columna vertebral de coordinación para todo lo que sigue s07: ejecución en background (s08), equipos multi-agente (s09+), y aislamiento worktree (s12) todos leen y escriben en esta misma estructura.

Cómo Funciona

  1. TaskManager: un archivo JSON por tarea, CRUD con grafo de dependencias.
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. Resolución de dependencias: completar una tarea borra su ID de la lista blockedBy de cada otra tarea, desbloqueando automáticamente dependientes.
(self, completed_iddef _clear_dependency):
    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. Cableado de estado + dependencias: update maneja transiciones y aristas de dependencias.
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. Cuatro herramientas de tarea van al mapa de distribución.
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"]),
}

Desde s07 en adelante, el grafo de tareas es el predeterminado para trabajo multi-pasos. El Todo de s03 permanece para listas de verificación rápidas de una sola sesión.

Qué Cambió Respecto a s06

ComponenteAntes (s06)Después (s07)
Herramientas58 (task_create/update/list/get)
Modelo de planificaciónLista de verificación plana (en memoria)Grafo de tareas con dependencias (en disco)
RelacionesNoneAristas blockedBy + blocks
Seguimiento de estadoHecho o nopending -> in_progress -> completed
PersistenciaPerdido en compresiónSobrevive a compresión y reinicios

Pruébalo

python agents/s07_task_system.py
  1. Crear 3 tareas: "Setup project", "Write code", "Write tests". Hacerlas depender unas de otras en orden.
  2. Listar todas las tareas y mostrar el grafo de dependencias
  3. Completar la tarea 1 luego listar tareas para ver la tarea 2 desbloqueada
  4. Crear un tablero de tareas para refactoring: parse -> transform -> emit -> test, donde transform y emit pueden ejecutarse en paralelo después de parse