别再半夜爬起来修定时任务了！我用 LangGraph 给它装上了“自纠错”大脑

1. 引言定时任务是后端系统里最常见的“基础设施”之一每天凌晨同步数据、每小时拉取一次第三方接口、每周生成一次报表……它们看起来简单却在真实生产环境中频繁出问题——上游接口字段变了、数据量突增导致超时、偶发网络抖动让任务失败。传统定时任务失败后只能告警、人工介入、重跑而重跑往往又会遇到同样的错误。LangGraph 提供了一种新的思路把定时任务改造成一个“自纠错 Agent”。它不仅能执行任务还能在失败时自己分析原因、调整策略、重新尝试甚至把无法解决的问题带着完整上下文抛给人类。这篇文章会从理论讲起再带你一步步用 LangGraph 实现一个可运行的自纠错定时任务最后复盘我在改造过程中踩过的坑和思考。2. 为什么定时任务需要“自纠错”2.1 传统定时任务的痛点一个典型的定时任务通常长这样defsync_orders():datafetch_orders_from_upstream()save_to_db(data)它的问题在于每一步失败都没有恢复路径。fetch_orders_from_upstream()抛异常任务就死了。你只能靠告警发现然后手动修复、手动重跑。更麻烦的是很多错误是“间歇性”的——网络抖动、上游限流、临时字段缺失重跑一次可能就好了但人工介入的延迟往往让数据滞后很久。2.2 自纠错 Agent 的核心思想自纠错 Agent 的本质是把“执行”和“决策”分离执行层仍然调用原来的业务函数拉数据、写库。决策层当执行失败时Agent 根据错误信息、上下文、历史经验决定下一步动作——重试、换参数、跳过、还是上报人工。这就像把“运维工程师处理告警”的流程自动化了先看日志判断原因选择对策执行再验证。2.3 LangGraph 为什么适合做这件事LangGraph 的核心抽象是图节点Node是计算单元边Edge是流转逻辑状态State在节点间传递。这天然适合表达“任务执行 → 失败 → 分析 → 重试 → 再验证”这种带循环和分支的流程。相比直接用 LangChain 的 Agent 循环LangGraph 让你能精确控制每一步的流转条件而不是把控制权完全交给模型。3. 理论LangGraph 的核心概念在动手之前先梳理几个 LangGraph 的关键概念后面代码都基于它们。3.1 State状态State 是在图节点之间传递的数据结构通常用 TypedDict 定义。自纠错任务里State 至少需要包含任务输入、执行结果、错误信息、重试次数、最终状态。fromtypingimportTypedDict,OptionalclassTaskState(TypedDict):input_data:dictresult:Optional[dict]error:Optional[str]retry_count:intmax_retries:intstatus:str# running / success / failed / needs_human3.2 Node节点节点是图里的一个计算单元接收 State返回 State 的更新。自纠错任务里我们会设计几个核心节点执行任务、分析错误、决定策略、上报人工。3.3 Edge边与条件边边决定节点之间的流转。普通边是固定流转条件边则根据 State 内容动态决定下一个节点——这正是“自纠错”的关键根据错误类型走不同的处理分支。3.4 图Graph把节点和边组装起来编译成可执行的图。LangGraph 支持循环所以“重试”可以表达为一条回到执行节点的边。4. 实践把一个定时任务改造成自纠错 Agent下面我们用一个真实的例子每天同步第三方平台的订单数据。原始定时任务很简单我们一步步把它改造成自纠错 Agent。4.1 原始定时任务# tasks.pyimportrequestsdefsync_orders():从第三方平台拉取订单并写入本地库伪代码resprequests.get(https://api.example.com/orders,timeout30)resp.raise_for_status()ordersresp.json()[data]save_to_db(orders)returnlen(orders)这个任务挂在 cron 里每天凌晨 2 点跑。它的问题很明显任何一步失败任务就终止只能靠告警和人工。4.2 定义状态与节点首先定义 State 和各个节点。注意我们把“执行”和“决策”分开业务函数保持原样Agent 只负责编排。# agent.pyfromtypingimportTypedDict,Optionalfromlanggraph.graphimportStateGraph,ENDclassTaskState(TypedDict):input_data:dictresult:Optional[dict]error:Optional[str]retry_count:intmax_retries:intstatus:strdefexecute_task(state:TaskState)-TaskState:执行真正的业务逻辑try:resultsync_orders()return{**state,result:result,status:success}exceptExceptionase:return{**state,error:str(e),status:failed}defanalyze_error(state:TaskState)-TaskState:分析错误决定下一步策略errorstate[error]or# 这里可以接入 LLM也可以先用规则判断iftimeoutinerror.lower():strategyretry_with_longer_timeoutelifrate limitinerror.lower()or429inerror:strategywait_and_retryelifauthinerror.lower()or401inerroror403inerror:strategyneeds_humanelse:strategyretryreturn{**state,strategy:strategy}4.3 定义条件边条件边是自纠错的核心。根据分析和重试次数决定下一个节点。defshould_retry(state:TaskState)-str:ifstate[status]success:returnendifstate[retry_count]state[max_retries]:returnneeds_humanreturnretry4.4 组装图defbuild_graph():gStateGraph(TaskState)g.add_node(execute,execute_task)g.add_node(analyze,analyze_error)g.add_node(human,human_intervention)g.set_entry_point(execute)g.add_edge(execute,analyze)g.add_conditional_edges(analyze,should_retry,{retry:execute,needs_human:human,end:END,})g.add_edge(human,END)returng.compile()这里的关键是add_conditional_edges分析完错误后如果决定重试就回到execute节点形成循环如果重试次数耗尽就进入人工处理节点。4.5 接入 LLM 做更智能的错误分析规则判断能覆盖常见错误但真实世界的错误千奇百怪。我们可以让 LLM 参与错误分析把错误信息、任务上下文、历史处理经验一起交给模型让它给出策略。fromlangchain_openaiimportChatOpenAI llmChatOpenAI(modelgpt-4o-mini)defanalyze_error_with_llm(state:TaskState)-TaskState:errorstate[error]orpromptf 定时任务执行失败错误信息如下{error}任务上下文{state[input_data]}已重试次数{state[retry_count]}请判断下一步策略只返回以下选项之一 - retry可以重试 - retry_with_backoff等待后重试 - skip跳过本次 - needs_human需要人工介入 respllm.invoke(prompt)strategyresp.content.strip()return{**state,strategy:strategy}4.6 在定时任务中调用 Agent改造后的定时任务入口变得非常简单# scheduler.pyfromagentimportbuild_graphdefscheduled_job():graphbuild_graph()initial_state{input_data:{date:2026-09-01},result:None,error:None,retry_count:0,max_retries:3,status:running,}final_stategraph.invoke(initial_state)iffinal_state[status]needs_human:notify_human(final_state[error])returnfinal_statecron 配置不变还是每天凌晨跑但内部逻辑已经从“一条直线”变成了“带反馈的循环”。5. 过程中遇到的问题与解决改造过程不是一帆风顺的这里记录几个我实际踩到的坑。5.1 问题一LangGraph 版本 API 差异现象按照网上教程写StateGraph结果add_conditional_edges的签名对不上报TypeError。原因LangGraph 0.2 和 0.3 的 API 有变化旧教程用的是add_conditional_edges(analyze, should_retry, {...})新版本要求传入路径映射的方式略有不同。解决锁定版本并查阅对应版本的官方文档。我最终固定使用langgraph0.3.x并按照新版签名调整g.add_conditional_edges(analyze,should_retry,{retry:execute,needs_human:human,end:END,})复盘遇到 API 报错第一反应应该是查当前安装版本的文档而不是照搬旧教程。建议在项目里用requirements.txt锁死版本。5.2 问题二State 更新导致的数据丢失现象重试几次后发现input_data变成了None。原因我在节点里返回{**state, ...}时某个节点错误地覆盖了input_data字段。LangGraph 的 State 更新是“按返回的 dict 合并”如果某个节点返回了{input_data: None}就会覆盖原值。解决在每个节点里显式保留不需要修改的字段或者用operator.add之类的 reducer 来合并。更稳妥的做法是节点只返回需要更新的字段不要返回整个 state。defexecute_task(state:TaskState)-dict:try:resultsync_orders()return{result:result,status:success}exceptExceptionase:return{error:str(e),status:failed}复盘LangGraph 的 State 是“部分更新”语义不是“整体替换”。理解这一点能避免很多隐蔽 bug。5.3 问题三LLM 分析延迟太高现象每次失败都要调一次 LLM一次分析要 2-3 秒重试 3 次就是近 10 秒的额外延迟。解决分层策略——先用规则快速判断常见错误超时、限流、鉴权规则命中就直接走对应分支只有规则无法判断的“未知错误”才调 LLM。这样 90% 的情况都不需要 LLM 参与。defanalyze_error(state:TaskState)-TaskState:errorstate[error]or# 快速规则判断iftimeoutinerror.lower():return{**state,strategy:retry_with_longer_timeout}if429inerrororrate limitinerror.lower():return{**state,strategy:wait_and_retry}if401inerroror403inerror:return{**state,strategy:needs_human}# 未知错误才交给 LLMreturnanalyze_error_with_llm(state)复盘LLM 不是万能的也不是必须的。能用规则解决的问题不要用模型成本和延迟都更低。5.4 问题四重试导致的数据重复现象任务第一次执行时已经写入了部分数据失败重试后又把同一批数据写了一遍造成重复。解决给任务加幂等性。在写入前先检查是否已存在或者用“业务主键 去重表”的方式保证同一批数据只写一次。defsave_to_db(orders):fororderinorders:# 以 order_id 为主键存在则跳过ifnotdb.exists(order[order_id]):db.insert(order)复盘自纠错 Agent 让“重试”变得频繁幂等性从“可选优化”变成了“必须项”。改造定时任务时一定要先审视业务函数是否幂等。5.5 问题五死循环风险现象某个错误一直无法解决Agent 陷入“执行→失败→重试→再失败”的死循环。解决max_retries是硬性上限达到后强制进入needs_human分支。同时在重试之间加入退避backoff避免对上游造成压力。importtimedefexecute_task(state:TaskState)-dict:ifstate[retry_count]0:time.sleep(2**state[retry_count])# 指数退避try:resultsync_orders()return{result:result,status:success}exceptExceptionase:return{error:str(e),status:failed}复盘任何带循环的系统都要有“终止条件”。max_retries就是自纠错 Agent 的安全阀。6. 后期复盘与思考6.1 收益减少人工介入常见的超时、限流问题Agent 能自动重试解决告警量明显下降。错误处理更规范所有失败都走同一条“分析→决策→执行”路径日志更完整排查更容易。可观测性提升每次重试、每次策略选择都记录在 State 里相当于自动生成了“处理日志”。6.2 代价与风险复杂度上升从“一个函数”变成“一张图”理解和维护成本增加。LLM 成本如果大量使用 LLM 做分析token 费用不可忽视。建议用规则兜底LLM 只处理未知情况。调试难度图的流转不像线性代码那么直观需要借助 LangGraph 的调试工具或自己打印 State 流转。6.3 什么场景适合改造不是所有定时任务都值得改造成 Agent。我的判断标准是任务失败成本高数据同步、对账、报表失败影响大。错误类型多样且不可预测上游接口不稳定、字段经常变。重试有实际意义很多错误是间歇性的重试能解决。如果任务简单、失败影响小、错误类型单一用传统的“重试 告警”就够了没必要引入 Agent。6.4 后续优化方向引入记忆让 Agent 记住“上次遇到这个错误是怎么解决的”下次直接复用策略。多任务共享经验多个定时任务共用一个“错误处理知识库”。人工反馈闭环当 Agent 把问题抛给人工后人工的解决方案可以回填到知识库让 Agent 越用越聪明。7. 总结把定时任务改造成自纠错 Agent本质上是把“执行”和“决策”分离用一张带循环和分支的图来编排“执行→失败→分析→重试→验证”的流程。LangGraph 的 State、Node、条件边正好提供了这种表达能力。改造过程中我最大的体会是Agent 不是银弹它只是把“重试”这件事做得更聪明、更有章法。真正的难点不在 LangGraph 本身而在于你的业务函数是否幂等、你的错误分类是否清晰、你的终止条件是否可靠。把这些基础打牢Agent 才能在上面稳定运行。如果你也在维护一堆脆弱的定时任务不妨从其中一个开始试着用 LangGraph 给它装上“自纠错”的能力。