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

Tarefas

Planejamento e Coordenacao

Task Graph + Dependencies

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

"Divida objetivos grandes em pequenas tarefas, ordene-as, persista em disco" -- um grafo de tarefas baseado em arquivos com dependências, lançando as bases para colaboração multiagente.

Problema

O TodoManager do s03 é uma lista de verificação plana na memória: sem ordenação, sem dependências, sem status além de feito ou não. Objetivos reais têm estrutura -- tarefa B depende da tarefa A, tarefas C e D podem rodar em paralelo, tarefa E espera por ambas C e D.

Sem relacionamentos explícitos, o agente não consegue dizer o que está pronto, o que está bloqueado, ou o que pode rodar simultaneamente. E porque a lista existe apenas na memória, a compressão de contexto (s06) a apaga completamente.

Solução

Promova a lista de verificação para um grafo de tarefas persistido em disco. Cada tarefa é um arquivo JSON com status, dependências (blockedBy), e dependentes (blocks). O grafo responde três perguntas a qualquer momento:

  • O que está pronto? -- tarefas com status pending e blockedBy vazio.
  • O que está bloqueado? -- tarefas esperando em dependências não finalizadas.
  • O que está feito? -- tarefas completed, cuja conclusão automaticamente desbloqueia dependentes.
.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"}

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

Ordering:     task 1 must finish before 2 and 3
Parallelism:  tasks 2 and 3 can run at the same time
Dependencies: task 4 waits for both 2 and 3
Status:       pending -> in_progress -> completed

Este grafo de tarefas se torna a espinha dorsal de coordenação para tudo depois de s07: execução em segundo plano (s08), equipes multiagente (s09+), e isolamento de worktree (s12) todos leem e escrevem nesta mesma estrutura.

Como Funciona

  1. TaskManager: um arquivo JSON por tarefa, CRUD com grafo de dependências.
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. Resolução de dependências: completar uma tarefa remove seu ID de cada blockedBy de outras tarefas, automaticamente desbloqueando dependentes.
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. Status + fiação de dependências: update gerencia transições e arestas de dependência.
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. Quatro ferramentas de tarefa vão para o mapa de 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"]),
}

A partir de s07, o grafo de tarefas é o padrão para trabalho multi-etapa. O Todo do s03 permanece para checklists rápidos de sessão única.

O Que Mudou Desde s06

ComponenteAntes (s06)Depois (s07)
Ferramentas58 (task_create/update/list/get)
Modelo de planejamentoLista plana (na memória)Grafo de tarefas com dependências (em disco)
RelacionamentosNenhumarestas blockedBy + blocks
Rastreamento de statusFeito ou nãopending -> in_progress -> completed
PersistênciaPerdido na compressãoSobrevive compressão e reinicializações

Experimente

python agents/s07_task_system.py
  1. Crie 3 tarefas: "Setup project", "Write code", "Write tests". Faça-as depender umas das outras em ordem.
  2. Liste todas as tarefas e mostre o grafo de dependências
  3. Complete a tarefa 1 e depois liste as tarefas para ver a tarefa 2 desbloqueada
  4. Crie um quadro de tarefas para refatoração: parse -> transform -> emit -> test, onde transform e emit podem rodar em paralelo após parse