Агенты с планированием и запуском по Cron
APScheduler, задания cron и автономное выполнение агентов по временным триггерам.
«Агенты с планированием и запуском по Cron» — бесплатный урок AI Agents на CoddyKit. Это урок 3 из 4. Ты можешь прочитать весь урок бесплатно ниже — а потом практиковать его прямо в браузере с встроенным редактором кода и ИИ-репетитором 24/7. Это часть пути обучения AI Agents, и твой прогресс синхронизируется между веб-версией и приложением CoddyKit. Курс AI Agents содержит 4 уроков всего.
Зачем планировать запуск агентов
Некоторые задачи агентов не нужно выполнять непрерывно или по запросу — они запускаются по расписанию. Ежедневные отчёты, еженедельные сводки, ежечасные проверки данных и периодическая очистка — подходящие задачи для агентов с плановым запуском.
Основы APScheduler
APScheduler (расширенный планировщик Python) выполняет задания внутри процесса Python. Он поддерживает три типа триггеров: date (однократный запуск), interval (повторяющийся запуск) и cron (на основе календаря).
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.schedulers.background import BackgroundScheduler
from datetime import datetime
# BlockingScheduler: takes over the main thread
# BackgroundScheduler: runs in background thread
scheduler = BackgroundScheduler()
def my_agent_job():
print(f'Agent running at {datetime.now()}')
# Add a simple interval job
scheduler.add_job(my_agent_job, 'interval', minutes=5)
scheduler.start()
print('Scheduler started in background')
# Your app continues running here
import time
time.sleep(15)
scheduler.shutdown()
print('Scheduler stopped')Синтаксис триггера cron
Триггер cron использует знакомые поля cron: year, month, day, week, day_of_week, hour, minute, second. Можно использовать числа, диапазоны, списки и подстановочные символы.
from apscheduler.schedulers.background import BackgroundScheduler
scheduler = BackgroundScheduler()
def morning_briefing():
print('Good morning! Running daily briefing agent')
def weekly_report():
print('Running weekly summary')
def every_business_hour():
print('Hourly check during business hours')
# Every day at 9:00 AM
scheduler.add_job(morning_briefing, 'cron', hour=9, minute=0)
# Every Monday at 8:30 AM
scheduler.add_job(weekly_report, 'cron', day_of_week='mon', hour=8, minute=30)
# Every hour from 9am to 5pm, weekdays only
scheduler.add_job(every_business_hour, 'cron',
day_of_week='mon-fri',
hour='9-17',
minute=0
)
scheduler.start()
print('Jobs scheduled:', len(scheduler.get_jobs()))Синтаксис выражений cron
Стандартные выражения cron используют 5 полей: minute hour day month weekday. APScheduler также принимает их в виде одной строки с помощью CronTrigger.from_crontab().
0 9 * * *— ежедневно в 9:000 9 * * 1— каждый понедельник в 9:00*/15 * * * *— каждые 15 минут
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.cron import CronTrigger
scheduler = BackgroundScheduler()
def agent_task():
print('Running scheduled agent')
# From crontab string: every day at 9am
trigger = CronTrigger.from_crontab('0 9 * * *')
scheduler.add_job(agent_task, trigger)
# Equivalent explicit form
scheduler.add_job(
agent_task,
'cron',
minute=0,
hour=9
)
# Every 15 minutes
scheduler.add_job(agent_task, CronTrigger.from_crontab('*/15 * * * *'))
print('Scheduled jobs:')
for job in scheduler.get_jobs():
print(f' {job.id}: next run {job.next_run_time}')Триггер с интервалом
Триггер с интервалом запускает задание каждые N единиц времени. Используйте его для опроса, контрольных сигналов или любой задачи, которую нужно повторять с фиксированной частотой.
from apscheduler.schedulers.background import BackgroundScheduler
from datetime import datetime, timedelta
scheduler = BackgroundScheduler()
def check_for_updates():
print(f'Checking for updates at {datetime.now()}')
# Agent logic: poll API, check for new items
# Every 30 minutes
scheduler.add_job(check_for_updates, 'interval', minutes=30)
# Every 2 hours, starting 10 minutes from now
start_time = datetime.now() + timedelta(minutes=10)
scheduler.add_job(
check_for_updates,
'interval',
hours=2,
start_date=start_time
)
# Run once in the future (date trigger)
from apscheduler.triggers.date import DateTrigger
run_at = datetime.now() + timedelta(minutes=5)
scheduler.add_job(check_for_updates, DateTrigger(run_date=run_at))
scheduler.start()
print('All jobs scheduled')Параметры заданий и идентификаторы
Назначайте заданиям идентификаторы, чтобы позднее обращаться к ним, приостанавливать или удалять их. Передавайте аргументы функции задания через args или kwargs.
from apscheduler.schedulers.background import BackgroundScheduler
scheduler = BackgroundScheduler()
def fetch_report(report_type, user_id):
print(f'Fetching {report_type} report for user {user_id}')
# Named job with arguments
scheduler.add_job(
fetch_report,
'cron',
hour=9,
minute=0,
id='daily_report_user_42',
kwargs={'report_type': 'daily', 'user_id': 42},
replace_existing=True # Update if job already exists
)
scheduler.start()
# Pause a specific job
scheduler.pause_job('daily_report_user_42')
print('Job paused')
# Resume it
scheduler.resume_job('daily_report_user_42')
print('Job resumed')
# Remove it
scheduler.remove_job('daily_report_user_42')
print('Job removed')Сохранение заданий после перезапуска
По умолчанию APScheduler хранит задания в памяти, поэтому после перезапуска они теряются. Используйте хранилище заданий SQLAlchemy, чтобы сохранять задания в базе данных и не терять их после перезапуска.
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from apscheduler.executors.pool import ThreadPoolExecutor
jobstores = {
'default': SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')
}
executors = {
'default': ThreadPoolExecutor(20)
}
scheduler = BackgroundScheduler(
jobstores=jobstores,
executors=executors
)
def persistent_agent():
print('Running persisted scheduled agent')
# This job survives restarts
scheduler.add_job(
persistent_agent,
'cron',
hour=8,
minute=0,
id='morning_agent',
replace_existing=True
)
scheduler.start()
print('Scheduler started with SQLite persistence')Обработка исключений заданий
Оберните функции заданий, выполняемых по расписанию, в обработку исключений, чтобы один неудачный запуск незаметно не остановил будущие запуски. Записывайте сбои в журнал и при необходимости отправляйте уведомления.
import logging
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.events import EVENT_JOB_EXECUTED, EVENT_JOB_ERROR
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger('scheduler')
scheduler = BackgroundScheduler()
def job_listener(event):
if event.exception:
logger.error(f'Job {event.job_id} failed: {event.exception}')
# Optionally send alert: email, Slack, PagerDuty
else:
logger.info(f'Job {event.job_id} completed successfully')
scheduler.add_listener(job_listener, EVENT_JOB_EXECUTED | EVENT_JOB_ERROR)
def my_agent_job():
# Errors here are caught by the listener
raise ValueError('Something went wrong in the agent')
scheduler.add_job(my_agent_job, 'interval', seconds=10, id='test_job')
scheduler.start()Предотвращение перекрытия заданий
Если выполнение задания занимает больше времени, чем его интервал, следующий запуск может начаться до завершения предыдущего. Установите max_instances=1 (значение по умолчанию) или используйте объединение запусков, чтобы пропускать пропущенные запуски.
from apscheduler.schedulers.background import BackgroundScheduler
import time
scheduler = BackgroundScheduler()
def slow_agent():
print('Agent started')
time.sleep(45) # Takes 45 seconds
print('Agent finished')
# max_instances=1: only one run at a time (default)
# coalesce=True: if multiple runs were missed, fire only once when caught up
scheduler.add_job(
slow_agent,
'interval',
minutes=1,
max_instances=1,
coalesce=True,
id='slow_agent'
)
scheduler.start()
print('Slow agent scheduled (max 1 concurrent run)')Интеграция планирования с FastAPI
Используйте BackgroundScheduler в FastAPI вместе с контекстом lifespan, чтобы корректно запускать и останавливать планировщик в соответствии с жизненным циклом сервера.
from fastapi import FastAPI
from apscheduler.schedulers.background import BackgroundScheduler
from contextlib import asynccontextmanager
scheduler = BackgroundScheduler()
def morning_agent_job():
print('Morning agent running')
@asynccontextmanager
async def lifespan(app: FastAPI):
# Start scheduler when app starts
scheduler.add_job(morning_agent_job, 'cron', hour=9, minute=0)
scheduler.start()
print('Scheduler started')
yield # App runs here
# Stop scheduler when app shuts down
scheduler.shutdown()
print('Scheduler stopped')
app = FastAPI(lifespan=lifespan)
@app.get('/jobs')
def list_jobs():
return [
{'id': job.id, 'next_run': str(job.next_run_time)}
for job in scheduler.get_jobs()
]Планирование с учётом часового пояса
Всегда указывайте часовые пояса для агентов с плановым запуском. Задания cron без контекста часового пояса могут запускаться в неправильное время после перехода на летнее или зимнее время.
from apscheduler.schedulers.background import BackgroundScheduler
import pytz
# Scheduler with timezone
scheduler = BackgroundScheduler(timezone='America/New_York')
def ny_morning_agent():
print('Running at 9am New York time')
def london_eod_agent():
print('Running at 5pm London time')
# 9am New York (handles EST/EDT automatically)
scheduler.add_job(
ny_morning_agent,
'cron',
hour=9,
minute=0,
timezone=pytz.timezone('America/New_York')
)
# 5pm London time
scheduler.add_job(
london_eod_agent,
'cron',
hour=17,
minute=0,
timezone=pytz.timezone('Europe/London')
)
scheduler.start()
print('Timezone-aware scheduler running')Проверка знаний: планирование
Проверьте, насколько хорошо Вы понимаете APScheduler и агентов на основе cron.
Лучшие практики планирования
Основные правила для надёжных агентов с плановым запуском:
- Всегда используйте планирование с учётом часового пояса
- Сохраняйте задания в базе данных, чтобы они переживали перезапуск
- Устанавливайте
max_instances=1для долгих заданий - Добавляйте обработчики ошибок, чтобы выявлять незаметные сбои
- Записывайте в журнал каждый запуск по расписанию: время начала, время окончания и результат
Часто задаваемые вопросы
Урок «Агенты с планированием и запуском по Cron» бесплатный?
Да — полный текст урока «Агенты с планированием и запуском по Cron» бесплатно доступен здесь в веб-версии. Чтобы практиковать его интерактивно (встроенный редактор кода и ИИ-репетитор 24/7) и разблокировать остальной курс AI Agents, подпишись на CoddyKit PRO. Курс AI Agents содержит 4 уроков всего.
Чему я научусь в уроке «Агенты с планированием и запуском по Cron»?
APScheduler, задания cron и автономное выполнение агентов по временным триггерам. Ты практикуешь AI Agents с помощью реального кода, который запускаешь прямо в браузере, и ИИ-репетитор 24/7 отвечает на твои вопросы во время урока.
Нужен ли мне опыт, чтобы начать AI Agents?
Предыдущий опыт не требуется. AI Agents на CoddyKit структурирован для всех уровней — от новичков до продвинутых, поэтому ты можешь начать отсюда или с самого начала и учиться в своем темпе. Это урок 3 из 4.
Сколько времени занимает урок «Агенты с планированием и запуском по Cron»?
Большинство уроков CoddyKit занимают около 5–10 минут. Каждый из них компактный и интерактивный, поэтому ты постоянно делаешь прогресс и продолжаешь с того же места в веб-версии и приложении.
Можно ли писать и запускать код в этом уроке AI Agents?
Да. Каждый урок AI Agents включает встроенный редактор кода, поэтому ты пишешь и запускаешь реальный код прямо в браузере и получаешь моментальную обратную связь от AI — локальная установка не требуется.
Все уроки этого курса
- Шаблоны агентов «триггер — действие»
- Подключение агентов к вебхукам
- Агенты с планированием и запуском по Cron
- Создание конвейера автоматизации нескольких приложений