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

Agenti Autonomi

Collaborazione

Scan Board, Claim Tasks

499 LOC14 strumentiTask 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

"I compagni scansionano la bacheca e prendono i task da soli" -- non c'è bisogno che il lead assegni ciascuno.

Problema

In s09-s10, i compagni lavorano solo quando esplicitamente detto. Il lead deve spawnare ciascuno con un prompt specifico. 10 task non presi sulla bacheca? Il lead assegna ciascuno manualmente. Non scala.

Autonomia vera: i compagni scansionano la bacheca dei task da soli, prendono i task non presi, ci lavorano, poi cercano altro.

Una sottigliezza: dopo la compressione del contesto (s06), l'agente potrebbe dimenticare chi è. La re-iniezione di identità risolve questo.

Soluzione

Ciclo di vita del compagno 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-iniezione identità dopo compressione:
  if len(messages) <= 3:
    messages.insert(0, identity_block)

Come Funziona

  1. Il loop del compagno ha due fasi: WORK e IDLE. Quando il LLM smette di chiamare strumenti (o chiama idle), il compagno entra in 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 pollinga inbox e bacheca task in un loop.
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. Scansione bacheca task: trova task pending, senza owner, non bloccati.
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-iniezione identità: quando il contesto è troppo corto (compressione avvenuta), inserisci un blocco identità.
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."})

Cosa è Cambiato da s10

ComponentePrima (s10)Dopo (s11)
Strumenti1214 (+idle, +claim_task)
AutonomiaLead-directedAuto-organizzante
Fase idleNessunaPoll inbox + bacheca task
Presa taskSolo manualeAuto-claim task non presi
IdentitàPrompt sistema+ re-iniezione dopo compress
TimeoutNessuno60s idle -> shutdown auto

Provalo

python agents/s11_autonomous_agents.py
  1. Crea 3 task sulla bacheca, poi spawna alice e bob. Guardali auto-prendere.
  2. Spawna un compagno coder e falli trovare lavoro dalla bacheca task
  3. Crea task con dipendenze. Guarda i compagni rispettare l'ordine bloccato.
  4. Digita /tasks per vedere la bacheca task con i proprietari
  5. Digita /team per monitorare chi sta lavorando vs idle