How to Build an AI Agent
How to Build an AI Agent
s09

Agenten-Teams

Zusammenarbeit

Teammates + Mailboxes

348 LOC10 ToolsTeammateManager + file-based mailbox
When one agent can't finish, delegate to persistent teammates via async mailboxes

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

"Wenn die Aufgabe zu groß für einen ist, delegiere an Teamkollegen" -- persistente Teamkollegen + asynchrone Mailboxen.

Problem

Subagenten (s04) sind Einweg: spawnen, arbeiten, Zusammenfassung zurückgeben, sterben. Keine Identität, kein Gedächtnis zwischen Aufrufen. Hintergrundaufgaben (s08) führen Shell-Befehle aus, können aber keine LLM-gesteuerten Entscheidungen treffen.

Echte Teamarbeit braucht: (1) persistente Agenten, die einen einzelnen Prompt überleben, (2) Identitäts- und Lebenszyklusmanagement, (3) einen Kommunikationskanal zwischen Agenten.

Lösung

Teamkollegen-Lebenszyklus:
  spawn -> WORKING -> IDLE -> WORKING -> ... -> SHUTDOWN

Kommunikation:
  .team/
    config.json           <- Team-Roster + Status
    inbox/
      alice.jsonl        <- nur anhängen, lesen und leeren
      bob.jsonl
      lead.jsonl

              +--------+    send("alice","bob","...")    +--------+
              | alice  | -----------------------------> |  bob   |
              | loop   |    bob.jsonl << {json_line}     |  loop  |
              +--------+                                +--------+
                   ^                                         |
                   |        BUS.read_inbox("alice")          |
                   +---- alice.jsonl -> read + drain ---------+

Wie es funktioniert

  1. TeammateManager verwaltet config.json mit dem Team-Roster.
class TeammateManager:
    def __init__(self, team_dir: Path):
        self.dir = team_dir
        self.dir.mkdir(exist_ok=True)
        self.config_path = self.dir / "config.json"
        self.config = self._load_config()
        self.threads = {}
  1. spawn() erstellt einen Teamkollegen und startet seine Agenten-Schleife in einem Thread.
def spawn(self, name: str, role: str, prompt: str) -> str:
    member = {"name": name, "role": role, "status": "working"}
    self.config["members"].append(member)
    self._save_config()
    thread = threading.Thread(
        target=self._teammate_loop,
        args=(name, role, prompt), daemon=True)
    thread.start()
    return f"Spawned teammate '{name}' (role: {role})"
  1. MessageBus: Nur-anhängen JSONL-Postfächer. send() hängt eine JSON-Zeile an; read_inbox() liest alles und leert.
class MessageBus:
    def send(self, sender, to, content, msg_type="message", extra=None):
        msg = {"type": msg_type, "from": sender,
               "content": content, "timestamp": time.time()}
        if extra:
            msg.update(extra)
        with open(self.dir / f"{to}.jsonl", "a") as f:
            f.write(json.dumps(msg) + "\n")

    def read_inbox(self, name):
        path = self.dir / f"{name}.jsonl"
        if not path.exists(): return "[]"
        msgs = [json.loads(l) for l in path.read_text().strip().splitlines() if l]
        path.write_text("")  # drain
        return json.dumps(msgs, indent=2)
  1. Jeder Teamkollege prüft sein Postfach vor jedem LLM-Aufruf und fügt empfangene Nachrichten in den Kontext ein.
def _teammate_loop(self, name, role, prompt):
    messages = [{"role": "user", "content": prompt}]
    for _ in range(50):
        inbox = BUS.read_inbox(name)
        if inbox != "[]":
            messages.append({"role": "user",
                "content": f"<inbox>{inbox}</inbox>"})
            messages.append({"role": "assistant",
                "content": "Noted inbox messages."})
        response = client.messages.create(...)
        if response.stop_reason != "tool_use":
            break
        # execute tools, append results...
    self._find_member(name)["status"] = "idle"

Was sich von s08 geändert hat

KomponenteVorher (s08)Nachher (s09)
Tools69 (+spawn/send/read_inbox)
AgentsSingleLead + N teammates
PersistenzKeineconfig.json + JSONL inboxes
ThreadsHintergrund-BefehleVollständige Agenten-Schleifen pro Thread
LebenszyklusFire-and-forgetidle -> working -> idle
KommunikationKeinemessage + broadcast

Ausprobieren

python agents/s09_agent_teams.py
  1. Spawn alice (coder) und bob (tester). Lasse alice eine Nachricht an bob senden.
  2. Broadcast "status update: phase 1 complete" an alle Teamkollegen
  3. Prüfe das Lead-Postfach auf Nachrichten
  4. Tippe /team um den Team-Roster mit Status zu sehen
  5. Tippe /inbox um das Lead-Postfach manuell zu prüfen