Zeitplanung und Cron-basierte Agenten
APScheduler, Cron-Jobs und zeitgesteuerte autonome Agentenausführung.
Zeitplanung und Cron-basierte Agenten ist eine kostenlose AI Agents-Lektion auf CoddyKit. Dies ist Lektion 3 von 4. Du kannst die komplette Lektion unten kostenlos lesen – dann übst du sie direkt im Browser mit einem integrierten Code-Editor und einem KI-Tutor rund um die Uhr. Sie ist Teil des AI Agents-Lernpfads, und dein Fortschritt wird über Web und CoddyKit-App synchronisiert. Der AI Agents-Kurs umfasst insgesamt 4 Lektionen.
Warum Agenten zeitgesteuert ausführen?
Manche Agentenaufgaben müssen nicht kontinuierlich oder auf Abruf ausgeführt werden — sie laufen nach einem Zeitplan. Tagesberichte, wöchentliche Zusammenfassungen, stündliche Datenprüfungen und regelmäßige Bereinigungen eignen sich gut für zeitgesteuerte Agenten.
APScheduler-Grundlagen
APScheduler (Advanced Python Scheduler) führt Jobs innerhalb Ihres Python-Prozesses aus. Es unterstützt drei Triggertypen: date (einmalig), interval (wiederholt) und cron (kalenderbasiert).
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')Syntax von Cron-Triggern
Der Cron-Trigger verwendet vertraute Cron-Felder: year, month, day, week, day_of_week, hour, minute, second. Sie können Zahlen, Bereiche, Listen und Platzhalter verwenden.
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()))Syntax von Cron-Ausdrücken
Standard-Cron-Ausdrücke verwenden 5 Felder: minute hour day month weekday. APScheduler akzeptiert diese auch als einzelnen String mit CronTrigger.from_crontab().
0 9 * * *— täglich um 9 Uhr0 9 * * 1— montags um 9 Uhr*/15 * * * *— alle 15 Minuten
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}')Intervall-Trigger
Der Intervall-Trigger führt einen Job alle N Zeiteinheiten aus. Verwenden Sie ihn für Polling, Heartbeats oder Aufgaben, die sich in einem festen Rhythmus wiederholen sollen.
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')Job-Parameter und IDs
Weisen Sie Jobs IDs zu, damit Sie später auf sie verweisen, sie pausieren oder entfernen können. Übergeben Sie Argumente an die Jobfunktion mit args oder 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')Jobs über Neustarts hinweg speichern
Standardmäßig speichert APScheduler Jobs im Arbeitsspeicher, sodass sie bei einem Neustart verloren gehen. Verwenden Sie einen SQLAlchemy-Jobstore, um Jobs in einer Datenbank zu speichern und über Neustarts hinweg zu erhalten.
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')Ausnahmen in Jobs behandeln
Umschließen Sie Funktionen für geplante Jobs mit try/except, damit ein fehlgeschlagener Lauf zukünftige Läufe nicht unbemerkt beendet. Protokollieren Sie Fehler und senden Sie optional Benachrichtigungen.
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()Überlappungen von Jobs verhindern
Wenn ein Job länger als sein Intervall läuft, kann der nächste Lauf beginnen, bevor der vorherige beendet ist. Setzen Sie max_instances=1 (die Standardeinstellung) oder verwenden Sie Coalescing, um verpasste Läufe zu überspringen.
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)')Zeitplanung in FastAPI integrieren
Verwenden Sie den BackgroundScheduler von APScheduler in FastAPI zusammen mit einem Lifespan-Kontext, um den Scheduler passend zum Lebenszyklus des Servers ordnungsgemäß zu starten und zu beenden.
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()
]Zeitzonenabhängige Zeitplanung
Geben Sie für zeitgesteuerte Agenten stets Zeitzonen an. Cronjobs ohne Zeitzonenkontext können nach Änderungen an der Sommerzeit zur falschen Zeit ausgeführt werden.
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')Wissensüberprüfung: Zeitplanung
Testen Sie Ihr Verständnis von APScheduler und Cron-basierten Agenten.
Best Practices für die Zeitplanung
Wichtige Regeln für zuverlässige zeitgesteuerte Agenten:
- Verwenden Sie stets eine zeitzonenabhängige Zeitplanung
- Speichern Sie Jobs in einer Datenbank, damit sie Neustarts überstehen
- Setzen Sie
max_instances=1für lang laufende Jobs - Fügen Sie Fehler-Listener hinzu, um unbemerkte Fehler zu erkennen
- Protokollieren Sie jeden geplanten Lauf mit Startzeit, Endzeit und Ergebnis
Häufig gestellte Fragen
Ist die Lektion „Zeitplanung und Cron-basierte Agenten“ kostenlos?
Ja — der vollständige Text von „Zeitplanung und Cron-basierte Agenten“ ist hier im Web kostenlos zu lesen. Um sie interaktiv zu üben (integrierter Code-Editor und 24/7 KI-Tutor) und den Rest des AI Agents-Kurses freizuschalten, upgrade auf CoddyKit PRO. Der AI Agents-Kurs umfasst insgesamt 4 Lektionen.
Was lerne ich in „Zeitplanung und Cron-basierte Agenten“?
APScheduler, Cron-Jobs und zeitgesteuerte autonome Agentenausführung. Du übst AI Agents mit praktischem Code, den du direkt im Browser ausführst, und ein 24/7 KI-Tutor beantwortet deine Fragen während du die Lektion bearbeitest.
Brauche ich Erfahrung, um AI Agents zu starten?
Keine Vorkenntnisse erforderlich. AI Agents auf CoddyKit ist für Anfänger bis fortgeschrittene Lernende strukturiert, sodass du hier starten oder von Anfang an beginnen und in deinem eigenen Tempo voranschreiten kannst. Dies ist Lektion 3 von 4.
Wie lange dauert die Lektion „Zeitplanung und Cron-basierte Agenten“?
Die meisten CoddyKit-Lektionen dauern etwa 5–10 Minuten. Jede ist kompakt und interaktiv, sodass du stetig Fortschritte machst und genau dort weitermachst, wo du aufgehört hast – im Web und in der App.
Kann ich in dieser AI Agents-Lektion Code schreiben und ausführen?
Ja. Jede AI Agents-Lektion enthält einen integrierten Code-Editor, sodass du echten Code direkt in deinem Browser schreibst und ausführst und sofort KI-Feedback erhältst — ohne lokale Einrichtung erforderlich.
Alle Lektionen in diesem Kurs
- Muster für Trigger-Action-Agenten
- Agenten mit Webhooks verbinden
- Zeitplanung und Cron-basierte Agenten
- Eine Multi-App-Automatisierungspipeline erstellen