s11
Agenti Autonomi
CollaborazioneScan 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
- 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")
- 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
- 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
- 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
| Componente | Prima (s10) | Dopo (s11) |
|---|---|---|
| Strumenti | 12 | 14 (+idle, +claim_task) |
| Autonomia | Lead-directed | Auto-organizzante |
| Fase idle | Nessuna | Poll inbox + bacheca task |
| Presa task | Solo manuale | Auto-claim task non presi |
| Identità | Prompt sistema | + re-iniezione dopo compress |
| Timeout | Nessuno | 60s idle -> shutdown auto |
Provalo
python agents/s11_autonomous_agents.py
Crea 3 task sulla bacheca, poi spawna alice e bob. Guardali auto-prendere.Spawna un compagno coder e falli trovare lavoro dalla bacheca taskCrea task con dipendenze. Guarda i compagni rispettare l'ordine bloccato.- Digita
/tasksper vedere la bacheca task con i proprietari - Digita
/teamper monitorare chi sta lavorando vs idle
