让 Agent 真正自动干活Python 脚本化的三根支柱摘要Agent 能看懂数据、“听懂需求之后还差最后一口气——自动干活。每天盯着开源项目有没有发新版、日志里有没有冒出异常这些重复、有规律的事正是脚本化的主场。支撑这套自动化的是三根支柱批量处理循环连接复用、自动化逻辑if-elsetry-except、定时执行schedule→cron。本文用真实代码把它们一块块拼成完整的开源情报 Agent”。目录第一根批量处理——一次办妥而不是一个个来第二根自动化逻辑——会判断、会兜底第三根定时执行——到点自己动三根支柱搭起来一个开源情报 Agent两件小事决定脚本是玩具还是工具Agent 能看懂数据、听懂需求之后还差最后一口气——自动干活。每天盯着几个开源项目有没有发新版、日志里有没有冒出异常、关注的信息源有没有更新这些重复、有规律、需要定时或批量跑的事正是脚本化编程的主场也和 Agent 的执行能力天然契合。说白了脚本化就是给 Agent 写一份操作手册什么时候做、做什么、怎么做写清楚剩下交给它自己跑。支撑这套自动化的是三根支柱。第一根批量处理——一次办妥而不是一个个来Agent 经常要同时处理多个对象多个数据源、多个文件、多个接口。批量处理的核心是**“循环 连接复用”**。一个原创场景你关注了一批开源项目想一次性知道它们最近有没有发新版。逐个手动查太蠢写个脚本批量拉取 GitHub Release 信息几行就搞定。importhttpximportos repos[python/cpython,pallets/flask,psf/requests,encode/httpx]tokenos.getenv(GITHUB_TOKEN)defbatch_check_releases(repos:list)-list:updates[]# 关键用 httpx.Client 复用连接同一会话内多次请求共享 TCP 连接# 省掉重复握手开销这是批量调用的核心优化withhttpx.Client(timeout10)asclient:forrepoinrepos:try:respclient.get(fhttps://api.github.com/repos/{repo}/releases/latest,headers{Authorization:fBearer{token}})resp.raise_for_status()dataresp.json()updates.append({repo:repo,tag:data[tag_name],published:data[published_at][:10]})exceptExceptionase:updates.append({repo:repo,error:str(e)})returnupdates打个比方批量处理就像工厂的流水线——不是一个个零件搬来搬去而是一条传送带依次过效率天差地别。httpx.Client就是那条传送带多次请求共享同一个 TCP 连接省掉了反复握手的开销。顺带说一句列表推导式。它通常比for循环加append更快因为 CPython 会预分配内存、省掉方法调用开销但幅度视操作复杂度而定一般在 20% 到 50% 之间不是固定数字。逻辑简单时用它又快又优雅逻辑复杂到要写多行表达式时老老实实用循环可读性更重要。第二根自动化逻辑——会判断、会兜底脚本不能一根筋。真实世界里文件可能不存在、接口可能超时、数据可能缺字段只会一条路走到底的脚本跑两次就崩了。if-else处理不同情况try-except接住异常这和 Agent 根据决策执行、遇错重试或降级的逻辑是一回事。换一个原创场景扫描服务日志命中关键词就收集起来触发告警。fromdatetimeimportdatetime KEYWORDS[ERROR,Traceback,FATAL]defscan_logs(log_file:str)-list:try:alerts[]withopen(log_file,encodingutf-8)asf:forlineno,lineinenumerate(f,1):forkwinKEYWORDS:ifkwinline:alerts.append({line:lineno,keyword:kw,text:line.strip()})ifalerts:print(f[{datetime.now():%F %T}] 发现{len(alerts)}条告警准备通知...)# 这里接通知逻辑发邮件、推 webhook、写数据库都行else:print(f[{datetime.now():%F %T}] 日志干净无异常)returnalertsexceptFileNotFoundError:print(f日志文件不存在{log_file})return[]exceptExceptionase:print(f扫描异常{e})return[]自动化逻辑就像流水线上的质检员——正常的放行有问题的拦下来归类质检员自己生病了也不至于把整条流水线搞停。条件判断让脚本有了脑子——知道什么该管、什么该跳过异常处理让脚本有了韧性——错了不崩兜底继续。这两样加起来脚本才算从能跑变成可靠地跑。第三根定时执行——到点自己动很多任务不需要人触发到点就该跑。最轻量的选择是schedule库pip install schedule稳定版 1.2.x语法直白importscheduleimporttimefromdatetimeimportdatetimedefdaily_report():print(f[{datetime.now():%F %T}] 开始生成每日情报...)schedule.every().day.at(09:00).do(daily_report)whileTrue:schedule.run_pending()time.sleep(60)定时执行就是给脚本装了个闹钟——不等人喊到点自己醒来干活。但心里要有数schedule是单线程串行的一个任务跑得久会堵住后面的。原型和小脚本用它没问题真要上生产任务多、需要并发或要持久化重试时换 APScheduler或者干脆用系统级的cron、systemd timer——进程崩了系统帮你拉起比常驻一个 Python 进程省心得多。这就像从手机闹钟升级成工厂的自动排班系统一个管一个人一个管一整条产线。三根支柱搭起来一个开源情报 Agent把上面三块拼到一起就是一个完整的开源情报 Agent每天到点批量检查关注的项目有没有新版本顺带扫一眼服务日志有没有异常最后汇总成一条通知发给你。核心结构大概是这样classReleaseWatchAgent:def__init__(self,repos,log_file):self.reposrepos self.log_filelog_filedefrun(self):# 批量拉取 自动化逻辑异常处理在函数内部releasesbatch_check_releases(self.repos)alertsscan_logs(self.log_file)# 汇总成通知self.notify(releases,alerts)defstart(self):schedule.every().day.at(09:00).do(self.run)whileTrue:schedule.run_pending()time.sleep(60)批量处理一次性消化多个项目和日志自动化逻辑过滤出真正值得关心的信息定时执行让它每天自己醒来干活。Agent 就从你喊一声它动一下变成了自己盯着、自己干——这才是自动干活该有的样子。两件小事决定脚本是玩具还是工具第一密钥别写死在代码里。用python-dotenv从.env读环境变量既能保护敏感信息也方便开发生产环境切换。密钥硬编码就像把家门钥匙贴在门框上——方便了自己也方便了别人。第二日志别只用print换标准库logging能落文件、能分级过滤长跑脚本出问题时这是你唯一的排查线索。print是口头交代logging是白纸黑字留档——口头交代转身就忘留档的东西随时能翻。这两件小事是脚本从能跑到能维护的分水岭。