第 14 章 / 共 21 章
定时调度：让 Agent 自己醒过来
上一章留下的问题：后台任务解决了”不阻塞”，但触发方式仍然是人。你想要”每天早上九点跑一次测试并把失败汇总发我”，现在只能每天早上自己打开终端说一句。
有人会想：让模型记住不就行了？不行——模型没有时钟，也没有在你不说话时自己运行的能力。周期性工作应该由 harness 创建，而不是让模型记住。
14.1 闹钟模型
一个闹钟为什么可靠？因为它有一个独立于你的计时器，时间到了就响，不管你在不在看它。照这个思路搭四层：
| 层 | 干什么 | 频率 |
|---|---|---|
| 调度线程 | 每秒检查一遍所有任务的 cron 表达式是否匹配当前时间 | 1 秒 |
| 队列 | 匹配上的任务写进 cron_queue | — |
| 队列处理器 | 队列非空且 Agent 空闲时，启动一轮 Agent | 0.2 秒 |
| 消费层 | 循环从队列取出任务，作为 user 消息注入 | 每轮 |
14.2 数据结构与 cron 匹配
from dataclasses import dataclass
@dataclass
class CronJob:
id: str
cron: str # 标准五段式："0 9 * * *"
prompt: str # 触发时注入的内容
recurring: bool # True 重复，False 一次性
durable: bool # True 落盘，重启后恢复五段依次是：分 时 日 月 周。支持 *（任意）、*/N（每 N 个）、N（具体值）、N-M（区间）、N,M（列表）。
| 表达式 | 含义 |
|---|---|
0 9 * * * | 每天 9 |
*/5 * * * * | 每 5 分钟 |
0 9 * * 1-5 | 工作日 9 |
30 18 1 * * | 每月 1 号 18 |
匹配函数里有一处反直觉的地方：
def cron_matches(cron_expr: str, dt) -> bool:
fields = cron_expr.strip().split()
if len(fields) != 5:
return False
minute, hour, dom, month, dow = fields
dow_val = (dt.weekday() + 1) % 7 # Python 周一=0，cron 周日=0
if not (_field(minute, dt.minute) and _field(hour, dt.hour) and _field(month, dt.month)):
return False
dom_free, dow_free = dom == "*", dow == "*"
if dom_free and dow_free:
return True
if dom_free:
return _field(dow, dow_val)
if dow_free:
return _field(dom, dt.day)
return _field(dom, dt.day) or _field(dow, dow_val) # 两者都限定时是「或」最后那行的”或”是 Unix cron 的标准语义，也是最常被搞错的一点：当日期和星期都被限定时，满足任意一个就触发，不是两个都要满足。0 9 1 * 1 的意思是”每月 1 号或每周一的 9 点”，不是”1 号且是周一”。
14.3 调度线程
import time, threading
from datetime import datetime
scheduled_jobs: dict[str, CronJob] = {}
cron_queue: list[CronJob] = []
cron_lock = threading.Lock()
_last_fired: dict[str, str] = {}
def cron_scheduler_loop():
while True:
time.sleep(1)
now = datetime.now()
minute_marker = now.strftime("%Y-%m-%d %H:%M")
with cron_lock:
for job in list(scheduled_jobs.values()):
try:
if cron_matches(job.cron, now):
if _last_fired.get(job.id) != minute_marker: # 同一分钟只触发一次
cron_queue.append(job)
_last_fired[job.id] = minute_marker
if not job.recurring:
scheduled_jobs.pop(job.id, None)
if job.durable:
save_durable_jobs()
except Exception as e:
print(f"[cron 错误] {job.id}: {e}") # 单个任务出错不影响其他两处细节值得停一下：
**minute_marker 防重复触发。**线程每秒跑一次，一分钟内会检查 60 次，cron_matches 都会返回 True。用”年-月-日 时
**每个任务单独 try/except。**一个写错的 cron 表达式不该把整个调度线程干掉——那会导致所有定时任务集体静默失效，而且很难发现。
14.4 队列处理器与消费
agent_lock = threading.Lock()
def queue_processor_loop():
while True:
time.sleep(0.2)
with cron_lock:
has_work = bool(cron_queue)
if not has_work:
continue
if not agent_lock.acquire(blocking=False): # Agent 正忙，等下一轮
continue
try:
run_agent_turn_locked() # Agent 空闲，启动一轮
finally:
agent_lock.release()循环里消费：
with cron_lock:
fired, cron_queue[:] = list(cron_queue), []
for job in fired:
messages.append({"role": "user", "content": f"[定时任务] {job.prompt}"})那个 [定时任务] 前缀是给模型看的：它需要知道这条输入不是人打的，而是系统按计划注入的——两者该有不同的应答风格（比如不需要说”好的，我这就去做”）。
14.5 走一遍：注册到触发
你：每天早上 9 点跑一次测试，如果有失败就整理成清单给我。
Agent：调用 schedule_cron(cron="0 9 * * *",
prompt="运行测试套件，把失败用例整理成清单",
recurring=True, durable=True)
→ "已创建定时任务 cron_0001，每天 09:00 执行"第二天 09:00
，调度线程匹配上，任务入队；队列处理器发现 Agent 空闲，启动一轮；循环取出任务，注入[定时任务] 运行测试套件……；Agent 跑测试、整理清单、输出结果。全程没有人参与。
14.6 一个必须说清楚的限制
durable=True 的含义常被误解。它保证的是任务定义落盘到 .scheduled_tasks.json，进程重启后能被重新加载。
它不保证进程关闭期间任务会执行。调度线程活在你的 Agent 进程里，进程没跑，就没人看时间。晚上关了电脑，早上 9 点的任务不会执行；你 10 点开机，它也不会补跑（因为那一分钟已经过去了）。
要做到真正与应用状态无关的定时执行，得把触发权交给操作系统——系统的 crontab、systemd timer，或者一个常驻的服务端调度器。教学版的定时器是”应用内闹钟”，不是”操作系统闹钟”。
14.7 工程现实
生产系统在这一层还会做这些事：
- 文件锁防止同一项目的多个会话重复触发同一个任务；
- 抖动：重复任务按周期的一定比例（有上限）随机延后，避免所有实例在整点同时打过来；
- 任务数量上限（比如 50 个）和自动过期（比如重复任务默认 7 天后失效），防止无人清理的任务无限累积；
- 优先级标记：定时触发的请求打上标签，在容量紧张时以较低优先级服务，不影响交互式请求；
- 时区：表达式按本地时区解释，跨时区团队要显式约定。