Planificacion de tareas para agentes AI: dependencias, eventos y todos jerarquicos

Pidele a un agente AI que “construya una REST API con autenticacion” y observa lo que pasa. Sin planificacion estructurada, saltara directamente a escribir codigo - omitiendo el diseno de base de datos, olvidandose de crear las migraciones e implementando el middleware de autenticacion antes de que exista el modelo de usuario.

El problema no es falta de inteligencia. Es que el agente no tiene forma de dividir tareas complejas en pasos, rastrear el progreso o entender dependencias. Simplemente ejecuta lo que parece correcto en el momento.

TL;DR

• Sin una lista de tareas, los agentes omiten pasos, repiten trabajo y pierden el hilo. Darles planificacion estructurada mejora dramaticamente la fiabilidad.
• pydantic-ai-todo es un toolset independiente de planificacion de tareas para Pydantic AI con backends in-memory, async memory y PostgreSQL.
• Subtareas y dependencias permiten planificacion jerarquica y respetan el orden de ejecucion - “el sistema de autenticacion necesita el modelo de usuario” se convierte en una restriccion firme, no una sugerencia.
• Deteccion de ciclos mediante busqueda en profundidad previene deadlocks cuando el agente crea dependencias circulares.
• Multitenencia en PostgreSQL con almacenamiento por sesion lo hace listo para produccion en aplicaciones web con usuarios concurrentes.

Hemos visto este patron destruir la fiabilidad de agentes en Vstorm. La solucion es sorprendentemente simple: dale al agente una lista de tareas. No una metaforica - una herramienta real que crea, rastrea y gestiona tareas estructuradas con estados, subtareas y dependencias.

Eso es lo que hace pydantic-ai-todo. Es un toolset independiente de planificacion de tareas para Pydantic AI con backends in-memory, async memory y PostgreSQL.

Configuracion basica: Una linea, planificacion completa

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

El agente ahora tiene herramientas de planificacion: read_todos, write_todos, add_todo, update_todo_status y remove_todo. El par principal - read_todos y write_todos - cubre la mayoria de los casos de uso. El agente crea listas de tareas estructuradas, actualiza estados mientras trabaja y mantiene el control de lo que esta hecho y lo que queda pendiente.

El modelo Todo

Cada elemento todo tiene una estructura clara:

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

El campo active_form esta en presente continuo - “Implementing JWT tokens” en lugar de “Implement JWT tokens.” Se usa para spinners de estado y pantallas de progreso que muestran lo que el agente esta haciendo actualmente.

Acceder a los todos despues de las ejecuciones del agente

Pasa una instancia de almacenamiento para acceder a los todos fuera del agente:

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

Subtareas y dependencias

Activa la planificacion jerarquica con 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)
""")

Con las subtareas activadas, el agente obtiene tres herramientas adicionales:

• add_subtask - crear una tarea vinculada a una tarea padre
• set_dependency - declarar que la tarea B depende de la tarea A
• get_available_tasks - listar tareas sin dependencias bloqueantes

Deteccion de ciclos

Cuando el agente establece dependencias, los ciclos se detectan automaticamente mediante busqueda en profundidad:

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)

Si la Tarea A depende de la Tarea B, y el agente intenta hacer que la Tarea B dependa de la Tarea A, recibe un mensaje de error explicando la dependencia circular. Sin deadlocks.

El sistema de eventos

Para integraciones en produccion, el emisor de eventos notifica sobre cambios en los todos:

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

Los eventos se disparan para: CREATED, UPDATED, STATUS_CHANGED, COMPLETED, DELETED. Puedes conectarlos a dashboards, notificaciones, webhooks o registros de auditoria.

Backend PostgreSQL multiinquilino

Para aplicaciones web donde multiples usuarios ejecutan agentes concurrentemente, el backend PostgreSQL proporciona persistencia con alcance por sesion:

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)

Todas las operaciones estan delimitadas por session_id. Alice nunca ve las tareas de Bob, y viceversa. El esquema se crea automaticamente en 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);

El indice en session_id asegura consultas rapidas por usuario incluso con millones de todos entre todos los usuarios.

Storage Factory

La factory create_storage() ofrece una API limpia para elegir 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",
)

El backend PostgreSQL soporta tanto connection_string (crea su propio pool) como pool (reutiliza uno existente). Cuando pasas un pool externo, close() no lo cerrara - solo los pools creados internamente se limpian.

Integracion con el System Prompt

Genera un system prompt dinamico que incluye el estado actual de los todos:

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)

Esto inyecta la lista actual de todos en el system prompt del agente, para que sepa lo que ya esta planificado y lo que esta en progreso - incluso entre reinicios de conversacion.

Conclusiones clave

• La planificacion estructurada reduce los pasos alucinados. Sin una lista de tareas, los agentes omiten pasos, repiten trabajo y pierden el hilo. Con una, planifican sistematicamente y ejecutan en orden.
• Las dependencias previenen errores de orden. “El sistema de autenticacion necesita el modelo de usuario” es una dependencia, no una sugerencia. El agente ve el estado blocked y trabaja primero en las tareas disponibles.
• La deteccion de ciclos es automatica. La deteccion de ciclos basada en DFS previene deadlocks cuando el agente crea dependencias circulares. El mensaje de error explica que salio mal.
• La multitenencia en PostgreSQL esta lista para produccion. Almacenamiento por sesion, connection pooling, creacion automatica de esquema y soporte para pools externos - la infraestructura aburrida que las aplicaciones web necesitan.
• Los eventos habilitan integraciones. Conecta eventos CREATED, COMPLETED, STATUS_CHANGED a dashboards, notificaciones y registros de auditoria.

Pruebalo tu mismo

pydantic-ai-todo - Toolset de planificacion de tareas para agentes Pydantic AI con subtareas, dependencias, eventos y soporte multiinquilino en PostgreSQL.

pip install pydantic-ai-todo