在学习 Agent-harness 的过程中我卡得最久的不是模型调用也不是工具接入反而是定时任务调度这一小块。之前做单次对话式 Agent 总觉得已经够了直到开始让 Agent 每天自动写行业日报、定时巡检线上服务、凌晨拉取数据并生成分析报告时才真正意识到让 Agent 跑起来和让 Agent 按点跑起来完全是两码事。如果你也在折腾 Agent-harness并且马上要给自己的 Agent 加上按时上班的能力那这篇文章应该能帮你少走点弯路。Agent-harness 本质上是一个把大模型、工具调用、记忆、权限控制组合在一起的编排框架它解决的核心问题是让 Agent 不再只是一次性的问答脚本而是能被你统一调度、持续运行的一等公民。定时任务调度就是其中最关键的一环cron 表达式怎么选、调度器怎么设计、任务执行超时怎么办、进程重启后任务会不会丢、多实例部署时会不会重复触发……这些问题在单机开发时几乎遇不到但一旦放到真实环境里跑的每一个都能让你查到怀疑人生。我把整个学习过程里的设计思路、代码实现、踩坑记录都整理在下面了。不是纯理论都是我能跑通的方案。1. 为什么 Agent 要按时上班先想清楚调度到底调什么1.1 Agent 不是定时脚本它有状态、有上下文、有外部依赖很多人第一反应是定时任务调度不就是个 cron 吗Linux 上写一行30 8 * * *就能每天 8 点半跑脚本有什么好学的。这个理解放到普通脚本场景里没有问题但放到 Agent-harness 里就完全不够用了。普通 cron 脚本是确定性的执行一个固定命令得到预期输出跑完就退出。但 Agent 不一样Agent 是一个有状态、有决策逻辑的流程。同样一个任务今天可能需要调搜索工具明天可能需要调数据库后天可能还要调用某个内部 API 获取数据。它的执行时间是不确定的外部依赖也是不确定的网络抖动、模型服务超时、工具返回格式变化都会让任务执行时间从几秒钟拉到几分钟。这意味着调度器不能简单粗暴地说到点了执行命令而是要负责在正确的时间点把 Agent 任务实例化并下发给任务带上完整的上下文参数、目标、相关数据跟踪任务从开始到结束的完整生命周期处理超时、失败、重试、并发控制。所以Agent 的定时调度本质上是任务编排而不是简单的定时触发。这是理解整个体系的前提。1.2 典型应用场景与触发粒度选择我目前把 Agent-harness 的定时任务分为四类每一类对调度能力的要求都不一样场景触发方式典型周期核心诉求每日定时报告cron 表达式每天一次准点触发结果可靠异步任务处理延迟触发用户提交后 5 分钟不阻塞主流程周期性巡检固定间隔每 10 分钟错峰执行避免堆积补数/回填一次性定时指定时间执行可重复执行幂等触发粒度也是需要提前想清楚的。如果你只是让 Agent 每天生成一份报告秒级触发完全没有意义分钟级就足够了。但如果你要做实时性较强的监控任务可能就需要秒级调度。这里我踩过一个坑任务处理耗时本身就超过了调度周期但调度器还在往队列里塞任务导致任务越积越多。事后总结了一个经验调度的最小周期最好不要小于任务平均耗时的三倍否则并发堆积几乎是必然的。2. Agent-harness 中调度的完整生命周期从任务注册到结果回传2.1 一个调度任务应该包含哪些字段在 Agent-harness 里第一步不是写调度逻辑而是定义任务模型。我建议不要把调度配置散落在代码里而是给调度任务单独建一个结构统一管理。下面是我在项目里使用的任务定义示例{ task_id: daily_report_001, agent_id: report_agent, trigger: { type: cron, expr: 0 30 8 * * *, timezone: Asia/Shanghai }, execution: { timeout_seconds: 600, max_retries: 3, concurrency_limit: 2 }, context: { params: { region: cn-north, output_format: markdown } }, notify: { on_success: [webhook://report-sink], on_failure: [webhook://alert] } }为什么任务定义要单独维护因为 Agent 的执行上下文非常容易膨胀。你可能会在后期给任务加触发条件、加过滤器、加数据源配置如果这些全都写在 Agent 内部最后会变成一团乱麻。把任务定义和 Agent 执行逻辑解耦之后调度器只需要关心什么时间该触发哪个任务Agent 只关心拿到 context 后怎么执行各司其职维护成本会低很多。2.2 任务状态机调度器必须能回答这个任务现在到哪了调度器的核心能力是状态管理。一个任务从被触发到最终完成会经历多个状态。我设计的简化状态流转如下Pending任务已经到了触发时间等待 worker 接收Runningworker 正在执行 Agent 流程SucceededAgent 执行成功结果已存储FailedAgent 执行失败等待重试Retrying进入退避重试流程Timeout任务执行超过最长时限强制终止Skipped由于并发限制或上游依赖未满足本次任务不执行。这里最容易被忽略的是Skipped状态。之前我做过一个巡检任务每次执行需要 20 秒但 cron 设置了每 10 秒触发一次造成大量任务排队。后来加了并发控制超出限制的任务不是丢弃而是进入 Skipped 状态并记录原因。这样可以保留现场排查问题时不会两眼一抹黑。我建议在 Agent-harness 中实现一个简单的状态机不管用数据库字段还是内存对象都行。目的是让调度器随时能回答三个问题今天哪些任务触发了哪些还在跑哪些已经失败失败是重试了还是完全放弃了做不到这一点调度器就只能算是一个高级定时器。2.3 漏执行风险没有触发记录一切等于没发生定时任务调度最隐蔽的问题是漏执行。如果调度器进程挂了或者线程池满了任务根本没被触发但外部没有感知。普通脚本任务漏跑一次可能问题不大Agent 的巡检任务漏跑一次可能就会错过一个线上告警。所以我后来强制要求每个任务触发时都要写一条触发记录包含scheduled_at计划触发时间、actual_started_at实际开始时间、finished_at结束时间、status状态。这样即使出现问题也能通过记录发现问题。我还会定期统计实际开始时间 - 计划触发时间这个差值如果超过阈值就说明调度器本身有延迟需要排查是否线程池耗尽或者任务队列积压。3. 手写一个轻量调度器从 cron 解析到任务下发的核心原理3.1 为什么我选择自己封装一层而不是直接用现成框架Agent-harness 里的定时任务调度和普通后端服务的定时任务有一个很大的区别普通服务调度的往往是短平快的函数而 Agent 任务可能要跑很久还伴随着模型调用、工具调用等不可控因素。成熟的调度框架比如 APScheduler 确实很好用但直接在 Agent-harness 里用会遇到几个问题Agent 任务的超时控制非常关键但框架默认的 misfire 策略不一定符合你的预期我们需要精确控制 worker 并发数防止多个 Agent 同时抢占外部资源框架的 job store 是持久化机制但 Agent 任务的上下文比普通函数参数复杂得多。所以我的做法是用 APScheduler 做底层触发但外面自己包了一层任务车间。下面为了把原理讲清楚我会直接手写一个最小可运行的调度循环你会发现核心逻辑其实不复杂。3.2 cron 表达式解析不要每一秒都遍历任务要用下次触发时间很多初学者写定时器时会犯一个错误搞一个 while True 循环每秒钟扫一遍所有任务判断当前时间是否匹配 cron 表达式。这种做法在小规模任务下没问题但任务一多每分钟要执行几千次字符串匹配纯属浪费资源。正确做法是利用croniter库直接计算每个任务的下一次触发时间然后让调度线程 sleep 到最近的那个时间点。这样不管有多少个任务调度循环本身的消耗都非常低。from croniter import croniter from datetime import datetime def get_next_run_time(cron_expr: str, base: datetime | None None) - datetime: base base or datetime.now() itr croniter(cron_expr, base) return itr.get_next(datetime)这里有一个关键细节get_next(datetime)返回的是严格大于 base 的下一个时间点。所以如果任务应该立即执行我不会用当前时间作为 base而是用上一次计划时间减一秒钟否则会漏掉当前这一轮。3.3 调度主循环与 worker 线程池当任务注册进来之后调度器需要维护一个按下次触发时间排序的队列。主循环不断获取最近的任务到期后放进执行队列由 worker 线程池去真正执行 Agent 流程。下面是最小实现的关键代码import queue import threading import time from datetime import datetime class Scheduler: def __init__(self): self.tasks [] self.task_queue queue.Queue() self.running True def register(self, task): # task.next_run_time 由 croniter 计算得出 self.tasks.append(task) def _tick(self): 检查到期的任务放进执行队列 now datetime.now() for task in self.tasks: if task.next_run_time and task.next_run_time now: # 如果任务设置了并发独占这里可以加判断 self.task_queue.put(task) task.next_run_time task.compute_next_run_time() def run(self): while self.running: self._tick() time.sleep(0.5) def worker_loop(self): while self.running: task self.task_queue.get() try: result task.agent.run(task.context) self._handle_success(task, result) except Exception as e: self._handle_failure(task, e) finally: self.task_queue.task_done() def start(self): threading.Thread(targetself.run, daemonTrue).start() for _ in range(4): threading.Thread(targetself.worker_loop, daemonTrue).start()调度线程只负责把到期的任务投递到队列worker 线程池负责真正执行。为什么要分离因为 Agent 任务耗时不可控如果调度线程直接去执行任务下一个任务就没人在意了。分离之后调度线程永远只做轻量级的检查工作就算 worker 全部阻塞调度本身也不会停止。3.4 任务结果如何回传和存储Agent 任务执行完结果不能直接丢在函数返回值里因为定时任务是异步的。我通常把结果写入任务表同时在任务完成时触发回调。回调可以是 webhook、消息队列也可以只是更新状态字段。关键在于结果存储和状态更新必须和任务执行解耦。我用过的方案是任务执行成功后worker 把结果存到对象存储或者数据库随后发送一个完成事件到内部消息通道。订阅者收到事件后可以推送报告、更新看板、发送通知。这样即使后续要扩展更多下游处理也不用改动调度器本身。4. 真实部署里最容易踩的坑时区、阻塞、重试风暴4.1 时区问题cron 里的每天 8 点到底是谁的 8 点这是我踩过的第一个大坑。有一次我在服务器上配置了每天 8 点的报告任务结果早上 6 点就收到了报告推送。排查了半天发现问题出在容器内时区是 UTC而业务用的是东八区。cron 表达式0 8 * * *匹配的是 UTC 8 点换算成东八区就成了下午 4 点。我配置时以为服务器用的东八区实际跑起来却完全对不上。解决方式有两层。第一层是统一存储标准任务定义里的时间一律存 UTC展示和配置界面再做时区转换。Agent-harness 的任务触发时间最终执行时也统一用 UTC 计算不依赖服务器本地时区。第二层是给每个任务显式指定时区字段。上面任务定义里我写了timezone: Asia/Shanghai调度器在计算下一次触发时间之前会把基准时间先转换到任务指定时区算完再转回 UTC这样无论调度器部署在哪个区域都不会出错。4.2 worker 阻塞一个 Agent 卡住后续任务全部排队Agent 任务和普通任务最大的不同是不确定性。模型 API 可能响应很慢工具调用可能一直挂着外部服务可能无响应。刚开始我把 worker 线程池开到 8想着足够用了结果有一天一个 Agent 在调外部搜索接口时整个线程卡住紧接着后面十几个任务全部堆积在队列里调度延迟越来越大。这个问题要从两个方向解决给 Agent 执行过程设置总超时。在 Agent-harness 的任务配置里加上timeout_secondsworker 执行时用future.result(timeouttimeout)或者 asyncio.wait_for 控制。超时后强制终止任务标记为 Timeout。用信号量控制 Agent 并发数。尤其是多个任务会调用同一个下游服务时一定要做并发限制。比如规定同一时刻最多有 3 个任务在调用外部搜索服务超出就排队或跳过。还有一个经验不要只给调度器设置超时给 Agent 内部的每一步工具调用也要设超时。模型调用、HTTP 请求、数据库连接全都加上明确的超时时间。否则一个 Agent 卡在某个工具调用上整个任务的超时控制就形同虚设。4.3 重试风暴失败任务不断重试反而把下游接口打挂第一个版本上线不久我遇到了一个比阻塞更麻烦的问题凌晨某个下游数据源连接池满了Agent 执行失败后立刻重试重试又失败又重试直接把下游服务打到雪崩。后来我给所有任务的重试逻辑加了两个限定最大重试次数我一般设置 3 次指数退避 随机抖动避免所有失败任务在同一时间反复重试。退避时间的计算方式很简单import random def retry_delay(attempt: int, base_seconds: int 2, cap_seconds: int 60) - float: exp_backoff min(base_seconds * (2 ** attempt), cap_seconds) jitter random.uniform(0, 0.5) return exp_backoff jitter第 1 次失败等待约 2 秒第 2 次约 4 秒第 3 次约 8 秒封顶 60 秒。随机抖动可以避免多个任务恰好同一时刻重试。但这里有一个更关键的隔离策略当某种类型的任务连续失败超过阈值时应该触发熔断直接停止该类任务一段时间而不是继续快速重试。比如巡检任务连续 5 次失败说明下游大概率挂了再重试只会加重问题。这时候把任务置为 Failed发一次告警就够了等人工介入或系统恢复后再由补偿任务重新执行。5. 从单机定时到分布式调度扩展路径与关键设计5.1 为什么必须持久化任务进程重启后任务不能丢用纯内存队列跑调度开发时很方便但一旦部署成服务问题就来了调度器进程只要一重启所有待执行任务、下一次触发时间、历史记录全部丢失。对 Agent-harness 这种长期运行的服务来说这是不可接受的。所以做分布式调度之前第一步是持久化任务定义和任务状态。我用的方案是给调度器加一个数据库表存储任务定义、下一次触发时间、最近触发时间、状态等字段。调度器启动时从数据库加载所有启用中的任务恢复各自的下一次触发时间然后重新进入调度循环。持久化的意义不只是防止丢失更是让多个调度器实例可以共享同一份任务状态。5.2 多实例部署时的重复调度问题当调度器运行多个副本时一个新的问题出现了如果两个实例都在各自的内存里维护同一个任务的下一次触发时间那么任务到点时两个实例都会触发一次导致 Agent 任务重复执行。解决重复调度的核心是分布式锁 原子更新。比较简单的实现方式是用 Redis 分布式锁每次任务触发前先尝试获取锁锁的 key 设计为schedule:lock:{task_id}:{scheduled_time}只有拿到锁的实例才能把任务投递到执行队列锁的过期时间要大于任务最长执行时间否则执行到一半锁过期另一个实例又会重复执行。还有一种方式是数据库唯一约束在触发记录表中为task_id scheduled_time建唯一索引。任务触发时先插入触发记录如果插入冲突说明其他实例已经处理过这个时间点了当前实例直接跳过。这个方案不需要额外引入 Redis实现简单在高并发下也够用。5.3 worker 心跳与任务重新分配任务被某个 worker 实例领取后如果该实例宕机了这个任务就会永远卡在 Running 状态。要解决这个问题需要引入心跳机制。每个 worker 定期写入自己的心跳时间调度器如果发现某个 worker 超过阈值没有心跳就把该 worker 上正在运行的任务重新标记为 Pending让其他 worker 再次领取。不过这里有一个非常重要的前提Agent 任务必须是幂等的。因为对方 worker 可能并不是真的挂了只是网络分区导致心跳没有上报。如果你盲目把任务重新分配就可能两个 worker 同时执行同一个任务。Agent 任务往往涉及外部系统调用比如发通知、写数据、调用支付接口不幂等的话会造成严重事故。我的建议是在 Agent-harness 里给每个任务关联一个全局唯一的执行 ID结果回写带上这个 ID。外部系统如果收到重复请求可以根据执行 ID 去重。如果做不到那就必须接受至少一次的语义在业务层处理好重复执行的后果。6. 定时调度之后向事件驱动演化6.1 定时触发只是起点事件流才是 Agent 的常态如果把 Agent-harness 的调度体系只限定在 cron 表达式上会发现很多场景非常尴尬。比如用户提交了一个数据清洗需求处理完成后要通知下游系统这时候难道要等下一次 cron 触发吗显然不合理。定时任务适合固定节奏的场景但真实业务里Agent 更多是被事件驱动的用户提交、消息到达、系统告警、数据更新都是一个个事件。我现在的做法是定时触发 事件触发混用。定时触发作为兜底比如每天凌晨做一次全量巡检事件触发作为主力比如收到 webhook 后立即启动 Agent 流程。Agent-harness 里可以抽象出一个触发源的概念一个任务可以有多个触发源{ task_id: data_pipeline_001, triggers: [ {type: cron, expr: 0 2 * * *}, {type: webhook, path: /hooks/data-pipeline}, {type: queue, topic: data.updated} ] }6.2 设计一个触发器路由器的思路从架构上看调度器在事件驱动阶段会演变成一个触发器路由器它不再主动计算时间而是监听多个输入源把任何形式的触发统一转换成标准任务事件。cron 只是其中一个输入源和 webhook、消息队列、数据库 binlog 都平起平坐。这样设计的收益很明显Agent 任务的注册方不需要关心触发源的具体实现只需要声明我这个任务在什么条件下应该执行。调度器内部维护一套统一的任务执行管道不管是定时触发还是事件触发最终都走同一条队列、同一套 worker、同一套状态管理。从定时任务调度到事件驱动这个演进过程不是推翻重来而是把调度器从看时间升级为看事件。每一步都是在前面踩坑的基础上一点点补齐的。对 Agent-harness 来说定时任务不只是一个功能点它其实是理解 Agent 工程化的一个很好的入口状态机、持久化、并发、幂等、失败恢复这些分布式系统的基本功都会在调度模块里完整地过一遍。
Agent定时任务调度实战：从cron到分布式锁的完整指南
在学习 Agent-harness 的过程中我卡得最久的不是模型调用也不是工具接入反而是定时任务调度这一小块。之前做单次对话式 Agent 总觉得已经够了直到开始让 Agent 每天自动写行业日报、定时巡检线上服务、凌晨拉取数据并生成分析报告时才真正意识到让 Agent 跑起来和让 Agent 按点跑起来完全是两码事。如果你也在折腾 Agent-harness并且马上要给自己的 Agent 加上按时上班的能力那这篇文章应该能帮你少走点弯路。Agent-harness 本质上是一个把大模型、工具调用、记忆、权限控制组合在一起的编排框架它解决的核心问题是让 Agent 不再只是一次性的问答脚本而是能被你统一调度、持续运行的一等公民。定时任务调度就是其中最关键的一环cron 表达式怎么选、调度器怎么设计、任务执行超时怎么办、进程重启后任务会不会丢、多实例部署时会不会重复触发……这些问题在单机开发时几乎遇不到但一旦放到真实环境里跑的每一个都能让你查到怀疑人生。我把整个学习过程里的设计思路、代码实现、踩坑记录都整理在下面了。不是纯理论都是我能跑通的方案。1. 为什么 Agent 要按时上班先想清楚调度到底调什么1.1 Agent 不是定时脚本它有状态、有上下文、有外部依赖很多人第一反应是定时任务调度不就是个 cron 吗Linux 上写一行30 8 * * *就能每天 8 点半跑脚本有什么好学的。这个理解放到普通脚本场景里没有问题但放到 Agent-harness 里就完全不够用了。普通 cron 脚本是确定性的执行一个固定命令得到预期输出跑完就退出。但 Agent 不一样Agent 是一个有状态、有决策逻辑的流程。同样一个任务今天可能需要调搜索工具明天可能需要调数据库后天可能还要调用某个内部 API 获取数据。它的执行时间是不确定的外部依赖也是不确定的网络抖动、模型服务超时、工具返回格式变化都会让任务执行时间从几秒钟拉到几分钟。这意味着调度器不能简单粗暴地说到点了执行命令而是要负责在正确的时间点把 Agent 任务实例化并下发给任务带上完整的上下文参数、目标、相关数据跟踪任务从开始到结束的完整生命周期处理超时、失败、重试、并发控制。所以Agent 的定时调度本质上是任务编排而不是简单的定时触发。这是理解整个体系的前提。1.2 典型应用场景与触发粒度选择我目前把 Agent-harness 的定时任务分为四类每一类对调度能力的要求都不一样场景触发方式典型周期核心诉求每日定时报告cron 表达式每天一次准点触发结果可靠异步任务处理延迟触发用户提交后 5 分钟不阻塞主流程周期性巡检固定间隔每 10 分钟错峰执行避免堆积补数/回填一次性定时指定时间执行可重复执行幂等触发粒度也是需要提前想清楚的。如果你只是让 Agent 每天生成一份报告秒级触发完全没有意义分钟级就足够了。但如果你要做实时性较强的监控任务可能就需要秒级调度。这里我踩过一个坑任务处理耗时本身就超过了调度周期但调度器还在往队列里塞任务导致任务越积越多。事后总结了一个经验调度的最小周期最好不要小于任务平均耗时的三倍否则并发堆积几乎是必然的。2. Agent-harness 中调度的完整生命周期从任务注册到结果回传2.1 一个调度任务应该包含哪些字段在 Agent-harness 里第一步不是写调度逻辑而是定义任务模型。我建议不要把调度配置散落在代码里而是给调度任务单独建一个结构统一管理。下面是我在项目里使用的任务定义示例{ task_id: daily_report_001, agent_id: report_agent, trigger: { type: cron, expr: 0 30 8 * * *, timezone: Asia/Shanghai }, execution: { timeout_seconds: 600, max_retries: 3, concurrency_limit: 2 }, context: { params: { region: cn-north, output_format: markdown } }, notify: { on_success: [webhook://report-sink], on_failure: [webhook://alert] } }为什么任务定义要单独维护因为 Agent 的执行上下文非常容易膨胀。你可能会在后期给任务加触发条件、加过滤器、加数据源配置如果这些全都写在 Agent 内部最后会变成一团乱麻。把任务定义和 Agent 执行逻辑解耦之后调度器只需要关心什么时间该触发哪个任务Agent 只关心拿到 context 后怎么执行各司其职维护成本会低很多。2.2 任务状态机调度器必须能回答这个任务现在到哪了调度器的核心能力是状态管理。一个任务从被触发到最终完成会经历多个状态。我设计的简化状态流转如下Pending任务已经到了触发时间等待 worker 接收Runningworker 正在执行 Agent 流程SucceededAgent 执行成功结果已存储FailedAgent 执行失败等待重试Retrying进入退避重试流程Timeout任务执行超过最长时限强制终止Skipped由于并发限制或上游依赖未满足本次任务不执行。这里最容易被忽略的是Skipped状态。之前我做过一个巡检任务每次执行需要 20 秒但 cron 设置了每 10 秒触发一次造成大量任务排队。后来加了并发控制超出限制的任务不是丢弃而是进入 Skipped 状态并记录原因。这样可以保留现场排查问题时不会两眼一抹黑。我建议在 Agent-harness 中实现一个简单的状态机不管用数据库字段还是内存对象都行。目的是让调度器随时能回答三个问题今天哪些任务触发了哪些还在跑哪些已经失败失败是重试了还是完全放弃了做不到这一点调度器就只能算是一个高级定时器。2.3 漏执行风险没有触发记录一切等于没发生定时任务调度最隐蔽的问题是漏执行。如果调度器进程挂了或者线程池满了任务根本没被触发但外部没有感知。普通脚本任务漏跑一次可能问题不大Agent 的巡检任务漏跑一次可能就会错过一个线上告警。所以我后来强制要求每个任务触发时都要写一条触发记录包含scheduled_at计划触发时间、actual_started_at实际开始时间、finished_at结束时间、status状态。这样即使出现问题也能通过记录发现问题。我还会定期统计实际开始时间 - 计划触发时间这个差值如果超过阈值就说明调度器本身有延迟需要排查是否线程池耗尽或者任务队列积压。3. 手写一个轻量调度器从 cron 解析到任务下发的核心原理3.1 为什么我选择自己封装一层而不是直接用现成框架Agent-harness 里的定时任务调度和普通后端服务的定时任务有一个很大的区别普通服务调度的往往是短平快的函数而 Agent 任务可能要跑很久还伴随着模型调用、工具调用等不可控因素。成熟的调度框架比如 APScheduler 确实很好用但直接在 Agent-harness 里用会遇到几个问题Agent 任务的超时控制非常关键但框架默认的 misfire 策略不一定符合你的预期我们需要精确控制 worker 并发数防止多个 Agent 同时抢占外部资源框架的 job store 是持久化机制但 Agent 任务的上下文比普通函数参数复杂得多。所以我的做法是用 APScheduler 做底层触发但外面自己包了一层任务车间。下面为了把原理讲清楚我会直接手写一个最小可运行的调度循环你会发现核心逻辑其实不复杂。3.2 cron 表达式解析不要每一秒都遍历任务要用下次触发时间很多初学者写定时器时会犯一个错误搞一个 while True 循环每秒钟扫一遍所有任务判断当前时间是否匹配 cron 表达式。这种做法在小规模任务下没问题但任务一多每分钟要执行几千次字符串匹配纯属浪费资源。正确做法是利用croniter库直接计算每个任务的下一次触发时间然后让调度线程 sleep 到最近的那个时间点。这样不管有多少个任务调度循环本身的消耗都非常低。from croniter import croniter from datetime import datetime def get_next_run_time(cron_expr: str, base: datetime | None None) - datetime: base base or datetime.now() itr croniter(cron_expr, base) return itr.get_next(datetime)这里有一个关键细节get_next(datetime)返回的是严格大于 base 的下一个时间点。所以如果任务应该立即执行我不会用当前时间作为 base而是用上一次计划时间减一秒钟否则会漏掉当前这一轮。3.3 调度主循环与 worker 线程池当任务注册进来之后调度器需要维护一个按下次触发时间排序的队列。主循环不断获取最近的任务到期后放进执行队列由 worker 线程池去真正执行 Agent 流程。下面是最小实现的关键代码import queue import threading import time from datetime import datetime class Scheduler: def __init__(self): self.tasks [] self.task_queue queue.Queue() self.running True def register(self, task): # task.next_run_time 由 croniter 计算得出 self.tasks.append(task) def _tick(self): 检查到期的任务放进执行队列 now datetime.now() for task in self.tasks: if task.next_run_time and task.next_run_time now: # 如果任务设置了并发独占这里可以加判断 self.task_queue.put(task) task.next_run_time task.compute_next_run_time() def run(self): while self.running: self._tick() time.sleep(0.5) def worker_loop(self): while self.running: task self.task_queue.get() try: result task.agent.run(task.context) self._handle_success(task, result) except Exception as e: self._handle_failure(task, e) finally: self.task_queue.task_done() def start(self): threading.Thread(targetself.run, daemonTrue).start() for _ in range(4): threading.Thread(targetself.worker_loop, daemonTrue).start()调度线程只负责把到期的任务投递到队列worker 线程池负责真正执行。为什么要分离因为 Agent 任务耗时不可控如果调度线程直接去执行任务下一个任务就没人在意了。分离之后调度线程永远只做轻量级的检查工作就算 worker 全部阻塞调度本身也不会停止。3.4 任务结果如何回传和存储Agent 任务执行完结果不能直接丢在函数返回值里因为定时任务是异步的。我通常把结果写入任务表同时在任务完成时触发回调。回调可以是 webhook、消息队列也可以只是更新状态字段。关键在于结果存储和状态更新必须和任务执行解耦。我用过的方案是任务执行成功后worker 把结果存到对象存储或者数据库随后发送一个完成事件到内部消息通道。订阅者收到事件后可以推送报告、更新看板、发送通知。这样即使后续要扩展更多下游处理也不用改动调度器本身。4. 真实部署里最容易踩的坑时区、阻塞、重试风暴4.1 时区问题cron 里的每天 8 点到底是谁的 8 点这是我踩过的第一个大坑。有一次我在服务器上配置了每天 8 点的报告任务结果早上 6 点就收到了报告推送。排查了半天发现问题出在容器内时区是 UTC而业务用的是东八区。cron 表达式0 8 * * *匹配的是 UTC 8 点换算成东八区就成了下午 4 点。我配置时以为服务器用的东八区实际跑起来却完全对不上。解决方式有两层。第一层是统一存储标准任务定义里的时间一律存 UTC展示和配置界面再做时区转换。Agent-harness 的任务触发时间最终执行时也统一用 UTC 计算不依赖服务器本地时区。第二层是给每个任务显式指定时区字段。上面任务定义里我写了timezone: Asia/Shanghai调度器在计算下一次触发时间之前会把基准时间先转换到任务指定时区算完再转回 UTC这样无论调度器部署在哪个区域都不会出错。4.2 worker 阻塞一个 Agent 卡住后续任务全部排队Agent 任务和普通任务最大的不同是不确定性。模型 API 可能响应很慢工具调用可能一直挂着外部服务可能无响应。刚开始我把 worker 线程池开到 8想着足够用了结果有一天一个 Agent 在调外部搜索接口时整个线程卡住紧接着后面十几个任务全部堆积在队列里调度延迟越来越大。这个问题要从两个方向解决给 Agent 执行过程设置总超时。在 Agent-harness 的任务配置里加上timeout_secondsworker 执行时用future.result(timeouttimeout)或者 asyncio.wait_for 控制。超时后强制终止任务标记为 Timeout。用信号量控制 Agent 并发数。尤其是多个任务会调用同一个下游服务时一定要做并发限制。比如规定同一时刻最多有 3 个任务在调用外部搜索服务超出就排队或跳过。还有一个经验不要只给调度器设置超时给 Agent 内部的每一步工具调用也要设超时。模型调用、HTTP 请求、数据库连接全都加上明确的超时时间。否则一个 Agent 卡在某个工具调用上整个任务的超时控制就形同虚设。4.3 重试风暴失败任务不断重试反而把下游接口打挂第一个版本上线不久我遇到了一个比阻塞更麻烦的问题凌晨某个下游数据源连接池满了Agent 执行失败后立刻重试重试又失败又重试直接把下游服务打到雪崩。后来我给所有任务的重试逻辑加了两个限定最大重试次数我一般设置 3 次指数退避 随机抖动避免所有失败任务在同一时间反复重试。退避时间的计算方式很简单import random def retry_delay(attempt: int, base_seconds: int 2, cap_seconds: int 60) - float: exp_backoff min(base_seconds * (2 ** attempt), cap_seconds) jitter random.uniform(0, 0.5) return exp_backoff jitter第 1 次失败等待约 2 秒第 2 次约 4 秒第 3 次约 8 秒封顶 60 秒。随机抖动可以避免多个任务恰好同一时刻重试。但这里有一个更关键的隔离策略当某种类型的任务连续失败超过阈值时应该触发熔断直接停止该类任务一段时间而不是继续快速重试。比如巡检任务连续 5 次失败说明下游大概率挂了再重试只会加重问题。这时候把任务置为 Failed发一次告警就够了等人工介入或系统恢复后再由补偿任务重新执行。5. 从单机定时到分布式调度扩展路径与关键设计5.1 为什么必须持久化任务进程重启后任务不能丢用纯内存队列跑调度开发时很方便但一旦部署成服务问题就来了调度器进程只要一重启所有待执行任务、下一次触发时间、历史记录全部丢失。对 Agent-harness 这种长期运行的服务来说这是不可接受的。所以做分布式调度之前第一步是持久化任务定义和任务状态。我用的方案是给调度器加一个数据库表存储任务定义、下一次触发时间、最近触发时间、状态等字段。调度器启动时从数据库加载所有启用中的任务恢复各自的下一次触发时间然后重新进入调度循环。持久化的意义不只是防止丢失更是让多个调度器实例可以共享同一份任务状态。5.2 多实例部署时的重复调度问题当调度器运行多个副本时一个新的问题出现了如果两个实例都在各自的内存里维护同一个任务的下一次触发时间那么任务到点时两个实例都会触发一次导致 Agent 任务重复执行。解决重复调度的核心是分布式锁 原子更新。比较简单的实现方式是用 Redis 分布式锁每次任务触发前先尝试获取锁锁的 key 设计为schedule:lock:{task_id}:{scheduled_time}只有拿到锁的实例才能把任务投递到执行队列锁的过期时间要大于任务最长执行时间否则执行到一半锁过期另一个实例又会重复执行。还有一种方式是数据库唯一约束在触发记录表中为task_id scheduled_time建唯一索引。任务触发时先插入触发记录如果插入冲突说明其他实例已经处理过这个时间点了当前实例直接跳过。这个方案不需要额外引入 Redis实现简单在高并发下也够用。5.3 worker 心跳与任务重新分配任务被某个 worker 实例领取后如果该实例宕机了这个任务就会永远卡在 Running 状态。要解决这个问题需要引入心跳机制。每个 worker 定期写入自己的心跳时间调度器如果发现某个 worker 超过阈值没有心跳就把该 worker 上正在运行的任务重新标记为 Pending让其他 worker 再次领取。不过这里有一个非常重要的前提Agent 任务必须是幂等的。因为对方 worker 可能并不是真的挂了只是网络分区导致心跳没有上报。如果你盲目把任务重新分配就可能两个 worker 同时执行同一个任务。Agent 任务往往涉及外部系统调用比如发通知、写数据、调用支付接口不幂等的话会造成严重事故。我的建议是在 Agent-harness 里给每个任务关联一个全局唯一的执行 ID结果回写带上这个 ID。外部系统如果收到重复请求可以根据执行 ID 去重。如果做不到那就必须接受至少一次的语义在业务层处理好重复执行的后果。6. 定时调度之后向事件驱动演化6.1 定时触发只是起点事件流才是 Agent 的常态如果把 Agent-harness 的调度体系只限定在 cron 表达式上会发现很多场景非常尴尬。比如用户提交了一个数据清洗需求处理完成后要通知下游系统这时候难道要等下一次 cron 触发吗显然不合理。定时任务适合固定节奏的场景但真实业务里Agent 更多是被事件驱动的用户提交、消息到达、系统告警、数据更新都是一个个事件。我现在的做法是定时触发 事件触发混用。定时触发作为兜底比如每天凌晨做一次全量巡检事件触发作为主力比如收到 webhook 后立即启动 Agent 流程。Agent-harness 里可以抽象出一个触发源的概念一个任务可以有多个触发源{ task_id: data_pipeline_001, triggers: [ {type: cron, expr: 0 2 * * *}, {type: webhook, path: /hooks/data-pipeline}, {type: queue, topic: data.updated} ] }6.2 设计一个触发器路由器的思路从架构上看调度器在事件驱动阶段会演变成一个触发器路由器它不再主动计算时间而是监听多个输入源把任何形式的触发统一转换成标准任务事件。cron 只是其中一个输入源和 webhook、消息队列、数据库 binlog 都平起平坐。这样设计的收益很明显Agent 任务的注册方不需要关心触发源的具体实现只需要声明我这个任务在什么条件下应该执行。调度器内部维护一套统一的任务执行管道不管是定时触发还是事件触发最终都走同一条队列、同一套 worker、同一套状态管理。从定时任务调度到事件驱动这个演进过程不是推翻重来而是把调度器从看时间升级为看事件。每一步都是在前面踩坑的基础上一点点补齐的。对 Agent-harness 来说定时任务不只是一个功能点它其实是理解 Agent 工程化的一个很好的入口状态机、持久化、并发、幂等、失败恢复这些分布式系统的基本功都会在调度模块里完整地过一遍。
