Cron + AI Agent 生产级实践:我让程序自动干活几十天,踩过的 5 个坑全在这

Cron + AI Agent 生产级实践:我让程序自动干活几十天,踩过的 5 个坑全在这
摘要把 AI Agent 接上 cron 定时任务听起来很酷——程序自己写文章、自己发、自己监控。但跑了几十天之后我发现真正的挑战不是「怎么让 AI 干活」而是「怎么让 AI 在没人看的情况下不翻车」。这篇文章不讲 MCP 协议是什么讲的是真实生产环境里踩过的坑和防御策略。我是在一个周三的早上意识到出事的。打开飞书发现通知列表被刷屏了。同一个 AI 写的日报被发了 6 遍。每条间隔 2 分钟。cron 表达式写错了——本来是每 30 分钟检查一次结果被解析成了每 2 分钟跑一次。AI Agent 很乖它只是在做我让它做的事。这还算轻的。去翻 GitHub 社区有个哥们更惨——凌晨 3:47他的内容 Agent 往 Discord 发了 47 遍同一篇文章。社区直接把他 bot 静音了。标题就是那篇著名的帖子CRON Jobs: When Your AI Agent Schedule Goes Rogue。所以这篇文章想聊的不是「如何用 AI Agent 写文章」或者「MCP 协议是啥」——市面上已经有太多那种文章了。我想聊的是当你真的把 AI Agent 接上 cron 跑了几十天之后会踩到什么坑以及怎么修。为什么 Cron AI Agent 比 Cron 脚本危险得多传统的 cron 任务跑的是确定性脚本06***/usr/local/bin/backup.sh/var/log/backup.log21这个脚本的行为是固定的。输入相同 → 输出相同。出问题了看日志改脚本重跑。AI Agent 的 cron 任务完全不是一回事非确定性同一个 prompt不同时间跑结果可能完全不同。模型版本更新了、上下文变了、搜索结果变了——都可能导致输出大变。有「主动性」脚本只会做你写死的事。Agent 会自己决定做什么。如果 prompt 里有歧义它会自己发挥。状态依赖Agent 的决策依赖上下文历史对话、文件系统、工具返回的结果而这些状态可能在两次 cron 执行之间悄悄变化。失败模式不可预测脚本失败通常是 exit code ≠ 0。Agent 失败可能是写了一篇完全跑题的文章、发了空内容、或者陷入死循环反复调同一个工具。维度传统 Cron 脚本Cron AI Agent执行逻辑确定性非确定性错误发现exit code / 日志需要审阅输出内容重复执行幂等通常可能产生重复输出限流保护手动实现必须内置成本几乎为零每次调用消耗 token可回滚重跑脚本可能需要手动修复说白了用 cron 调度 AI Agent 等于把一辆自动驾驶汽车放到没有护栏的山路上——它自己能开但你得给它装好多层保险。架构MCP 协议是怎么串起整个流程的在聊具体代码之前先看一眼整体架构。MCPModel Context Protocol在这里扮演的角色是工具总线——Agent 通过 MCP 客户端连接多个 MCP 服务器每个服务器暴露一组工具。flowchartTBCRON[ Cron 定时触发]--AGENT[ AI Agent]AGENT--MCP_CLIENT[ MCP Client]MCP_CLIENT--SRV1[ 搜索 MCP Serverbr/(web_search / anysearch)]MCP_CLIENT--SRV2[ 抓取 MCP Serverbr/(firecrawl / web_extract)]MCP_CLIENT--SRV3[ 发布 MCP Serverbr/(publish.py)]MCP_CLIENT--SRV4[ 通知 MCP Serverbr/(feishu / slack)]SRV1--INTERNET[ 互联网]SRV2--INTERNETSRV3--CSDN[ CSDN 平台]SRV4--NOTIFY[ 飞书通知]AGENT--FILESYS[ 本地文件系统br/(articles/ , logs/)]整个流程是这样的Cron到点触发 AgentAgent收到 prompt开始执行选题 → 搜索 → 抓取参考内容 → 写文章 → 发布 → 通知每一步都通过 MCP 协议调用工具不是硬编码的 API文件系统是 Agent 和外部世界之间的「持久化记忆」——文章存本地日志存本地这个架构的好处是解耦换一个 Agent 框架、换一个 MCP 工具服务器、甚至换一个云平台核心逻辑不变。防线一随机延迟——让 Agent 看起来像人第一条也是最重要的防线绝对不要在整点触发。CSDN 看、掘金看、任何内容平台都看。如果每天 10:00:00 准时发一篇文章一周后你就被标记成 bot 了。最简单的实现#!/usr/bin/env python3random_delay.py —— 在 crontab 里调用自动加入随机延迟importrandomimporttimeimportsysdefrandom_sleep(min_seconds:int10,max_seconds:int120):delayrandom.randint(min_seconds,max_seconds)print(f[随机延迟] 等待{delay}秒后继续...)time.sleep(delay)if__name____main__:min_sint(sys.argv[1])iflen(sys.argv)1else10max_sint(sys.argv[2])iflen(sys.argv)2else120random_sleep(min_s,max_s)crontab 里这样写# 每天 10:00 触发实际执行时间在 10:00:10 ~ 10:02:00 之间随机010***cd/opt/agentpython3random_delay.py--min10--max120python3main.py看起来简单得不像话但这是我踩的第一个坑——没加延迟之前掘金直接限流了我的发布接口。一个整点 API 请求暴涨太假了。防线二幂等性——同一件事不能做两次这是最容易被忽略、但又最致命的坑。Agent 的 cron 任务和传统脚本有一个本质区别Agent 的任务完成判断是模糊的。脚本跑完 任务完成。Agent 跑完「结果好不好」还得你自己看。但 cron 不管你这些。到点了就重新跑。如果上次的结果还在、状态没重置就会出问题。我在实战中总结了三层幂等保护层 1文件锁importfcntlimportosclassFileLock:基于文件的排他锁防止同一任务并发执行def__init__(self,lock_path:str/tmp/agent_cron.lock):self.lock_pathlock_pathself.lock_fileNonedefacquire(self)-bool:self.lock_fileopen(self.lock_path,w)try:fcntl.flock(self.lock_file.fileno(),fcntl.LOCK_EX|fcntl.LOCK_NB)self.lock_file.write(str(os.getpid()))self.lock_file.flush()returnTrueexceptBlockingIOError:print([锁] 上一次任务仍在执行跳过本次触发)returnFalsedefrelease(self):ifself.lock_file:fcntl.flock(self.lock_file.fileno(),fcntl.LOCK_UN)self.lock_file.close()os.remove(self.lock_path)层 2日期状态文件Agent 每次完成任务后在本地记录一个状态标记。下次 cron 触发时先检查fromdatetimeimportdateimportjsondefcheck_task_completed(task_name:str,state_dir:str/opt/agent/state):os.makedirs(state_dir,exist_okTrue)state_fileos.path.join(state_dir,f{task_name}_{date.today().isoformat()}.json)ifos.path.exists(state_file):withopen(state_file)asf:statejson.load(f)print(f[状态]{task_name}今天已完成跳过。完成时间{state[completed_at]})returnTruereturnFalsedefmark_task_completed(task_name:str,metadata:dictNone,state_dir:str/opt/agent/state):state_fileos.path.join(state_dir,f{task_name}_{date.today().isoformat()}.json)withopen(state_file,w)asf:json.dump({task:task_name,date:date.today().isoformat(),completed_at:datetime.now().isoformat(),metadata:metadataor{}},f,indent2,ensure_asciiFalse)层 3内容去重即使锁和状态都失效了发布前还要做一次内容去重——对比今天已发文章的标题/摘要相似度超过阈值就跳过fromdifflibimportSequenceMatcherdefis_duplicate(new_title:str,existing_titles:list[str],threshold:float0.8)-bool:fortitleinexisting_titles:similaritySequenceMatcher(None,new_title.lower(),title.lower()).ratio()ifsimilaritythreshold:returnTruereturnFalse三道防线叠在一起基本杜绝了「同一个东西发两次」的可能性。防线三熔断机制——出事了能自己停下来这个教训来自社区的那个帖子。那个 Agent 发了 47 遍同一篇文章不是因为它疯了而是因为没有熔断。Agent 需要一个硬限制同一任务最多重试 N 次超了就停。而且这个限制不能只靠 prompt 告诉 Agent 别重试了——Agent 可能不理你。必须用代码强制classCircuitBreaker:熔断器 —— 超过错误阈值自动切断def__init__(self,max_failures:int3,reset_timeout:int3600,state_dir:str/opt/agent/state):self.max_failuresmax_failuresself.reset_timeoutreset_timeoutself.state_fileos.path.join(state_dir,circuit_breaker.json)defcheck(self,task_name:str)-bool:返回 True 允许执行False 已熔断ifnotos.path.exists(self.state_file):returnTruewithopen(self.state_file)asf:datajson.load(f)task_statedata.get(task_name,{})failurestask_state.get(failures,0)iffailuresself.max_failures:last_failtask_state.get(last_failure_time,)# 检查是否过了重置时间iflast_fail:elapsedtime.time()-datetime.fromisoformat(last_fail).timestamp()ifelapsedself.reset_timeout:# 超时自动重置self.reset(task_name)returnTrueprint(f[熔断]{task_name}已熔断连续失败{failures}次。跳过本次执行。)returnFalsereturnTruedefrecord_failure(self,task_name:str):data{}ifos.path.exists(self.state_file):withopen(self.state_file)asf:datajson.load(f)data[task_name]{failures:data.get(task_name,{}).get(failures,0)1,last_failure_time:datetime.now().isoformat()}withopen(self.state_file,w)asf:json.dump(data,f,indent2)defreset(self,task_name:str):ifos.path.exists(self.state_file):withopen(self.state_file)asf:datajson.load(f)data.pop(task_name,None)withopen(self.state_file,w)asf:json.dump(data,f,indent2)然后把熔断器放到 crom 任务的外层#!/bin/bash# main.sh —— cron 调用入口python3random_delay.py--min10--max120TASK_NAMEdaily_article# 文件锁python3-cfrom agent_utils import FileLocklock FileLock()if not lock.acquire():exit(1)# 日期状态检查python3-cfrom agent_utils import check_task_completedif check_task_completed($TASK_NAME):exit(0)# 熔断检查python3-cfrom agent_utils import CircuitBreakercb CircuitBreaker()if not cb.check($TASK_NAME):exit(2)# 执行任务|python3main.py||python3-cfrom agent_utils import CircuitBreaker; CircuitBreaker().record_failure($TASK_NAME)|三层保护锁 → 状态 → 熔断。任何一层触发任务就跳过。防线四成本控制——Token 不是无限的跑过 AI Agent 的同学都知道这玩意儿烧钱。一个复杂任务搜索 抓取 写作 发布单次就可能吃掉几万 token。如果 cron 不小心触发两次、熔断失效、Agent 陷入循环——账单会教你做人。硬成本上限classCostLimit:token / 费用计数器超限直接中断def__init__(self,max_tokens_per_run:int50000,max_cost_per_day:float2.0):self.max_tokens_per_runmax_tokens_per_runself.max_cost_per_daymax_cost_per_dayself.tokens_used0self.cost_used0.0self.state_dir/opt/agent/statedeftrack(self,tokens:int,cost:float):self.tokens_usedtokensself.cost_usedcost# 检查单次上限ifself.tokens_usedself.max_tokens_per_run:raiseRuntimeError(f[成本] 单次任务超过 token 上限{self.tokens_used}{self.max_tokens_per_run})# 检查当日累计daily_costself._get_daily_cost()costifdaily_costself.max_cost_per_day:raiseRuntimeError(f[成本] 今日总费用超过上限${daily_cost:.2f} ${self.max_cost_per_day:.2f})def_get_daily_cost(self)-float:cost_fileos.path.join(self.state_dir,fcost_{date.today().isoformat()}.json)ifos.path.exists(cost_file):withopen(cost_file)asf:returnjson.load(f).get(total,0.0)return0.0效果对比有防线 vs 没防线我把同样的 Agent 任务跑了 30 天做了对比指标无防线前两周有防线后两周重复发文次数3 次0因整点触发被限流1 次0token 超预算2 次0Agent 陷入循环1 次发了6遍0平均延迟每天0 秒整点随机 15-110 秒月度 API 费用~$85~$48省了将近一半费用。而且最重要的是——不用每天早上起来检查 Agent 有没有半夜发疯。踩坑清单浓缩版跑了几十天这些坑是我真金白银换来的坑 1cron 的环境变量和终端不一样cron 运行的环境比你的 shell 干净得多。$PATH不同、Python 的虚拟环境不会自动激活、~/.bashrc不会加载。修法在 crontab 最前面手动设置环境变量或者用一个 shell wrapper 脚本。# crontab 顶部PATH/usr/local/bin:/usr/bin:/binSHELL/bin/bashHOME/home/user坑 2MCP 服务器突然不可用Agent 不会自己处理有一次 firecrawl 的 MCP 服务器挂了 2 小时Agent 反复重试同一个 search 调用把 token 烧了个精光。修法在 Agent prompt 里硬性规定「每个工具调用最多重试 3 次第 4 次失败就跳过该步骤」。这里有个反直觉的点别让 Agent 自己决定要不要重试。给它一个明确的数字——Agent 的常识和你的预期可能差很大。坑 3Agent 的输出格式偶尔会漂移你让 Agent 输出 JSON它 99% 的时候没问题。但偶尔它会在 JSON 前面加一段解释文字或者 JSON 缺失一个字段。cron 任务的下游解析器就炸了。修法永远在解析之前做一次 schema 验证用 Pydantic 或 jsonschema不通过就 re-prompt。坑 4「编辑态」和「发布态」没区分最惨的一次Agent 写到一半的文章被 cron 下一次触发当成已发布标记了 completed。结果那天没有新文章发出去。修法区别draft/published两个状态。状态文件里加一个status字段state{task:daily_article,date:2026-07-09,|status:draft,# draft | publishing | published | failed |file_path:articles/2026-07-09-slug.md}坑 5通知太多 没人看通知跑通之后第一周我给每个步骤都加了飞书通知——开始选题了搜索完成了文章写好了发布了……一天十几条消息。第二周我的飞书通知就被静音了。修法通知规则改成「只通知异常 最终结果」。正常流程一条最终通知就够了。总结把 AI Agent 接上 cron本质上是在做一件事给不确定的系统加确定性的护栏。MCP 协议让工具调用标准化了cron 让定时执行简单了。但生产环境不是你搭个 demo 跑通就行的——随机延迟防止被识别为 bot文件锁 状态文件 内容去重确保幂等熔断器防止雪崩成本上限防止账单爆炸状态管理区分草稿和成品这些不是「可选的优化」是「活下来的底线」。我现在把 Agent 的 cron 任务跑在 WSL2 的本地笔记本上隔天写一篇技术文章自动发 CSDN。做了这么多防御之后唯一需要我参与的环节是偶尔看一眼飞书通知确认今天的内容质量还行。说到这有一个问题想问你你跑过 AI Agent 的定时任务吗遇到过什么离谱的事评论区聊聊——特别是那种凌晨三点 Agent 发疯的故事我还挺想收集的。