04. nanobot 源码解读：cron 与 heartbeat

文档内容基于 HKUDS/nanobot: “🐈 nanobot: The Ultra-Lightweight Personal AI Agent” 的 main 分支 6bfb75e 提交进行说明。

目录

• • 04. nanobot 源码解读：cron 与 heartbeat
• • 目录
• • cron
• • CronService
• • cron job
• • heartbeat
• • HEARTBEAT.md
• • cron 与 heartbeat 的对比

前面介绍了用户可以通过 IM 发送消息给 nanobot，然后 nanobot 通过 Agent Loop 响应用户消息。这个流程从 nanobot 角度来看，属于被动操作。主动执行任务则由 cron 和 heartbeat 承载。

• • cron：管理定时任务，支持一次性任务和周期性任务的调度和执行
• • heartbeat：周期性检查配置文件，动态发现并执行待办任务

cron

nanobot 的 cron 模块实现了一个定时任务管理系统。nanobot 的 cron 将任务委托给 Agent Loop 处理，而不是执行固定的 shell 命令。

nanobot 定义了三种类型的定时规则：

• • at：一次性任务，在具体的时间点触发执行
• • every：周期性任务，每隔多长时间触发一次执行
• • cron：周期性任务，通过 cron 表达式决定下一次触发执行的时间

nanobot 使用了 croniter 库解析 cron 表达式，可以支持 5~7 个字段：minute、hour、day of month、month、day of week、second、year，分别表示分、时、日、月、星期、秒、年。例如 0 2 * * * 表示每天凌晨 2 点执行，*/30 * * * * 表示每 30 分钟执行一次。

nanobot 的 cron 模块主要包含以下功能：

• • 任务调度：根据配置的定时规则，计算每个任务下一次的执行时间
• • 任务持久化：所有的任务配置保存在 {workspace}/cron/jobs.json 文件中，并支持动态添加和删除
• • 执行委托：通过 CronService.on_job 回调将任务执行委托给外部处理逻辑

CronService

CronService（nanobot/cron/service.py）是定时任务（后文称 cron job）的管理器，包含了对 cron job 的增删改查以及执行触发等操作。

注意，CronService 本身不包含 cron job 的实际执行流程，而是通过一个类型为 Callable[[CronJob], Coroutine[Any, Any, str | None]] 的成员变量 on_job 来执行的。nanobot 启动（nanobot/cli/command.py）时会设置该参数，默认的 cron job 执行流程如下：

可以看到，nanobot 定义的 cron job 除了 dream 是内部的一个固定流程，其他的 cron job 都是通过 Agent Loop 来实现的。

dream 对应的操作是 nanobot 内置的 Dream 机制，用于实现持久化记忆的提取和更新。它的作用是定期分析历史对话，提取重要的信息（如用户偏好、bot 人设、上下文知识等）并持久化到对应的文件中，从而实现记忆的长期保存和检索。

dream 任务在 nanobot 启动时会自动注册（代码示例）：

# nanobot/cli/command.py# nanobot 提供的默认任务配置内容是空的，但是会在启动的时候注册 dream 定时任务from nanobot.cron.types import CronJob, CronPayloadcron.register_system_job(CronJob(    id="dream",    name="dream",    schedule=dream_cfg.build_schedule(config.agents.defaults.timezone),    payload=CronPayload(kind="system_event"),))

cron job

nanobot 触发 cron job 执行的流程如下：

• • 启动时加载所有 cron job，并计算每一个 cron job 的下一次执行时间
• • 等待最近的 cron job 执行时间到达
• • 唤醒执行 cron job，执行完毕后检查
• • at 型任务执行后会标记为失效（如果配置了执行后删除，则直接从配置文件中删除）
• • 非 at 型任务会计算下一次执行时间，然后等待唤醒执行

cron job 的来源有两种：

• • nanobot 启动前在配置文件（默认 {workspace}/cron/jobs.json）中写入了任务信息
• • nanobot 运行期间通过 CronTool（nanobot/agent/tool/cron.py，暴露 CronService 的 add_job、remove_job、list_jobs 能力）添加

cron job 的任何变更都会重新写入到配置文件（默认 {workspace}/cron/jobs.json）

heartbeat

heartbeat 相关代码在 nanobot/heartbeat/ 包下。在计算机领域，heartbeat 一般是一个周期性发送的消息或者信号，用于连接保活、健康检查等。

nanobot 的 heartbeat 则是通过定期检查配置文件，主动发现并处理待办事项，而不是被动等待用户指令。heartbeat（HeartbeatService）触发后会执行如下两个阶段的操作：

• • 第一阶段：读取 {workspace}/HEARTBEAT.md 文件内容，然后询问 LLM 是否存在需要执行的任务，如果判断存在则进入第二阶段
• • 第二阶段：接收第一阶段提取的 tasks 信息，通过类型为 Callable[[str], Coroutine[Any, Any, str]] 的成员变量 on_execute 执行

async def _tick(self) -> None:    """Execute a single heartbeat tick."""    from nanobot.utils.evaluator import evaluate_response    # 每次触发执行都会重新读取 {workspace}/HEARTBEAT.md 内容    content = self._read_heartbeat_file()    ...    try:        # 第一阶段任务，判断 HEARTBEAT.md 文件内容中是否存在需要执行的任务        # 直接调用 LLM 接口，而不是走比较重的 Agent Loop 流程        action, tasks = await self._decide(content)        ...        if self.on_execute:            # nanobot 启动（nanobot/cli/command.py）的时候设置            # 使用 Agent Loop 流程来执行任务            response = await self.on_execute(tasks)            if response:                # 判断任务执行结果是否需要发送通知给用户                # 也是直接调用 LLM 接口，传入 tasks 信息和执行结果                should_notify = await evaluate_response(                    response, tasks, self.provider, self.model,                )                if should_notify and self.on_notify:                    logger.info("Heartbeat: completed, delivering response")                    await self.on_notify(response)                else:                    logger.info("Heartbeat: silenced by post-run evaluation")    except Exception:        logger.exception("Heartbeat execution failed")

HEARTBEAT.md

nanobot 提供的 HEARTBEAT.md（nanobot/templates/HEARTBEAT.md）默认内容如下：

# Heartbeat TasksThis file is checked every 30 minutes by your nanobot agent.Add tasks below that you want the agent to work on periodically.If this file has no tasks (only headers and comments), the agent will skip the heartbeat.## Active Tasks<!-- Add your periodic tasks below this line -->## Completed<!-- Move completed tasks here or delete them -->

一个明显的倾向是，nanobot 建议用户手动维护 HEARTBEAT.md 文件内容，如添加需要执行的任务、删除已经完成的任务等，而不是依赖 Agent Loop 来管控。

cron 与 heartbeat 的对比