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

Autonome Agenten

Zusammenarbeit

Scan Board, Claim Tasks

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

"Teamkollegen scannen die Wand und beanspruchen Aufgaben selbst" -- kein Bedarf für den Lead, jedem einzelnen eine zuzuweisen.

Problem

In s09-s10 arbeiten Teamkollegen nur, wenn sie explizit dazu aufgefordert werden. Der Lead muss jeden mit einem spezifischen Prompt spawnen. 10 nicht beanspruchte Aufgaben an der Wand? Der Lead weist jeden manuell zu. Skaliert nicht.

Echte Autonomie: Teamkollegen scannen selbst die Aufgabenwand, beanspruchen nicht beanspruchte Aufgaben, arbeiten daran, dann suchen sie weiter.

Eine Feinheit: Nach der Kontextkomprimierung (s06) könnte der Agent vergessen, wer er ist. Identitäts-Wiedereinschreibung behebt das.

Lösung

Teamkollegen-Lebenszyklus mit Leerlauf-Zyklus:

+-------+
| spawn |
+---+---+
    |
    v
+-------+   tool_use     +-------+
| WORK  | <------------- |  LLM  |
+---+---+                +-------+
    |
    | stop_reason != tool_use (oder idle tool aufgerufen)
    v
+--------+
|  IDLE  |  poll alle 5s für bis zu 60s
+---+----+
    |
    +---> check inbox --> Nachricht? ----------> WORK
    |
    +---> scan .tasks/ --> nicht beansprucht? --> beanspruchen -> WORK
    |
    +---> 60s Timeout ----------------------> SHUTDOWN

Identitäts-Wiedereinschreibung nach Komprimierung:
  if len(messages) <= 3:
    messages.insert(0, identity_block)

Wie es funktioniert

  1. Die Teamkollegen-Schleife hat zwei Phasen: WORK und IDLE. Wenn der LLM aufhört, Tools aufzurufen (oder idle aufruft), geht der Teamkollege in IDLE über.
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. Die IDLE-Phase pollt Postfach und Aufgabenwand in einer Schleife.
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. Aufgabenwand-Scanning: Finde ausstehende, nicht beanspruchte, nicht blockierte Aufgaben.
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. Identitäts-Wiedereinschreibung: Wenn der Kontext zu kurz ist (Komprimierung passiert), füge einen Identitäts-Block ein.
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."})

Was sich von s10 geändert hat

KomponenteVorher (s10)Nachher (s11)
Tools1214 (+idle, +claim_task)
AutonomieLead-gesteuertSelbstorganisierend
IDLE-PhaseKeinePoll inbox + Aufgabenwand
AufgabenbeanspruchungNur manuellAuto-Claim nicht beanspruchte Aufgaben
IdentitätSystem-Prompt+ Wiedereinschreibung nach Komprimierung
TimeoutKeine60s Idle -> automatisches Herunterfahren

Ausprobieren

python agents/s11_autonomous_agents.py
  1. Erstelle 3 Aufgaben an der Wand, dann spawn alice und bob. Beobachte, wie sie auto-beanspruchen.
  2. Spawn einen Coder-Teamkollegen und lasse ihn die Arbeit von der Aufgabenwand selbst finden
  3. Erstelle Aufgaben mit Abhängigkeiten. Beobachte, wie Teamkollegen die blockierte Reihenfolge respektieren.
  4. Tippe /tasks um die Aufgabenwand mit Besitzern zu sehen
  5. Tippe /team um zu überwachen, wer arbeitet vs. im Leerlauf