0Pricing
AI Agents · Урок

Шаблоны агентов «триггер — действие»

Обнаружение события → принятие решения → действие: основной цикл автоматизации.

«Шаблоны агентов «триггер — действие»» — бесплатный урок AI Agents на CoddyKit. Это урок 1 из 4. Ты можешь прочитать весь урок бесплатно ниже — а потом практиковать его прямо в браузере с встроенным редактором кода и ИИ-репетитором 24/7. Это часть пути обучения AI Agents, и твой прогресс синхронизируется между веб-версией и приложением CoddyKit. Курс AI Agents содержит 4 уроков всего.

Что такое агент «триггер — действие»

Агент «триггер — действие» отслеживает события и реагирует на них действиями. Цикл состоит из трёх частей: обнаружение события → LLM выбирает действие → выполнение действия.

Примеры: получено письмо → создать краткое содержание и ответить; загружен файл → проверить и обработать; каждый день в 9:00 → сформировать ежедневный отчёт.

Типы триггеров

Триггеры делятся на три категории:

  • На основе событий: вебхук срабатывает при получении письма или загрузке файла
  • На основе времени: расписание cron запускает агента через заданные интервалы
  • На основе опроса: агент многократно обращается к API в поисках новых данных

Выбор подходящего типа триггера определяет задержку агента и расход ресурсов.

Этап обнаружения

Обнаружение означает получение или распознавание события. Для вебхуков сервер получает запрос POST. При опросе агент запрашивает API и сравнивает результаты с последним известным состоянием.

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

Этап принятия решения

Обнаружив событие, агент передаёт контекст в LLM и спрашивает, какое действие следует выполнить. LLM либо выбирает инструмент, либо возвращает прямой ответ.

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)

Этап выполнения

На этапе выполнения запускается выбранное действие. Оно может вызывать API, записывать файл, отправлять сообщение или запускать другой рабочий процесс. Всегда обрабатывайте ошибки и записывайте результаты в журнал.

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)

Модель автомата состояний

Автомат состояний — мощная модель для агентов автоматизации. Состояния могут быть такими: IDLE, DETECTING, DECIDING, EXECUTING, ERROR. Переходы происходят при возникновении событий или выполнении условий.

Автоматы состояний делают поведение агента предсказуемым и упрощают отладку.

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

Шаблон запуска при получении электронной почты

Push-уведомления Gmail используют Pub/Sub. Когда приходит новое письмо, Google публикует сообщение в Вашей теме. Агент получает вебхук, извлекает письмо и обрабатывает его.

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

Шаблон запуска при загрузке файла

Уведомления о событиях S3 или наблюдатели локальной файловой системы могут запускать агентов при появлении файлов. Библиотека watchdog отслеживает новые файлы в каталогах.

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

Шаблон запуска по времени

Триггеры на основе времени запускают агентов по расписанию. Используйте APScheduler для планирования внутри процесса или системное задание cron для планирования на уровне процесса.

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

Идемпотентное выполнение действий

Агенты автоматизации должны быть идемпотентными: повторный запуск одного и того же действия не должен приводить к дублирующим эффектам. Используйте ключи идемпотентности и шаблоны с проверкой перед выполнением.

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

Состояния ошибок и восстановление

Надёжные агенты корректно обрабатывают сбои. Если выполнение завершается ошибкой, агент может повторить попытку с экспоненциальной задержкой, уведомить человека или переместить задачу в очередь недоставленных сообщений для ручной проверки.

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)

Проверка знаний: шаблоны «триггер — действие»

Проверьте, насколько хорошо Вы понимаете шаблоны агентов «триггер — действие».

Собираем всё вместе

Полноценный агент «триггер — действие» объединяет все компоненты: источник триггера (электронная почта, файл, таймер), уровень обнаружения, принятие решений на основе LLM, идемпотентное выполнение, логику повторных попыток и отслеживание состояния.

Начните с простого: один тип триггера и одно действие. Постепенно добавляйте сложность по мере того, как будете лучше понимать поведение агента.

Часто задаваемые вопросы

Урок «Шаблоны агентов «триггер — действие»» бесплатный?

Да — полный текст урока «Шаблоны агентов «триггер — действие»» бесплатно доступен здесь в веб-версии. Чтобы практиковать его интерактивно (встроенный редактор кода и ИИ-репетитор 24/7) и разблокировать остальной курс AI Agents, подпишись на CoddyKit PRO. Курс AI Agents содержит 4 уроков всего.

Чему я научусь в уроке «Шаблоны агентов «триггер — действие»»?

Обнаружение события → принятие решения → действие: основной цикл автоматизации. Ты практикуешь AI Agents с помощью реального кода, который запускаешь прямо в браузере, и ИИ-репетитор 24/7 отвечает на твои вопросы во время урока.

Нужен ли мне опыт, чтобы начать AI Agents?

Предыдущий опыт не требуется. AI Agents на CoddyKit структурирован для всех уровней — от новичков до продвинутых, поэтому ты можешь начать отсюда или с самого начала и учиться в своем темпе. Это урок 1 из 4.

Сколько времени занимает урок «Шаблоны агентов «триггер — действие»»?

Большинство уроков CoddyKit занимают около 5–10 минут. Каждый из них компактный и интерактивный, поэтому ты постоянно делаешь прогресс и продолжаешь с того же места в веб-версии и приложении.

Можно ли писать и запускать код в этом уроке AI Agents?

Да. Каждый урок AI Agents включает встроенный редактор кода, поэтому ты пишешь и запускаешь реальный код прямо в браузере и получаешь моментальную обратную связь от AI — локальная установка не требуется.

Все уроки этого курса

  1. Шаблоны агентов «триггер — действие»
  2. Подключение агентов к вебхукам
  3. Агенты с планированием и запуском по Cron
  4. Создание конвейера автоматизации нескольких приложений
← Назад к AI Agents