0Pricing
AI Agents · Lekcja

Agenci planowani i oparte na Cronie

APScheduler, zadania cron i autonomiczne wykonywanie agentów wyzwalane czasowo.

Agenci planowani i oparte na Cronie to bezpłatna lekcja AI Agents na CoddyKit. To lekcja 3 z 4. Możesz przeczytać całą lekcję poniżej za darmo — a potem ćwiczyć ją interaktywnie w przeglądarce z wbudowanym edytorem kodu i tutorem AI dostępnym 24/7. To część ścieżki edukacyjnej AI Agents, a Twój postęp synchronizuje się między webem a aplikacją CoddyKit. Kurs AI Agents zawiera 4 lekcji w sumie.

Dlaczego planować pracę agentów?

Niektóre zadania agentów nie muszą działać stale ani na żądanie — uruchamiają się zgodnie z harmonogramem. Raporty dzienne, podsumowania tygodniowe, cogodzinne sprawdzanie danych i okresowe czyszczenie to dobre zastosowania dla agentów uruchamianych według harmonogramu.

Podstawy APScheduler

APScheduler (Advanced Python Scheduler) uruchamia zadania wewnątrz procesu Python. Obsługuje trzy rodzaje wyzwalaczy: date (jednorazowy), interval (powtarzający się) i cron (oparty na kalendarzu).

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

Składnia wyzwalacza cron

Wyzwalacz cron korzysta ze znanych pól cron: year, month, day, week, day_of_week, hour, minute, second. Można używać liczb, zakresów, list i symboli wieloznacznych.

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

Składnia wyrażeń cron

Standardowe wyrażenia cron korzystają z 5 pól: minute hour day month weekday. APScheduler akceptuje je również jako pojedynczy ciąg znaków za pomocą CronTrigger.from_crontab().

  • 0 9 * * * — codziennie o 9:00
  • 0 9 * * 1 — w każdy poniedziałek o 9:00
  • */15 * * * * — co 15 minut
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}')

Wyzwalacz interwałowy

Wyzwalacz interwałowy uruchamia zadanie co N jednostek czasu. Należy używać go do odpytywania, wysyłania sygnałów heartbeat lub dowolnego zadania, które powinno powtarzać się ze stałą częstotliwością.

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

Parametry i identyfikatory zadań

Należy przypisywać zadaniom identyfikatory, aby można było później się do nich odwoływać, wstrzymywać je lub usuwać. Argumenty do funkcji zadania należy przekazywać za pomocą args lub 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')

Utrwalanie zadań po ponownym uruchomieniu

Domyślnie APScheduler przechowuje zadania w pamięci, więc są one tracone po ponownym uruchomieniu. Należy użyć magazynu zadań SQLAlchemy, aby utrwalać zadania w bazie danych i zachować je po restarcie.

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

Obsługa wyjątków zadań

Należy opakować funkcje zaplanowanych zadań w konstrukcję try/except, aby jedno nieudane uruchomienie nie zatrzymało po cichu kolejnych. Błędy należy rejestrować i w razie potrzeby wysyłać powiadomienia.

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

Zapobieganie nakładaniu się zadań

Jeśli wykonanie zadania trwa dłużej niż jego interwał, kolejne uruchomienie może rozpocząć się przed zakończeniem poprzedniego. Należy ustawić max_instances=1 (wartość domyślna) albo użyć mechanizmu coalescing, aby pomijać opuszczone uruchomienia.

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

Integracja harmonogramowania z FastAPI

Należy użyć BackgroundScheduler wewnątrz FastAPI, korzystając z kontekstu lifespan, aby prawidłowo uruchamiać i zatrzymywać harmonogram zgodnie z cyklem życia serwera.

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

Harmonogramowanie uwzględniające strefę czasową

Należy zawsze określać strefy czasowe dla agentów uruchamianych według harmonogramu. Zadania cron bez określonej strefy czasowej mogą uruchamiać się o niewłaściwych porach po przejściu na czas letni lub zimowy.

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

Sprawdzenie wiedzy: harmonogramowanie

Sprawdź swoją wiedzę na temat APScheduler i agentów opartych na cron.

Najlepsze praktyki harmonogramowania

Najważniejsze zasady niezawodnych agentów uruchamianych według harmonogramu:

  • Zawsze stosuj harmonogramowanie uwzględniające strefę czasową
  • Utrwalaj zadania w bazie danych, aby przetrwały ponowne uruchomienie
  • Ustaw max_instances=1 dla zadań długotrwałych
  • Dodawaj listenery błędów, aby wykrywać awarie następujące po cichu
  • Rejestruj każde zaplanowane uruchomienie wraz z czasem rozpoczęcia, czasem zakończenia i wynikiem

Często zadawane pytania

Czy lekcja „Agenci planowani i oparte na Cronie” jest bezpłatna?

Tak — pełny tekst „Agenci planowani i oparte na Cronie” jest dostępny za darmo tutaj w sieci. Aby ćwiczyć ją interaktywnie (wbudowany edytor kodu i tutor AI dostępny 24/7) i odblokować resztę kursu AI Agents, przejdź na CoddyKit PRO. Kurs AI Agents zawiera 4 lekcji w sumie.

Co nauczysz się w „Agenci planowani i oparte na Cronie”?

APScheduler, zadania cron i autonomiczne wykonywanie agentów wyzwalane czasowo. Ćwiczysz AI Agents z praktycznym kodem, który uruchamiasz bezpośrednio w przeglądarce, a tutor AI dostępny 24/7 odpowiada na Twoje pytania podczas pracy nad lekcją.

Czy potrzebuję doświadczenia, aby zacząć AI Agents?

Nie wymagamy żadnego doświadczenia. AI Agents w CoddyKit jest strukturyzowany dla początkujących i zaawansowanych użytkowników, więc możesz zacząć tutaj lub od początku i uczyć się w swoim tempie. To lekcja 3 z 4.

Ile czasu zajmuje lekcja „Agenci planowani i oparte na Cronie”?

Większość lekcji CoddyKit trwa około 5–10 minut. Każda lekcja to mały, interaktywny krok, dzięki czemu robisz systematyczne postępy i zawsze wracasz dokładnie do tego samego miejsca — na webie i w aplikacji.

Czy mogę pisać i uruchamiać kod w tej lekcji AI Agents?

Tak. Każda lekcja AI Agents zawiera wbudowany edytor kodu, więc piszesz i uruchamiasz prawdziwy kod bezpośrednio w przeglądarce i od razu otrzymujesz sprzężenie zwrotne od AI — bez konfiguracji na komputerze.

Wszystkie lekcje w tym kursie

  1. Wzorce agentów trigger-action
  2. Łączenie agentów z webhookami
  3. Agenci planowani i oparte na Cronie
  4. Budowanie potoku automatyzacji obejmującego wiele aplikacji
← Powrót do AI Agents