ตัวแทนตามกำหนดเวลาและ Cron

เหตุใดจึงต้องจัดตารางเวลาให้เอเจนต์

พื้นฐาน APScheduler

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

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

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

ตัวกระตุ้นตามช่วงเวลา

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 ของงาน

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

การคงงานไว้หลังการเริ่มระบบใหม่

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

การป้องกันงานทำงานซ้อนกัน

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

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

การจัดตารางเวลาที่คำนึงถึงเขตเวลา

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

ตรวจสอบความรู้: การจัดตารางเวลา

แนวทางปฏิบัติที่ดีที่สุดสำหรับการจัดตารางเวลา