Command Queue: Agenten-Runs effizient verwalten

Kennst du das? Mehrere Nachrichten kommen gleichzeitig rein, dein Agent startet für jede Nachricht einen eigenen LLM-Call und plötzlich knallt es. Session-Dateien werden überschrieben, Logs vermischen sich und das Upstream Rate Limit ist sofort erreicht. Es ist frustrierend, wenn parallele Anfragen die Zuverlässigkeit deines Systems zerstören, nur weil die Synchronisierung fehlt.

Hier hilft der Command Queue. Er serialisiert eingehende Runs über eine kleine In-Process Queue, um Kollisionen zu vermeiden, während er gleichzeitig sichere Parallelität über verschiedene Sessions hinweg erlaubt.

Voraussetzungen

• Inbound-Kanäle, die die Gateway Reply Pipeline nutzen (z. B. WhatsApp, Telegram, Slack, Discord).
• runEmbeddedPiAgent für das Enqueueing via Session Key.

Schnellstart

In der Standardeinstellung werden alle Nachrichten über den Modus collect verarbeitet. Das System wartet kurz und fasst Nachrichten zusammen, bevor ein Run startet.

Du kannst das Verhalten global in deiner Config anpassen:

{
  messages: {
    queue: {
      mode: "collect",
      debounceMs: 1000,
      cap: 20,
      drop: "summarize",
      byChannel: { discord: "collect" },
    },
  },
}

Wenn du das Verhalten nur für eine bestimmte Session ändern willst, sende einfach einen Befehl direkt im Chat:

• /queue collect (Empfohlen: Sammelt Nachrichten für einen Followup-Run)
• /queue steer (Injiziert die Nachricht sofort in den aktuellen Run)

Funktionsweise und Modi

Der Queue arbeitet Lane-aware. Das bedeutet, jede Session hat ihre eigene Lane (session:<key>), damit sich Sessions nicht gegenseitig blockieren. Diese landen dann in einer globalen Lane (standardmäßig main), deren Parallelität du über agents.defaults.maxConcurrent steuerst.

Ich empfehle collect oder steer, wenn du genau eine Antwort pro eingehender Nachricht erwartest. Hier sind die wichtigsten Modi:

• collect: Fasst alle wartenden Nachrichten zu einem einzigen Followup-Turn zusammen (Standard).
• steer: Injiziert die Nachricht sofort in den laufenden Run und bricht anstehende Tool-Calls ab.
• followup: Reiht die Nachricht für den nächsten Turn ein, nachdem der aktuelle Run beendet ist.
• steer-backlog: Kombiniert steer und followup.

Die Steuerung erfolgt ohne externe Abhängigkeiten oder Worker-Threads – es ist pures TypeScript mit Promises. Typing indicators werden beim Enqueueing sofort ausgelöst, sodass die User Experience flüssig bleibt.

Queue Optionen

Für die Modi followup, collect und steer-backlog kannst du das Verhalten feinjustieren:

• debounceMs: Wartezeit auf weitere Nachrichten (Standard: 1000).
• cap: Maximale Anzahl an Nachrichten in der Queue pro Session (Standard: 20).
• drop: Richtlinie bei Überlauf. summarize erstellt eine Liste der verworfenen Nachrichten und fügt sie als Prompt an.

Fehlerbehebung

• Befehle scheinen festzustecken: Aktiviere verbose Logs. Suche nach Zeilen wie „queued for …ms“, um zu prüfen, ob der Queue korrekt abgearbeitet wird.
• Queue-Tiefe prüfen: Wenn du wissen willst, wie viele Aufgaben warten, aktiviere verbose Logs und beobachte die Timing-Zeilen des Queue-Systems.

Nutze den AI Setup Assistant, falls du spezifische Fragen zur Integration in dein System hast.

Nächste Schritte

• Gateway Pipeline Dokumentation
• Agent Configuration Reference