Mein Ausblick für 2026: Vereinfachung des AI-Agenten-Klebcodes

1. Ereignisse: Einfache Datenstrukturen, die beschreiben, was passiert ist (z.B. `ToolCalled`, `LLMResponseReceived`, `UserQueryReceived`).
2. Ein Ereignis-Dispatcher: Ein zentrales Element, das ein Ereignis entgegennimmt und es an alle interessierten Parteien weiterleitet.
3. Ereignis-Handler: Funktionen oder Klassen, die auf spezifische Ereignisarten „lauschen“ und etwas Logik ausführen, wenn sie eines erhalten.

Schauen wir uns ein vereinfachtes Beispiel an. Stellen Sie sich unseren Schreib-Pipeline-Agenten vor. Anstatt einer riesigen Funktion haben wir:

• Ein `UserQueryReceived`-Ereignis (wenn eine neue E-Mail eintrifft).
• Ein `LLMInputGenerated`-Ereignis (wenn wir einen Prompt für das LLM erstellt haben).
• Ein `LLMResponseReceived`-Ereignis (wenn das LLM seine Ausgabe zurücksendet).
• Ein `ToolCallRequested`-Ereignis (wenn das LLM vorschlägt, ein Tool zu verwenden).
• Ein `ToolCallSucceeded` / `ToolCallFailed`-Ereignis (nach einer Tool-Interaktion).
• Ein `DraftResponseReady`-Ereignis (wenn ein Entwurf zur Überprüfung bereit ist).

Jedes dieser Ereignisse trägt relevante Daten – den Inhalt der E-Mail, den LLM-Prompt/Antwort, den Toolnamen und die Argumente usw.

Bausteine: Ein Pythonischer Ansatz

Für einfache Agentensysteme benötigen Sie keine leistungsstarke Nachrichtenwarteschlange wie Kafka (obwohl Sie dies für Produktionsumgebungen mit verteilten Agenten definitiv könnten!). Für einen Einzelprozess-Agenten funktioniert ein einfacher speicherinterner Ereignis-Dispatcher Wunder.

Schritt 1: Definieren Sie Ihre Ereignisse

Ich verwende gerne `dataclasses` für Ereignisse, weil sie sauber und explizit 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 für die Reihenfolge und das Debuggen hinzu
 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 die `query_id`. Dies ist entscheidend! Sie ermöglicht es uns, Ereignisse, die zur gleichen Nutzung oder Aufgabe gehören, miteinander zu verknüpfen. Ohne sie wird Ihr Ereignisstream ein chaotisches Durcheinander.

Schritt 2: Erstellen Sie einen Ereignis-Dispatcher

Hier werden Ereignisse weitergeleitet. Ein einfaches Dictionary, das Ereignistypen auf Listen von Handlern abbildet, 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]):
 """Registriert eine Funktion, die einen bestimmten Ereignistyp bearbeitet."""
 self._handlers[event_type].append(handler)

 def dispatch(self, event: AgentEvent):
 """Sendet ein Ereignis an alle registrierten Handler."""
 # Sicherstellen, dass der Zeitstempel gesetzt ist, falls nicht bereits
 if not hasattr(event, 'timestamp') or event.timestamp is None:
 event.timestamp = time.time()
 
 # An Handler spezifisch für den Ereignistyp weiterleiten
 for handler in self._handlers[type(event)]:
 try:
 handler(event)
 except Exception as e:
 print(f"Fehler im Handler {handler.__name__} für Ereignis {type(event).__name__}: {e}")
 # Eventuell hier ein Fehlerereignis zur Robustheit weiterleiten
 
 # Auch an Handler weiterleiten, die für den Basis-AgentEvent-Typ registriert sind
 # Das ermöglicht generelles Logging oder Monitoring
 for handler in self._handlers[AgentEvent]:
 try:
 handler(event)
 except Exception as e:
 print(f"Fehler im allgemeinen Handler {handler.__name__} für 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 weiterleiten.

Lassen Sie uns einige Handler für unseren Schreib-Agenten skizzieren:

# Angenommen, 'dispatcher' ist eine Instanz von EventDispatcher

# --- Handler für die anfä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 anfängliche Absicht zu bestimmen
 # Zur Vereinfachung nehmen wir an, es geht immer an das LLM zum Entwurf
 prompt = f"Sie sind ein hilfreicher Assistent für einen freiberuflichen Schriftsteller. Entwerfen Sie eine anfängliche, höfliche Antwort auf die folgende Kundenanfrage und schlagen Sie eine Folgemaßnahme vor (z.B. 'schedule_call', 'search_knowledge_base'):\n\n{event.content}\n\nAusgabe im JSON-Format mit 'draft_response' und 'suggested_action'-Feldern."
 
 # Dispatchen eines Ereignisses, um es an das LLM zu senden
 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 parsen (dies wäre mit Pydantic solider)
 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"LLM vorgeschlagene Maßnahme: {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 erforderlich
 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 die Suchanfrage bereitgestellt
 dispatcher.dispatch(ToolCallRequested(
 query_id=event.query_id,
 tool_name="knowledge_base_search",
 tool_args={"query": "verwandte Artikel über KI-Agenten"} # Platzhalter
 ))

 except json.JSONDecodeError:
 print(f"[{event.query_id}] LLM-Antwort ist kein gültiges JSON. Wird zur menschlichen Überprüfung gesendet.")
 dispatcher.dispatch(FinalResponseReady(
 query_id=event.query_id,
 response_content="Parsing der LLM-Ausgabe fehlgeschlagen, benötigt menschliche Überprüfung. 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}")
 
 # Simulierung der Werkzeugausführung
 if event.tool_name == "calendar_scheduler":
 # In einem realen System würde dies eine tatsächliche API aufrufen
 print(f"Anruf für {event.tool_args.get('client_email')} planen...")
 time.sleep(1) # Netzwerkverzögerung simulieren
 if random.random() > 0.1: # 90% Erfolgsquote
 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="Kalender-API-Fehler oder besetzt"
 ))
 # ... andere Werkzeuge ...

# --- Generischer Logger-Handler ---
def log_all_events(event: AgentEvent):
 print(f"LOG: {type(event).__name__} - {event.query_id} - {event.timestamp}")

# --- Registrieren von Handlern ---
dispatcher = EventDispatcher()
dispatcher.register_handler(UserQueryReceived, handle_user_query)
dispatcher.register_handler(LLMResponseReceived, handle_llm_response)
dispatcher.register_handler(ToolCallRequested, handle_tool_call_request)
# ... andere Handler für ToolCallSucceeded, ToolCallFailed, usw.
dispatcher.register_handler(AgentEvent, log_all_events) # Generischer Handler für alle Ereignisse

Dies 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 es ein `LLMRequestSent`-Ereignis auslösen muss. Ähnlich kümmert sich der `handle_llm_response` nicht darum, wer den ursprünglichen Prompt gesendet hat; er verarbeitet einfach die Antwort und entscheidet, was als Nächstes zu tun ist.

Simulieren 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 wird `LLMResponseReceived` dispatcht, wenn das Ergebnis zurückkommt. Hier kommt `asyncio` oder ein einfacher Thread-Pool ins Spiel, um konkurrierende LLM-Aufrufe oder Werkzeugausführungen durchzuführen, ohne die Ereignisschleife zu blockieren.

import asyncio
import json
import random
import time

# ... (Ereignisdefinitionen und EventDispatcher von oben) ...

# Mock LLM API
async def mock_llm_call(prompt: str) -> str:
 print(f" [Mock LLM] Verarbeite Prompt: {prompt[:80]}...")
 await asyncio.sleep(random.uniform(1.0, 3.0)) # Simuliere LLM-Latenz
 
 # Sehr grundlegende Mock-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, mehr zu plaudern. Wie wäre es, wenn wir nächste Woche einen kurzen Anruf planen?",
 "suggested_action": "schedule_call"
 })
 elif "search_knowledge_base" in prompt:
 return json.dumps({
 "draft_response": "Tolle Frage! Ich habe eine Antwort entworfen und auch einige relevante Artikel nachgeschlagen.",
 "suggested_action": "search_knowledge_base"
 })
 else:
 return json.dumps({
 "draft_response": "Danke, dass Sie sich gemeldet haben! Ich habe Ihre Anfrage überprüft und eine anfängliche Antwort entworfen.",
 "suggested_action": "none"
 })

# LLM-Agentenkomponente (lauscht auf LLMRequestSent, dispatcht LLMResponseReceived)
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 aus der LLM-Antwort parsen
 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}
 ))

# Registrieren des asynchronen Handlers
dispatcher.register_handler(LLMRequestSent, lambda e: asyncio.create_task(llm_agent_component(e, dispatcher)))

# ... (andere Handler von 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()
 ))
 
 # Geben Sie etwas Zeit für die Verarbeitung der Ereignisse
 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 vergangenen Artikel über tiefes Lernen 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, um der `llm_agent_component` zu ermöglichen, gleichzeitig mit anderen Handlern oder nachfolgenden Dispatches zu arbeiten. Hier glänzen ereignisgesteuerte Architekturen wirklich hinsichtlich Leistung und Reaktionsfähigkeit in KI-Agenten.

Handlungsfähige Erkenntnisse für Ihr nächstes Agentenprojekt

1. Beginnen Sie einfach, denken Sie in Ereignissen: Selbst für einen kleinen Agenten skizzieren Sie die wichtigsten Ereignisse, die passieren. Was löst was aus? Welche Informationen müssen weitergegeben werden?
2. Definieren Sie klare Ereignisschemas: Verwenden Sie `dataclasses` oder Pydantic-Modelle für Ihre Ereignisse. Dies gewährleistet Konsistenz und erleichtert das Debugging. Immer eine `query_id` oder `correlation_id` einfügen.
3. Trennen Sie die Anliegen: Jeder Handler sollte eine Aufgabe gut ausführen. Versuchen Sie nicht, zu viel Logik in einen einzelnen Handler zu quetschen. Wenn ein Handler einen externen Aufruf benötigt, sollte er ein Anfrageereignis dispatchen und auf ein entsprechendes Antwortereignis warten.
4. Umarmen Sie die Asynchronität: Interaktionen von KI-Agenten (LLM-Aufrufe, Werkzeugausführung) sind von Natur aus asynchron. Verwenden Sie `asyncio` oder ein ähnliches Framework, um diese gleichzeitig zu bearbeiten, ohne Ihre Ereignisschleife zu blockieren.
5. Integrieren Sie Beobachtbarkeit: Ein generischer Ereignislogger (wie mein `log_all_events`) ist unglaublich wertvoll. Sie können diese Ereignisse leicht an ein Überwachungssystem weiterleiten oder einfach für die Entwicklung ausgeben. Dieser Ereignisstrom wird zum internen „Denkprozess“-Protokoll Ihres Agenten.
6. Fehlerbehandlung mit Ereignissen: Anstelle von tief verschachtelten `try/except` sollten Sie `ErrorEvent` oder `ToolCallFailed`-Ereignisse dispatchen. Andere Handler können dann spezifisch darauf hören, um Wiederholungslogik, Fallbacks oder Anfragen für menschliches Eingreifen umzusetzen.

Der Übergang zu einem ereignisgesteuerten Modell hat meine Denkweise über den Bau von Agenten völlig verändert. Es hat mich davon abgehalten, jeden möglichen Pfad in einem linearen Fluss vorherzusehen, und hin zu einem System bewegt, das intelligent auf seine Umgebung und seine eigenen internen Abläufe reagiert. Es ist eine widerstandsfähigere, skalierbare und ehrlicherweise angenehmere Art, komplexe KI-Agenten zu bauen.

Probieren Sie es für Ihr nächstes Agentenprojekt aus. Sie könnten feststellen, dass Sie diesen Spaghetti-Code schneller entwirren, als Sie denken!

🕒 Published: March 28, 2026