ARTICLE DETAIL

资讯详情

深耕商务建站与企业官网运营的一线实战洞察。

Agent Zero 定时任务完全指南:精通 scheduled-tasks 技能与 scheduler 工具

Agent Zero 定时任务完全指南:精通 scheduled-tasks 技能与 scheduler 工具 Agent Zero 定时任务完全指南精通 scheduled-tasks 技能与 scheduler 工具【免费下载链接】agent-zeroAgent Zero AI framework项目地址: https://gitcode.com/GitHub_Trending/ag/agent-zero导读本文围绕 Agent Zero 内置的scheduled-tasks技能及其底层的scheduler工具展开系统讲解 Agent Zero 中定时任务scheduled、计划任务planned与临时任务adhoc的创建、查询、修改、运行与删除全流程。读完本文你将掌握 cron 风格schedule字段与 ISOplan字段的正确用法、IANA 时区处理机制、专用上下文dedicated context与安全规则并能结合源码理解任务调度器helpers/task_scheduler.py的底层实现原理。一、scheduled-tasks 技能是什么scheduled-tasks是 Agent Zero 内置技能之一其说明为Use for complex Agent Zero scheduler work, including creating, updating, deleting, running, waiting for, timezone-correcting, or auditing scheduled, planned, and adhoc tasks.即凡是涉及 Agent Zero 调度器的复杂工作——创建、更新、删除、运行、等待、时区校正、审计定时/计划/临时任务——都应加载该技能。技能包由两个文件组成skills/scheduled-tasks/SKILL.md技能主体拥有调度器动作action描述、调度字段schedule fields指引、安全规则与示例的“所有权”Ownershipskills/scheduled-tasks/AGENTS.md技能的维护型 DOX规定技能的职责边界与本地契约Local Contracts。同时Agent Zero 的主提示词 prompts/agent.system.tool.scheduler.md 中明确约定复杂调度任务应加载scheduled-tasks技能而在技能加载之前Agent 也需要遵循其核心规则先查后建、不递归调度、专用上下文等。二、调度器核心工作流先查后动技能与系统提示词都强调同一条铁律Always inspect existing tasks before creating, updating, deleting, or running one.在创建create_*、更新update_task、删除delete_task或运行run_task任何任务之前必须先通过find_task_by_name或list_tasks检查已有任务。这一契约在 skills/scheduled-tasks/SKILL.md 与 prompts/agent.system.tool.scheduler.md 中均有明确表述。其典型调用格式技能中的示例{ tool_name: scheduler, tool_args: { action: find_task_by_name, name: daily backup } }三、scheduler 工具的动作清单scheduler工具共支持 10 个动作每个动作对应 tools/scheduler.py 中SchedulerTool类的一个异步方法execute方法通过action参数分发到具体实现见 tools/scheduler.py。动作及参数如下动作必填参数可选参数说明list_tasks—state[]、type[]、next_run_within、next_run_after按状态、类型、距下次运行时间分钟过滤列出任务find_task_by_namename—按名称模糊查找任务名称不区分大小写的子串匹配show_taskuuid—查看单个任务详情run_taskuuidcontext立即运行任务可附加上下文update_taskuuidname、system_prompt、prompt、attachments[]、schedule、timezone、plan[]、state、dedicated_context更新任务字段delete_taskuuid—删除任务破坏性操作需先按 UUID 定位create_scheduled_taskname、system_prompt、promptattachments[]、schedule、timezone、dedicated_context创建按 cron 周期重复的任务create_adhoc_taskname、system_prompt、promptattachments[]、dedicated_context创建一次性临时任务create_planned_taskname、system_prompt、promptattachments[]、plan[]、dedicated_context创建按计划时间列表触发的任务wait_for_taskuuid—等待任务完成并返回其结果如果传入了未知动作execute会返回错误提示并列出自上而下的全部受支持动作tools/scheduler.py。3.1 三种任务类型的底层模型从源码看三种任务分别对应 helpers/task_scheduler.py 中继承自BaseTask的三个 Pydantic 模型通过type字段做判别联合见 helpers/task_scheduler.pyScheduledTasktype: scheduled携带TaskSchedule按 cron 表达式周期触发AdHocTasktype: adhoc一次性临时任务额外带有token字段用于身份校验PlannedTasktype: planned携带TaskPlantodo/in_progress/done三个时间列表按指定时刻逐个触发。任务状态TaskState共有四种枚举值idle、running、disabled、errorhelpers/task_scheduler.py。任务被创建后默认处于idle运行中被置为running运行失败置为error用户可显式禁用为disabled。四、Schedule 字段详解cron 风格而非 ISO技能明确规定schedule使用 cron 风格的字段严禁把 ISO 日期时间塞进scheduleDo not put ISO datetimes intoschedule。schedule包含以下字段minute分钟hour小时day日month月weekday星期timezone时区这些字段在TaskSchedule模型中被定义为 5 个 cron 字段加一个时区字段并通过to_crontab()拼装为标准 5 段 cron 表达式helpers/task_scheduler.pydef to_crontab(self) - str: return f{self.minute} {self.hour} {self.day} {self.month} {self.weekday}计划任务的日期时间则放进plan字段且应为 ISO 字符串例如2026-05-09T18:25:00。_task_plan_from_input会逐个解析plan列表中的 ISO 时间戳任何一个无法解析都会返回 Invalid datetime: ... 错误tools/scheduler.py。4.1 时区规范使用 IANA 时区名称例如Europe/Rome省略timezone时使用当前用户时区任务创建时的时区解析逻辑位于 tools/scheduler.py_normalize_timezone会先检查别名local、user、default、current、current_timezone均映射到当前用户时区再通过pytz.timezone()校验合法性非法时区会抛出明确的ValueError提示信息底层normalize_schedule_timezone也实现了同样的别名归一化与非法时区回退打印错误日志后回退到用户时区见 helpers/task_scheduler.py。测试用例 tests/test_task_scheduler_timezone.py 验证了时区行为Europe/Rome时区下 5 月 10 日 09:30 的定时任务其 UTC 下次运行时间为 07:30test_scheduled_task_next_run_uses_schedule_timezone同时验证了local这类历史遗留时区别名会被归一化为用户时区test_scheduled_task_normalizes_legacy_local_timezone。4.2 cron 表达式的防幻觉校验_validate_task_schedule使用正则表达式对拼装出的 crontab 进行校验注释明确指出“agent might hallucinate”Agent 可能幻觉生成非法 cron非法表达式会返回 Invalid cron expression: ...tools/scheduler.py。此外_task_schedule_from_input同时支持把字符串形式的schedule空格分隔的 5 段 cron解析为字段字典tools/scheduler.py因此在工具层与 api/scheduler_task_create.py 的 API 层都兼容字符串与字典两种schedule格式。五、实战示例三种创建动作5.1 单个未来提醒create_planned_task对于一次性未来提醒技能推荐create_planned_task{ action: create_planned_task, name: drink water, prompt: Remind the user to drink water., plan: [2026-05-11T09:15:00], dedicated_context: true }plan是 ISO 日期时间数组任务会在所列时刻逐一触发每完成一个时间点就将其从todo移入done见下文TaskPlan进度推进。PlannedTask.check_schedule()通过plan.should_launch()判断是否有已到期的计划时刻helpers/task_scheduler.py。5.2 周期重复任务create_scheduled_task对于周期性、cron 形态的重复任务使用create_scheduled_task{ action: create_scheduled_task, name: weekday stretch, prompt: Remind the user to stretch., schedule: { minute: 15, hour: 9, day: *, month: *, weekday: 1-5, timezone: Europe/Rome }, dedicated_context: true }该示例的含义是每个工作日周一至周五weekday: 1-5的 09:15minute: 15、hour: 9在罗马时区触发任务。技能系统提示词中还给出了“明天罗马时间 9:15”的具体写法prompts/agent.system.tool.scheduler.md{ schedule: { minute: 15, hour: 9, day: 11, month: 5, weekday: *, timezone: Europe/Rome } }5.3 一次性临时任务create_adhoc_taskcreate_adhoc_task不携带schedule或plan创建后由用户显式触发。从源码看AdHocTask额外生成一个 19 位随机tokenhelpers/task_scheduler.pySchedulerTaskList.save()甚至在写盘前专门校验 adhoc 任务的 token 非空为空时自动补生成helpers/task_scheduler.py。5.4 完整的 JSON 调用结构技能中的完整示例展示了工具调用包裹结构{ tool_name: scheduler, tool_args: { action: find_task_by_name, name: daily backup } }SchedulerTool.execute会先将action参数统一转为小写并把连字符替换为下划线_current_actiontools/scheduler.py再做动作分发。六、安全规则Safety技能 skills/scheduled-tasks/SKILL.md 明确列出四条安全红线不要创建递归的任务提示词Do not create recursive task prompts that schedule more tasks即任务自身的 prompt 不应再触发创建新的调度任务防止任务无限自我繁衍不要因为任务已到点就运行它只有用户要求时才运行Do not run a task just because it is scheduled; run only if the user asksAgent 不应主动抢跑新创建的任务默认使用专用上下文除非dedicated_context显式为false从源码看三个创建动作的dedicated_context默认值都是True为True时context_idNone之后BaseTask.__init__会把context_id回填为任务自身的uuidis_dedicated()判断context_id uuid为False时则绑定到当前 Agent 的上下文tools/scheduler.py、helpers/task_scheduler.py破坏性操作必须先用 UUID 定位任务再执行delete_task、update_task都要求先通过find_task_by_name或list_tasks拿到目标任务的uuid再以 UUID 操作。此外prompts/agent.system.tool.scheduler.md 还补充了一条规则涉及计划/定时任务时若用户点名了某个时区必须把该时区写入timezone字段。七、源码级原理任务如何被触发、执行与持久化7.1 触发机制tick 轮询任务到期检测由TaskScheduler.tick()完成——它从任务列表取出所有“到期且空闲”的任务并逐个运行helpers/task_scheduler.pyasync def tick(self): for task in await self._tasks.get_due_tasks(): await self._run_task(task)get_due_tasks()内部先reload()最新状态再筛选check_schedule()为真且state IDLE的任务helpers/task_scheduler.py。ScheduledTask.check_schedule()用crontab库计算参考时刻到下次执行的秒数是否落在轮询频率默认 60 秒窗口内helpers/task_scheduler.py。对外暴露的轮询入口是 API 端点 api/scheduler_tick.py它支持传入timezone设置本地化、打印任务清单与统计并返回序列化后的全部任务。7.2 执行机制后台 DeferredTask 线程_run_task将任务的执行包装进DeferredTask线程thread_nameTaskScheduler中异步运行并登记到_running_deferred_tasks字典以便支持取消cancel_running_task、cancel_tasks_by_contexthelpers/task_scheduler.py。执行流程为快照检查任务是否存在、是否已在运行防重复触发原子地把状态置为runningupdate_task_checked通过验证函数防止竞态条件调用on_run()PlannedTask会在此把即将执行的时刻从todo移到in_progress获取/创建任务上下文_get_chat_context组装带附件的用户消息附件支持本地路径与 http/https 等 URL注入system_prompt并调用agent.monologue()运行 Agent成功后on_success将状态置回idle并记录last_result失败后on_error将状态置为error并记录last_result ERROR: ...on_finish统一刷新updated_at时间戳。run_task_by_uuid还处理了特殊状态转换已运行的任务再次运行会报错禁用任务拒绝运行error 状态任务会先重置回 idle 再运行helpers/task_scheduler.py。7.3 计划任务的进度推进TaskPlan维护三个时间列表todo待执行、in_progress进行中、done已完成。add_todo插入并排序set_in_progress把时刻从todo移到in_progressset_done在任务结束后把in_progress移到donehelpers/task_scheduler.py。PlannedTask.on_run/on_finish将这一推进过程与调度器生命周期挂钩并强制持久化helpers/task_scheduler.py。这样plan: [2026-05-11T09:15:00, 2026-05-11T15:00:00]这样的任务会在两个时刻各触发一次第二次触发只会在第一次完成后发生。7.4 持久化usr/scheduler/tasks.json所有任务通过SchedulerTaskList单例持久化到usr/scheduler/tasks.json常量SCHEDULER_FOLDER usr/schedulerhelpers/task_scheduler.py。SchedulerTaskList.get()在首次访问时若文件不存在会自动创建空任务列表helpers/task_scheduler.py。每次增删改都会落盘save()每次读取前都会reload()重新校验最新状态update_task_by_uuid通过“先 reload、再更新、再保存”并配合线程锁实现原子更新避免多进程并发竞争helpers/task_scheduler.py。所有增删改操作后还会调用mark_dirty_all通知状态监控模块state monitor刷新界面。7.5 任务序列化字段serialize_task输出的标准任务结构包含uuid、name、state、system_prompt、prompt、attachments、project_name、project_color、created_at、updated_at、last_run、next_run由 cron/plan 计算、last_result、context_id、dedicated_context以及按类型附加的schedulescheduled/tokenadhoc/planplanned字段helpers/task_scheduler.py。list_tasks动作的过滤参数next_run_within、next_run_after正是基于序列化中的get_next_run_minutes()计算值实现的tools/scheduler.py。7.6 WebUI 与 API 层除 Agent 工具外调度功能还通过 WebUI API 暴露SchedulerTaskCreateapi/scheduler_task_create.py负责 UI 端创建任务自动区分schedule、plan、adhoc 三种分支并为 adhoc 任务自动生成 tokenSchedulerTaskUpdateapi/scheduler_task_update.py支持按task_id更新字段、状态与调度并明确禁止修改项目绑定Project changes are not allowed。整套 API 均复用helpers/task_scheduler.py的解析与序列化函数保证工具层与 UI 层行为一致。八、运行与等待run_task 与 wait_for_taskrun_task按 UUID 立即运行任务可选context参数注入附加上下文task_context会被拼入任务的 user 消息中。值得注意的细节是如果任务运行在当前上下文非专用上下文工具会返回break_loopTrue以中断当前对话循环避免同一窗口出现两个对话tools/scheduler.pywait_for_task阻塞等待任务在独立专用上下文中完成默认超时 300 秒DEFAULT_WAIT_TIMEOUT。如果任务运行在调用者自己的上下文中会直接返回错误 You can only wait for tasks running in their own dedicated context.。等待期间每秒轮询一次任务状态超时返回 Task wait timeout (300 seconds)tools/scheduler.py。等待结束时返回任务 UUID、状态、上次运行时间与结果内容。九、删除与清理delete_task是破坏性操作删除前会检查任务状态若任务正在运行running则先重置其上下文与状态为idle若任务持有独立的专用上下文还会同时移除AgentContext与对应的持久化聊天记录persist_chat.remove_chat最后从任务列表中移除并落盘tools/scheduler.py。这也是技能强调“破坏性操作先按 UUID 定位”的原因——删除不仅影响任务本身还连带清理其上下文与历史。十、技能维护约定根据 skills/scheduled-tasks/AGENTS.md 的约定该技能由 Agent Zero 开发者负责维护核心职责包括保持调度器工具动作、字段名、时区行为的准确性技能示例必须始终是合法的 JSON 工具参数Keep examples valid JSON tool arguments当调度器工具动作、字段或时区行为变化时同步更新技能并手动复查 SKILL.md 中是否存在过期的动作名与日期指引。技能目录结构验证scheduled-tasks技能目录下共两个文件AGENTS.md与SKILL.md无子 DOX与 AGENTS.md 中 No child DOX files 的声明一致符合 skills/AGENTS.md 中“每个技能目录必须包含SKILL.md”的全局契约。结语scheduled-tasks技能与scheduler工具共同构成了 Agent Zero 的完整任务调度能力cron 风格的schedule支撑周期任务ISO 格式的plan支撑精确时刻的一次性提醒adhoc支撑临时执行wait_for_task支撑同步等待结果而“先查后动”、非递归提示、专用上下文与 UUID 定位等安全规则则保证了调度行为的可控与可审计。理解 tools/scheduler.py 与 helpers/task_scheduler.py 的底层实现可以帮助你在使用调度能力时准确预判其行为并在扩展或排障时快速定位问题。【免费下载链接】agent-zeroAgent Zero AI framework项目地址: https://gitcode.com/GitHub_Trending/ag/agent-zero创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考
返回列表
PREV
查看更多资讯
NEXT
返回资讯列表