Ciao a tutti, Alex qui da agntai.net! Siamo a marzo 2026 e ho passato troppo tempo recentemente a riflettere su come costruiamo agenti IA. In particolare, ho difficoltà con il “codice di collegamento” – gli elementi che collegano tutte le sofisticate uscite LLM, le chiamate agli strumenti e la gestione dello stato. Abbiamo tutti visto le dimostrazioni impressionanti, vero? Agenti che fanno cose incredibili. Ma poi, provi a costruirne uno per un problema del mondo reale e ti scontri con un muro di promemoria, logica condizionale e aggiornamenti di stato. Sembra meno la costruzione di un sistema intelligente e più la gestione di una fabbrica molto complessa di spaghetti.
Quindi oggi voglio parlare di qualcosa che ha guadagnato popolarità discretamente e, a dire il vero, preserva la mia salute mentale: le Architetture Orientate agli Eventi per gli Agenti IA. Non è un concetto nuovo nell’ingegneria del software, tutt’altro, ma applicarlo in modo riflessivo agli agenti IA, in particolare a quelli che orchestrano più interazioni LLM e strumenti esterni, sembra essere una boccata d’aria fresca. Dimentichiamo per un momento il pensiero lineare e passo dopo passo. Pensiamo ai sistemi reattivi.
La Mia Lotta Personale con i Monoliti degli Agenti
Qualche mese fa, stavo lavorando su un agente progettato per aiutarmi a gestire il mio pipeline di scrittura freelance. L’idea era semplice: avrebbe monitorato la mia casella di posta per nuove richieste, redatto risposte iniziali, suggerito articoli passati pertinenti dalla mia base di conoscenze e persino aiutato a pianificare chiamate di follow-up. Sembrava abbastanza semplice.
Il mio approccio iniziale era abbastanza tipico: un ciclo principale. Ricevere un’email. Analizzare l’email. Determinare l’azione (scrivere, pianificare, ricercare). Chiamare LLM. Elaborare l’uscita LLM. Chiamare uno strumento (API calendario, API email, API base di conoscenze). Aggiornare lo stato interno. Ripetere.
È iniziato bene, ma man mano che aggiungevo più “intelligenza” e più strumenti, stava diventando un incubo. Cosa succedeva se la chiamata all’API del calendario falliva? Cosa succedeva se il LLM allucinava un contatto che non esisteva? Cosa succedeva se dovevo fare una pausa e chiedere un parere umano per una decisione critica? Il mio script dell’agente unico e monolitico si era rapidamente trasformato in un labirinto annidato di `if/else` con blocchi `try/except` ovunque. Il debug era un incubo. Modificare una parte spesso rompeva un’altra. Sembrava che dovessi costantemente tappare le perdite in una nave che affondava.
Ricordo una notte tarda, cercando di capire perché il mio agente continuava a redigere risposte per email che aveva già trattato. Si è scoperto che l’aggiornamento dello stato per “email trattata” avveniva *dopo* una possibile nuova esecuzione di LLM in un percorso di errore. Era una condizione di concorrenza classica in un sistema non progettato per gestire le operazioni asincrone e non deterministiche in modo appropriato. È stato allora che ho iniziato a cercare un modo migliore di fare.
Perché gli Agenti Orientati agli Eventi Hanno Senso
Pensa a come funzionano gli esseri umani. Di solito non seguiamo un copione rigoroso e predeterminato per ogni interazione. Reagiamo alle cose. Qualcuno fa una domanda – è un evento. Noi lo trattiamo e rispondiamo – è un altro evento. Riceviamo una nuova informazione – evento. Decidiamo di usare uno strumento (come aprire un browser) – evento. Il nostro “stato” interno cambia costantemente in base a questi eventi.
Un’architettura orientata agli eventi (AOE) per gli agenti IA riflette questo modello di interazione naturale. Invece di un flusso di controllo rigido, i componenti emettono eventi quando succede qualcosa di significativo. Altri componenti (ascoltatori, gestori) reagiscono a questi eventi. Ciò porta a diversi vantaggi chiave:
- Modularità: I componenti diventano debolmente accoppiati. Un esecutore di strumenti non ha bisogno di sapere chi lo ha chiamato o cosa succederà dopo; emette semplicemente un evento come “tool_call_succeeded” o “tool_call_failed.”
- Flessibilità: È molto più facile aggiungere nuove funzionalità o modificare quelle esistenti. Vuoi un nuovo strumento? Aggiungi semplicemente un gestore che ascolta un evento di intenzione specifica. Hai bisogno di registrare ogni chiamata LLM? Aggiungi un registratore che ascolta “llm_response_received.”
- Resilienza: Se un componente fallisce, è meno probabile che faccia crollare l’intero sistema. Un evento può essere riprovato, o un gestore alternativo può gestirlo. Puoi integrare code per gli eventi che non possono essere trattati.
- Concorrenza: Più eventi possono essere trattati in parallelo, sia da diversi gestori, sia dallo stesso gestore su diverse istanze di eventi. Questo è cruciale per gli agenti che devono gestire più compiti contemporaneamente.
- Osservabilità: Il flusso di eventi fornisce un registro chiaro e auditabile di tutto ciò che fa l’agente. Puoi facilmente risalire al flusso di informazioni e decisioni.
L’Idea Principale: Eventi, Dispatcher e Gestori
Al centro di una AOE ci sono tre cose:
- Eventi: Strutture dati semplici che descrivono qualcosa che è accaduto (ad esempio, `ToolCalled`, `LLMResponseReceived`, `UserQueryReceived`).
- Un Dispatcher di Eventi: Un meccanismo centrale che prende un evento e lo indirizza a tutte le parti interessate.
- Gestori di Eventi: Funzioni o classi che “ascoltano” per particolari tipi di eventi e eseguono una logica quando ne ricevono uno.
Guardiamo un esempio semplificato. Immagina il nostro agente del pipeline di scrittura. Invece di una grande funzione, abbiamo:
- Un evento `UserQueryReceived` (quando arriva una nuova email).
- Un evento `LLMInputGenerated` (quando abbiamo elaborato un prompt per il LLM).
- Un evento `LLMResponseReceived` (quando il LLM restituisce la sua uscita).
- Un evento `ToolCallRequested` (quando il LLM suggerisce di usare uno strumento).
- Un evento `ToolCallSucceeded` / `ToolCallFailed` (dopo un’interazione con uno strumento).
- Un evento `DraftResponseReady` (quando una bozza è pronta per essere esaminata).
Ognuno di questi eventi trasmette dati pertinenti – il contenuto dell’email, il prompt/riposta LLM, il nome dello strumento e gli argomenti, ecc.
Elementi di Base: Un Approccio Pythonico
Non hai bisogno di una coda di messaggi pesante come Kafka per semplici sistemi di agenti (anche se per agenti distribuiti in produzione, potrebbe essere necessario!). Per un agente a processo unico, un semplice dispatcher di eventi in memoria funziona meravigliosamente.
Passo 1: Definisci i Tuoi Eventi
Mi piace usare le `dataclasses` per gli eventi poiché sono chiare ed esplicite.
from dataclasses import dataclass
from typing import Any, Dict, Optional
@dataclass
class AgentEvent:
"""Classe base per tutti gli eventi dell'agente."""
timestamp: float # Aggiungi un timestamp per l'ordinamento e il debug
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
Nota il `query_id`. Questo è fondamentale! Ci permette di correlare eventi appartenenti alla stessa interazione o compito utente globale. Senza di esso, il tuo flusso di eventi diventa un disastro caotico.
Passo 2: Crea un Dispatcher di Eventi
È qui che gli eventi vengono indirizzati. Un semplice dizionario che associa tipi di eventi a liste di gestori funziona bene.
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]):
"""Registrare una funzione per gestire un tipo di evento specifico."""
self._handlers[event_type].append(handler)
def dispatch(self, event: AgentEvent):
"""Inviare un evento a tutti i gestori registrati."""
# Assicurati che il timestamp sia impostato se non è già presente
if not hasattr(event, 'timestamp') or event.timestamp is None:
event.timestamp = time.time()
# Dispatcher ai gestori specifici per il tipo di evento
for handler in self._handlers[type(event)]:
try:
handler(event)
except Exception as e:
print(f"Errore nel gestore {handler.__name__} per l'evento {type(event).__name__}: {e}")
# Eventualmente dispatchare un evento di errore qui per la solidità
# Dispatcher anche ai gestori registrati per il tipo di AgentEvent di base
# Questo permette di registrare o monitorare in modo generico
for handler in self._handlers[AgentEvent]:
try:
handler(event)
except Exception as e:
print(f"Errore nel gestore generico {handler.__name__} per l'evento {type(event).__name__}: {e}")
Passo 3: Definisci i Tuoi Gestori
Ciascun gestore è una semplice funzione che prende un oggetto evento. Svolge il suo compito specifico e, ciò che è cruciale, può dispatchare nuovi eventi.
Schizziamo alcuni gestori per il nostro agente di scrittura:
# Supponiamo che 'dispatcher' sia un'istanza di EventDispatcher
# --- Gestore per la richiesta iniziale dell'utente ---
def handle_user_query(event: UserQueryReceived):
print(f"[{event.query_id}] Richiesta utente ricevuta: {event.content[:50]}...")
# Qui, di solito utilizzeremmo un LLM per decidere l'intento iniziale
# Per semplificare, supponiamo che passi sempre tramite il LLM per la scrittura
prompt = f"Sei un assistente utile per uno scrittore freelance. Scrivi una risposta iniziale, educata alla seguente richiesta del cliente, e suggerisci un'azione da intraprendere (ad esempio, 'schedule_call', 'search_knowledge_base'):\n\n{event.content}\n\nUscita in JSON con i campi 'draft_response' e 'suggested_action'."
# Emettere un evento da inviare al LLM
dispatcher.dispatch(LLMRequestSent(
query_id=event.query_id,
model_name="gpt-4",
prompt=prompt,
metadata={"previous_event": type(event).__name__}
))
# --- Gestore per le risposte LLM ---
def handle_llm_response(event: LLMResponseReceived):
print(f"[{event.query_id}] Risposta LLM ricevuta: {event.response_text[:50]}...")
# Analizzare la risposta LLM (questo sarebbe più robusto con 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"Azione suggerita dal LLM: {action}"
))
if draft:
dispatcher.dispatch(DraftResponseReady(
query_id=event.query_id,
response_content=draft,
action_taken="drafted_initial_response"
))
if action == "schedule_call":
# Supponiamo che il LLM abbia anche fornito dettagli sull'appuntamento se necessario
dispatcher.dispatch(ToolCallRequested(
query_id=event.query_id,
tool_name="calendar_scheduler",
tool_args={"client_email": "[email protected]", "duration": "30min"} # Placeholder
))
elif action == "search_knowledge_base":
# Supponiamo che il LLM abbia fornito una query di ricerca
dispatcher.dispatch(ToolCallRequested(
query_id=event.query_id,
tool_name="knowledge_base_search",
tool_args={"query": "articoli correlati sugli agenti IA"} # Placeholder
))
except json.JSONDecodeError:
print(f"[{event.query_id}] Risposta LLM non è un JSON valido. Inviato per revisione umana.")
dispatcher.dispatch(FinalResponseReady(
query_id=event.query_id,
response_content="L'analisi dell'output LLM è fallita, bisogno di un umano. Risposta LLM originale: " + event.response_text,
action_taken="human_review_needed"
))
# --- Gestore per le richieste di strumenti ---
def handle_tool_call_request(event: ToolCallRequested):
print(f"[{event.query_id}] Richiesta di chiamata al tool: {event.tool_name} con args {event.tool_args}")
# Simulare l'esecuzione dello strumento
if event.tool_name == "calendar_scheduler":
# In un sistema reale, questo chiamerebbe una vera API
print(f"Programmazione di una chiamata per {event.tool_args.get('client_email')}...")
time.sleep(1) # Simula un ritardo di rete
if random.random() > 0.1: # Tasso di successo del 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="Errore API del calendario o occupato"
))
# ... altri strumenti ...
# --- Gestore generico di registrazione ---
def log_all_events(event: AgentEvent):
print(f"LOG: {type(event).__name__} - {event.query_id} - {event.timestamp}")
# --- Registrazione dei gestori ---
dispatcher = EventDispatcher()
dispatcher.register_handler(UserQueryReceived, handle_user_query)
dispatcher.register_handler(LLMResponseReceived, handle_llm_response)
dispatcher.register_handler(ToolCallRequested, handle_tool_call_request)
# ... altri gestori per ToolCallSucceeded, ToolCallFailed, ecc.
dispatcher.register_handler(AgentEvent, log_all_events) # Gestore generico per tutti gli eventi
Questo è un esempio molto semplificato, ma puoi vedere come ogni elemento sia indipendente. Il `handle_user_query` non sa *come* la richiesta LLM sarà inviata, solo che deve emettere un evento `LLMRequestSent`. Allo stesso modo, `handle_llm_response` non si preoccupa di chi ha inviato il prompt originale; si occupa semplicemente della risposta e decide cosa fare dopo.
Simulazione delle chiamate LLM e degli strumenti
Per un sistema reale, `LLMRequestSent` attiverebbe un componente che chiama effettivamente l’API LLM, quindi emette `LLMResponseReceived` quando il risultato torna. È qui che `asyncio` o un semplice pool di thread possono essere utili per le chiamate LLM concorrenti o per l’esecuzione di strumenti senza bloccare il ciclo di eventi.
import asyncio
import json
import random
import time
# ... (Definizioni degli eventi e EventDispatcher sopra) ...
# API LLM fittizia
async def mock_llm_call(prompt: str) -> str:
print(f" [Mock LLM] Elaborazione del prompt: {prompt[:80]}...")
await asyncio.sleep(random.uniform(1.0, 3.0)) # Simulare la latenza LLM
# Logica fittizia molto basilare per il nostro caso d'uso
if "schedule_call" in prompt:
return json.dumps({
"draft_response": "Grazie per la tua richiesta! Mi piacerebbe discuterne ulteriormente. Che ne dici di programmare una veloce chiamata la prossima settimana?",
"suggested_action": "schedule_call"
})
elif "search_knowledge_base" in prompt:
return json.dumps({
"draft_response": "Ottima domanda! Ho redatto una risposta e ho anche consultato articoli pertinenti.",
"suggested_action": "search_knowledge_base"
})
else:
return json.dumps({
"draft_response": "Grazie per averci contattato! Ho esaminato la tua richiesta e redatto una risposta iniziale.",
"suggested_action": "none"
})
# Componente Agent LLM (ascolta le LLMRequestSent, emette LLMResponseReceived)
async def llm_agent_component(event: LLMRequestSent, dispatcher: EventDispatcher):
response_text = await mock_llm_call(event.prompt)
# In un sistema reale, analizzeresti per chiamate di strumenti nella risposta LLM
tool_calls = [] # Placeholder
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}
))
# Registrare il gestore asincrono
dispatcher.register_handler(LLMRequestSent, lambda e: asyncio.create_task(llm_agent_component(e, dispatcher)))
# ... (altri gestori sopra) ...
# Per eseguire un esempio:
async def main():
query_id = "user_email_123"
dispatcher.dispatch(UserQueryReceived(
query_id=query_id,
content="Ho bisogno di un articolo sugli agenti IA guidati da eventi e di una chiamata di follow-up.",
timestamp=time.time()
))
# Dare tempo per elaborare gli eventi
await asyncio.sleep(10)
print("\n--- Elaborazione completata per user_email_123 ---\n")
query_id_2 = "user_email_456"
dispatcher.dispatch(UserQueryReceived(
query_id=query_id_2,
content="Puoi riassumere i miei articoli precedenti sulle architetture di deep learning?",
timestamp=time.time()
))
await asyncio.sleep(10)
print("\n--- Elaborazione completata per user_email_456 ---\n")
if __name__ == "__main__":
asyncio.run(main())
Ho introdotto `asyncio.create_task` per permettere al `llm_agent_component` di eseguire in parallelo con altri gestori o dispatch successivi. È qui che le architetture orientate agli eventi brillano davvero per la performance e la reattività degli agenti IA.
Lezioni concrete per il tuo prossimo progetto di agente
- Inizia semplice, pensa agli eventi: Anche per un piccolo agente, delinea gli eventi principali che si verificano. Cosa innesca cosa? Quali informazioni devono essere trasmesse?
- Definisci schemi di eventi chiari: Usa `dataclasses` o modelli Pydantic per i tuoi eventi. Questo garantisce coerenza e semplifica il debug. Includi sempre un `query_id` o un `correlation_id`.
- Separa le preoccupazioni: Ogni gestore deve fare bene una cosa. Non cercare di comprimere troppa logica in un unico gestore. Se un gestore deve effettuare una chiamata esterna, deve emettere un evento di richiesta e attendere un evento di risposta corrispondente.
- Adotta l’asincronicità: Le interazioni degli agenti IA (chiamate LLM, esecuzione di strumenti) sono intrinsecamente asincrone. Usa `asyncio` o un framework simile per gestire queste operazioni in parallelo senza bloccare il tuo ciclo di eventi.
- Integra l’osservabile: Un logger di eventi generico (come il mio `log_all_events`) è incredibilmente prezioso. Puoi facilmente reindirizzare questi eventi a un sistema di monitoraggio o semplicemente stamparli per lo sviluppo. Questo flusso di eventi diventa il log del “processo di pensiero” interno del tuo agente.
- Gestione degli errori con eventi: Invece di un `try/except` profondamente annidato, emetti eventi `ErrorEvent` o `ToolCallFailed`. Altri gestori possono quindi ascoltare specificamente questi per implementare logiche di riprova, soluzioni di emergenza o richieste di intervento umano.
Passare a un modello orientato agli eventi ha completamente cambiato il mio modo di pensare alla costruzione di agenti. Mi ha allontanato dal tentativo di anticipare ogni possibile percorso in un flusso lineare e mi ha portato a costruire un sistema che reagisce in modo intelligente al suo ambiente e alle proprie operazioni interne. È un modo più resiliente, scalabile, e francamente, più piacevole di costruire agenti IA complessi.
Provalo per il tuo prossimo progetto di agente. Potresti ritrovarti a districare quel codice spaghetti più velocemente di quanto pensi!
🕒 Published: