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

Aufgaben

Planung & Koordination

Task Graph + Dependencies

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

"Große Ziele in kleine Aufgaben aufteilen, sie ordnen und auf Disk persistieren" -- ein dateibasiertes Aufgaben-Graph mit Abhängigkeiten, das die Grundlage für die Multi-Agenten-Zusammenarbeit bildet.

Problem

s03's TodoManager ist eine flache Checkliste im Speicher: keine Ordnung, keine Abhängigkeiten, kein Status außer erledigt oder nicht. Echte Ziele haben Struktur -- Aufgabe B hängt von Aufgabe A ab, Aufgaben C und D können parallel laufen, Aufgabe E wartet auf sowohl C als D.

Ohne explizite Beziehungen kann der Agent nicht sagen, was bereit ist, was blockiert ist, oder was parallel laufen kann. Und da die Liste nur im Speicher lebt, löscht die Kontextkomprimierung (s06) sie vollständig.

Lösung

Erweitern Sie die Checkliste zu einem Aufgaben-Graph, der auf Disk persistiert wird. Jede Aufgabe ist eine JSON-Datei mit Status, Abhängigkeiten (blockedBy) und Abhängige (blocks). Der Graph beantwortet jederzeit drei Fragen:

  • Was ist bereit? -- Aufgaben mit pending-Status und leerer blockedBy-Liste.
  • Was ist blockiert? -- Aufgaben, die auf unvollständige Abhängigkeiten warten.
  • Was ist erledigt? -- completed-Aufgaben, deren Abschluss automatisch Abhängige freigibt.
.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"}

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

Reihenfolge:    task 1 muss vor 2 und 3 fertig sein
Parallelität:    task 2 und 3 können gleichzeitig laufen
Abhängigkeiten:  task 4 wartet auf sowohl 2 und 3
Status:          pending -> in_progress -> completed

Dieser Aufgaben-Graph wird zum Koordinations-Rückgrat für alles nach s07: Hintergrundausführung (s08), Multi-Agenten-Teams (s09+) und Worktree-Isolation (s12) lesen und schreiben alle in dieselbe Struktur.

Wie es funktioniert

  1. TaskManager: Eine JSON-Datei pro Aufgabe, CRUD mit Abhängigkeits-Graph.
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. Abhängigkeitsauflösung: Das Abschließen einer Aufgabe entfernt ihre ID aus jeder anderen Aufgabe's blockedBy-Liste und gibt automatisch Abhängige frei.
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 + Abhängigkeits-Verdrahtung: update behandelt Übergänge und Abhängigkeitskanten.
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. Vier Aufgaben-Tools kommen in die Dispatch-Map.
TOOL_HANDLERS = {
    # ...basis 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"]),
}

Ab s07 ist der Aufgaben-Graph die Standardmethode für mehrstufige Arbeiten. s03's Todo bleibt für schnelle einzelne Sitzungs-Checklisten.

Was sich von s06 geändert hat

KomponenteVorher (s06)Nachher (s07)
Tools58 (task_create/update/list/get)
PlanungsmodellFlache Checkliste (im Speicher)Aufgaben-Graph mit Abhängigkeiten (auf Disk)
BeziehungenKeineblockedBy + blocks Kanten
Status-VerfolgungErledigt oder nichtpending -> in_progress -> completed
PersistenzBei Komprimierung verlorenÜberlebt Komprimierung und Neustarts

Ausprobieren

python agents/s07_task_system.py
  1. Erstelle 3 Aufgaben: "Setup project", "Write code", "Write tests". Lasse sie in dieser Reihenfolge voneinander abhängen.
  2. Liste alle Aufgaben auf und zeige den Abhängigkeitsgraphen
  3. Erledige Aufgabe 1 und liste dann die Aufgaben auf, um zu sehen, dass Aufgabe 2 freigegeben wurde
  4. Erstelle eine Aufgabenwand für Refactoring: parse -> transform -> emit -> test, wobei transform und emit nach parse parallel laufen können