【OpenClaw】通过 Nanobot 源码学习架构---（9）周期性执行：CronService 与 CronTool 配置骨架

1. 从一次“Agent 到点没动静”说起如果你正在读 Nanobot 源码大概率已经跑通了 AgentLoop 的单轮对话也见过 CronTool 在工具列表里挂着。但真正把cron(actionadd, every_seconds60)敲进去之后很多人会卡在同一个地方任务注册成功了cron(actionlist)也能看到 job可时间到了什么也没发生。这不是玄学而是 CronService 的调度链路里有一环没接上——on_job回调没有正确绑定到 AgentLoop 的执行入口。Nanobot 是香港大学数据科学实验室开源的超轻量级个人 AI 助手框架定位为“Ultra-Lightweight OpenClaw”代码量小、结构清晰非常适合拿来拆 Agent 架构。它的周期性执行能力由两个组件配合完成CronTool 是面向 LLM 的接口层负责把自然语言参数翻译成结构化调用CronService 是真正的业务执行者负责加载任务、计算下次触发时间、调用回调、记录状态。两者通过依赖注入串起来形成一条从“用户说一句话”到“Agent 主动执行”的完整链路。这篇会沿着源码把这条链路拆开先看 CronService 的三种调度类型怎么算时间再看 CronTool 怎么把参数塞进去然后给出一份可以直接复制的config.toml与settings.json配置骨架最后用一次真实的定时触发日志验证整条链路。适合已经能跑起 Nanobot、想搞懂定时任务内部机制的人。2. TaoToken 前置给 Agent 一个稳定的模型出口在拆 CronService 之前得先保证 AgentLoop 本身能正常跑。Nanobot 的process_direct()在触发定时任务时会走一次完整的模型调用如果模型出口不稳定你会看到任务触发了但结果为空误以为是调度问题。我本地是用 TaoToken 做模型接入的它的 API 地址是https://taotoken.net/api兼容 OpenAI 风格的调用方式Nanobot 的 provider 配置里直接填 base_url 就能用。官网在https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content注册后在控制台生成 API Key 即可。需要提前准备的东西不多一个可用的 API Key、Nanobot 源码目录、Python 3.10 环境。如果你还没拿到 Key可以先去控制台创建https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentcron_service_consoleutm_campaignrewrite。Key 生成页面在https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentcron_service_apikeysutm_campaignrewrite创建后复制保存后面配置里要用。注意API Key 只显示一次建议直接写进环境变量而不是硬编码进配置文件避免提交到仓库。3. CronService 源码拆解三种调度类型怎么算时间3.1 CronJob 的数据结构Nanobot 把每个定时任务抽象成CronJob核心字段包括id、name、enabled、schedule_kind、schedule_config、payload和consecutive_errors。其中schedule_kind只有三个取值at、every、cron分别对应一次性、固定间隔、标准表达式三种模式。payload里放的是{kind: agent_turn, message: ...}也就是触发时要交给 AgentLoop 执行的内容。consecutive_errors这个字段值得单独说。它不是装饰品而是自动熔断的依据连续失败 5 次后任务会被自动置为enabled False避免一个坏任务无限刷错误日志。3.2_compute_next的时间计算逻辑调度器的核心是_compute_next(job, now)它根据schedule_kind分三条分支at模式最简单把 ISO 时间字符串解析成时间戳如果大于当前时间就返回否则返回 0 表示不再触发。every模式用的是锚点对齐算法steps int((now - anchor) / every) 1然后返回anchor steps * every。这样做的目的是让触发时间可预测不会因为进程重启而漂移。cron模式直接交给croniter库传入表达式和当前时间取get_next(datetime)。def _compute_next(self, job, now): if job.schedule_kind at: ts datetime.fromisoformat(cfg.get(at, )).timestamp() return ts if ts now else 0.0 if job.schedule_kind every: every cfg.get(every_seconds, 3600) steps int((now - anchor) / every) 1 return anchor steps * every if job.schedule_kind cron: return croniter(expr, datetime.fromtimestamp(now)).get_next(datetime).timestamp()3.3 主循环与错误熔断CronService 跑在一个后台线程里每秒 tick 一次。每次 tick 遍历所有 job先判断enabled再判断是否 duedue 了就调_run_job()。执行结果分两种成功则把consecutive_errors归零失败则自增达到 5 次就自动禁用。执行记录会追加到cron-runs.jsonl方便事后排查。if status error: job.consecutive_errors 1 if job.consecutive_errors 5: job.enabled False else: job.consecutive_errors 0这段逻辑看着简单但它决定了你的定时任务在模型超时、网络抖动时会不会把整个调度器拖垮。4. CronTool 与依赖注入参数怎么从 LLM 流到 ServiceCronTool 是 LLM 能看到的工具它的职责是把action、message、every_seconds、cron_expr、at、tz、job_id这些参数整理成 CronService 能消费的结构。它本身不存任务只持有CronService的引用通过_cron成员变量调用add_job、list_jobs、remove_job。依赖注入的链路是这样的AgentLoop 在_register_default_tools()里创建 CronTool把 CronService 传进去CronService 在初始化时接收一个on_job回调这个回调由 Gateway 设置指向AgentLoop.process_direct()。所以当任务触发时实际执行路径是CronService._run_job()→on_job(job)→AgentLoop.process_direct()→ 模型调用 → 结果回写。CronTool 还持有_channel和_chat_id通过set_context()设置用来保存用户上下文。这样任务触发后结果能推回正确的会话而不是丢进虚空。5. 可复制配置骨架config.toml 与 settings.json下面这份配置是我本地验证过的骨架直接改路径和 Key 就能用。config.toml负责服务级参数settings.json负责 Agent 与工具级参数。# config.toml [gateway] host 127.0.0.1 port 8080 [cron] enabled true tick_interval 1 store_path ./data/cron.json runs_log ./data/cron-runs.jsonl max_consecutive_errors 5 [provider] base_url https://taotoken.net/api api_key ${TAOTOKEN_API_KEY} model gpt-4o-mini timeout 60{ agent: { name: nanobot-local, system_prompt_path: ./prompts/SOUL.md, memory_path: ./data/MEMORY.md }, tools: { cron: { enabled: true, default_tz: Asia/Shanghai, allow_every_seconds: true, allow_cron_expr: true, allow_at: true } }, loop: { max_turns: 8, single_turn_timeout: 45 } }cron.store_path指向任务持久化文件runs_log是执行记录。default_tz建议显式设置否则cron_expr会按服务器本地时区算跨时区部署时容易错点。single_turn_timeout要小于provider.timeout避免任务卡死。环境变量里设置 Keyexport TAOTOKEN_API_KEY你的Key6. 验证一次定时触发从注册到日志配置就绪后启动 Nanobot然后在对话里注册一个 60 秒间隔的任务cron(actionadd, messageCheck HKUDS/nanobot GitHub stars and report, every_seconds60)注册成功后cron(actionlist)应该能看到 job。接下来观察cron-runs.jsonl正常触发时会出现类似记录{job_id:abc123,status:ok,started_at:2025-01-01T10:00:00,finished_at:2025-01-01T10:00:03,result_preview:nanobot has 1.2k stars}如果status是error先看result_preview里的报错。常见的是模型调用超时或 Key 无效而不是调度器本身的问题。连续 5 次 error 后job 的enabled会变成false这时候list里还能看到它但不会再触发。想验证cron_expr模式可以注册一个工作日早上 9 点的任务cron(actionadd, messageMorning standup, cron_expr0 9 * * 1-5, tzAsia/Shanghai)at模式适合一次性提醒传入 ISO 时间字符串即可触发后任务自动删除不会留冗余。7. 本篇常见错排查任务注册了但不触发先确认config.toml里cron.enabled true再看store_path是否可写。如果on_job回调没绑定任务会 due 但执行结果为空日志里只有 started 没有 finished。cron_expr时间对不上九成是时区问题。tz不填时用服务器本地时区容器里通常是 UTC和你的预期差 8 小时。显式写tz Asia/Shanghai最稳。连续 error 后任务消失不是消失是被自动禁用了。去cron.json里把enabled改回true同时排查根因否则 5 次之后还会被禁。every_seconds触发时间漂移Nanobot 用的是锚点对齐不会漂移。如果你看到间隔忽长忽短检查是不是有多个 CronService 实例在跑重复触发。模型调用超时导致任务失败把loop.single_turn_timeout调到小于provider.timeout并确认 API 出口稳定。我这边用 TaoToken 的https://taotoken.net/api做接入超时情况少很多。8. 继续往下走把 CronService 和 CronTool 的链路跑通之后你会发现 Nanobot 的定时任务设计其实很克制没有分布式调度、没有复杂依赖就是一个后台线程加一个 JSON 存储。但正是这种轻量让它特别适合拿来学 Agent 架构——每个组件的职责边界都很清楚。如果你还没拿到 API Key可以从这里开始https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentcron_service_apikeysutm_campaignrewrite。接入文档在https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentcron_service_docutm_campaignrewrite里面有完整的 provider 配置说明。想先验证模型对话是否正常可以用https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentcron_service_modelsutm_campaignrewrite试一轮。如果你打算长期跑编码类 AgentCoding Plan 在https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcron_service_codingplanutm_campaignrewrite按周期计费更适合高频调用场景。下一步可以试着改_compute_next的every分支把锚点换成任务创建时间观察触发行为的变化。这种小改动最能帮你理解调度器的边界。