0基础学会Agent Harness工程（14）：用Cron Scheduler让Agent在未来自动醒来

1. 从“后台跑得动”到“到点自己醒”Agent 缺的那根时间轴如果你已经跟着这个系列走到第 13 篇大概率手上有一个能跑后台任务的 Agent Harness慢工具丢进 daemon thread主循环用占位 tool result 先配对等后台跑完再把真实结果作为新 observation 回填。这套机制解决的是“慢操作不阻塞当前轮”但它有一个前提——得有人先发起这一轮。没有用户输入、没有新的 tool result前台 Agent Loop 就静静待在那里不会自己动。我试过把“每天 09:00 检查构建状态”直接塞进后台线程结果发现三个问题绕不过去第一谁来在 09:00 创建这份工作后台线程只执行已经提交的调用它不保存未来计划。第二进程重启后这个“每天 09:00”还在不在内存里的线程肯定没了。第三就算线程活着它凭什么去调用 Agent Loop它没有权限也没有入口。所以第 14 篇要补的是 Agent Harness 里最容易被忽略的一层Cron Scheduler定时唤醒。它让 Agent 能在未来某个时间点由系统自己产生一条新输入从而开启一轮全新的模型交互。核心检索词先摆出来Cron Scheduler 是 Agent Harness 中负责“时间命中→产生事件→唤醒 Agent”的调度组件CronJob 是它管理的持久计划对象Python threading 提供调度线程与互斥锁OpenAI Function Calling 负责当前轮注册计划时的 tool call 配对。适合谁适合已经能让 Agent 跑后台任务、现在想让它“到点自己醒来”的开发者。这里必须先把四个对象拆开否则后面代码一定读乱schedule是五段式时间规则CronJob是包含规则、prompt 和生命周期的持久计划fired event是某一分钟已经命中的具体事件Agent turn是消费事件后开始的一轮模型交互。四者不能缩成一个“定时任务”词汇。CronJob 可以周期性产生多个 fired event每个 event 可能唤醒一次 Agent turnAgent 又可能在该轮提交一个 background job。把“计划”和“运行状态”混在一起重启后两种信息会一起丢。2. TaoToken 前置用统一 Key 打通 Function Calling 通道在写调度代码之前先把模型通道准备好。因为本篇要演示“未来唤醒后Agent 基于新输入发起一次 OpenAI Function Calling”你需要一个稳定的、兼容 OpenAI 协议的端点。我用的是 TaoToken 的统一 Key/API 通道好处是同一个 Key 既能跑模型对话也能跑 Function Calling不用为不同能力维护多套凭证。官网入口在这里https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。注册后在控制台创建 API Key控制台地址是 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite Key 管理页在 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 。API 基地址统一用 https://taotoken.net/api 注意这个地址后面不加 UTM 参数直接作为 base_url 使用。拿到 Key 之后先确认两件事一是模型对话通道能通可以在 https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite 里选一个支持 Function Calling 的模型二是接入文档里关于 tool call 的字段说明文档地址 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。如果你后面要做长期编码或 Agent 常驻任务可以了解 Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 它更适合持续运行的场景。注意TaoToken 在这里的角色是“统一的模型调用通道”它不提供定时器、不提供持久队列、也不提供未来唤醒。Cron Scheduler、CronJob、cron_queue 全部是你 Harness 内部的组件。别把本地调度能力说成 API 自带功能这是两回事。3. 可复制配置CronJob 骨架与 Python threading 定时触发先把数据结构定下来。CronJob 只需要五个字段id 提供稳定引用cron 保存五段式表达式prompt 是触发后要注入的消息recurring 区分周期与首次命中后删除durable 决定是否写入 .scheduled_tasks.json。from dataclasses import dataclass, asdict from datetime import datetime import json import threading import time from pathlib import Path DURABLE_PATH Path(.scheduled_tasks.json) dataclass class CronJob: id: str cron: str prompt: str recurring: bool durable: bool scheduled_jobs: dict[str, CronJob] {} cron_queue: list[CronJob] [] cron_lock threading.Lock() agent_lock threading.Lock() _last_fired: dict[str, str] {}代码先把“定义”与“运行时容器”并列放置CronJob 描述一条规则四个全局对象记录规则当前怎样被管理。字段少不代表状态少同一个 job 的定义可以长期留在 scheduled_jobs某次命中又同时出现在 cron_queue而 _last_fired 只记录最近一次触发水位。区分定义、事件和水位才能解释重启时究竟恢复了什么。接下来是五段式匹配。cron 按“分钟、小时、月内日期、月份、星期”排列本地实现支持*、*/N、单值、范围和逗号列表。前三区域使用 ANDDOMday of month与 DOWday of week则根据是否为*选择单侧判断或 OR。def _cron_field_matches(field: str, value: int) - bool: if field *: return True for part in field.split(,): if part.startswith(*/): step int(part[2:]) if step 0 and value % step 0: return True elif - in part: start, end part.split(-) if int(start) value int(end): return True else: if int(part) value: return True return False def cron_matches(cron_expr: str, dt: datetime) - bool: minute, hour, dom, month, dow cron_expr.strip().split() dow_value (dt.weekday() 1) % 7 minute_ok _cron_field_matches(minute, dt.minute) hour_ok _cron_field_matches(hour, dt.hour) month_ok _cron_field_matches(month, dt.month) dom_ok _cron_field_matches(dom, dt.day) dow_ok _cron_field_matches(dow, dow_value) if not (minute_ok and hour_ok and month_ok): return False if dom * and dow *: return True if dom *: return dow_ok if dow *: return dom_ok return dom_ok or dow_ok这段函数按两级门禁阅读。第一层先淘汰分钟、小时、月份任一不符的时间点第二层只解决两个“日期选择器”怎样组合。把第二层直接改成dom_ok and dow_ok虽然更符合直觉却会改变常见 cron 语义。比如0 9 1 * 1不是“每月 1 日且星期一 09:00”而是月内日期为 1 或星期一时命中。这一点如果不说明表达式看似合法实际运行频率却可能远高于预期。持久化只保存 durableTrue 的 CronJobdef save_durable_jobs(): durable_jobs [asdict(job) for job in scheduled_jobs.values() if job.durable] DURABLE_PATH.write_text(json.dumps(durable_jobs, indent2)) def load_durable_jobs(): if not DURABLE_PATH.exists(): return for item in json.loads(DURABLE_PATH.read_text()): job CronJob(**item) scheduled_jobs[job.id] job保存和加载形成的是规则快照而不是触发事务。save_durable_jobs() 只筛选 durable job 并覆盖 JSONload_durable_jobs() 只重建对象索引这里没有保存某次 scheduled time 是否已产生事件、事件是否被 Agent 消费或上次执行是否成功。所以“重启后还能看到计划”只能证明定义恢复不能证明调度进度恢复。调度线程每秒轮询把命中点写入队列def cron_scheduler_loop(): while True: time.sleep(1) now datetime.now() minute_marker now.strftime(%Y-%m-%d %H:%M) with cron_lock: for job in list(scheduled_jobs.values()): if not cron_matches(job.cron, now): continue if _last_fired.get(job.id) minute_marker: continue cron_queue.append(job) _last_fired[job.id] minute_marker if not job.recurring: scheduled_jobs.pop(job.id, None) if job.durable: save_durable_jobs()循环中的顺序非常关键先判断当前分钟是否命中再检查 _last_fired随后把事件入队并更新水位。若先更新水位、入队前进程崩溃该分钟会被永久视为已处理若先入队、更新水位前崩溃重启或下一次轮询可能重复入队。教学代码用同一把进程锁缩小窗口却没有事务因此只能提供 best-effort 的进程内一致性。4. 验证请求一次未来唤醒如何变成模型可见输入调度线程只负责产生事件真正把它变成模型输入的是 queue processor 和 agent_loop。queue processor 轮询队列用 agent_lock 尝试抢占前台执行权若用户轮次正在运行它不阻塞等锁而是稍后再试。def has_cron_queue() - bool: with cron_lock: return len(cron_queue) 0 def consume_cron_queue() - list[dict]: with cron_lock: jobs list(cron_queue) cron_queue.clear() return [ {role: user, content: f[Scheduled] {job.prompt}} for job in jobs ] def queue_processor_loop(agent_loop): while True: time.sleep(0.5) if not has_cron_queue(): continue if not agent_lock.acquire(blockingFalse): continue try: if not has_cron_queue(): continue messages consume_cron_queue() agent_loop(messages) finally: agent_lock.release()从模型的视角看[Scheduled]消息是一条新环境输入从 Harness 的视角看它是时间事件到 message 的适配。适配后的roleuser不表示真实的人在这一刻键入文字它表示这是一条需要模型处理的新输入而不是对旧 tool call 的回答。这里存在两个不同的协议交点。当前轮注册计划时模型可能输出schedule_cron(cron, prompt, recurring, durable)Harness 执行 handler 后用原 tool_call_id 回填“已注册”。未来时间命中时早已没有那个待回答的 tool call因此不能复用旧 ID而要建立新消息。下面用 TaoToken 通道验证一次未来唤醒后的 Function Callingfrom openai import OpenAI client OpenAI( api_key你的_TaoToken_Key, base_urlhttps://taotoken.net/api ) tools [{ type: function, function: { name: schedule_cron, description: 注册一个未来定时唤醒计划, parameters: { type: object, properties: { cron: {type: string, description: 五段式 cron 表达式}, prompt: {type: string, description: 触发后注入的消息}, recurring: {type: boolean}, durable: {type: boolean} }, required: [cron, prompt] } } }] def agent_loop(messages): resp client.chat.completions.create( modelgpt-4o-mini, messagesmessages, toolstools, tool_choiceauto ) msg resp.choices[0].message if msg.tool_calls: for call in msg.tool_calls: if call.function.name schedule_cron: args json.loads(call.function.arguments) job CronJob( idfjob-{int(time.time())}, cronargs[cron], promptargs[prompt], recurringargs.get(recurring, True), durableargs.get(durable, True) ) with cron_lock: scheduled_jobs[job.id] job if job.durable: save_durable_jobs() messages.append({ role: tool, tool_call_id: call.id, content: f已注册计划 {job.id} }) return messages启动调度线程和消费线程threading.Thread(targetcron_scheduler_loop, daemonTrue).start() threading.Thread(targetqueue_processor_loop, args(agent_loop,), daemonTrue).start()成功结果长这样你给 Agent 发一句“帮我注册一个每 5 分钟检查构建状态的计划”模型返回schedule_crontool callHarness 写入 scheduled_jobs 并落盘。等到下一个 5 分钟边界cron_scheduler_loop 命中cron_queue 收到事件queue_processor_loop 抢到 agent_lockagent_loop 收到[Scheduled] 检查构建状态模型基于这条新输入继续发起新的 tool call。整条链跑通说明未来唤醒生效。5. 本篇常见错排查错误一重启后同一分钟重复触发。_last_fired 没有写盘在已命中的同一分钟重启后job 可以再入队。解法不是只把字典写盘而是用(job_id, scheduled_time)构造唯一触发键将产生事件和去重水位放进同一事务。错误二宕机期间的触发被静默跳过。重启只加载 CronJob不回放上次检查到当前时间之间的命中点。你需要明确 misfire policy立即补跑、仅补最近一次、在 deadline 内补跑或全部跳过。别默认“没触发就是没到点”。错误三时区不明确。datetime.now()使用运行主机的本地时间CronJob 没有 timezone 字段。迁移主机、容器时区不同或夏令时切换都可能让触发时刻变化。生产计划应显式保存 IANA timezone内部时间比较使用带时区对象。错误四多实例同时扫描同一批计划。多个进程或机器会同时扫描同一批计划各自有自己的 _last_fired同一分钟会各自入队一次。需要 leader election、数据库锁、分区所有权或支持幂等的多消费者队列而不是跨机器共享 threading.Lock。错误五把 [Scheduled] 当成 tool result 复用旧 ID。未来事件早已没有待回答的 tool call复用旧 tool_call_id 会导致协议配对失败。正确做法是建立新的 user message保留 job_id 和 scheduled_time 作为结构化元数据。错误六队列没有 durable acknowledgement。cron_queue 是内存 list取出后在模型调用前崩溃会丢事件。生产队列需要可见性超时、acknowledgement、死信和重放机制。提示Kubernetes CronJob 文档也明确把调度描述为近似行为某些情况下可能创建两个 Job 或不创建因此实际工作必须幂等。本地 Harness 更不应假设精确一次。6. 把时间轴接上模型通道下一步是并发到这里持久运行模块形成了三层连续关系第 12 篇保存“要做什么”第 13 篇处理“慢操作如何不阻塞”第 14 篇补上“何时主动开始”。CronJob 定义未来计划cron_matches() 定义命中规则.scheduled_tasks.json 让 durable job 跨进程恢复scheduler 把命中点写入 cron_queuequeue processor 使用 agent_lock 等待前台空闲Agent Loop 最后把事件变成新 user message。如果你还没把模型通道接上建议先去 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 拿一个 Key再对照 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 里的 Function Calling 字段说明跑一遍上面的验证代码。想先确认模型是否支持 tool call可以直接在 https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite 里试一轮对话。长期跑常驻 Agent 的话Coding Plan 会更省心https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 。但所有事件最终仍进入同一个 session_history 和同一把 agent_lock。当多项工作同时到来时单 Agent 会成为并发瓶颈第 15 篇将引入多个独立 Agent、邮箱和消息路由开始处理“谁来做”。