Agenti pianificati e basati su cron
APScheduler, cron job ed esecuzione autonoma degli agenti basata sul tempo
Agenti pianificati e basati su cron è una lezione AI Agents gratuita su CoddyKit. Questa è la lezione 3 di 4. Puoi leggere la lezione completa qui gratuitamente — poi esercitati direttamente nel browser con un editor di codice integrato e un tutor IA disponibile 24/7. Fa parte del percorso di apprendimento AI Agents, e i tuoi progressi si sincronizzano tra il web e l'app CoddyKit. Il corso AI Agents include 4 lezioni in totale.
Perché pianificare gli agenti
Alcune attività degli agenti non devono essere eseguite continuamente o su richiesta: vengono eseguite secondo una pianificazione. Rapporti giornalieri, riepiloghi settimanali, controlli orari dei dati e operazioni periodiche di pulizia sono tutti ottimi candidati per gli agenti pianificati.
Nozioni di base su APScheduler
APScheduler (Advanced Python Scheduler) esegue i job all'interno del processo Python. Supporta tre tipi di trigger: date (una volta), interval (ripetuto) e cron (basato sul calendario).
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')Sintassi del trigger cron
Il trigger cron utilizza i noti campi cron: year, month, day, week, day_of_week, hour, minute, second. È possibile utilizzare numeri, intervalli, elenchi e caratteri jolly.
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()))Sintassi delle espressioni cron
Le espressioni cron standard utilizzano 5 campi: minute hour day month weekday. APScheduler accetta anche questi campi come un'unica stringa con CronTrigger.from_crontab().
0 9 * * *— ogni giorno alle 90 9 * * 1— ogni lunedì alle 9*/15 * * * *— ogni 15 minuti
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}')Trigger a intervallo
Il trigger a intervallo esegue un job ogni N unità di tempo. Lo utilizzi per il polling, gli heartbeat o qualsiasi attività che debba ripetersi a una frequenza fissa.
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')Parametri e ID dei job
Assegni degli ID ai job per poterli identificare, mettere in pausa o rimuovere in seguito. Passi gli argomenti alla funzione del job tramite args o 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')Persistenza dei job tra i riavvii
Per impostazione predefinita, APScheduler memorizza i job in memoria, che vengono persi al riavvio. Utilizzi SQLAlchemy job store per rendere persistenti i job in un database, così che sopravvivano ai riavvii.
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')Gestione delle eccezioni dei job
Racchiuda le funzioni dei job pianificati in try/except per impedire che un'esecuzione non riuscita interrompa in modo silenzioso le esecuzioni future. Registri gli errori e, facoltativamente, invii degli avvisi.
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()Prevenzione della sovrapposizione dei job
Se un job richiede più tempo del proprio intervallo, l'esecuzione successiva potrebbe iniziare prima che quella precedente sia terminata. Imposti max_instances=1 (il valore predefinito) oppure utilizzi il coalescing per saltare le esecuzioni mancate.
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)')Integrazione della pianificazione con FastAPI
Utilizzi BackgroundScheduler di APScheduler all'interno di FastAPI con un contesto lifespan, per avviare e arrestare correttamente lo scheduler in accordo con il ciclo di vita del server.
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()
]Pianificazione con fuso orario
Specifichi sempre i fusi orari per gli agenti pianificati. I job cron senza un contesto di fuso orario possono essere eseguiti agli orari sbagliati dopo i cambiamenti dell'ora legale.
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')Verifica delle conoscenze: pianificazione
Verifichi la Sua comprensione di APScheduler e degli agenti basati su cron.
Buone pratiche per la pianificazione
Regole fondamentali per agenti pianificati affidabili:
- Utilizzi sempre una pianificazione consapevole del fuso orario
- Renda persistenti i job in un database, così che sopravvivano ai riavvii
- Imposti
max_instances=1per i job di lunga durata - Aggiunga listener degli errori per rilevare gli errori silenziosi
- Registri ogni esecuzione pianificata con ora di inizio, ora di fine ed esito
Domande Frequenti
La lezione «Agenti pianificati e basati su cron» è gratuita?
Sì — il testo completo di «Agenti pianificati e basati su cron» è gratuito qui sul web. Per esercitarvi in modo interattivo (un editor di codice integrato e un tutor IA 24/7) e sbloccare il resto del corso AI Agents, passa a CoddyKit PRO. Il corso AI Agents include 4 lezioni in totale.
Cosa imparerò in «Agenti pianificati e basati su cron»?
APScheduler, cron job ed esecuzione autonoma degli agenti basata sul tempo Eserciti AI Agents con codice pratico che esegui direttamente nel browser, e un tutor IA 24/7 risponde alle tue domande mentre lavori sulla lezione.
Ho bisogno di esperienza per iniziare AI Agents?
Non è richiesta alcuna esperienza precedente. AI Agents su CoddyKit è strutturato per principianti e studenti avanzati, quindi puoi iniziare da qui o dall'inizio e procedere al tuo ritmo. Questa è la lezione 3 di 4.
Quanto tempo richiede la lezione «Agenti pianificati e basati su cron»?
La maggior parte delle lezioni CoddyKit richiede circa 5–10 minuti. Ogni lezione è breve e interattiva, quindi fai progressi costanti e riprendi esattamente da dove hai lasciato su web e app.
Posso scrivere ed eseguire codice in questa lezione AI Agents?
Sì. Ogni lezione AI Agents include un editor di codice integrato, quindi scrivi ed esegui codice reale direttamente nel tuo browser e ricevi feedback istantaneo dall'IA — nessuna configurazione locale necessaria.
Tutte le lezioni di questo corso
- Pattern di agenti trigger-azione
- Connessione degli agenti ai webhook
- Agenti pianificati e basati su cron
- Creazione di una pipeline di automazione multi-app