Agen Berbasis Penjadwalan dan Cron
APScheduler, tugas cron, dan eksekusi agen otonom berdasarkan pemicu waktu.
Agen Berbasis Penjadwalan dan Cron adalah pelajaran AI Agents gratis di CoddyKit. Ini adalah pelajaran 3 dari 4. Kamu bisa membaca pelajaran lengkapnya di bawah secara gratis — lalu praktikkan langsung di browser dengan editor kode bawaan dan tutor AI 24/7. Ini adalah bagian dari jalur belajar AI Agents, dan progresmu tersinkronisasi di web dan aplikasi CoddyKit. Kursus AI Agents mencakup 4 pelajaran total.
Mengapa Menjadwalkan Agen?
Beberapa tugas agen tidak perlu dijalankan terus-menerus atau sesuai permintaan — tugas tersebut berjalan berdasarkan jadwal. Laporan harian, rangkuman mingguan, pemeriksaan data setiap jam, dan pembersihan berkala merupakan kandidat yang baik untuk agen terjadwal.
Dasar-Dasar APScheduler
APScheduler (Penjadwal Python Lanjutan) menjalankan pekerjaan di dalam proses Python Anda. APScheduler mendukung tiga jenis pemicu: date (satu kali), interval (berulang), dan cron (berdasarkan kalender).
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')Sintaks Pemicu Cron
Pemicu cron menggunakan bidang cron yang umum: year, month, day, week, day_of_week, hour, minute, second. Anda dapat menggunakan angka, rentang, daftar, dan karakter pengganti.
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()))Sintaks Ekspresi Cron
Ekspresi cron standar menggunakan 5 bidang: minute hour day month weekday. APScheduler juga menerima bidang tersebut sebagai satu string dengan CronTrigger.from_crontab().
0 9 * * *— setiap hari pukul 09.000 9 * * 1— setiap Senin pukul 09.00*/15 * * * *— setiap 15 menit
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}')Pemicu Interval
Pemicu interval menjalankan pekerjaan setiap N satuan waktu. Gunakan pemicu ini untuk pemeriksaan berkala, sinyal denyut, atau tugas apa pun yang harus diulang pada frekuensi tetap.
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')Parameter dan ID Pekerjaan
Tetapkan ID pada pekerjaan agar Anda dapat merujuk, menjeda, atau menghapusnya nanti. Teruskan argumen ke fungsi pekerjaan melalui args atau 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')Mempertahankan Pekerjaan Setelah Mulai Ulang
Secara bawaan, APScheduler menyimpan pekerjaan di memori dan pekerjaan tersebut hilang saat dimulai ulang. Gunakan penyimpanan pekerjaan SQLAlchemy untuk mempertahankan pekerjaan di basis data agar tetap ada setelah dimulai ulang.
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')Menangani Pengecualian Pekerjaan
Bungkus fungsi pekerjaan terjadwal dalam penanganan pengecualian agar satu proses yang gagal tidak diam-diam menghentikan proses berikutnya. Catat kegagalan dan, jika perlu, kirim peringatan.
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()Mencegah Tumpang Tindih Pekerjaan
Jika pekerjaan memerlukan waktu lebih lama daripada intervalnya, proses berikutnya dapat dimulai sebelum proses sebelumnya selesai. Tetapkan max_instances=1 (nilai bawaan) atau gunakan penggabungan untuk melewati proses yang terlewat.
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)')Mengintegrasikan Penjadwalan dengan FastAPI
Gunakan BackgroundScheduler dari APScheduler di dalam FastAPI bersama konteks lifespan untuk memulai dan menghentikan penjadwal dengan baik mengikuti siklus hidup 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()
]Penjadwalan Berbasis Zona Waktu
Selalu tentukan zona waktu untuk agen terjadwal. Pekerjaan cron tanpa konteks zona waktu dapat berjalan pada waktu yang salah setelah perubahan waktu musim panas.
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')Uji Pemahaman: Penjadwalan
Uji pemahaman Anda tentang APScheduler dan agen berbasis cron.
Praktik Terbaik Penjadwalan
Aturan utama untuk agen terjadwal yang andal:
- Selalu gunakan penjadwalan yang memperhitungkan zona waktu
- Simpan pekerjaan di basis data agar tetap ada setelah dimulai ulang
- Tetapkan
max_instances=1untuk pekerjaan yang berjalan lama - Tambahkan pendengar kesalahan untuk menangkap kegagalan diam-diam
- Catat setiap proses terjadwal beserta waktu mulai, waktu selesai, dan hasilnya
Pertanyaan yang Sering Diajukan
Apakah pelajaran “Agen Berbasis Penjadwalan dan Cron” gratis?
Ya — teks lengkap “Agen Berbasis Penjadwalan dan Cron” gratis dibaca di sini di web. Untuk praktiknya secara interaktif (editor kode bawaan dan tutor AI 24/7) dan buka sisa kursus AI Agents, upgrade ke CoddyKit PRO. Kursus AI Agents mencakup 4 pelajaran total.
Apa yang akan aku pelajari di “Agen Berbasis Penjadwalan dan Cron”?
APScheduler, tugas cron, dan eksekusi agen otonom berdasarkan pemicu waktu. Kamu berlatih AI Agents dengan kode praktik yang langsung kamu jalankan di browser, dan tutor AI 24/7 menjawab pertanyaanmu saat kamu mengerjakan pelajaran ini.
Apakah aku perlu pengalaman untuk memulai AI Agents?
Tidak diperlukan pengalaman sebelumnya. AI Agents di CoddyKit dirancang untuk pemula hingga pelajar tingkat lanjut, jadi kamu bisa memulai di sini atau dari awal dan belajar sesuai kecepatan kamu sendiri. Ini adalah pelajaran 3 dari 4.
Berapa lama pelajaran “Agen Berbasis Penjadwalan dan Cron” memakan waktu?
Sebagian besar pelajaran CoddyKit memakan waktu sekitar 5–10 menit. Setiap pelajaran ringkas dan interaktif, jadi kamu membuat kemajuan stabil dan melanjutkan dari tempat kamu tinggalkan di web dan aplikasi.
Bisakah aku menulis dan menjalankan kode dalam pelajaran AI Agents ini?
Ya. Setiap pelajaran AI Agents menyertakan editor kode bawaan, jadi kamu menulis dan menjalankan kode nyata langsung di browser dan mendapatkan umpan balik AI instan — tidak diperlukan penyiapan lokal.
Semua pelajaran dalam kursus ini
- Pola Agen Pemicu-Tindakan
- Menghubungkan Agen ke Webhook
- Agen Berbasis Penjadwalan dan Cron
- Membangun Pipeline Otomatisasi Multi-Aplikasi