スケジューリングとCronベースのエージェント
APScheduler、cronジョブ、時間をトリガーとする自律エージェントの実行を学びます。
「スケジューリングとCronベースのエージェント」はCoddyKit上の無料AI Agentsレッスンです。 これはレッスン3/4です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。 これはAI Agents学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 AI Agentsコースには全4レッスンが含まれています。
エージェントをスケジュール実行する理由
エージェントのタスクには、継続的またはオンデマンドで実行する必要がなく、スケジュールに従って実行すればよいものがあります。日次レポート、週次サマリー、1時間ごとのデータチェック、定期的なクリーンアップは、スケジュール実行に適しています。
APSchedulerの基礎
APScheduler（Advanced Python Scheduler）は、Pythonプロセス内でジョブを実行します。3種類のトリガー、date（1回限り）、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()を使って、これらを1つの文字列として指定することもできます。
0 9 * * *— 毎日午前9時0 9 * * 1— 毎週月曜日の午前9時*/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')ジョブのパラメーターとID
ジョブにIDを割り当てると、後からジョブを参照、一時停止、削除できます。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 job storeを使用してジョブをデータベースに永続化すると、再起動後もジョブを保持できます。
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')ジョブの例外処理
スケジュールされたジョブ関数をtry/exceptで囲み、1回の実行失敗によって以降の実行が何も通知されずに停止するのを防いでください。失敗をログに記録し、必要に応じてアラートを送信します。
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の統合
FastAPIの内部でAPSchedulerのBackgroundSchedulerをライフスパンコンテキストとともに使用し、サーバーのライフサイクルに合わせてスケジューラーを適切に開始・停止してください。
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時間対応のAIチューター）、AI Agentsコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 AI Agentsコースには全4レッスンが含まれています。
「スケジューリングとCronベースのエージェント」で何を学びますか？
APScheduler、cronジョブ、時間をトリガーとする自律エージェントの実行を学びます。 ブラウザで直接実行するハンズオンコードでAI Agentsを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。
AI Agentsを始めるのに経験は必要ですか？
事前経験は必要ありません。CoddyKitのAI Agentsは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン3/4です。
「スケジューリングとCronベースのエージェント」レッスンにはどのくらい時間がかかりますか？
ほとんどのCoddyKitレッスンは約5～10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。
このAI Agentsレッスンでコードを書いて実行できますか？
はい。すべてのAI Agentsレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。
このコースのすべてのレッスン
- トリガー・アクション型エージェントのパターン
- エージェントとWebhookの接続
- スケジューリングとCronベースのエージェント
- 複数アプリの自動化パイプラインを構築する