想让 Nanobot 自动跑一件事？

哈，咱说点啥呢？

Ru果你对 AI Agent 感兴趣，却又想把它变成每天自动提醒、定时抓数据的帮手，那就得搞懂 Nanobot 的周期任务。

别担心，接下来我会像老朋友一样，把源码里实现周期性执行的套路拆开来聊。

CronTool：把自然语言变成调度指令

CronTool 就是个桥梁，拿着用户说好的“每晚十点给我发个喝水提醒”，转交给后端去排期。

它的 execute 方法里先判断 action 是 add、list 还是 remove，然后再调用对应的内部方法。

比如 add 时你必须先给它一个 message，不然就报错：没有消息干啥啊。

另外还得在调用前 set_context 把渠道和聊天 ID 塞进去，否则 Bot 就不知道该把提醒送到哪儿去了。

CronService：真正跑任务的引擎

CronService 不会去操作系统的 crontab，而是自己用 asyncio 写了个小调度器。

它维护一个存储文件，里面记录所有 Job 的 id、名字、是否启用、调度规则等信息。

当你调用 add_job 时它会先验证调度规则合法性，再生成一个唯一 id，Zui后把 Job 加进列表并写回磁盘。

启动服务时它会读取所有 Job，计算下次运行时间，然后设好一个 Timer 等待到那刻才触发。

三种调度模式

1️⃣ 每日固定时间点——at。你给 ISO 格式时间，等到那刻就跑一次然后可选删掉自己。

2️⃣ 每隔固定秒数——every。Zui简单，只要传 every_seconds，就按间隔轮番执行。

3️⃣ cron 表达式——cron_expr。geng强大，Ke以指定分钟、小时、星期等细节，也Ke以配 tz 指定时区。

如何避免堆积与错误失效

CronService 内部有两种计时方式：scheduleAtFixedRate 和 scheduleWithFixedDelay。前者从上一次开始计，后者从结束算起。大多数场景建议用后者，因为不会出现堆积。

CronJob 执行前一定要 try/except 捕获异常，否则整个定时器可Neng静默失效，让你连天dou不知道没跑完。

代码片段速览

async def execute(self, action: str, message="", every_seconds=None,
                  cron_expr=None, tz=None, at=None):
    if action == "add":
        return self._add_job(message, every_seconds,
                            cron_expr, tz, at)
    elif action == "list":
        return self._list_jobs
    elif action == "remove":
        return self._remove_job
    else:
        return f"Unknown action: {action}"

_add_job 内部流程简述

def _add_job:
    # 校验
    if not message: raise ValueError
    # 判断是 which 调度类型
    if every_seconds:
        schedule = CronSchedule(kind="every",
                                every_ms=every_seconds*1000)
    elif cron_expr:
        schedule = CronSchedule(kind="cron",
                                expr=cron_expr,
                                tz=tz)
    elif at:
        dt = datetime.fromisoformat
        schedule = CronSchedule(kind="at",
                                at_ms=int*1000))
    else:
        raise ValueError
    # 调用底层添加
    job = self._cron.add_job(name=message,
                             schedule=schedule,
                             message=message,
                             deliver=True,
                             channel=self._channel,
                             to=self._chat_id,
                             delete_after_run=)
    return f"Created job '{job.name}' "

Cron 与 AgentLoop 的协作方式

AIGC 对话结束后你往往需要后台继续Zuo事。这时候 AgentLoop 会监听一个 Heartbeat 队列，每秒检查一次是否有任务需要跑。

CronService 在 tick 时把到期的 Job 逐个取出来通过 on_job 回调让 AgentLoop 去执行实际业务逻辑，比如查询 GitHub stars 或者推送一条提醒消息。

为什么不用传统 crontab 呢？

Nano 本身就是轻量级 AI 助手框架，用外部 crontab 会导致耦合过深；而自建 Scheduler Ke以随时 功Neng，比如动态切换时区或支持一次性任务删除，让使用体验geng友好。

状态持久化与恢复示例

def _save_store:
    data = {"jobs": }
    self.store_path.write_text, encoding='utf-8')
def _load_store:
    if not self.store_path.exists:
        return 
    data = json.loads)
    jobs = 
    self._store.jobs = jobs

Semi‑Automated Dream 模式与周期任务交互

DREAM 是 Nano 里的“睡眠”机制，用来批量分析历史记录并geng新长期记忆文件。但它不走正常的 AgentLoop 路径，而是直接触发一些内部 Job 来Zuo批处理工作。

所以Ru果你想让某个报告每天凌晨自动生成，就Ke以写一个 cron_expr 为 “0 1 * * *” 的 Job，然后在 on_job 回调里调用 DREAM API Zuo分析并推送结果给用户。

Nano 场景实战：喝水提醒 & 数据同步示例

"喝水提醒": 用 “cron” 添加每小时提醒，一键搞定。不需要手动干预，Bot 自己跑。

"每日 GitHub 星数": 用 “cron”。Bot 在每天早上九点半抓星数，然后回推给你一条信息。

"会议临时通知": 用 “cron”。单次任务完成后自动删除，不会留下残余记录。

MVP 快速搭建步骤`

text 1) 在项目根目录下创建 `crons.json` 用于持久化。

text 2) 在你的工具集注册 CronTool 并注入 CronService。

text 3) 启动 Nano，并在对话中输入 `cron`。

Nano 开发者注意事项与Zui佳实践`

"异常捕获": 确保每个 Runnable dou有 try/except，否则定时器可Neng彻底卡住。

`

"延迟计算": 使用 scheduleWithFixedDelay 而非 fixedRate，以免累积堆积导致 CPU 飙高。

`

"JSON 文件大小": Ru果任务太多，可考虑分割文件或迁移至数据库，但一般来说几百行足够用了。

`

"日志监控": 给每次执行加日志标签，例如 logger.info，方便排查问题。

`

一下吧…`

`Nano 的周期性执行机制其实hen像一个微型的 Linux crontab，但专门为 AI Agent Zuo了适配和优化。

1) 用户通过自然语言或工具调用添加/删除/查询 Job； 2) CronTool 把命令转成结构化参数； 3) CronService 持久化存储、计算下一次运行时间，并用 asyncio Timer 定期检查； 4) 到期后通过 on_job 回调让 AgentLoop 去真正执行业务逻辑； 5) 所有过程dou围绕 JSON 存档和异常捕获Zuo稳健处理。

只要记住这几个环节，你就Neng让 Nano 成为个人或团队的小助手，把重复劳动全托付给它去Zuo。

好了我这就先停下来你kan不懂别生气，Ru果还有啥细节想知道，就直接跟我聊，我随时准备继续补充。

祝编码愉快~ 哈哈！

---