OpenClaw Cron 深度解读让 AI Agent 学会自主定时工作一句话总结OpenClaw 的 Cron 系统让 AI Agent 具备了设闹钟的能力——不仅能定时提醒用户还能自己悄悄去执行后台任务干完活再汇报结果。 为什么 Agent 需要定时任务想象一下这个场景你让 AI 助手帮你每天早上9点检查一下服务器状态。传统的做法是什么你得自己设个闹钟到点了打开对话框再敲一遍帮我检查服务器。这跟没有 AI 助手有什么区别真正智能的 Agent 应该能够自主调度记住用户的需求到点自动执行后台执行不打扰用户悄悄干活主动汇报干完了告诉你结果这就是 OpenClaw Cron 系统要解决的问题。它让 Agent 从被动响应升级为主动服务。️ 系统架构总览图1Cron 系统由三个核心组件构成——CronStore 负责持久化、CronOps 处理增删改查、CronTimer 调度执行。最关键的是 executeJob() 执行引擎它决定任务是注入主会话还是启动独立会话。整个 Cron 系统的设计思路很清晰组件职责关键方法CronStore持久化存储load / saveCronOpsCRUD 操作add / remove / updateCronTimer定时调度armTimer / onTimerexecuteJob任务执行main vs isolated这套架构的精妙之处在于一个定时器管所有任务。不是给每个任务都开一个定时器那样内存会爆而是只维护一个指向最近要执行的任务的定时器。每次触发后再计算下一个最近的任务。 三种调度类型图2at 用于一次性提醒every 用于周期性任务cron 则支持复杂的时间表达式。at一次性定时{kind:at,atMs:Date.now()3600_000}// 1小时后执行这是最简单的调度——在指定时间点执行一次。执行完成后job.enabled自动设为false任务就算完结了。适用场景“30分钟后提醒我开会”“明天早上8点叫我起床”every间隔执行{kind:every,everyMs:60_000,anchorMs?:1706745600000}固定间隔重复执行。有个可选的anchorMs参数很有意思——它是对齐锚点。比如你想让任务每小时整点执行而不是从现在开始每小时执行就可以设置一个整点时间戳作为锚点。计算公式是nextRun anchor Math.ceil((now - anchor) / everyMs) * everyMscron表达式调度{kind:cron,expr:0 9 * * 1-5,tz:Asia/Shanghai}标准 cron 格式还支持时区设置。上面这个表达式的意思是工作日每天早9点上海时间。cron 表达式的格式秒 分 时 日 月 周位置含义示例1分钟0-592小时0-233日期1-314月份1-125星期0-7 (0和7都是周日)OpenClaw 使用croniter库解析表达式。这个库在 Python 生态里很成熟处理各种边界情况比如闰年、夏令时都很稳定。⚡ 两种执行模式Main vs Isolated这是 Cron 系统最有趣的设计。同样是定时任务执行方式完全不同Main Session注入主会话{sessionTarget:main,payload:{kind:systemEvent,text:每日提醒检查邮件},wakeMode:now}任务不是执行而是注入。系统把消息塞进主会话的消息队列就像有人在对话框里发了一条系统消息。这种模式适合简单提醒不需要 Agent 做复杂操作需要用户看到并响应的任务依赖现有上下文的任务wakeMode参数控制是否立即触发 Agent 心跳now立刻触发Agent 马上处理这条消息next-heartbeat等下次自然心跳时再处理Isolated Session独立会话执行{sessionTarget:isolated,payload:{kind:agentTurn,message:检查服务器健康状态并生成报告,model:claude-3-5-sonnet,timeoutSeconds:300,deliver:true,channel:telegram,to:user123}}这才是真正的后台执行。系统会启动一个全新的 Agent 会话专门执行这个任务。执行完后把结果汇报回主会话。这种模式适合复杂任务需要多轮思考和工具调用耗时任务用户不想等不需要用户介入的任务关键参数解释参数作用model指定执行任务的模型timeoutSeconds超时限制deliver是否把结果推送给用户channel推送渠道telegram/email/…to收件人 定时器核心逻辑OpenClaw 的定时器实现有几个精巧的细节单一定时器模式constMAX_TIMEOUT_MS2**31-1;// JS setTimeout 最大值exportfunctionarmTimer(state:CronServiceState){// 1. 清除旧定时器if(state.timer)clearTimeout(state.timer);state.timernull;// 2. 找到最近的待执行时间constnextAtnextWakeAtMs(state);if(!nextAt)return;// 3. 设置新定时器注意延迟上限constdelayMath.max(nextAt-state.deps.nowMs(),0);constclampedDelayMath.min(delay,MAX_TIMEOUT_MS);state.timersetTimeout((){voidonTimer(state);},clampedDelay);state.timer.unref?.();// 允许进程在定时器未触发时退出}几个要点单一定时器永远只有一个活跃的定时器指向最近的任务。这避免了定时器泛滥。延迟上限处理JavaScript 的setTimeout最大只支持约 24.8 天2^31-1 毫秒。如果任务在更远的未来先设一个最大延迟到时候再重新计算。unref 调用timer.unref()让这个定时器不阻止 Node.js 进程退出。如果用户关闭了应用不会因为还有待执行的定时任务而卡住。并发控制exportasyncfunctiononTimer(state:CronServiceState){if(state.running)return;// 防止并发执行state.runningtrue;try{awaitlocked(state,async(){awaitensureLoaded(state);awaitrunDueJobs(state);awaitpersist(state);armTimer(state);});}finally{state.runningfalse;}}用一个简单的running标志位防止重入。如果定时器触发时上一次执行还没结束直接跳过。 任务执行流程图3任务执行的完整流程——从 Timer 触发到最终完成中间根据 sessionTarget 分叉为 Main 和 Isolated 两条路径。让我们跟踪一个完整的执行流程阶段1Timer 触发定时器到期onTimer()被调用。阶段2筛选到期任务constduejobs.filter(jj.enabledj.state.runningAtMsnull// 没有在执行中j.state.nextRunAtMs!nullnowMsj.state.nextRunAtMs// 已经到期);注意runningAtMs检查——如果一个任务正在执行比如上次还没跑完不会重复触发。阶段3执行任务根据sessionTarget分叉Main 路径// 注入系统事件state.deps.enqueueSystemEvent(text,{agentId:job.agentId});// 如果 wakeMode 是 now立即触发心跳if(job.wakeModenow){constresultawaitstate.deps.runHeartbeatOnce({reason:cron:${job.id}});}Isolated 路径// 启动独立 Agent 会话constresawaitstate.deps.runIsolatedAgentJob({job,message:job.payload.message});// 把结果汇报到主会话state.deps.enqueueSystemEvent(${prefix}:${res.summary},{agentId:job.agentId});阶段4更新状态job.state.lastRunAtMsstartedAt;job.state.lastStatusstatus;// ok | error | skippedjob.state.lastDurationMsendedAt-startedAt;// 计算下次执行时间if(job.schedule.kindatstatusok){job.enabledfalse;// 一次性任务完成后禁用}elseif(job.enabled){job.state.nextRunAtMscomputeNextRunAtMs(job.schedule,nowMs);}阶段5持久化 重新调度保存任务状态到存储然后armTimer()重新设置下一个定时器。 任务状态机一个 Cron Job 的状态流转┌─────────────────────────────────────────────────────────────┐ │ │ │ ┌──────────┐ add() ┌──────────┐ │ │ │ 创建 │ ──────── │ enabled │ ────┐ │ │ └──────────┘ └────┬─────┘ │ │ │ │ │ │ │ 到期触发 │ │ │ │ │ │ │ v │ │ │ ┌──────────┐ │ │ │ │ running │ │ │ │ └────┬─────┘ │ │ │ │ │ │ │ ┌───────────────┼───────────────┐ │ │ │ │ │ │ │ v v v │ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │ │ ok │ │ error │ │ skipped │ │ │ └────┬─────┘ └────┬─────┘ └────┬─────┘ │ │ │ │ │ │ │ └───────────────┴───────────────┘ │ │ │ │ │ ┌─────────┴─────────┐ │ │ │ │ │ │ at ok? 其他情况 │ │ │ │ │ │ v v │ │ ┌──────────┐ ┌──────────┐ │ │ │ disabled │ │ 等待 │ ──── 下次执行 │ │ └──────────┘ │ 下次触发 │ │ │ └──────────┘ │ │ │ └────────────────────────────────────────────────────────────┘关键状态说明状态字段含义enabled任务是否激活nextRunAtMs下次执行时间戳runningAtMs当前执行开始时间null 表示未在执行lastStatus上次执行结果lastDurationMs上次执行耗时 设计要点总结1. 单一定时器 vs 多定时器OpenClaw 选择了单一定时器模式。为什么多定时器的问题内存占用每个任务一个定时器1000 个任务就是 1000 个定时器精度问题大量定时器可能导致事件循环拥堵难以管理取消、更新操作复杂单一定时器的优势内存恒定永远只有一个活跃定时器逻辑清晰所有调度逻辑集中在armTimer()易于调试只需关注一个定时器的行为2. Main vs Isolated 的权衡维度MainIsolated上下文共享主会话独立会话适用任务简单提醒复杂操作用户感知立即可见执行完再通知资源消耗低高新建会话什么时候用 Main只是提醒用户做某事需要用户确认或响应任务依赖当前对话上下文什么时候用 Isolated任务可能耗时较长不需要用户介入需要干净的执行环境3. 结果汇报机制Isolated 任务执行完后通过enqueueSystemEvent()把结果注入主会话。用户会看到类似Cron: 服务器健康检查完成所有服务正常运行这个Cron:前缀是可配置的postToMainPrefix字段。 Python 复现建议如果你想用 Python 实现类似的 Cron 系统核心依赖是pip install croniter# cron 表达式解析关键实现点1. 调度计算fromcroniterimportcroniterfromzoneinfoimportZoneInfodefcompute_next_run(schedule,now_ms):ifschedule.kindcron:tzZoneInfo(schedule.tz)ifschedule.tzelsetimezone.utc base_timedatetime.fromtimestamp(now_ms/1000,tztz)croncroniter(schedule.expr,base_time)next_timecron.get_next(datetime)returnint(next_time.timestamp()*1000)2. 异步定时器asyncdefarm_timer(self):ifself.timer_task:self.timer_task.cancel()next_atself.next_wake_at_ms()ifnext_atisNone:returndelaymax(0,(next_at-self.now_ms())/1000)self.timer_taskasyncio.create_task(self.timer_tick(delay))3. 依赖注入classCronService:def__init__(self,on_system_event:Callable[[str],None],run_agent_turn:Callable[[Job,str],dict],run_heartbeat:Callable[[str],dict],):self.on_system_eventon_system_event self.run_agent_turnrun_agent_turn self.run_heartbeatrun_heartbeat把执行 Agent、触发心跳等操作作为依赖注入让 Cron 模块可以独立测试。 我的思考这套设计解决了什么问题传统的定时任务系统比如 crontab、APScheduler只管到点执行。但 Agent 场景下执行本身是个复杂的过程——需要上下文、需要推理、需要调用工具、还需要汇报结果。OpenClaw 的 Cron 系统把这些都考虑进去了上下文隔离Isolated 模式避免污染主对话结果回传执行完自动汇报灵活调度三种调度类型覆盖常见场景还有什么可以改进任务依赖当前任务之间是独立的。如果 A 任务失败了B 任务是否还执行没有依赖图的概念。重试机制任务失败后没有自动重试。对于网络请求类任务这可能是个问题。优先级调度所有任务平等。如果同一时刻有多个任务到期按什么顺序执行分布式支持单机单定时器的设计在分布式场景下需要改造。实际应用场景这套 Cron 系统特别适合个人助理 Agent每日提醒、定期汇总、自动检查监控 Agent定时巡检、异常告警内容 Agent定时抓取、自动发布想象一下你对 AI 说每天晚上10点帮我总结一下今天的邮件它就真的每天10点自动干活干完了发个总结给你。这才是真正有用的 AI 助手。 相关资源croniter 文档https://github.com/kiorky/cronitercron 表达式在线测试https://crontab.guru/Python asyncio 官方文档https://docs.python.org/3/library/asyncio.html如果你正在构建自己的 AI Agent 系统强烈建议把 Cron 模块纳入规划。它不复杂但能让你的 Agent 从被动工具变成主动助手。