Aufgabenplanung fuer AI-Agenten: Abhaengigkeiten, Events und hierarchische Todos

Bitte einen AI-Agenten, “eine REST API mit Authentifizierung zu bauen” und schau, was passiert. Ohne strukturierte Planung springt er direkt zum Code schreiben - ueberspringt das Datenbankdesign, vergisst die Migrations zu erstellen und implementiert die Auth-Middleware, bevor das User-Model existiert.

Das Problem ist nicht fehlende Intelligenz. Es liegt daran, dass der Agent keine Moeglichkeit hat, komplexe Aufgaben in Schritte aufzuteilen, Fortschritt zu verfolgen oder Abhaengigkeiten zu verstehen. Er fuehrt einfach das aus, was im Moment richtig erscheint.

TL;DR

• Ohne eine Todo-Liste ueberspringt der Agent Schritte, wiederholt Arbeit und verliert den Ueberblick. Strukturierte Aufgabenplanung verbessert die Zuverlaessigkeit von Agenten drastisch.
• pydantic-ai-todo ist ein eigenstaendiges Aufgabenplanungs-Toolset fuer Pydantic AI mit In-Memory-, Async-Memory- und PostgreSQL-Backends.
• Unteraufgaben und Abhaengigkeiten ermoeglichen hierarchische Planung und erzwingen die Ausfuehrungsreihenfolge - “Auth-System braucht User-Model” wird zu einer harten Bedingung, nicht nur einem Vorschlag.
• Zykluserkennung per Tiefensuche verhindert Deadlocks, wenn der Agent zirkulaere Abhaengigkeiten erstellt.
• PostgreSQL-Mandantenfaehigkeit mit sitzungsbezogenem Speicher macht es produktionsreif fuer Webanwendungen mit gleichzeitigen Benutzern.

Wir haben bei Vstorm gesehen, wie dieses Muster die Zuverlaessigkeit von Agenten zerstoert. Die Loesung ist ueberraschend einfach: Gib dem Agenten eine Todo-Liste. Keine metaphorische - eine echte Tool-Funktion, die strukturierte Aufgaben mit Status, Unteraufgaben und Abhaengigkeiten erstellt, verfolgt und verwaltet.

Genau das macht pydantic-ai-todo. Es ist ein eigenstaendiges Aufgabenplanungs-Toolset fuer Pydantic AI mit In-Memory-, Async-Memory- und PostgreSQL-Backends.

Grundlegende Einrichtung: Eine Zeile, volle Planung

from pydantic_ai import Agent
from pydantic_ai_todo import create_todo_toolset

agent = Agent(
    "openai:gpt-4o",
    toolsets=[create_todo_toolset()],
)

result = await agent.run(
    "Create a todo list for building a REST API with user authentication"
)

Der Agent hat jetzt Planungs-Tools: read_todos, write_todos, add_todo, update_todo_status und remove_todo. Das Kernpaar - read_todos und write_todos - deckt die meisten Anwendungsfaelle ab. Der Agent erstellt strukturierte Aufgabenlisten, aktualisiert den Status waehrend der Arbeit und behaelt den Ueberblick, was erledigt und was noch offen ist.

Das Todo-Model

Jedes Todo-Element hat eine klare Struktur:

class Todo(BaseModel):
    id: str              # 8-char random ID
    content: str         # "Implement JWT token generation"
    status: str          # pending | in_progress | completed | blocked
    active_form: str     # "Implementing JWT token generation"
    parent_id: str | None  # Link to parent task
    depends_on: list[str]  # IDs of blocking tasks

Das Feld active_form steht im Present Continuous - “Implementing JWT tokens” statt “Implement JWT tokens.” Dies wird fuer Status-Spinner und Fortschrittsanzeigen verwendet, die zeigen, was der Agent gerade tut.

Zugriff auf Todos nach Agent-Laeufen

Uebergib eine Storage-Instanz, um ausserhalb des Agenten auf Todos zuzugreifen:

from pydantic_ai_todo import create_todo_toolset, TodoStorage

storage = TodoStorage()
toolset = create_todo_toolset(storage=storage)

agent = Agent(
    "openai:gpt-4o",
    toolsets=[toolset],
    system_prompt="""When asked to plan something:
1. Break it down into specific tasks
2. Use write_todos to create each task
3. Summarize the plan""",
)

result = await agent.run("Plan the implementation of a blog application")

# Access tasks directly
for todo in storage.todos:
    status_icon = "done" if todo.status == "completed" else "pending"
    print(f"  [{status_icon}] {todo.content}")

Unteraufgaben und Abhaengigkeiten

Aktiviere hierarchische Planung mit enable_subtasks=True:

from pydantic_ai_todo import create_todo_toolset, AsyncMemoryStorage

storage = AsyncMemoryStorage()
toolset = create_todo_toolset(
    async_storage=storage,
    enable_subtasks=True,
)

agent = Agent(
    "openai:gpt-4o",
    toolsets=[toolset],
    system_prompt="""When planning projects:
1. Create main tasks with write_todos
2. Break them into subtasks with add_subtask
3. Set dependencies where tasks must wait for others
4. Use get_available_tasks to see what can start now""",
)

result = await agent.run("""
Plan building a REST API with:
- Database design (must be done first)
- User model (needs database)
- Auth system (needs user model)
- API endpoints (needs auth)
""")

Mit aktivierten Unteraufgaben erhaelt der Agent drei zusaetzliche Tools:

• add_subtask - eine Aufgabe erstellen, die mit einer uebergeordneten verknuepft ist
• set_dependency - festlegen, dass Aufgabe B von Aufgabe A abhaengt
• get_available_tasks - Aufgaben auflisten, die keine blockierenden Abhaengigkeiten haben

Zykluserkennung

Wenn der Agent Abhaengigkeiten setzt, werden Zyklen automatisch per Tiefensuche erkannt:

def _has_cycle(todo_id: str, dependency_id: str, todos: list[Todo]) -> bool:
    """Check if adding dependency would create a cycle."""
    visited: set[str] = set()

    def visit(current_id: str) -> bool:
        if current_id == todo_id:
            return True  # Cycle found
        if current_id in visited:
            return False
        visited.add(current_id)
        todo = _get_todo_by_id(current_id)
        if todo:
            for dep_id in todo.depends_on:
                if visit(dep_id):
                    return True
        return False

    return visit(dependency_id)

Wenn Aufgabe A von Aufgabe B abhaengt und der Agent versucht, Aufgabe B von Aufgabe A abhaengig zu machen, erhaelt er eine Fehlermeldung, die die zirkulaere Abhaengigkeit erklaert. Keine Deadlocks.

Das Event-System

Fuer Produktions-Integrationen benachrichtigt der Event-Emitter ueber Todo-Aenderungen:

from pydantic_ai_todo import create_storage, TodoEventEmitter

emitter = TodoEventEmitter()

@emitter.on_completed
async def notify_completion(event):
    # Send notification, update dashboard, etc.
    print(f"Task completed: {event.todo.content}")

storage = create_storage(
    "postgres",
    connection_string="postgresql://user:pass@localhost/db",
    session_id="user-123",
    event_emitter=emitter,
)

Events werden ausgeloest fuer: CREATED, UPDATED, STATUS_CHANGED, COMPLETED, DELETED. Du kannst sie fuer Dashboards, Benachrichtigungen, Webhooks oder Audit-Logging nutzen.

PostgreSQL Multi-Tenant-Backend

Fuer Webanwendungen, in denen mehrere Benutzer gleichzeitig Agenten ausfuehren, bietet das PostgreSQL-Backend sitzungsbezogene Persistenz:

from pydantic_ai_todo import create_storage, create_todo_toolset

# User A's session
storage_a = create_storage(
    "postgres",
    connection_string="postgresql://user:pass@localhost/mydb",
    session_id="user-alice",
)
await storage_a.initialize()

# User B's session
storage_b = create_storage(
    "postgres",
    connection_string=connection_string,
    session_id="user-bob",
)
await storage_b.initialize()

# Each user has separate todos
toolset_a = create_todo_toolset(async_storage=storage_a)
toolset_b = create_todo_toolset(async_storage=storage_b)

Alle Operationen sind durch session_id begrenzt. Alice sieht niemals Bobs Aufgaben und umgekehrt. Das Schema wird automatisch bei initialize() erstellt:

CREATE TABLE IF NOT EXISTS todos (
    id VARCHAR(8) PRIMARY KEY,
    session_id VARCHAR(255) NOT NULL,
    content TEXT NOT NULL,
    status VARCHAR(20) NOT NULL,
    active_form TEXT NOT NULL,
    parent_id VARCHAR(8),
    depends_on TEXT[] DEFAULT '{}',
    created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
    updated_at TIMESTAMP WITH TIME ZONE DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS idx_todos_session_id ON todos(session_id);

Der Index auf session_id stellt schnelle benutzerbezogene Abfragen sicher, selbst bei Millionen von Todos ueber alle Benutzer hinweg.

Storage Factory

Die create_storage()-Factory bietet eine saubere API zur Auswahl der Backends:

from pydantic_ai_todo import create_storage

# In-memory (development, testing)
storage = create_storage("memory")

# PostgreSQL (production)
storage = create_storage(
    "postgres",
    connection_string="postgresql://...",
    session_id="user-123",
)
await storage.initialize()

# With existing connection pool
import asyncpg
pool = await asyncpg.create_pool("postgresql://...")
storage = create_storage(
    "postgres",
    pool=pool,
    session_id="user-123",
)

Das PostgreSQL-Backend unterstuetzt sowohl connection_string (erstellt einen eigenen Pool) als auch pool (nutzt einen bestehenden). Wenn du einen externen Pool uebergibst, schliesst close() ihn nicht - nur intern erstellte Pools werden aufgeraeumt.

System-Prompt-Integration

Generiere einen dynamischen System-Prompt, der den aktuellen Todo-Status enthaelt:

from pydantic_ai_todo import get_todo_system_prompt

prompt = get_todo_system_prompt(storage)
# Returns base todo instructions + current todo list

# For async storage:
from pydantic_ai_todo import get_todo_system_prompt_async
prompt = await get_todo_system_prompt_async(async_storage)

Dies injiziert die aktuelle Todo-Liste in den System-Prompt des Agenten, sodass er weiss, was bereits geplant ist und was gerade bearbeitet wird - selbst ueber Gespraeche hinweg.

Wichtigste Erkenntnisse

• Strukturierte Planung reduziert halluzinierte Schritte. Ohne Todo-Liste ueberspringt der Agent Schritte, wiederholt Arbeit und verliert den Ueberblick. Mit einer plant er systematisch und fuehrt in der richtigen Reihenfolge aus.
• Abhaengigkeiten verhindern Reihenfolgefehler. “Auth-System braucht User-Model” ist eine Abhaengigkeit, kein Vorschlag. Der Agent sieht den Status blocked und arbeitet zuerst an verfuegbaren Aufgaben.
• Zykluserkennung ist automatisch. DFS-basierte Zykluserkennung verhindert Deadlocks, wenn der Agent zirkulaere Abhaengigkeiten erstellt. Die Fehlermeldung erklaert, was schiefgelaufen ist.
• PostgreSQL-Mandantenfaehigkeit ist produktionsreif. Sitzungsbezogener Speicher, Connection-Pooling, automatische Schema-Erstellung und Unterstuetzung externer Pools - die langweilige Infrastruktur, die Webanwendungen brauchen.
• Events ermoeglichen Integrationen. Nutze CREATED, COMPLETED, STATUS_CHANGED-Events fuer Dashboards, Benachrichtigungen und Audit-Trails.

Probiere es selbst aus

pydantic-ai-todo - Aufgabenplanungs-Toolset fuer Pydantic AI Agenten mit Unteraufgaben, Abhaengigkeiten, Events und PostgreSQL Multi-Tenant-Support.

pip install pydantic-ai-todo