闹钟不需要你盯着它才会响。
你设好 7:00，去睡觉，去洗澡，去做饭，到点它自己响。这是一件理所当然的事情，理所当然到你从来不会去想：闹钟是怎么在你不在的时候知道时间到了的。
但当你开始构建 Agent 系统的时候，这个问题就不理所当然了。
前面一个版本的 Agent 支持后台执行——耗时几分钟的操作可以异步跑，主线程继续响应用户。但所有操作仍然是人触发的。你说一句，Agent 动一下。"每天早上 9 点跑一次测试"、"每 30 分钟检查一次 CI 状态"——这类任务，没有人去推，Agent 就永远不会动。
本文讲的就是：如何给 Agent 加一个会自己看表的调度器。
问题的本质：触发权在哪里
手动触发和定时触发，表面上是"方便不方便"的差异，本质上是触发权的归属不同。
手动触发的链路是：人 → Agent。触发权在人手里，人不在，链路断掉。
定时触发的链路是：调度器 → Agent。人退出了这条链路，调度器成了触发者。
一旦触发权从人转移到调度器，两个工程问题就浮出来了：
第一，谁来判断"时间到了"？
必须有一个东西持续运行，持续检查当前时间是否匹配某个触发条件。这个东西不能是 Agent Loop 本身——Agent Loop 在没有输入的时候是停着的，它不能一边等用户输入一边还去看表。
第二，判断完之后，怎么把这个信息交给 Agent？
调度器不能直接调用 Agent Loop。一方面，Agent Loop 可能正在处理其他事情；另一方面，调度器和 Agent Loop 跑在不同的线程里，直接调用会有竞态问题。
这两个问题的答案，指向同一个架构决策：调度与执行必须解耦。调度器只管判断时间，执行由另一套机制负责——中间用队列隔开。
四层模型
整个调度系统分四层，职责边界非常清晰：
┌─────────────────────────────────────────┐
│ Layer 1: Scheduler │
│ 独立 daemon 线程，每秒轮询 │
│ 判断时间 → 触发任务写入队列 │
└────────────────────┬────────────────────┘
│ 写入
▼
┌─────────────────────────────────────────┐
│ Layer 2: Queue (cron_queue) │
│ 跨线程通信的唯一通道 │
│ 调度器写，Queue Processor 触发读 │
└────────────────────┬────────────────────┘
│ 检测到有任务
▼
┌─────────────────────────────────────────┐
│ Layer 3: Queue Processor │
│ 独立线程，每 200ms 检查一次 │
│ 条件：队列非空 + Agent 空闲 │
│ 满足则唤醒 Agent Loop │
└────────────────────┬────────────────────┘
│ 唤醒
▼
┌─────────────────────────────────────────┐
│ Layer 4: Consumer (agent_loop) │
│ 从队列消费已触发的任务 │
│ 注入为 user message，交给 LLM 处理 │
└─────────────────────────────────────────┘
每一层只做自己的事：
- Scheduler 只管时间，不管 Agent 在不在、忙不忙
- Queue 是唯一的跨线程通信通道，不做任何逻辑
- Queue Processor 只管"有活 + Agent 空闲"，不管时间
- Consumer 只管消费队列里的内容，不管时间，也不管来源
为什么要四层而不是两层？
因为解耦让每一层可以单独失败而不拖垮其他层。Scheduler 出了一个 bug，Queue Processor 还在跑。Agent Loop 崩了，Scheduler 还在计时，Queue 里的任务还在等着。任何一层的问题都被限制在那一层内。
数据结构：CronJob
每一个定时任务，抽象为一个 CronJob：
@dataclass
class CronJob:
id: str
cron: str # "0 9 * * *" — 五段式 cron 表达式
prompt: str # 触发时注入给 Agent 的消息
recurring: bool # True=周期性，False=一次性
durable: bool # True=持久化到磁盘，跨会话保留
五个字段，每个都有设计理由。
cron 字段用五段式表达式，而不是"每 N 分钟"这种简化接口。
五段式是 Unix cron 用了 50 年的语义：
分钟 小时 日 月 星期
* * * * * 每分钟
0 9 * * * 每天早上 9:00
*/5 * * * * 每 5 分钟
0 9 * * 1-5 工作日早上 9:00
支持 *（任意）、*/N（步进）、N（精确值）、N-M（范围）、N,M,...（列表）。
简化接口表达力有限。"第一个工作日早上 9 点"用五段式写是 0 9 1-7 * 1，用"每 N 分钟"的接口根本表达不了。
recurring 区分两种生命周期。
周期性任务触发后继续保留在 scheduled_jobs 里，等待下一次触发。一次性任务触发后立即从 scheduled_jobs 删除。这个区别很重要，不能靠业务逻辑去判断，必须在数据结构层面固化。
durable 定义持久化边界，也定义了一个容易被误解的概念。
durable=True 的任务，定义会写入磁盘（.scheduled_tasks.json）。Agent 重启后加载文件，任务恢复。
但有一点必须说清楚：durable 不等于"进程关闭后也能定时触发"。调度器跑在 Agent 进程内。进程关闭，调度线程跟着退，没有任何东西在看表。Durable 只意味着任务定义跨重启保留——下次启动 Agent，调度器才重新接管。
如果需要"即使应用关闭也能定时跑"，那是系统层面的问题，用 crontab 或 systemd timer，不是应用内调度器能解决的。
cron 表达式匹配：实现细节
匹配逻辑是整个调度器里细节最多的部分，也是最容易被误实现的部分。
单字段匹配比较直接：
def _cron_field_matches(field: str, value: int) -> bool:
if field == "*":
return True
if field.startswith("*/"):
step = int(field[2:])
return step > 0 and value % step == 0
if "," in field:
return any(_cron_field_matches(f.strip(), value)
for f in field.split(","))
if "-" in field:
lo, hi = field.split("-", 1)
return int(lo) <= value <= int(hi)
return value == int(field)
五段式整体匹配的核心逻辑：
def cron_matches(cron_expr: str, dt: datetime) -> bool:
fields = cron_expr.strip().split()
if len(fields) != 5:
return False
minute, hour, dom, month, dow = fields
# Python 的 weekday()：Monday=0，cron 语义：Sunday=0
dow_val = (dt.weekday() + 1) % 7
m = _cron_field_matches(minute, dt.minute)
h = _cron_field_matches(hour, dt.hour)
dom_ok = _cron_field_matches(dom, dt.day)
month_ok = _cron_field_matches(month, dt.month)
dow_ok = _cron_field_matches(dow, dow_val)
if not (m and h and month_ok):
return False
dom_unconstrained = dom == "*"
dow_unconstrained = dow == "*"
if dom_unconstrained and dow_unconstrained:
return True
if dom_unconstrained:
return dow_ok
if dow_unconstrained:
return dom_ok
return dom_ok or dow_ok
这里有两个值得重点讲的地方。
Python weekday 的转换。
Python 的 datetime.weekday() 返回 Monday=0，Sunday=6。而标准 cron 语义是 Sunday=0，Saturday=6。差一位，不转换就会算错。
转换公式是 (dt.weekday() + 1) % 7：Python 的 Monday(0) → cron 的 1，Python 的 Sunday(6) → cron 的 0。
DOM 和 DOW 的 OR 语义。
这是标准 cron 里最容易被误实现的地方。
当日（DOM）和星期（DOW）同时被约束时（两者都不是 *），标准 cron 的语义是 OR，不是 AND。
比如 0 9 1 * 1 表示：每月 1 号早上 9 点，或者每个周一早上 9 点。不是"每月 1 号且是周一的早上 9 点"。
如果你按 AND 实现，0 9 1 * 1 就变成了每年只触发几次（凑巧周一是每月 1 号的时候），而不是预期的每月 1 号或每周一。这个 bug 在测试里很难发现，因为触发频率低。
代码里的处理：两者都有约束时，dom_ok or dow_ok。如果只有一方被约束（另一方是 *），就只用那一方的结果。
调度线程：独立 daemon，每秒轮询
调度器是整个系统里第一个开口说话的人。它不等任何人叫它，它自己在后台数着秒。
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 error] {job.id}: {e}")
threading.Thread(target=cron_scheduler_loop, daemon=True).start()
三个设计决策值得细说。
date-aware minute_marker，不是简单的 HH:MM。
_last_fired 记录每个 job 上次触发的时间标记，用来防止同一分钟内因为每秒轮询而重复触发。
标记用 "%Y-%m-%d %H:%M" 而不是 "%H:%M"。
差别在哪里？如果用 "%H:%M"，每天的 09:00 在 _last_fired 里的值是一样的，都是 "09:00"。那么第一天 9 点触发后，第二天 9 点检查，_last_fired["job_id"] 还是 "09:00"，等于已触发，直接跳过。
每日任务就这样静默地在第二天被跳过了，日志里什么也看不到。这个 bug 非常隐蔽，因为第一天是正常的。
加上日期之后，"2025-01-01 09:00" 和 "2025-01-02 09:00" 是不同的字符串，问题消失。
单 job try/except：一个坏任务不能杀死整个调度线程。
cron_matches 里如果某个任务的 cron 表达式有问题，或者任务的其他字段有异常，try/except 会捕获它，打印日志，然后继续处理下一个任务。
如果没有这个保护，一个坏任务会让整个 for 循环崩掉，调度线程停止工作，所有任务都不再触发。
daemon=True：正确的生命周期设计，不是 workaround。
daemon 线程的行为是：主线程退出时，所有 daemon 线程跟着退，不等它们完成。
这正是调度线程需要的语义。调度线程是 Agent 进程的附属，Agent 退了，调度器没有理由继续存在。如果调度线程是非 daemon 的，主线程退出后进程会卡在那里等调度线程，而调度线程在无限循环，永远不会主动退出。
Queue Processor：Agent 空闲检测
Queue Processor 是第三层，跑在独立线程里，每 200ms 检查一次：
def queue_processor_loop():
while True:
time.sleep(0.2)
if not has_cron_queue():
continue
if not agent_lock.acquire(blocking=False):
continue
try:
if not has_cron_queue():
continue
print("[queue processor] delivering scheduled work")
run_agent_turn_locked()
finally:
agent_lock.release()
blocking=False 是这里最值得注意的细节。
agent_lock.acquire(blocking=False) 的语义是：尝试获取锁，拿不到就立即返回 False，不等待。
如果用阻塞模式（blocking=True，也是默认值），当 Agent 正在处理用户输入时，Queue Processor 会卡在 acquire() 这一行，直到 Agent 空闲。卡着的时候，200ms 的轮询周期失效，下一个定时任务到来时这个线程根本跑不到检查的地方。
非阻塞模式下，拿不到锁就放弃，200ms 后再来。每次循环都是轻量的，不会积压。
另一个细节：进入 try 块后，有一个二次检查 if not has_cron_queue()。原因是在 Queue Processor 获取锁的过程中（即使是非阻塞的，也有极短的时间差），cron_queue 可能已经被上一轮消费掉了。二次检查避免了空跑一次 Agent Loop。
Agent Loop 消费队列的逻辑很简单：
fired = consume_cron_queue()
for job in fired:
messages.append({
"role": "user",
"content": f"[Scheduled] {job.prompt}"
})
定时任务被注入为 user message，从 LLM 的视角看，它和用户输入没有区别，只是内容前面有一个 [Scheduled] 标记。
持久化：跨重启保留任务定义
Durable 任务写入 .scheduled_tasks.json：
def save_durable_jobs():
durable = [asdict(j) for j in scheduled_jobs.values() if j.durable]
DURABLE_PATH.write_text(json.dumps(durable, indent=2))
启动时加载：
def load_durable_jobs():
if not DURABLE_PATH.exists():
return
try:
jobs = json.loads(DURABLE_PATH.read_text())
for j in jobs:
job = CronJob(**j)
err = validate_cron(job.cron)
if err:
print(f"[cron] skipping invalid job {job.id}: {err}")
continue
scheduled_jobs[job.id] = job
except Exception:
pass
加载时做校验而不是直接信任磁盘文件，是因为文件可能被手动编辑过，或者由旧版本写入。跳过非法表达式而不是整体失败，原因和单 job try/except 一样：一个坏任务不能阻止其他任务恢复。
注册时也先校验：
def schedule_job(cron: str, prompt: str,
recurring: bool = True, durable: bool = True) -> CronJob | str:
err = validate_cron(cron)
if err:
return err # 返回错误字符串，让 LLM 看到并告知用户
job = CronJob(
id=f"cron_{random.randint(0, 999999):06d}",
cron=cron, prompt=prompt,
recurring=recurring, durable=durable,
)
with cron_lock:
scheduled_jobs[job.id] = job
if durable:
save_durable_jobs()
return job
注册失败返回字符串而不是抛异常，是为了让 LLM 能读到错误内容，自行纠正 cron 表达式后重试。
整体运行时序
把四层拼在一起，一次定时任务从注册到执行的完整时序如下：
启动：
load_durable_jobs()
Thread(cron_scheduler_loop, daemon=True).start()
Thread(queue_processor_loop, daemon=True).start()
注册：
schedule_cron("*/2 * * * *", "check CI status", durable=True)
→ CronJob 写入 scheduled_jobs
→ 写入 .scheduled_tasks.json
每 2 分钟：
cron_scheduler_loop 检测到 cron_matches 返回 True
→ cron_queue.append(job)
→ minute_marker 更新，防重复
queue_processor_loop 检测到 has_cron_queue() = True
→ agent_lock.acquire(blocking=False) 成功
→ run_agent_turn_locked()
→ agent_loop 调用 consume_cron_queue()
→ messages.append({"role": "user", "content": "[Scheduled] check CI status"})
→ LLM 收到消息，处理任务
进程退出：
daemon 线程跟着退
.scheduled_tasks.json 留在磁盘
下次启动：
load_durable_jobs() → 任务恢复
调度器重新开始计时
生产系统做了什么
当前版本实现的是最小可用的调度器。Claude Code 在这个基础上加了几个值得学的生产级设计。
抖动（Jitter）。
如果 100 个任务都设置了 0 * * * *（每小时整点触发），它们会在同一秒全部进入队列，瞬间打出 100 个 LLM 请求。这叫惊群效应（thundering herd）。
Claude Code 的处理方式是给每个任务加一个随机延迟，但不是真正的随机——用任务 ID 做哈希，生成确定性的延迟值。好处是同一个任务每次触发的延迟是固定的，不会因为抖动而在不同时间偏移。重复任务的延迟上限是触发周期的 10%（最多 15 分钟），一次性任务最多提前 90 秒。
自动过期。
重复任务默认 7 天后自动过期，过期前触发最后一次，然后删除。防止用户忘记的僵尸任务无限积压。上限可以配置，但不超过 30 天。
作业数上限。
MAX_JOBS = 50。超限时直接报错："Too many scheduled jobs (max 50). Cancel one first." 不是静默丢弃，不是队列积压，就是返回错误让用户处理。上限的存在本身是一种设计意图的表达：调度器是为少量长期任务设计的，不是批处理引擎。
多 session 锁。
同一个项目可能被多个 IDE session 同时打开。如果每个 session 里都有调度器，同一个 durable 任务会被触发多次。Claude Code 用 .scheduled_tasks.lock 文件实现互斥：谁持有锁谁触发 durable 任务，session-only 任务各自触发。
QoS 降级。
cron 触发的 API 请求被标记为 WORKLOAD_CRON，在服务端容量紧张时以更低的优先级处理。背后的逻辑很清楚：定时任务通常不是用户正在等待的事情，可以等一等；交互式请求不能等。
这五点背后分别对应：惊群、资源泄漏、滥用、并发冲突、资源竞争。每一点都是从真实问题里蒸馏出来的，不是过度设计。
小结
给 Agent 加调度器，本质上是把触发权从人手里转移给机器。这件事听起来简单，实现起来有三个坑最容易踩：
一、调度和执行没有解耦，调度器直接调 Agent Loop，竞态问题随即出现。
二、minute_marker 没有带日期，每日任务从第二天起静默消失。
三、DOM/DOW 用了 AND 语义，周期性任务的触发频率远低于预期，而且没有明显报错。
这三个坑都是在代码里很难被测试发现的——它们需要时间才会暴露，通常是上线一两天之后。
四层模型把问题拆清楚之后，每一层的实现都很简单。复杂度来自层间的协调，而不是单层的逻辑。这是设计上应该追求的状态。
下一篇讲 Agent Teams：当任务大到一个 Agent 装不下，需要组队的时候，持久队友和异步收件箱怎么做。本系列内容由云栈社区持续更新，欢迎关注更多技术分享。
发表于 1 小时前
|
查看: 3|
回复: 0
