Patrones de agentes de tipo activador-acción
Detección de eventos → decisión → acción: el ciclo esencial de la automatización.
Patrones de agentes de tipo activador-acción es una lección gratuita de AI Agents en CoddyKit. Esta es la lección 1 de 4. Puedes leer la lección completa abajo gratuitamente — luego la practicas en el navegador con un editor de código integrado y un tutor de IA 24/7. Forma parte de la ruta de aprendizaje de AI Agents, y tu progreso se sincroniza en la web y la app de CoddyKit. El curso de AI Agents incluye 4 lecciones en total.
¿Qué es un agente basado en disparadores y acciones?
Un agente basado en disparadores y acciones supervisa eventos y responde ejecutando acciones. El ciclo de tres pasos es: detectar el evento → el LLM decide la acción → ejecutar la acción.
Ejemplos: se recibe un correo electrónico → resumirlo y responder; se carga un archivo → validarlo y procesarlo; todos los días a las 9:00 → generar un informe diario.
Tipos de disparadores
Los disparadores se clasifican en tres categorías:
- Basados en eventos: se activa un webhook cuando llega un correo electrónico o se carga un archivo
- Basados en tiempo: una programación de cron ejecuta el agente a intervalos fijos
- Basados en polling: el agente consulta repetidamente una API para buscar datos nuevos
Elegir el tipo de disparador adecuado determina la latencia y el uso de recursos del agente.
La fase de detección
Detectar significa recibir o reconocer un evento. En el caso de los webhooks, su servidor recibe una solicitud POST. En el caso del polling, su agente consulta una API y compara los resultados con el estado visto por última vez.
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 de decisión
Después de detectar un evento, el agente envía el contexto a un LLM y le pregunta qué acción debe realizar. El LLM selecciona una herramienta o devuelve una respuesta directa.
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 de ejecución
La ejecución lleva a cabo la acción elegida. Puede llamar a una API, escribir un archivo, enviar un mensaje o activar otro flujo de trabajo. Controle siempre los errores y registre los resultados.
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)
Modelo de máquina de estados
Una máquina de estados es un modelo eficaz para los agentes de automatización. Los estados pueden ser: IDLE, DETECTING, DECIDING, EXECUTING, ERROR. Las transiciones se producen a partir de eventos o condiciones.
Las máquinas de estados hacen que el comportamiento del agente sea predecible y más fácil de depurar.
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'})Patrón de disparador al recibir un correo electrónico
Las notificaciones push de Gmail utilizan Pub/Sub. Cuando llega un correo electrónico nuevo, Google publica un mensaje en su tema. Su agente recibe un webhook, obtiene el correo electrónico y lo procesa.
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'}Patrón de disparador al cargar un archivo
Las notificaciones de eventos de S3 o los observadores del sistema de archivos local pueden activar agentes cuando aparecen archivos. La biblioteca watchdog supervisa los directorios para detectar archivos nuevos.
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()Patrón de disparador basado en tiempo
Los disparadores basados en tiempo ejecutan agentes según una programación. Utilice APScheduler para la programación dentro del proceso o un trabajo de cron del sistema para la programación a nivel de proceso.
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()Ejecución idempotente de acciones
Los agentes de automatización deben ser idempotentes: ejecutar la misma acción dos veces no debería producir efectos duplicados. Utilice claves de idempotencia y patrones de comprobación previa a la acción.
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 - skippedEstados de error y recuperación
Los agentes robustos gestionan los fallos correctamente. Cuando la ejecución falla, el agente puede reintentar con un backoff exponencial, avisar a una persona o pasar el mensaje a una cola de mensajes fallidos para su revisión manual.
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)Comprobación de conocimientos: patrones de disparador y acción
Compruebe sus conocimientos sobre los patrones de los agentes basados en disparadores y acciones.
Cómo integrar todas las piezas
Un agente completo basado en disparadores y acciones combina todas las piezas: una fuente de disparadores (correo electrónico, archivo o temporizador), una capa de detección, toma de decisiones basada en LLM, ejecución idempotente, lógica de reintentos y seguimiento del estado.
Comience con algo sencillo: un tipo de disparador y una acción. Añada complejidad de forma gradual a medida que gane confianza en el comportamiento del agente.
Preguntas frecuentes
¿La lección «Patrones de agentes de tipo activador-acción» es gratis?
Sí — el texto completo de «Patrones de agentes de tipo activador-acción» es gratis para leer aquí en la web. Para practicarla de forma interactiva (editor de código integrado y tutor de IA 24/7) y desbloquear el resto del curso de AI Agents, actualiza a CoddyKit PRO. El curso de AI Agents incluye 4 lecciones en total.
¿Qué aprenderé en «Patrones de agentes de tipo activador-acción»?
Detección de eventos → decisión → acción: el ciclo esencial de la automatización. Practicas AI Agents con código real que ejecutas directamente en el navegador, y un tutor de IA 24/7 responde tus preguntas mientras trabajas en la lección.
¿Necesito experiencia previa para empezar AI Agents?
No se requiere experiencia previa. AI Agents en CoddyKit está estructurado para principiantes hasta estudiantes avanzados, así que puedes empezar aquí o desde el inicio y avanzar a tu ritmo. Esta es la lección 1 de 4.
¿Cuánto tiempo toma la lección «Patrones de agentes de tipo activador-acción»?
La mayoría de las lecciones de CoddyKit toman alrededor de 5–10 minutos. Cada una es compacta e interactiva, así que avanzas constantemente y retomas exactamente por donde dejaste en la web y la app.
¿Puedo escribir y ejecutar código en esta lección de AI Agents?
Sí. Cada lección de AI Agents incluye un editor de código integrado, así que escribes y ejecutas código real directamente en tu navegador y obtienes retroalimentación instantánea de IA — sin configuración local necesaria.
Todas las lecciones de este curso
- Patrones de agentes de tipo activador-acción
- Conexión de agentes a webhooks
- Agentes programados y basados en cron
- Creación de un pipeline de automatización multiaplicación