AI Agents · 강의

예약 및 Cron 기반 에이전트

APScheduler, cron 작업, 시간에 따라 실행되는 자율 에이전트를 다룹니다.

예약 및 Cron 기반 에이전트은(는) CoddyKit의 무료 AI Agents 강의입니다. 이것은 4개 중 3번째 강의입니다. 아래에서 전체 강의를 무료로 읽을 수 있으며, 내장 코드 에디터와 24/7 AI 튜터와 함께 브라우저에서 직접 실습할 수 있습니다. 이 강의는 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')

크론 트리거 구문

크론 트리거는 익숙한 크론 필드를 사용합니다. 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()))

크론 표현식 구문

표준 크론 표현식은 5개의 필드를 사용합니다. minute hour day month weekday입니다. APScheduler에서는 이러한 필드를 CronTrigger.from_crontab()과 함께 하나의 문자열로 사용할 수도 있습니다.

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 작업 저장소를 사용하십시오.

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와 스케줄링 통합

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

시간대 인식 스케줄링

예약 실행 에이전트에는 항상 시간대를 지정하십시오. 시간대 컨텍스트가 없는 크론 작업은 일광 절약 시간제 변경 후 잘못된 시간에 실행될 수 있습니다.

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와 크론 기반 에이전트에 대한 이해도를 테스트해 보십시오.

스케줄링 모범 사례

신뢰할 수 있는 예약 실행 에이전트를 위한 핵심 규칙은 다음과 같습니다.

항상 시간대 인식 스케줄링을 사용하십시오
재시작 후에도 작업이 유지되도록 데이터베이스에 저장하십시오
오래 실행되는 작업에는 max_instances=1을 설정하십시오
조용한 실패를 감지할 수 있도록 오류 수신기를 추가하십시오
모든 예약 실행에 시작 시간, 종료 시간, 결과를 기록하십시오

자주 묻는 질문

“예약 및 Cron 기반 에이전트” 강의는 무료인가요?

네 — “예약 및 Cron 기반 에이전트” 전체 내용을 이 웹사이트에서 무료로 읽을 수 있습니다. 인터랙티브하게 실습하려면(내장 코드 에디터와 24/7 AI 튜터), CoddyKit PRO로 업그레이드하면 AI Agents 강의 전체를 잠금 해제할 수 있습니다. AI Agents 강의에는 총 4개의 강의가 포함되어 있습니다.

“예약 및 Cron 기반 에이전트”에서 뭘 배우나요?

APScheduler, cron 작업, 시간에 따라 실행되는 자율 에이전트를 다룹니다. 브라우저에서 직접 실행하는 실습 코드로 AI Agents을(를) 배우며, 24/7 AI 튜터가 강의를 진행하면서 질문에 답변해줍니다.

AI Agents을(를) 시작하는 데 경험이 필요한가요?

사전 경험은 필요하지 않습니다. CoddyKit의 AI Agents은(는) 초급자부터 고급 학습자까지를 위해 구성되어 있으므로, 여기서 시작하거나 처음부터 시작할 수 있으며 자신의 속도대로 진행할 수 있습니다. 이것은 4개 중 3번째 강의입니다.

“예약 및 Cron 기반 에이전트” 강의는 얼마나 걸리나요?

대부분의 CoddyKit 강의는 약 5~10분이 소요됩니다. 각 강의는 간결하고 인터랙티브하여 꾸준한 진행이 가능하며, 웹과 앱에서 중단한 부분부터 바로 시작할 수 있습니다.

이 AI Agents 강의에서 코드를 작성하고 실행할 수 있나요?

네. 모든 AI Agents 강의에는 내장 코드 에디터가 포함되어 있으므로, 브라우저에서 바로 실제 코드를 작성하고 실행한 후 즉시 AI 피드백을 받을 수 있습니다 — 로컬 설정이 필요 없습니다.

이 강의의 모든 강의

← AI Agents(으)로 돌아가기