Hallo zusammen, hier ist Alex von agntai.net! Wir sind im März 2026 und ich habe in letzter Zeit viel zu viel Zeit damit verbracht, darüber nachzudenken, wie wir KI-Agenten aufbauen. Insbesondere habe ich Schwierigkeiten mit dem „Binding Code“ – den Elementen, die alle ausgefeilten LLM-Ausgaben, Toolaufrufe und das Statusmanagement verbinden. Wir haben alle die beeindruckenden Demos gesehen, oder? Agenten, die erstaunliche Dinge tun. Aber dann versucht man, einen für ein Problem aus der realen Welt zu bauen, und man stößt auf eine Mauer aus Rückrufen, Bedingungslogik und Statusaktualisierungen. Es fühlt sich weniger nach dem Bau eines intelligenten Systems an und mehr wie das Management einer sehr komplexen Spaghetti-Fabrik.
Also möchte ich heute über etwas sprechen, das heimlich an Beliebtheit gewonnen hat und, um ehrlich zu sein, meine mentale Gesundheit bewahrt: ereignisgesteuerte Architekturen für KI-Agenten. Dies ist kein neues Konzept in der Softwaretechnik, ganz im Gegenteil, aber es durchdacht auf KI-Agenten anzuwenden, besonders solche, die mehrere LLM-Interaktionen und externe Tools orchestrieren, scheint eine erfrischende Wendung zu sein. Lassen wir für einen Moment das lineare und schrittweise Denken beiseite. Lassen Sie uns an reaktive Systeme denken.
Mein persönlicher Kampf mit monolithischen Agenten
Vor ein paar Monaten arbeitete ich an einem Agenten, der mir helfen sollte, meinen Freiberufler-Schreibpipeline zu verwalten. Die Idee war einfach: Er würde mein Postfach auf neue Anfragen überwachen, erste Antworten verfassen, relevante frühere Artikel aus meiner Wissensdatenbank vorschlagen und mir sogar helfen, Follow-up-Anrufe zu planen. Das schien ziemlich einfach.
Mein ursprünglicher Ansatz war ziemlich typisch: eine Hauptschleife. Eine E-Mail empfangen. Die E-Mail analysieren. Die Aktion bestimmen (verfassen, planen, recherchieren). LLM aufrufen. Die LLM-Ausgabe verarbeiten. Ein Tool aufrufen (Kalender-API, E-Mail-API, Wissensdatenbank-API). Den internen Status aktualisieren. Wiederholen.
Das ging gut los, aber als ich mehr „Intelligenz“ und mehr Tools hinzufügte, wurde es zu einem Albtraum. Was passierte, wenn der Aufruf zur Kalender-API fehlschlug? Was passierte, wenn das LLM einen Kontakt halluzinierte, der nicht existierte? Was passierte, wenn ich eine Pause machen und um menschliches Feedback für eine kritische Entscheidung bitten musste? Mein einzigartiger monolithischer Agentenskript verwandelte sich schnell in ein verschachteltes Labyrinth von `if/else` mit `try/except`-Blöcken überall. Das Debuggen war ein Albtraum. Das Ändern eines Teils brach oft einen anderen. Es schien, als müsste ich ständig Lecks in einem sinkenden Schiff stopfen.
Ich erinnere mich an eine spätabendliche Sitzung, in der ich versuchte zu verstehen, warum mein Agent weiterhin Antworten für E-Mails verfasste, die er bereits bearbeitet hatte. Es stellte sich heraus, dass die Statusaktualisierung für „E-Mail bearbeitet“ *nach* einer möglichen neuen Ausführung des LLM in einem Fehlereingangsweg stattfand. Es war eine klassische Wettbewerbsbedingung in einem System, das nicht dafür ausgelegt war, asynchrone und nicht deterministische Operationen angemessen zu verarbeiten. Das war der Punkt, an dem ich begann, nach einer besseren Lösung zu suchen.
Warum ereignisgesteuerte Agenten sinnvoll sind
Denken Sie darüber nach, wie Menschen funktionieren. Wir befolgen normalerweise kein striktes, vorher festgelegtes Skript für jede Interaktion. Wir reagieren auf Dinge. Jemand stellt eine Frage – das ist ein Ereignis. Wir bearbeiten dieses Ereignis und antworten – das ist ein weiteres Ereignis. Wir erhalten neue Informationen – Ereignis. Wir entscheiden uns, ein Tool zu verwenden (z.B. einen Browser zu öffnen) – Ereignis. Unser internes „Status“ ändert sich ständig basierend auf diesen Ereignissen.
Eine ereignisgesteuerte Architektur (EDA) für KI-Agenten spiegelt dieses natürliche Interaktionsmodell wider. Anstelle eines starren Steuerungsflusses geben die Komponenten Ereignisse aus, wenn etwas Wichtiges passiert. Andere Komponenten (Listener, Handler) reagieren auf diese Ereignisse. Das bringt mehrere wichtige Vorteile mit sich:
- Modularität: Die Komponenten werden lose gekoppelt. Ein Tool-Executor muss nicht wissen, wer ihn aufgerufen hat oder was als Nächstes passieren wird; er gibt einfach ein Ereignis wie „tool_call_succeeded“ oder „tool_call_failed“ aus.
- Flexibilität: Es ist viel einfacher, neue Funktionen hinzuzufügen oder bestehende zu ändern. Möchten Sie ein neues Tool? Fügen Sie einfach einen Handler hinzu, der auf ein spezifisches Intentionsevent hört. Müssen Sie jeden LLM-Aufruf protokollieren? Fügen Sie einen Logger hinzu, der auf „llm_response_received“ hört.
- Resilienz: Wenn eine Komponente ausfällt, ist es weniger wahrscheinlich, dass sie das ganze System zum Absturz bringt. Ein Ereignis kann erneut versucht werden, oder ein alternativer Handler kann es übernehmen. Sie können Warteschlangen für Ereignisse integrieren, die nicht verarbeitet werden können.
- Parallelverarbeitung: Mehrere Ereignisse können parallel verarbeitet werden, entweder von verschiedenen Handlers oder vom gleichen Handler auf unterschiedlichen Instanzen von Ereignissen. Dies ist entscheidend für Agenten, die mehrere laufende Aufgaben verwalten müssen.
- Beobachtbarkeit: Der Ereignisfluss bietet ein klares und nachvollziehbares Protokoll von allem, was der Agent tut. Sie können den Informations- und Entscheidungsfluss leicht zurückverfolgen.
Der Hauptgedanke: Ereignisse, Dispatcher und Handler
Im Herzen einer EDA benötigt man drei Dinge:
- Ereignisse: Einfache Datenstrukturen, die beschreiben, was passiert ist (z.B. `ToolCalled`, `LLMResponseReceived`, `UserQueryReceived`).
- Ein Ereignis-Dispatcher: Ein zentrales Mechanismus, das ein Ereignis aufnimmt und es an alle interessierten Parteien weiterleitet.
- Ereignis-Handler: Funktionen oder Klassen, die auf spezifische Ereignistypen „hören“ und Logik ausführen, wenn sie eines empfangen.
Lassen Sie uns ein vereinfachtes Beispiel betrachten. Stellen Sie sich unseren Schreibpipeline-Agenten vor. Anstelle einer riesigen Funktion haben wir:
- Ein Ereignis `UserQueryReceived` (wenn eine neue E-Mail eintrifft).
- Ein Ereignis `LLMInputGenerated` (wenn wir einen Prompt für das LLM erstellt haben).
- Ein Ereignis `LLMResponseReceived` (wenn das LLM seine Ausgabe zurückgibt).
- Ein Ereignis `ToolCallRequested` (wenn das LLM vorschlägt, ein Tool zu verwenden).
- Ein Ereignis `ToolCallSucceeded` / `ToolCallFailed` (nach einer Interaktion mit einem Tool).
- Ein Ereignis `DraftResponseReady` (wenn ein Entwurf bereit zur Überprüfung ist).
Jedes dieser Ereignisse überträgt relevante Daten – den Inhalt der E-Mail, den Prompt/Antwort des LLM, den Namen des Tools und die Argumente usw.
Grundlagen: Ein pythonischer Ansatz
Sie benötigen keine schwere Nachrichtenwarteschlange wie Kafka für einfache Agentensysteme (obwohl dies für verteilte produktive Agenten erforderlich sein könnte!). Für einen Agenten mit einem einzigen Prozess funktioniert ein einfacher in-memory Ereignis-Dispatcher Wunder.
Schritt 1: Definieren Sie Ihre Ereignisse
Ich verwende gerne `dataclasses` für Ereignisse, da sie klar und eindeutig sind.
from dataclasses import dataclass
from typing import Any, Dict, Optional
@dataclass
class AgentEvent:
"""Basis-Klasse für alle Agentenereignisse."""
timestamp: float # Fügen Sie einen Zeitstempel hinzu für Sortierung und Debugging
metadata: Dict[str, Any] = None
def __post_init__(self):
if self.metadata is None:
self.metadata = {}
@dataclass
class UserQueryReceived(AgentEvent):
query_id: str
content: str
source: str = "email"
@dataclass
class LLMRequestSent(AgentEvent):
query_id: str
model_name: str
prompt: str
@dataclass
class LLMResponseReceived(AgentEvent):
query_id: str
model_name: str
response_text: str
tool_calls: Optional[list[Dict[str, Any]]] = None
@dataclass
class ToolCallRequested(AgentEvent):
query_id: str
tool_name: str
tool_args: Dict[str, Any]
@dataclass
class ToolCallSucceeded(AgentEvent):
query_id: str
tool_name: str
tool_args: Dict[str, Any]
result: Any
@dataclass
class ToolCallFailed(AgentEvent):
query_id: str
tool_name: str
tool_args: Dict[str, Any]
error_message: str
@dataclass
class AgentThoughtEvent(AgentEvent):
query_id: str
thought: str
@dataclass
class FinalResponseReady(AgentEvent):
query_id: str
response_content: str
action_taken: str
Beachten Sie das `query_id`. Das ist entscheidend! Es ermöglicht uns, Ereignisse, die zu derselben Interaktion oder Aufgabe eines Benutzers gehören, zu korrelieren. Ohne dies wird Ihr Ereignisfluss zu einem chaotischen Durcheinander.
Schritt 2: Erstellen Sie einen Ereignis-Dispatcher
Hier werden die Ereignisse geleitet. Ein einfaches Wörterbuch, das Ereignistypen mit Listen von Handlern verknüpft, funktioniert gut.
import time
from collections import defaultdict
from typing import Callable, Type, List, Union
class EventDispatcher:
def __init__(self):
self._handlers: defaultdict[Type[AgentEvent], List[Callable[[AgentEvent], None]]] = defaultdict(list)
def register_handler(self, event_type: Type[AgentEvent], handler: Callable[[AgentEvent], None]):
"""Registrieren Sie eine Funktion zur Handhabung eines bestimmten Ereignistyps."""
self._handlers[event_type].append(handler)
def dispatch(self, event: AgentEvent):
"""Ein Ereignis an alle registrierten Handler senden."""
# Stellen Sie sicher, dass der Zeitstempel gesetzt ist, wenn er noch nicht gesetzt ist
if not hasattr(event, 'timestamp') or event.timestamp is None:
event.timestamp = time.time()
# Dispatcher an die spezifischen Handler des Ereignistyps
for handler in self._handlers[type(event)]:
try:
handler(event)
except Exception as e:
print(f"Fehler im Handler {handler.__name__} für das Ereignis {type(event).__name__}: {e}")
# Gegebenenfalls hier ein Fehlerereignis dispatchen für die Robustheit
# Dispatcher auch an die registrierten Handler für den Basis-AgentEvent-Typ
# Dadurch können Sie generisch registrieren oder überwachen
for handler in self._handlers[AgentEvent]:
try:
handler(event)
except Exception as e:
print(f"Fehler im generischen Handler {handler.__name__} für das Ereignis {type(event).__name__}: {e}")
Schritt 3: Definieren Sie Ihre Handler
Jeder Handler ist eine einfache Funktion, die ein Ereignisobjekt entgegennimmt. Er führt seine spezifische Aufgabe aus und kann entscheidend neue Ereignisse dispatchen.
Skizzieren wir einige Handler für unseren Schreibagenten:
# Angenommen, 'dispatcher' ist eine Instanz von EventDispatcher
# --- Handler für die ursprüngliche Benutzeranfrage ---
def handle_user_query(event: UserQueryReceived):
print(f"[{event.query_id}] Benutzeranfrage erhalten: {event.content[:50]}...")
# Hier würden wir typischerweise ein LLM verwenden, um die ursprüngliche Absicht zu bestimmen
# Zur Vereinfachung nehmen wir an, dass sie immer über das LLM zur Erstellung weitergeleitet wird
prompt = f"Sie sind ein hilfreicher Assistent für einen freiberuflichen Schriftsteller. Verfassen Sie eine höfliche erste Antwort auf die folgende Kundenanfrage und schlagen Sie eine Aktion vor (z.B. 'schedule_call', 'search_knowledge_base'):\n\n{event.content}\n\nAusgabe im JSON-Format mit den Feldern 'draft_response' und 'suggested_action'."
# Ein Ereignis auslösen, das an das LLM gesendet werden soll
dispatcher.dispatch(LLMRequestSent(
query_id=event.query_id,
model_name="gpt-4",
prompt=prompt,
metadata={"previous_event": type(event).__name__}
))
# --- Handler für LLM-Antworten ---
def handle_llm_response(event: LLMResponseReceived):
print(f"[{event.query_id}] LLM-Antwort erhalten: {event.response_text[:50]}...")
# LLM-Antwort analysieren (dies wäre robuster mit Pydantic)
try:
llm_output = json.loads(event.response_text)
draft = llm_output.get("draft_response")
action = llm_output.get("suggested_action")
dispatcher.dispatch(AgentThoughtEvent(
query_id=event.query_id,
thought=f"Vom LLM vorgeschlagene Aktion: {action}"
))
if draft:
dispatcher.dispatch(DraftResponseReady(
query_id=event.query_id,
response_content=draft,
action_taken="drafted_initial_response"
))
if action == "schedule_call":
# Angenommen, das LLM hat auch Anrufdetails bereitgestellt, falls nötig
dispatcher.dispatch(ToolCallRequested(
query_id=event.query_id,
tool_name="calendar_scheduler",
tool_args={"client_email": "[email protected]", "duration": "30min"} # Platzhalter
))
elif action == "search_knowledge_base":
# Angenommen, das LLM hat eine Suchanfrage bereitgestellt
dispatcher.dispatch(ToolCallRequested(
query_id=event.query_id,
tool_name="knowledge_base_search",
tool_args={"query": "artikel über KI-Agenten"} # Platzhalter
))
except json.JSONDecodeError:
print(f"[{event.query_id}] LLM-Antwort ist kein gültiges JSON. Zur Überprüfung an den Menschen senden.")
dispatcher.dispatch(FinalResponseReady(
query_id=event.query_id,
response_content="Die Analyse der LLM-Ausgabe ist fehlgeschlagen, ein Mensch benötigt Hilfe. Ursprüngliche LLM-Antwort: " + event.response_text,
action_taken="human_review_needed"
))
# --- Handler für Werkzeugaufrufe ---
def handle_tool_call_request(event: ToolCallRequested):
print(f"[{event.query_id}] Werkzeugaufruf angefordert: {event.tool_name} mit args {event.tool_args}")
# Simulation der Ausführung des Werkzeugs
if event.tool_name == "calendar_scheduler":
# In einem echten System würde dies eine echte API aufrufen
print(f"Planung eines Anrufs für {event.tool_args.get('client_email')}...")
time.sleep(1) # Netzwerkverzögerung simulieren
if random.random() > 0.1: # Erfolgsquote von 90 %
dispatcher.dispatch(ToolCallSucceeded(
query_id=event.query_id,
tool_name=event.tool_name,
tool_args=event.tool_args,
result={"status": "scheduled", "meeting_link": "https://meet.google.com/abc-xyz"}
))
else:
dispatcher.dispatch(ToolCallFailed(
query_id=event.query_id,
tool_name=event.tool_name,
tool_args=event.tool_args,
error_message="API-Fehler im Kalender oder beschäftigt"
))
# ... weitere Werkzeuge ...
# --- Generischer Logging-Handler ---
def log_all_events(event: AgentEvent):
print(f"LOG: {type(event).__name__} - {event.query_id} - {event.timestamp}")
# --- Registrierung der Handler ---
dispatcher = EventDispatcher()
dispatcher.register_handler(UserQueryReceived, handle_user_query)
dispatcher.register_handler(LLMResponseReceived, handle_llm_response)
dispatcher.register_handler(ToolCallRequested, handle_tool_call_request)
# ... weitere Handler für ToolCallSucceeded, ToolCallFailed usw.
dispatcher.register_handler(AgentEvent, log_all_events) # Generischer Handler für alle Ereignisse
Das ist ein sehr vereinfachtes Beispiel, aber Sie können sehen, wie jedes Element unabhängig ist. Der `handle_user_query` weiß nicht, *wie* die LLM-Anfrage gesendet wird, sondern nur, dass er ein Ereignis `LLMRequestSent` auslösen muss. Ebenso kümmert sich `handle_llm_response` nicht darum, wer den ursprünglichen Prompt gesendet hat; es verarbeitet einfach die Antwort und entscheidet, was als Nächstes zu tun ist.
Simulation von LLM- und Werkzeugaufrufen
Für ein echtes System würde `LLMRequestSent` eine Komponente auslösen, die tatsächlich die LLM-API aufruft und dann `LLMResponseReceived` ausgibt, wenn das Ergebnis zurückkommt. Hier können `asyncio` oder ein einfacher Thread-Pool nützlich sein für gleichzeitige LLM-Abrufe oder die Ausführung von Werkzeugen, ohne die Ereignisschleife zu blockieren.
import asyncio
import json
import random
import time
# ... (Definitionen der Ereignisse und EventDispatcher oben) ...
# Fiktive LLM-API
async def mock_llm_call(prompt: str) -> str:
print(f" [Mock LLM] Verarbeiten des Prompts: {prompt[:80]}...")
await asyncio.sleep(random.uniform(1.0, 3.0)) # LLM-Latenz simulieren
# Sehr grundlegende fiktive Logik für unseren Anwendungsfall
if "schedule_call" in prompt:
return json.dumps({
"draft_response": "Danke für Ihre Anfrage! Ich würde mich freuen, weiter zu diskutieren. Was halten Sie von einem kurzen Anruf nächste Woche?",
"suggested_action": "schedule_call"
})
elif "search_knowledge_base" in prompt:
return json.dumps({
"draft_response": "Exzellente Frage! Ich habe eine Antwort verfasst und auch relevante Artikel konsultiert.",
"suggested_action": "search_knowledge_base"
})
else:
return json.dumps({
"draft_response": "Danke, dass Sie uns kontaktiert haben! Ich habe Ihre Anfrage geprüft und eine erste Antwort verfasst.",
"suggested_action": "none"
})
# Agent LLM-Komponente (lauscht auf LLMRequestSent, gibt LLMResponseReceived aus)
async def llm_agent_component(event: LLMRequestSent, dispatcher: EventDispatcher):
response_text = await mock_llm_call(event.prompt)
# In einem echten System würden Sie nach Werkzeugaufrufen in der LLM-Antwort suchen
tool_calls = [] # Platzhalter
dispatcher.dispatch(LLMResponseReceived(
query_id=event.query_id,
model_name=event.model_name,
response_text=response_text,
tool_calls=tool_calls,
metadata={"original_prompt_event": event.timestamp}
))
# Asynchronen Handler registrieren
dispatcher.register_handler(LLMRequestSent, lambda e: asyncio.create_task(llm_agent_component(e, dispatcher)))
# ... (weitere Handler oben) ...
# Um ein Beispiel auszuführen:
async def main():
query_id = "user_email_123"
dispatcher.dispatch(UserQueryReceived(
query_id=query_id,
content="Ich benötige einen Artikel über ereignisgesteuerte KI-Agenten und einen Folgetermin.",
timestamp=time.time()
))
# Zeit geben, damit die Ereignisse verarbeitet werden
await asyncio.sleep(10)
print("\n--- Verarbeitung abgeschlossen für user_email_123 ---\n")
query_id_2 = "user_email_456"
dispatcher.dispatch(UserQueryReceived(
query_id=query_id_2,
content="Könnten Sie meine alten Artikel über Deep Learning-Architekturen zusammenfassen?",
timestamp=time.time()
))
await asyncio.sleep(10)
print("\n--- Verarbeitung abgeschlossen für user_email_456 ---\n")
if __name__ == "__main__":
asyncio.run(main())
Ich habe `asyncio.create_task` eingeführt, damit sich die `llm_agent_component` parallel zu anderen Handlers oder nachfolgenden Dispatches ausführen lassen kann. Hier glänzen ereignisorientierte Architekturen wirklich in Bezug auf die Leistung und Reaktivität der KI-Agenten.
Konkrete Lektionen für Ihr nächstes Agentenprojekt
- Beginnen Sie einfach, denken Sie an Ereignisse: Selbst für einen kleinen Agenten skizzieren Sie die wichtigsten Ereignisse, die eintreten. Was löst was aus? Welche Informationen müssen übertragen werden?
- Definieren Sie klare Ereignismuster: Verwenden Sie `dataclasses` oder Pydantic-Modelle für Ihre Ereignisse. Das gewährleistet Konsistenz und erleichtert das Debugging. Fügen Sie immer eine `query_id` oder eine `correlation_id` hinzu.
- Trennen Sie die Anliegen: Jeder Handler sollte eine Sache gut machen. Versuchen Sie nicht, zu viel Logik in einen einzigen Handler zu quetschen. Wenn ein Handler einen externen Aufruf tätigen muss, sollte er ein Anfrageereignis auslösen und auf ein entsprechendes Antwortereignis warten.
- Übernehmen Sie die Asynchronität: Die Interaktionen von KI-Agenten (LLM-Aufrufe, Ausführung von Werkzeugen) sind intrinsisch asynchron. Verwenden Sie `asyncio` oder ein ähnliches Framework, um diese Operationen parallel zu verwalten, ohne Ihre Ereignisschleife zu blockieren.
- Integrieren Sie das Beobachtbare: Ein generischer Ereignislogger (wie mein `log_all_events`) ist unglaublich wertvoll. Sie können diese Ereignisse einfach an ein Überwachungssystem weiterleiten oder sie einfach für die Entwicklung drucken. Dieser Ereignisfluss wird zum Protokoll des „denkenden Prozesses“ Ihres Agenten.
- Fehlerbehandlung mit Ereignissen: Anstelle eines tief verschachtelten `try/except` sollten Sie `ErrorEvent` oder `ToolCallFailed`-Ereignisse auslösen. Andere Handler können dann speziell darauf hören, um eine Retry-Logik, Fallback-Lösungen oder Anfragen nach menschlichem Eingreifen umzusetzen.
Der Wechsel zu einem ereignisgesteuerten Modell hat meine Denkweise über den Bau von Agenten vollständig verändert. Es hat mich davon abgehalten, jeden möglichen Weg in einem linearen Fluss vorherzusehen, und dazu geführt, dass ich ein System aufgebaut habe, das intelligent auf seine Umgebung und seine eigenen internen Vorgänge reagiert. Es ist eine widerstandsfähigere, skalierbare und ehrlich gesagt angenehmere Art, komplexe KI-Agenten zu bauen.
Probieren Sie es für Ihr nächstes Agentenprojekt aus. Sie könnten schneller als gedacht diesen Spaghetti-Code entwirren!
🕒 Published: