Pattern di agenti trigger-azione
Rilevamento dell’evento → decisione → azione: il ciclo fondamentale dell’automazione
Pattern di agenti trigger-azione è una lezione AI Agents gratuita su CoddyKit. Questa è la lezione 1 di 4. Puoi leggere la lezione completa qui gratuitamente — poi esercitati direttamente nel browser con un editor di codice integrato e un tutor IA disponibile 24/7. Fa parte del percorso di apprendimento AI Agents, e i tuoi progressi si sincronizzano tra il web e l'app CoddyKit. Il corso AI Agents include 4 lezioni in totale.
Che cos'è un agente basato su trigger e azioni
Un agente basato su trigger e azioni monitora gli eventi e risponde eseguendo azioni. Il ciclo in tre parti è: rilevare l'evento → decidere l'azione con l'LLM → eseguire l'azione.
Esempi: email ricevuta → riassumere e rispondere; file caricato → convalidare ed elaborare; ogni giorno alle 9 → generare il rapporto giornaliero.
Tipi di trigger
I trigger rientrano in tre categorie:
- Basati su eventi: un webhook viene attivato quando arriva un'email o viene caricato un file
- Basati sul tempo: una pianificazione cron esegue l'agente a intervalli prestabiliti
- Basati sul polling: l'agente controlla ripetutamente un'API alla ricerca di nuovi dati
La scelta del tipo di trigger corretto determina la latenza e l'utilizzo delle risorse dell'agente.
La fase di rilevamento
Il rilevamento consiste nel ricevere o riconoscere un evento. Per i webhook, il server riceve una richiesta POST. Per il polling, l'agente interroga un'API e confronta i risultati con lo stato osservato più di recente.
import json
def detect_new_email(current_emails, last_seen_id):
new_emails = [
e for e in current_emails
if e['id'] > last_seen_id
]
return new_emails
# Simulate detection
current = [{'id': 3, 'subject': 'Meeting'}, {'id': 4, 'subject': 'Invoice'}]
new = detect_new_email(current, last_seen_id=2)
print('New emails:', [e['subject'] for e in new])La fase decisionale
Dopo aver rilevato un evento, l'agente invia il contesto a un LLM e chiede quale azione eseguire. L'LLM seleziona uno strumento oppure restituisce una risposta diretta.
import openai
client = openai.OpenAI(api_key='sk-...')
def decide_action(event_description):
response = client.chat.completions.create(
model='gpt-4o-mini',
messages=[
{'role': 'system', 'content': 'You are an automation agent. Decide what action to take for the event.'},
{'role': 'user', 'content': f'Event: {event_description}'}
],
tools=[
{'type': 'function', 'function': {'name': 'send_reply', 'description': 'Reply to email', 'parameters': {'type': 'object', 'properties': {'message': {'type': 'string'}}, 'required': ['message']}}}
]
)
return response.choices[0].message
result = decide_action('New email: Invoice for $500 from supplier')
print(result)La fase di esecuzione
L'esecuzione avvia l'azione scelta. Potrebbe chiamare un'API, scrivere un file, inviare un messaggio o attivare un altro workflow. Gestisca sempre gli errori e registri i risultati.
import logging
import sys
logging.basicConfig(level=logging.INFO, stream=sys.stdout)
logger = logging.getLogger('agent')
def execute_action(action_name, params):
try:
if action_name == 'send_reply':
# In real code, call Gmail API here
logger.info(f'Sending reply: {params["message"]}')
return {'status': 'success'}
elif action_name == 'create_task':
logger.info(f'Creating task: {params["title"]}')
return {'status': 'success'}
else:
raise ValueError(f'Unknown action: {action_name}')
except Exception as e:
logger.error(f'Action failed: {e}')
return {'status': 'error', 'message': str(e)}
if __name__ == '__main__':
result = execute_action('send_reply', {'message': 'Thanks for reaching out!'})
print('Result:', result)
Modello a macchina a stati
Una macchina a stati è un modello efficace per gli agenti di automazione. Gli stati possono essere: IDLE, DETECTING, DECIDING, EXECUTING, ERROR. Le transizioni avvengono in risposta a eventi o condizioni.
Le macchine a stati rendono il comportamento dell'agente prevedibile e più facile da sottoporre a debug.
from enum import Enum
class AgentState(Enum):
IDLE = 'idle'
DETECTING = 'detecting'
DECIDING = 'deciding'
EXECUTING = 'executing'
ERROR = 'error'
class AutomationAgent:
def __init__(self):
self.state = AgentState.IDLE
def transition(self, new_state):
print(f'State: {self.state.value} -> {new_state.value}')
self.state = new_state
def run_cycle(self, event=None):
self.transition(AgentState.DETECTING)
if event:
self.transition(AgentState.DECIDING)
self.transition(AgentState.EXECUTING)
self.transition(AgentState.IDLE)
agent = AutomationAgent()
agent.run_cycle(event={'type': 'email', 'subject': 'Test'})Pattern di trigger per email ricevute
Le notifiche push di Gmail utilizzano Pub/Sub. Quando arriva una nuova email, Google pubblica un messaggio sul topic. L'agente riceve un webhook, recupera l'email e la elabora.
from fastapi import FastAPI, Request
import base64, json
app = FastAPI()
@app.post('/gmail-push')
async def gmail_push(request: Request):
body = await request.json()
# Decode Pub/Sub message
message = body.get('message', {})
data = base64.b64decode(message.get('data', '')).decode('utf-8')
notification = json.loads(data)
email_address = notification.get('emailAddress')
history_id = notification.get('historyId')
print(f'New email for {email_address}, historyId: {history_id}')
# Fetch email details and run agent here
return {'status': 'ok'}Pattern di trigger per il caricamento di file
Le notifiche degli eventi S3 o i watcher del filesystem locale possono attivare gli agenti quando compaiono dei file. La libreria watchdog monitora le directory alla ricerca di nuovi file.
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import time
class UploadHandler(FileSystemEventHandler):
def on_created(self, event):
if event.is_directory:
return
print(f'New file detected: {event.src_path}')
self.process_file(event.src_path)
def process_file(self, filepath):
# Run agent logic on new file
print(f'Processing: {filepath}')
observer = Observer()
handler = UploadHandler()
observer.schedule(handler, path='/tmp/uploads/', recursive=False)
observer.start()
try:
time.sleep(30) # Watch for 30 seconds
finally:
observer.stop()
observer.join()Pattern di trigger basato sul tempo
I trigger basati sul tempo attivano gli agenti secondo una pianificazione. Utilizzi APScheduler per la pianificazione all'interno del processo oppure un job cron di sistema per la pianificazione a livello di processo.
from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime
scheduler = BlockingScheduler()
def daily_report_agent():
print(f'Daily report running at {datetime.now()}')
# Fetch data, call LLM, send report
pass
def hourly_check_agent():
print(f'Hourly check at {datetime.now()}')
pass
# Run at 8am every day
scheduler.add_job(daily_report_agent, 'cron', hour=8, minute=0)
# Run every 30 minutes
scheduler.add_job(hourly_check_agent, 'interval', minutes=30)
print('Scheduler started')
scheduler.start()Esecuzione idempotente delle azioni
Gli agenti di automazione devono essere idempotenti: eseguire due volte la stessa azione non dovrebbe produrre effetti duplicati. Utilizzi chiavi di idempotenza e pattern di verifica prima dell'azione.
import hashlib
processed_events = set() # In production, use Redis or DB
def compute_event_id(event):
content = f"{event['type']}:{event['source_id']}:{event['timestamp']}"
return hashlib.sha256(content.encode()).hexdigest()[:16]
def handle_event_idempotent(event):
event_id = compute_event_id(event)
if event_id in processed_events:
print(f'Skipping duplicate event: {event_id}')
return {'status': 'duplicate', 'event_id': event_id}
# Process event
print(f'Processing event: {event_id}')
processed_events.add(event_id)
return {'status': 'processed', 'event_id': event_id}
# Simulate duplicate event
event = {'type': 'email', 'source_id': 'abc123', 'timestamp': '2024-01-01T09:00:00'}
print(handle_event_idempotent(event))
print(handle_event_idempotent(event)) # Duplicate - skippedStati di errore e ripristino
Gli agenti robusti gestiscono gli errori in modo appropriato. Quando l'esecuzione non riesce, l'agente può riprovare con un backoff esponenziale, avvisare una persona oppure spostare l'evento in una coda dead-letter per la revisione manuale.
import time
def execute_with_retry(action_fn, max_retries=3):
for attempt in range(max_retries):
try:
result = action_fn()
print(f'Success on attempt {attempt + 1}')
return result
except Exception as e:
wait = 2 ** attempt # Exponential backoff: 1s, 2s, 4s
print(f'Attempt {attempt + 1} failed: {e}. Retrying in {wait}s')
if attempt < max_retries - 1:
time.sleep(wait)
else:
print('All retries exhausted. Moving to dead-letter queue.')
raise
# Example usage
call_count = [0]
def flaky_action():
call_count[0] += 1
if call_count[0] < 3:
raise ConnectionError('Service unavailable')
return 'Done'
execute_with_retry(flaky_action)Verifica delle conoscenze: pattern trigger-azione
Verifichi la Sua comprensione dei pattern degli agenti basati su trigger e azioni.
Combinare tutti i componenti
Un agente completo basato su trigger e azioni combina tutti i componenti: una sorgente del trigger (email, file, timer), un livello di rilevamento, un processo decisionale basato su LLM, un'esecuzione idempotente, una logica di retry e il monitoraggio dello stato.
Inizi in modo semplice: un tipo di trigger e un'azione. Aggiunga complessità gradualmente, man mano che acquisisce fiducia nel comportamento dell'agente.
Domande Frequenti
La lezione «Pattern di agenti trigger-azione» è gratuita?
Sì — il testo completo di «Pattern di agenti trigger-azione» è gratuito qui sul web. Per esercitarvi in modo interattivo (un editor di codice integrato e un tutor IA 24/7) e sbloccare il resto del corso AI Agents, passa a CoddyKit PRO. Il corso AI Agents include 4 lezioni in totale.
Cosa imparerò in «Pattern di agenti trigger-azione»?
Rilevamento dell’evento → decisione → azione: il ciclo fondamentale dell’automazione Eserciti AI Agents con codice pratico che esegui direttamente nel browser, e un tutor IA 24/7 risponde alle tue domande mentre lavori sulla lezione.
Ho bisogno di esperienza per iniziare AI Agents?
Non è richiesta alcuna esperienza precedente. AI Agents su CoddyKit è strutturato per principianti e studenti avanzati, quindi puoi iniziare da qui o dall'inizio e procedere al tuo ritmo. Questa è la lezione 1 di 4.
Quanto tempo richiede la lezione «Pattern di agenti trigger-azione»?
La maggior parte delle lezioni CoddyKit richiede circa 5–10 minuti. Ogni lezione è breve e interattiva, quindi fai progressi costanti e riprendi esattamente da dove hai lasciato su web e app.
Posso scrivere ed eseguire codice in questa lezione AI Agents?
Sì. Ogni lezione AI Agents include un editor di codice integrato, quindi scrivi ed esegui codice reale direttamente nel tuo browser e ricevi feedback istantaneo dall'IA — nessuna configurazione locale necessaria.
Tutte le lezioni di questo corso
- Pattern di agenti trigger-azione
- Connessione degli agenti ai webhook
- Agenti pianificati e basati su cron
- Creazione di una pipeline di automazione multi-app