Ehi a tutti, Alex qui da agntai.net! È marzo 2026 e ultimamente ho speso troppo tempo a pensare a come costruiamo agenti AI. In particolare, ho lottato con il “codice di collegamento” – la parte che collega tutti i fantastici output LLM, le chiamate agli strumenti e la gestione dello stato. Abbiamo tutti visto delle dimostrazioni impressionanti, giusto? Agenti che fanno cose straordinarie. Ma poi provi a costruirne uno per un problema reale e ti trovi di fronte a una parete di callback, logica condizionale e aggiornamenti di stato. Sembra meno costruire un sistema intelligente e più gestire una fabbrica di spaghetti molto complessa.
Quindi, oggi voglio parlare di qualcosa che ha guadagnato silenziosamente attenzione e, francamente, sta salvando la mia sanità mentale: Architetture a Eventi per Agenti AI. Non è un concetto nuovo nell’ingegneria del software, per niente, ma applicarlo con criterio agli agenti AI, specialmente a quelli che orchestrano interazioni multiple con LLM e strumenti esterni, sembra respirare aria fresca. Dimentichiamo per un momento il pensiero lineare, passo dopo passo. Pensiamo ai sistemi reattivi.
La Mia Lotta Personale con gli Agenti Monolitici
Qualche mese fa, stavo lavorando a un agente progettato per aiutarmi a gestire il mio flusso di lavoro di scrittura freelance. L’idea era semplice: avrebbe monitorato la mia inbox per nuove richieste, redatto risposte iniziali, suggerito articoli pertinenti dal mio knowledge base e persino aiutato a pianificare chiamate di follow-up. Sembrava abbastanza semplice.
Il mio approccio iniziale era abbastanza tipico: un ciclo principale. Ricevi email. Elabora email. Decidi l’azione (redigi, pianifica, cerca). Chiama LLM. Elabora l’output LLM. Chiama lo strumento (API calendario, API email, API knowledge base). Aggiorna lo stato interno. Ripeti.
È iniziato bene, ma man mano che aggiungevo più “intelligenza” e più strumenti, è diventato un incubo. E se la chiamata all’API del calendario fallisse? E se il LLM avesse allucinato un contatto che non esisteva? E se avessi bisogno di mettere in pausa e chiedere un input umano per una decisione critica? Il mio unico script agente monolitico si trasformava rapidamente in un labirinto di `if/else` annidati con blocchi `try/except` ovunque. Il debugging era un incubo. Modificare una parte spesso rompeva un’altra. Sembrava che stessi costantemente sistemando perdite in una nave affonda.
Ricordo una notte tardi, cercando di capire perché il mio agente continuasse a redigere risposte per email già elaborate. Si è rivelato che l’aggiornamento di stato per “email elaborata” avveniva *dopo* una potenziale riesecuzione di LLM in un percorso di errore. Era una classica condizione di gara in un sistema che non era progettato per gestire operazioni asincrone e non deterministiche con eleganza. È allora che ho iniziato a cercare un modo migliore.
Perché gli Agenti Basati sugli Eventi Hanno Senso
Pensa a come lavorano gli esseri umani. Di solito non seguiamo uno script rigoroso e predefinito per ogni interazione. Reagiamo agli eventi. Qualcuno fa una domanda: questo è un evento. Lo elaboriamo e rispondiamo: questo è un altro evento. Riceviamo un nuovo pezzo di informazione: evento. Decidiamo di utilizzare uno strumento (come aprire un browser): evento. Il nostro “stato” interno cambia costantemente in base a questi eventi.
Un’architettura a eventi (EDA) per gli agenti AI rispecchia questo schema di interazione naturale. Invece di un flusso di controllo rigido, i componenti emettono eventi quando accade qualcosa di significativo. Altri componenti (ascoltatori, gestori) reagiscono a questi eventi. Questo porta a diversi vantaggi chiave:
- Modularità: I componenti diventano debolmente accoppiati. Un esecutore di strumenti non deve sapere chi lo ha chiamato o cosa accadrà dopo; emette semplicemente un evento come “tool_call_succeeded” oppure “tool_call_failed.”
- Flessibilità: È molto più facile aggiungere nuove funzionalità o modificare quelle esistenti. Vuoi un nuovo strumento? Basta aggiungere un gestore che ascolta un evento di intento specifico. Vuoi registrare ogni chiamata LLM? Aggiungi un logger 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ò prenderlo in carico. Puoi integrare code di dead-letter per eventi che non possono essere elaborati.
- Concorrenza: Molti eventi possono essere elaborati in parallelo, sia da diversi gestori che dallo stesso gestore su istanze di eventi diversi. Questo è cruciale per gli agenti che devono gestire più attività contemporanee.
- Osservabilità: Il flusso di eventi fornisce un registro chiaro e verificabile di tutto ciò che l’agente sta facendo. Puoi facilmente tracciare il flusso di informazioni e decisioni.
L’Idea Centrale: Eventi, Dispatcher e Gestori
In sostanza, un EDA ha bisogno di tre cose:
- Eventi: Strutture dati semplici che descrivono qualcosa che è accaduto (es. `ToolCalled`, `LLMResponseReceived`, `UserQueryReceived`).
- Un Dispatcher di Eventi: Un meccanismo centrale che prende un evento e lo instrada a tutte le parti interessate.
- Gestori di Eventi: Funzioni o classi che “ascoltano” per specifici tipi di eventi ed eseguono della logica quando ne ricevono uno.
Guardiamo un esempio semplificato. Immagina il nostro agente per il flusso di scrittura. Invece di una grande funzione, abbiamo:
- Un evento `UserQueryReceived` (quando arriva una nuova email).
- Un evento `LLMInputGenerated` (quando creiamo un prompt per il LLM).
- Un evento `LLMResponseReceived` (quando il LLM restituisce il suo output).
- Un evento `ToolCallRequested` (quando il LLM suggerisce di utilizzare uno strumento).
- Un evento `ToolCallSucceeded` / `ToolCallFailed` (dopo un’interazione con uno strumento).
- Un evento `DraftResponseReady` (quando una bozza è pronta per la revisione).
Ognuno di questi eventi porta dati rilevanti: il contenuto dell’email, il prompt/risposta del LLM, nome e argomenti dello strumento, ecc.
Blocchi di Costruzione: Un Approccio Pythonico
Non hai bisogno di una coda di messaggi pesante come Kafka per sistemi di agenti semplici (anche se per agenti distribuiti in produzione, potresti averne bisogno!). Per un agente a processo singolo, un semplice dispatcher di eventi in memoria funziona meraviglie.
Passo 1: Definisci i Tuoi Eventi
Mi piace utilizzare `dataclasses` per gli eventi perché sono pulite ed esplicite.
from dataclasses import dataclass
from typing import Any, Dict, Optional
@dataclass
class AgentEvent:
"""Classe base per tutti gli eventi degli agenti."""
timestamp: float # Aggiungi un timestamp per ordinamento e 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
Nota il `query_id`. Questo è fondamentale! Ci consente di correlare eventi appartenenti alla stessa interazione o compito dell’utente. Senza di esso, il tuo flusso di eventi diventa un caos.
Passo 2: Crea un Dispatcher di Eventi
Qui è dove gli eventi vengono instradati. Un semplice dictionary che mappa i 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]):
"""Registra una funzione per gestire un tipo di evento specifico."""
self._handlers[event_type].append(handler)
def dispatch(self, event: AgentEvent):
"""Invia un evento a tutti i gestori registrati."""
# Assicurati che il timestamp sia impostato se non lo è già
if not hasattr(event, 'timestamp') or event.timestamp is None:
event.timestamp = time.time()
# Inoltra 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}")
# Potenzialmente inoltrare qui un evento di errore per solidità
# Inoltra anche ai gestori registrati per il tipo base AgentEvent
# Questo consente per la registrazione o monitoraggio 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
Ogni gestore è una semplice funzione che prende un oggetto evento. Esegue il suo compito specifico e, cosa fondamentale, può inviare nuovi eventi.
Scorriamo alcuni gestori per il nostro agente di scrittura:
# Assumendo che 'dispatcher' sia un'istanza di EventDispatcher
# --- Gestore per la query iniziale dell'utente ---
def handle_user_query(event: UserQueryReceived):
print(f"[{event.query_id}] Query dell'utente ricevuta: {event.content[:50]}...")
# Qui, di solito, utilizzeremmo un LLM per decidere l'intento iniziale
# Per semplicità, assumiamo che vada sempre a LLM per la redazione
prompt = f"Sei un assistente utile per uno scrittore freelance. Redigi una risposta iniziale, cortese, alla seguente richiesta del cliente e suggerisci un'azione successiva (ad es., 'schedule_call', 'search_knowledge_base'):\n\n{event.content}\n\nOutput in JSON con i campi 'draft_response' e 'suggested_action'."
# Invia un evento per 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]}...")
# Analizza la risposta LLM (questo sarebbe più solido 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"LLM ha suggerito l'azione: {action}"
))
if draft:
dispatcher.dispatch(DraftResponseReady(
query_id=event.query_id,
response_content=draft,
action_taken="drafted_initial_response"
))
if action == "schedule_call":
# Assumiamo che LLM abbia fornito anche i dettagli della chiamata se necessario
dispatcher.dispatch(ToolCallRequested(
query_id=event.query_id,
tool_name="calendar_scheduler",
tool_args={"client_email": "[email protected]", "duration": "30min"} # Segnaposto
))
elif action == "search_knowledge_base":
# Assumiamo che LLM abbia fornito la query di ricerca
dispatcher.dispatch(ToolCallRequested(
query_id=event.query_id,
tool_name="knowledge_base_search",
tool_args={"query": "articoli correlati sugli agenti AI"} # Segnaposto
))
except json.JSONDecodeError:
print(f"[{event.query_id}] Risposta LLM non valida JSON. Inviata per revisione umana.")
dispatcher.dispatch(FinalResponseReady(
query_id=event.query_id,
response_content="Parsing dell'output LLM fallito, necessita di un umano. Risposta LLM originale: " + event.response_text,
action_taken="human_review_needed"
))
# --- Gestore per le richieste di chiamata degli strumenti ---
def handle_tool_call_request(event: ToolCallRequested):
print(f"[{event.query_id}] Richiesta di chiamata agli strumenti: {event.tool_name} con args {event.tool_args}")
# Simula l'esecuzione dello strumento
if event.tool_name == "calendar_scheduler":
# In un sistema reale, questo chiamerebbe un'API effettiva
print(f"Programmazione della chiamata per {event.tool_args.get('client_email')}...")
time.sleep(1) # Simula il ritardo di rete
if random.random() > 0.1: # 90% di successo
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 calendario o occupato"
))
# ... altri strumenti ...
# --- Gestore generico dei registri ---
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 pezzo sia indipendente. Il `handle_user_query` non sa *come* verrà inviata la richiesta LLM, solo che deve emettere un evento `LLMRequestSent`. Allo stesso modo, `handle_llm_response` non si preoccupa di chi ha inviato il prompt originale; si limita a elaborare la risposta e decidere cosa fare dopo.
Simulazione di chiamate LLM e strumenti
Per un sistema reale, `LLMRequestSent` attiverebbe un componente che chiama effettivamente l’API LLM, e poi invia `LLMResponseReceived` quando il risultato viene restituito. Qui è dove `asyncio` o un semplice thread pool possono rivelarsi utili per chiamate LLM o esecuzioni di strumenti concorrenti senza bloccare il ciclo degli eventi.
import asyncio
import json
import random
import time
# ... (Definizioni degli eventi e EventDispatcher soprastanti) ...
# Mock API LLM
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)) # Simula la latenza LLM
# Logica mock 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 parlare di più. Che ne dici di programmare una chiamata rapida 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 cercato alcuni articoli pertinenti.",
"suggested_action": "search_knowledge_base"
})
else:
return json.dumps({
"draft_response": "Grazie per aver contattato! Ho esaminato la tua richiesta e redatto una risposta iniziale.",
"suggested_action": "none"
})
# Componente agente LLM (ascolta per LLMRequestSent, invia LLMResponseReceived)
async def llm_agent_component(event: LLMRequestSent, dispatcher: EventDispatcher):
response_text = await mock_llm_call(event.prompt)
# In un sistema reale, dovresti analizzare per le chiamate agli strumenti dalla risposta LLM
tool_calls = [] # Segnaposto
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 async
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 AI basati su eventi e di una chiamata di follow-up.",
timestamp=time.time()
))
# Dare un po' di tempo per l'elaborazione degli 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 passati 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 consentire al `llm_agent_component` di funzionare in modo concorrente con altri gestori o invii successivi. Qui è dove le architetture basate su eventi brillano davvero per prestazioni e reattività negli agenti AI.
Consigli pratici per il tuo prossimo progetto di agente
- Inizia semplice, pensa agli eventi: Anche per un piccolo agente, schizza gli eventi chiave che accadono. Cosa attiva cosa? Quali informazioni devono essere trasmesse?
- Definisci schemi di eventi chiari: Usa `dataclasses` o modelli Pydantic per i tuoi eventi. Questo garantisce coerenza e rende più facile il debug. Includi sempre un `query_id` o `correlation_id`.
- Separa le preoccupazioni: Ogni gestore dovrebbe fare bene una cosa. Non cercare di comprimere troppa logica in un singolo gestore. Se un gestore ha bisogno di fare una chiamata esterna, dovrebbe inviare un evento di richiesta e aspettare un evento di risposta corrispondente.
- Abbraccia l’asincronia: Le interazioni degli agenti AI (chiamate LLM, esecuzione di strumenti) sono intrinsecamente asincrone. Usa `asyncio` o un framework simile per gestirle in modo concorrente senza bloccare il tuo ciclo degli eventi.
- Costruisci l’osservabilità: Un logger generico degli eventi (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 “processo di pensiero” interno del tuo agente.
- Gestione degli errori con eventi: Invece di annidare profondamente `try/except`, invia eventi `ErrorEvent` o `ToolCallFailed`. Altri gestori possono quindi ascoltare specificamente questi eventi per implementare logica di retry, fallback o richieste di intervento umano.
Passare a un modello basato su eventi ha completamente cambiato il mio modo di pensare a come costruire gli agenti. Mi ha allontanato dal cercare di anticipare ogni possibile percorso in un flusso lineare e verso la costruzione di un sistema che reagisce in modo intelligente al proprio ambiente e alle proprie operazioni interne. È un modo più resiliente, scalabile e, francamente, più piacevole per costruire agenti AI complessi.
Provalo per il tuo prossimo progetto di agente. Potresti trovarti a districare quel codice spaghetti più velocemente di quanto pensi!
🕒 Published: