s09
Agenten-Teams
ZusammenarbeitTeammates + 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
- 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 = {}
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})"
- 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)
- 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
| Komponente | Vorher (s08) | Nachher (s09) |
|---|---|---|
| Tools | 6 | 9 (+spawn/send/read_inbox) |
| Agents | Single | Lead + N teammates |
| Persistenz | Keine | config.json + JSONL inboxes |
| Threads | Hintergrund-Befehle | Vollständige Agenten-Schleifen pro Thread |
| Lebenszyklus | Fire-and-forget | idle -> working -> idle |
| Kommunikation | Keine | message + broadcast |
Ausprobieren
python agents/s09_agent_teams.py
Spawn alice (coder) und bob (tester). Lasse alice eine Nachricht an bob senden.Broadcast "status update: phase 1 complete" an alle TeamkollegenPrüfe das Lead-Postfach auf Nachrichten- Tippe
/teamum den Team-Roster mit Status zu sehen - Tippe
/inboxum das Lead-Postfach manuell zu prüfen
