0Pricing
AI Agents · 课时

基于调度与 Cron 的代理

使用 APScheduler、Cron 任务和时间触发的自主代理执行。

基于调度与 Cron 的代理 是 CoddyKit 上的免费 AI Agents 课时。 这是第 3 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 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')

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

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

处理作业异常

请使用 try/except 包装已调度的作业函数，防止某次运行失败后悄然停止未来的运行。记录失败，并根据需要发送提醒。

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 中使用 BackgroundScheduler 和 lifespan 上下文，使调度器能够随服务器生命周期平稳启动和停止。

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 的代理」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Agents 课程的其余内容，请升级到 CoddyKit PRO。 AI Agents 课程共包含 4 节课。

「基于调度与 Cron 的代理」这节课中我会学到什么？

使用 APScheduler、Cron 任务和时间触发的自主代理执行。 你通过在浏览器中直接运行的动手代码来练习 AI Agents，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 AI Agents 需要有经验吗？

无需任何先前经验。CoddyKit 上的 AI Agents 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 3 节课，共 4 节。

「基于调度与 Cron 的代理」课时需要多长时间？

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。

我能在这节 AI Agents 课中编写并运行代码吗？

能。每节 AI Agents 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

  1. 触发器—操作代理模式
  2. 将代理连接到 Webhook
  3. 基于调度与 Cron 的代理
  4. 构建多应用自动化流程
← 返回 AI Agents