0Pricing
AI Agents · Lekcja

Wzorce agentów trigger-action

Wykrycie zdarzenia → decyzja → działanie: podstawowa pętla automatyzacji.

Wzorce agentów trigger-action to bezpłatna lekcja AI Agents na CoddyKit. To lekcja 1 z 4. Możesz przeczytać całą lekcję poniżej za darmo — a potem ćwiczyć ją interaktywnie w przeglądarce z wbudowanym edytorem kodu i tutorem AI dostępnym 24/7. To część ścieżki edukacyjnej AI Agents, a Twój postęp synchronizuje się między webem a aplikacją CoddyKit. Kurs AI Agents zawiera 4 lekcji w sumie.

Czym jest agent typu trigger-action?

Agent typu trigger-action obserwuje zdarzenia i reaguje działaniami. Pętla składa się z trzech części: wykryj zdarzenie → LLM wybiera działanie → wykonaj działanie.

Przykłady: otrzymano wiadomość e-mail → podsumuj ją i odpowiedz; przesłano plik → sprawdź go i przetwórz; codziennie o 9:00 → wygeneruj raport dzienny.

Rodzaje wyzwalaczy

Wyzwalacze dzielą się na trzy kategorie:

  • Oparte na zdarzeniach: webhook jest wywoływany po otrzymaniu wiadomości e-mail lub przesłaniu pliku
  • Oparte na czasie: harmonogram cron uruchamia agenta w stałych odstępach
  • Oparte na odpytywaniu: agent wielokrotnie sprawdza API w poszukiwaniu nowych danych

Wybór właściwego rodzaju wyzwalacza określa opóźnienie agenta i zużycie zasobów.

Etap wykrywania

Wykrywanie oznacza odebranie lub rozpoznanie zdarzenia. W przypadku webhooków serwer odbiera żądanie POST. W przypadku odpytywania agent wysyła zapytanie do API i porównuje wyniki z ostatnio znanym stanem.

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

Etap podejmowania decyzji

Po wykryciu zdarzenia agent przekazuje kontekst do LLM i pyta, jakie działanie należy podjąć. LLM wybiera narzędzie albo zwraca bezpośrednią odpowiedź.

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)

Etap wykonywania

Wykonywanie uruchamia wybrane działanie. Może ono wywołać API, zapisać plik, wysłać wiadomość lub uruchomić inny workflow. Należy zawsze obsługiwać błędy i rejestrować wyniki.

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)

Model maszyny stanów

Maszyna stanów to potężny model dla agentów automatyzacji. Stany mogą obejmować: IDLE, DETECTING, DECIDING, EXECUTING, ERROR. Przejścia następują w reakcji na zdarzenia lub warunki.

Maszyny stanów sprawiają, że zachowanie agenta jest przewidywalne i łatwiejsze do debugowania.

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

Wzorzec wyzwalania po otrzymaniu wiadomości e-mail

Powiadomienia push Gmail korzystają z Pub/Sub. Gdy nadejdzie nowa wiadomość e-mail, Google publikuje komunikat w topicu. Agent odbiera webhook, pobiera wiadomość e-mail i ją przetwarza.

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

Wzorzec wyzwalania po przesłaniu pliku

Powiadomienia o zdarzeniach S3 lub obserwatory lokalnego systemu plików mogą uruchamiać agentów, gdy pojawią się pliki. Biblioteka watchdog obserwuje katalogi w poszukiwaniu nowych plików.

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

Wzorzec wyzwalania opartego na czasie

Wyzwalacze oparte na czasie uruchamiają agentów zgodnie z harmonogramem. Należy użyć APScheduler do planowania wewnątrz procesu albo systemowego zadania cron do planowania na poziomie procesu.

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

Idempotentne wykonywanie działań

Agenci automatyzacji muszą być idempotentni: dwukrotne wykonanie tego samego działania nie powinno powodować zduplikowanych skutków. Należy używać kluczy idempotencji i wzorców sprawdzania przed wykonaniem działania.

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

Stany błędów i odzyskiwanie

Odporni agenci obsługują awarie w kontrolowany sposób. Gdy wykonanie się nie powiedzie, agent może ponowić próbę z wykładniczym zwiększaniem odstępów między próbami, powiadomić człowieka lub przenieść zadanie do kolejki dead-letter w celu ręcznego sprawdzenia.

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)

Sprawdzenie wiedzy: wzorce trigger-action

Sprawdź swoją wiedzę na temat wzorców agentów typu trigger-action.

Łączenie wszystkich elementów

Kompletny agent typu trigger-action łączy wszystkie elementy: źródło wyzwalacza (e-mail, plik, timer), warstwę wykrywania, podejmowanie decyzji na podstawie LLM, idempotentne wykonywanie, logikę ponawiania prób i śledzenie stanu.

Należy zacząć od prostego rozwiązania: jeden rodzaj wyzwalacza i jedno działanie. Złożoność należy zwiększać stopniowo, w miarę zdobywania pewności co do zachowania agenta.

Często zadawane pytania

Czy lekcja „Wzorce agentów trigger-action” jest bezpłatna?

Tak — pełny tekst „Wzorce agentów trigger-action” jest dostępny za darmo tutaj w sieci. Aby ćwiczyć ją interaktywnie (wbudowany edytor kodu i tutor AI dostępny 24/7) i odblokować resztę kursu AI Agents, przejdź na CoddyKit PRO. Kurs AI Agents zawiera 4 lekcji w sumie.

Co nauczysz się w „Wzorce agentów trigger-action”?

Wykrycie zdarzenia → decyzja → działanie: podstawowa pętla automatyzacji. Ćwiczysz AI Agents z praktycznym kodem, który uruchamiasz bezpośrednio w przeglądarce, a tutor AI dostępny 24/7 odpowiada na Twoje pytania podczas pracy nad lekcją.

Czy potrzebuję doświadczenia, aby zacząć AI Agents?

Nie wymagamy żadnego doświadczenia. AI Agents w CoddyKit jest strukturyzowany dla początkujących i zaawansowanych użytkowników, więc możesz zacząć tutaj lub od początku i uczyć się w swoim tempie. To lekcja 1 z 4.

Ile czasu zajmuje lekcja „Wzorce agentów trigger-action”?

Większość lekcji CoddyKit trwa około 5–10 minut. Każda lekcja to mały, interaktywny krok, dzięki czemu robisz systematyczne postępy i zawsze wracasz dokładnie do tego samego miejsca — na webie i w aplikacji.

Czy mogę pisać i uruchamiać kod w tej lekcji AI Agents?

Tak. Każda lekcja AI Agents zawiera wbudowany edytor kodu, więc piszesz i uruchamiasz prawdziwy kod bezpośrednio w przeglądarce i od razu otrzymujesz sprzężenie zwrotne od AI — bez konfiguracji na komputerze.

Wszystkie lekcje w tym kursie

  1. Wzorce agentów trigger-action
  2. Łączenie agentów z webhookami
  3. Agenci planowani i oparte na Cronie
  4. Budowanie potoku automatyzacji obejmującego wiele aplikacji
← Powrót do AI Agents