0Pricing
AI Agents · Aula

Padrões de agentes baseados em gatilho e ação

Detecção de evento → decisão → ação: o ciclo central da automação.

Padrões de agentes baseados em gatilho e ação é uma aula grátis de AI Agents no CoddyKit. Esta é a aula 1 de 4. Você pode ler a aula completa abaixo gratuitamente — depois pratica ao vivo no navegador com um editor de código integrado e um tutor de IA 24/7. Faz parte do caminho de aprendizado de AI Agents, e seu progresso é sincronizado entre a web e o app CoddyKit. O curso de AI Agents inclui 4 aulas no total.

O que é um agente de gatilho e ação?

Um agente de gatilho e ação monitora eventos e responde com ações. O ciclo de três partes é: detectar o evento → o LLM decide a ação → executar a ação.

Exemplos: e-mail recebido → resumir e responder; arquivo enviado → validar e processar; todos os dias às 9h → gerar um relatório diário.

Tipos de gatilhos

Os gatilhos se dividem em três categorias:

  • Baseados em eventos: um retorno de chamada HTTP é acionado quando um e-mail chega ou um arquivo é enviado
  • Baseados em tempo: um agendamento do cron executa o agente em intervalos fixos
  • Baseados em consultas periódicas: o agente consulta repetidamente uma API em busca de dados novos

Escolher o tipo certo de gatilho determina a latência e o uso de recursos do agente.

A fase de detecção

Detecção significa receber ou reconhecer um evento. Para retornos de chamada HTTP, seu servidor recebe uma solicitação POST. Para consultas periódicas, seu agente consulta uma API e compara os resultados com o último estado observado.

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])

A fase de decisão

Depois de detectar um evento, o agente envia o contexto a um LLM e pergunta qual ação deve ser executada. O LLM seleciona uma ferramenta ou retorna uma resposta direta.

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)

A fase de execução

A execução realiza a ação escolhida. Ela pode chamar uma API, gravar um arquivo, enviar uma mensagem ou acionar outro fluxo de trabalho. Sempre trate os erros e registre os 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

Uma máquina de estados é um modelo poderoso para agentes de automação. Os estados podem ser: IDLE, DETECTING, DECIDING, EXECUTING, ERROR. As transições ocorrem em resposta a eventos ou condições.

As máquinas de estados tornam o comportamento do agente previsível e mais 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'})

Padrão de gatilho para e-mails recebidos

As notificações push do Gmail usam o Pub/Sub. Quando chega um novo e-mail, o Google publica uma mensagem no seu tópico. Seu agente recebe um retorno de chamada HTTP, busca o e-mail e o processa.

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'}

Padrão de gatilho para envio de arquivos

As notificações de eventos do S3 ou os observadores do sistema de arquivos local podem acionar agentes quando aparecem arquivos. A biblioteca watchdog observa diretórios em busca de arquivos novos.

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()

Padrão de gatilho baseado em tempo

Os gatilhos baseados em tempo executam agentes conforme um agendamento. Use o APScheduler para agendamento dentro do processo ou um trabalho do cron do sistema para agendamento no nível do 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()

Execução idempotente de ações

Os agentes de automação devem ser idempotentes: executar a mesma ação duas vezes não deve causar efeitos duplicados. Use chaves de idempotência e padrões de verificação antes de agir.

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 - skipped

Estados de erro e recuperação

Agentes robustos lidam com falhas de forma adequada. Quando a execução falha, o agente pode tentar novamente com espera exponencial, alertar uma pessoa ou mover o item para uma fila de mensagens não entregues para revisão 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)

Verificação de conhecimentos: padrões de gatilho e ação

Teste sua compreensão dos padrões de agentes de gatilho e ação.

Reunindo tudo

Um agente de gatilho e ação completo combina todas as partes: uma fonte de gatilho (e-mail, arquivo ou temporizador), uma camada de detecção, tomada de decisão baseada em LLM, execução idempotente, lógica de novas tentativas e acompanhamento de estado.

Comece de forma simples: um tipo de gatilho e uma ação. Adicione complexidade gradualmente à medida que ganhar confiança no comportamento do agente.

Perguntas Frequentes

A aula “Padrões de agentes baseados em gatilho e ação” é grátis?

Sim — o texto completo de “Padrões de agentes baseados em gatilho e ação” é grátis para ler aqui na web. Para praticá-la interativamente (um editor de código integrado e um tutor de IA 24/7) e desbloquear o restante do curso de AI Agents, atualize para CoddyKit PRO. O curso de AI Agents inclui 4 aulas no total.

O que vou aprender em “Padrões de agentes baseados em gatilho e ação”?

Detecção de evento → decisão → ação: o ciclo central da automação. Você pratica AI Agents com código prático que executa diretamente no navegador, e um tutor de IA 24/7 responde suas dúvidas enquanto trabalha na aula.

Preciso ter experiência prévia para começar AI Agents?

Nenhuma experiência prévia é necessária. AI Agents no CoddyKit é estruturado para alunos iniciantes até avançados, então você pode começar aqui ou desde o início e aprender no seu ritmo. Esta é a aula 1 de 4.

Quanto tempo leva a aula “Padrões de agentes baseados em gatilho e ação”?

A maioria das aulas CoddyKit leva cerca de 5–10 minutos. Cada uma é compacta e interativa, então você faz progresso constante e retoma exatamente de onde parou entre web e app.

Posso escrever e executar código nesta aula de AI Agents?

Sim. Cada aula de AI Agents inclui um editor de código integrado, então você escreve e executa código real direto no navegador e recebe feedback de IA instantaneamente — nenhuma configuração local necessária.

Todas as aulas deste curso

  1. Padrões de agentes baseados em gatilho e ação
  2. Conectando agentes a webhooks
  3. Agentes programados e baseados em Cron
  4. Construindo um pipeline de automação entre vários aplicativos
← Voltar para AI Agents