Planowanie zadan dla agentow AI: zaleznosci, eventy i hierarchiczne listy todo

Popros agenta AI o “zbudowanie REST API z uwierzytelnianiem” i obserwuj, co sie stanie. Bez strukturalnego planowania, rzuci sie prosto do pisania kodu - pomijajac projektowanie bazy danych, zapominajac o tworzeniu migracji i implementujac middleware auth przed istnieniem modelu uzytkownika.

Problem nie polega na inteligencji. Chodzi o to, ze agent nie ma sposobu na rozbicie zlozonych zadan na kroki, sledzenie postepu ani rozumienie zaleznosci. Po prostu wykonuje to, co w danej chwili wydaje sie odpowiednie.

TL;DR

• Bez listy todo, agenci pomijaja kroki, powtarzaja prace i traca orientacje w postepie. Danie agentom strukturalnego planowania zadan dramatycznie poprawia niezawodnosc.
• pydantic-ai-todo to samodzielny zestaw narzedzi do planowania zadan dla Pydantic AI z backendami in-memory, async memory i PostgreSQL.
• Podzadania i zaleznosci pozwalaja agentom planowac hierarchicznie i respektowac kolejnosc wykonywania - “system auth potrzebuje modelu uzytkownika” staje sie twardym ograniczeniem, nie sugestia.
• Wykrywanie cykli za pomoca przeszukiwania w glab (DFS) zapobiega deadlockom, gdy agent tworzy cykliczne zaleznosci.
• Wielodostepnosc PostgreSQL ze storage scopowanym na sesje sprawia, ze jest gotowy produkcyjnie dla aplikacji webowych z rownoleglymi uzytkownikami.

Widzielismy, jak ten wzorzec zabija niezawodnosc agentow w Vstorm. Rozwiazanie jest zaskakujaco proste: daj agentowi liste todo. Nie metaforyczna - faktyczne narzedzie, ktore tworzy, sledzi i zarzadza strukturalnymi zadaniami ze statusami, podzadaniami i zaleznosciami.

To wlasnie robi pydantic-ai-todo. To samodzielny zestaw narzedzi do planowania zadan dla Pydantic AI z backendami in-memory, async memory i PostgreSQL.

Podstawowa konfiguracja: jedna linia, pelne planowanie

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"
)

Agent ma teraz narzedzia do planowania: read_todos, write_todos, add_todo, update_todo_status i remove_todo. Podstawowa para - read_todos i write_todos - obsluguje wiekszosc przypadkow uzycia. Agent tworzy strukturalne listy zadan, aktualizuje statusy w trakcie pracy i sledzi, co jest zrobione, a co czeka.

Model Todo

Kazdy element todo ma jasna strukture:

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

Pole active_form jest w formie ciaglej - “Implementing JWT tokens” zamiast “Implement JWT tokens.” Sluzy do spinnerow statusu i wyswietlaczy postepu pokazujacych, co agent aktualnie robi.

Dostep do todo po uruchomieniu agenta

Przekaz instancje storage, aby uzyskac dostep do todo poza agentem:

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}")

Podzadania i zaleznosci

Wlacz hierarchiczne planowanie za pomoca 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)
""")

Z wlaczonymi podzadaniami, agent otrzymuje trzy dodatkowe narzedzia:

• add_subtask - tworzy zadanie polaczone z rodzicem
• set_dependency - deklaruje, ze zadanie B zalezy od zadania A
• get_available_tasks - listuje zadania bez blokujacych zaleznosci

Wykrywanie cykli

Gdy agent ustawia zaleznosci, cykle sa wykrywane automatycznie za pomoca przeszukiwania w glab:

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)

Jesli Zadanie A zalezy od Zadania B, a agent probuje uzaleznic Zadanie B od Zadania A, otrzymuje komunikat bledu wyjasniajacy cykliczna zaleznosc. Brak deadlockow.

System eventow

Dla integracji produkcyjnych, event emitter powiadamia cie o zmianach todo:

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,
)

Eventy sa emitowane dla: CREATED, UPDATED, STATUS_CHANGED, COMPLETED, DELETED. Mozesz sie do nich podpiac dla dashboardow, powiadomien, webhookow lub logowania audytowego.

Backend PostgreSQL z wielodostepnoscia

Dla aplikacji webowych, gdzie wielu uzytkownikow uruchamia agentow rownolegle, backend PostgreSQL zapewnia persistence scopowany na sesje:

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)

Wszystkie operacje sa scopowane przez session_id. Alice nigdy nie widzi zadan Boba i odwrotnie. Schemat tworzy sie automatycznie przy initialize():

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);

Indeks na session_id zapewnia szybkie zapytania per-uzytkownik nawet z milionami todo wsrod wszystkich uzytkownikow.

Fabryka storage

Fabryka create_storage() daje czyste API do wyboru backendow:

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",
)

Backend PostgreSQL obsluguje zarowno connection_string (tworzy wlasna pule) jak i pool (wykorzystuje istniejaca). Gdy przekazujesz zewnetrzna pule, close() jej nie zamknie - tylko pule utworzone wewnetrznie sa czyszczone.

Integracja z System Prompt

Generuj dynamiczny system prompt zawierajacy aktualny stan todo:

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)

To wstrzykuje aktualna liste todo do system promptu agenta, wiec wie, co jest juz zaplanowane i co jest w toku - nawet miedzy restartami konwersacji.

Kluczowe wnioski

• Strukturalne planowanie redukuje halucynowane kroki. Bez listy todo, agenci pomijaja kroki, powtarzaja prace i traca orientacje w postepie. Z lista, planuja systematycznie i wykonuja w kolejnosci.
• Zaleznosci zapobiegaja bledom kolejnosci. “System auth potrzebuje modelu uzytkownika” to zaleznosc, nie sugestia. Agent widzi status blocked i pracuje najpierw nad dostepnymi zadaniami.
• Wykrywanie cykli jest automatyczne. Wykrywanie cykli oparte na DFS zapobiega deadlockom, gdy agent tworzy cykliczne zaleznosci. Komunikat bledu wyjasnia, co poszlo nie tak.
• Wielodostepnosc PostgreSQL jest gotowa produkcyjnie. Storage scopowany na sesje, connection pooling, automatyczne tworzenie schematu i obsluga zewnetrznej puli - nudna infrastruktura, ktorej potrzebuja aplikacje webowe.
• Eventy umozliwiaja integracje. Podepnij sie do eventow CREATED, COMPLETED, STATUS_CHANGED dla dashboardow, powiadomien i sciezek audytowych.

Wyprobuj sam

pydantic-ai-todo - Zestaw narzedzi do planowania zadan dla agentow Pydantic AI z podzadaniami, zaleznosciami, eventami i obsluga PostgreSQL multi-tenant.

pip install pydantic-ai-todo