How to Build an AI Agent
How to Build an AI Agent
s11

Agentes Autónomos

Colaboración

Scan Board, Claim Tasks

499 LOC14 herramientasTask board polling + timeout-based self-governance
Teammates scan the board and claim tasks themselves; no need for the lead to assign each one

s01 > s02 > s03 > s04 > s05 > s06 | s07 > s08 > s09 > s10 > [ s11 ] s12

"Los compañeros escanean el tablero y claman tareas ellos mismos" -- no es necesario que el lead asigne cada uno.

Problema

En s09-s10, los compañeros solo trabajan cuando se les dice explícitamente. El lead debe generar cada uno con un prompt específico. ¿10 tareas sin reclamar en el tablero? El lead asigna cada una manualmente. No escala.

Autonomía verdadera: los compañeros escanean el tablero de tareas ellos mismos, claman tareas sin reclamar, trabajan en ellas, luego buscan más.

Una sutileza: después de la compresión de contexto (s06), el agente podría olvidar quién es. La re-inyección de identidad corrige esto.

Solución

Ciclo de vida del compañero con ciclo idle:

+-------+
| spawn |
+---+---+
    |
    v
+-------+   tool_use     +-------+
| WORK  | <------------- |  LLM  |
+-------+                +-------+
    |
    | stop_reason != tool_use (or idle tool called)
    v
+--------+
|  IDLE  |  poll every 5s for up to 60s
+---+----+
    |
    +---> check inbox --> message? ----------> WORK
    |
    +---> scan .tasks/ --> unclaimed? -------> claim -> WORK
    |
    +---> 60s timeout ----------------------> SHUTDOWN

Re-inyección de identidad después de compresión:
  if len(messages) <= 3:
    messages.insert(0, identity_block)

Cómo Funciona

  1. El bucle del compañero tiene dos fases: WORK y IDLE. Cuando el LLM deja de llamar herramientas (o llama idle), el compañero entra en IDLE.
def _loop(self, name, role, prompt):
    while True:
        # -- WORK PHASE --
        messages = [{"role": "user", "content": prompt}]
        for _ in range(50):
            response = client.messages.create(...)
            if response.stop_reason != "tool_use":
                break
            # execute tools...
            if idle_requested:
                break

        # -- IDLE PHASE --
        self._set_status(name, "idle")
        resume = self._idle_poll(name, messages)
        if not resume:
            self._set_status(name, "shutdown")
            return
        self._set_status(name, "working")
  1. La fase idle hace polling de bandeja de entrada y tablero de tareas en bucle.
def _idle_poll(self, name, messages):
    for _ in range(IDLE_TIMEOUT // POLL_INTERVAL):  # 60s / 5s = 12
        time.sleep(POLL_INTERVAL)
        inbox = BUS.read_inbox(name)
        if inbox:
            messages.append({"role": "user",
                "content": f"<inbox>{inbox}</inbox>"})
            return True
        unclaimed = scan_unclaimed_tasks()
        if unclaimed:
            claim_task(unclaimed[0]["id"], name)
            messages.append({"role": "user",
                "content": f"<auto-claimed>Task #{unclaimed[0]['id']}: "
                           f"{unclaimed[0]['subject']}</auto-claimed>"})
            return True
    return False  # timeout -> shutdown
  1. Escaneo del tablero de tareas: encontrar tareas pending, sin owner, no bloqueadas.
def scan_unclaimed_tasks() -> list:
    unclaimed = []
    for f in sorted(TASKS_DIR.glob("task_*.json")):
        task = json.loads(f.read_text())
        if (task.get("status") == "pending"
                and not task.get("owner")
                and not task.get("blockedBy")):
            unclaimed.append(task)
    return unclaimed
  1. Re-inyección de identidad: cuando el contexto es muy corto (compresión ocurrió), insertar un bloque de identidad.
if len(messages) <= 3:
    messages.insert(0, {"role": "user",
        "content": f"<identity>You are '{name}', role: {role}, "
                   f"team: {team_name}. Continue your work.</identity>"})
    messages.insert(1, {"role": "assistant",
        "content": f"I am {name}. Continuing."})

Qué Cambió Respecto a s10

ComponenteAntes (s10)Después (s11)
Herramientas1214 (+idle, +claim_task)
AutonomíaDirigido por leadAuto-organizado
Fase idleNonePolling bandeja de entrada + tablero de tareas
Reclamo de tareasSolo manualAuto-reclamar tareas sin reclamar
IdentidadPrompt sistema+ re-inyección después de compresión
TimeoutNone60s idle -> apagado automático

Pruébalo

python agents/s11_autonomous_agents.py
  1. Crear 3 tareas en el tablero, luego generar alice y bob. Observarlos auto-reclamar.
  2. Generar un compañero codificador y dejar que encuentre trabajo desde el tablero de tareas
  3. Crear tareas con dependencias. Observar a los compañeros respetar el orden bloqueado.
  4. Escribir /tasks para ver el tablero de tareas con owners
  5. Escribir /team para monitorear quién está trabajando vs idle